Module: Legion::API::Routes::Llm

Defined in:

lib/legion/api/llm.rb

Class Method Summary

• .register_chat(app) ⇒ Object
• .register_inference(app) ⇒ Object
• .register_providers(app) ⇒ Object
• .registered(app) ⇒ Object

Class Method Details

.register_chat(app) ⇒ Object

# File 'lib/legion/api/llm.rb', line 131

def self.register_chat(app)
  register_inference(app)

  app.post '/api/llm/chat' do
    Legion::Logging.debug "API: POST /api/llm/chat params=#{params.keys}"
    require_llm!

    body = parse_request_body
    validate_required!(body, :message)

    message = body[:message]

    # Tier 0 check - serve from PatternStore if available
    if defined?(Legion::MCP::TierRouter)
      tier_result = Legion::MCP::TierRouter.route(
        intent:  message,
        params:  body.except(:message, :model, :provider, :request_id),
        context: {}
      )
      if tier_result[:tier]&.zero?
        return json_response({
                               response:           tier_result[:response],
                               tier:               0,
                               latency_ms:         tier_result[:latency_ms],
                               pattern_confidence: tier_result[:pattern_confidence]
                             })
      end
    end

    request_id = body[:request_id] || SecureRandom.uuid
    model      = body[:model]
    provider   = body[:provider]

    # Compatibility fallback for legacy gateway installs. Native legion-llm handles routing first.
    if !Legion::LLM.respond_to?(:chat) && gateway_available?
      ingress_result = Legion::Ingress.run(
        payload:      { message: message, model: model, provider: provider,
                        request_id: request_id },
        runner_class: 'Legion::Extensions::Llm::Gateway::Runners::Inference',
        function:     'chat',
        source:       'api'
      )

      unless ingress_result[:success]
        Legion::Logging.error "[api/llm/chat] ingress failed: #{ingress_result}"
        return json_response({ error: ingress_result[:error] || ingress_result[:status] },
                             status_code: 502)
      end

      result = ingress_result[:result]

      if result.nil?
        Legion::Logging.warn "[api/llm/chat] runner returned nil (status=#{ingress_result[:status]})"
        return json_response({ error: { code:    'empty_result',
                                        message: 'Gateway runner returned no result' } },
                             status_code: 502)
      end

      response_content = if result.respond_to?(:content)
                           result.content
                         elsif result.is_a?(Hash) && result[:error]
                           return json_response({ error: result[:error] }, status_code: 502)
                         elsif result.is_a?(Hash)
                           result[:response] || result[:content] || result.to_s
                         else
                           result.to_s
                         end

      meta = { routed_via: 'gateway' }
      meta[:model] = result.model.to_s if result.respond_to?(:model)
      meta[:tokens_in] = result.input_tokens if result.respond_to?(:input_tokens)
      meta[:tokens_out] = result.output_tokens if result.respond_to?(:output_tokens)

      return json_response({ response: response_content, meta: meta }, status_code: 201)
    end

    # Fallback: direct LLM call (no metering, no task tracking)
    if cache_available? && env['HTTP_X_LEGION_SYNC'] != 'true'
      llm = Legion::LLM
      rc  = Legion::LLM::ResponseCache
      rc.init_request(request_id)

      Thread.new do
        session  = llm.chat_direct(model: model, provider: provider)
        response = session.ask(message)
        rc.complete(
          request_id,
          response: response.content,
          meta:     {
            model:      session.model.to_s,
            tokens_in:  response.respond_to?(:input_tokens) ? response.input_tokens : nil,
            tokens_out: response.respond_to?(:output_tokens) ? response.output_tokens : nil
          }
        )
      rescue StandardError => e
        Legion::Logging.log_exception(e, payload_summary: 'api/llm/chat async failed', component_type: :api)
        rc.fail_request(request_id, code: 'llm_error', message: e.message)
      end

      Legion::Logging.info "API: LLM chat request #{request_id} queued async"
      json_response({ request_id: request_id, poll_key: "llm:#{request_id}:status" },
                    status_code: 202)
    else
      session  = Legion::LLM.chat(model: model, provider: provider,
                                  caller: { source: 'api', path: request.path })
      response = session.ask(message)
      Legion::Logging.info "API: LLM chat request #{request_id} completed sync model=#{session.model}"
      json_response(
        {
          response: response.content,
          meta:     {
            model:      session.model.to_s,
            tokens_in:  response.respond_to?(:input_tokens) ? response.input_tokens : nil,
            tokens_out: response.respond_to?(:output_tokens) ? response.output_tokens : nil
          }
        },
        status_code: 201
      )
    end
  end
end

.register_inference(app) ⇒ Object

# File 'lib/legion/api/llm.rb', line 253

def self.register_inference(app)
  app.post '/api/llm/inference' do
    require_llm!
    body = parse_request_body
    validate_required!(body, :messages)

    messages        = body[:messages]
    tools           = body[:tools] || []
    model           = body[:model]
    provider        = body[:provider]
    requested_tools = body[:requested_tools] || []

    unless messages.is_a?(Array)
      halt 400, { 'Content-Type' => 'application/json' },
           Legion::JSON.generate({ error: { code: 'invalid_messages', message: 'messages must be an array' } })
    end

    caller_identity = env['legion.tenant_id'] || 'api:inference'

    # GAIA bridge - push InputFrame to sensory buffer
    last_user = messages.select { |m| (m[:role] || m['role']).to_s == 'user' }.last
    prompt    = (last_user || {})[:content] || (last_user || {})['content'] || ''

    if defined?(Legion::Gaia) && Legion::Gaia.respond_to?(:started?) && Legion::Gaia.started? && prompt.length.positive?
      begin
        frame = Legion::Gaia::InputFrame.new(
          content:      prompt,
          channel_id:   :api,
          content_type: :text,
          auth_context: { identity: caller_identity },
          metadata:     { source_type: :human_direct, salience: 0.5 }
        )
        Legion::Gaia.ingest(frame)
      rescue StandardError => e
        Legion::Logging.log_exception(e, payload_summary: 'gaia ingest failed in inference', component_type: :api)
      end
    end

    # Build client-side tool classes from Interlink definitions
    tool_classes = tools.filter_map do |t|
      ts = t.respond_to?(:transform_keys) ? t.transform_keys(&:to_sym) : t
      build_client_tool_class(ts[:name].to_s, ts[:description].to_s, ts[:parameters] || ts[:input_schema])
    end

    Legion::Logging.debug "[llm][api] inference inbound client_tools=#{tool_classes.size} requested_tools=#{requested_tools.size}"

    # Detect streaming mode
    streaming = body[:stream] == true && env['HTTP_ACCEPT']&.include?('text/event-stream')

    # Executor handles all registry tool injection — API only passes client-defined tools
    require 'legion/llm/inference' unless defined?(Legion::LLM::Inference::Request) &&
                                          defined?(Legion::LLM::Inference::Executor)

    principal  = defined?(Legion::Identity::Request) && env['legion.principal']
    caller_ctx = if principal
                   principal.to_caller_hash
                 else
                   { requested_by: { identity: caller_identity, type: :user, credential: :api } }
                 end

    caller_metadata = body[:metadata].is_a?(Hash) ? body[:metadata] : {}
    req = Legion::LLM::Inference::Request.build(
      messages:        messages,
      system:          body[:system],
      routing:         { provider: provider, model: model },
      tools:           tool_classes,
      caller:          caller_ctx,
      conversation_id: body[:conversation_id],
      metadata:        caller_metadata.merge(requested_tools: requested_tools),
      stream:          streaming,
      cache:           { strategy: :default, cacheable: true }
    )
    executor = Legion::LLM::Inference::Executor.new(req)

    if streaming
      content_type 'text/event-stream'
      headers 'Cache-Control' => 'no-cache', 'Connection' => 'keep-alive',
              'X-Accel-Buffering' => 'no'

      stream do |out|
        # Wire up real-time tool-call / tool-result / tool-error / model-fallback SSE events.
        # The executor fires tool_event_handler for each event as it happens,
        # including accurate wall-clock startedAt/finishedAt/durationMs timing.
        emitted_tool_call_ids = Set.new
        executor.tool_event_handler = lambda do |event|
          case event[:type]
          when :tool_call
            emitted_tool_call_ids << event[:tool_call_id] if event[:tool_call_id]
            out << "event: tool-call\ndata: #{Legion::JSON.generate({
                                                                      toolCallId: event[:tool_call_id],
                                                                      toolName:   event[:tool_name],
                                                                      args:       event[:arguments] || {},
                                                                      startedAt:  event[:started_at]&.iso8601(3),
                                                                      timestamp:  event[:started_at]&.iso8601(3) || Time.now.iso8601(3)
                                                                    })}\n\n"
          when :tool_result
            out << "event: tool-result\ndata: #{Legion::JSON.generate({
                                                                        toolCallId: event[:tool_call_id],
                                                                        toolName:   event[:tool_name],
                                                                        result:     event[:result],
                                                                        startedAt:  event[:started_at]&.iso8601(3),
                                                                        finishedAt: event[:finished_at]&.iso8601(3) || Time.now.iso8601(3),
                                                                        durationMs: event[:duration_ms],
                                                                        timestamp:  event[:finished_at]&.iso8601(3) || Time.now.iso8601(3)
                                                                      })}\n\n"
          when :tool_error
            out << "event: tool-error\ndata: #{Legion::JSON.generate({
                                                                       toolCallId: event[:tool_call_id],
                                                                       toolName:   event[:tool_name],
                                                                       error:      (event[:error] || event[:result]).to_s,
                                                                       startedAt:  event[:started_at]&.iso8601(3),
                                                                       finishedAt: Time.now.iso8601(3),
                                                                       timestamp:  Time.now.iso8601(3)
                                                                     })}\n\n"
          when :model_fallback
            out << "event: model-fallback\ndata: #{Legion::JSON.generate({
                                                                           fromModel:  event[:from_model],
                                                                           toModel:    event[:to_model],
                                                                           toModelKey: event[:to_model],
                                                                           error:      event[:error] || 'Provider unavailable',
                                                                           reason:     event[:reason] || 'provider_fallback'
                                                                         })}\n\n"
          end
        end

        full_text = +''
        pipeline_response = executor.call_stream do |chunk|
          text = chunk.respond_to?(:content) ? chunk.content.to_s : chunk.to_s
          next if text.empty?

          full_text << text
          out << "event: text-delta\ndata: #{Legion::JSON.generate({ delta: text })}\n\n"
        end

        # Post-hoc safety net: emit any tool-calls that weren't fired in real-time
        # (e.g. non-streaming tool paths). Skip IDs already sent via tool_event_handler.
        if pipeline_response.tools.is_a?(Array) && !pipeline_response.tools.empty?
          pipeline_response.tools.each do |tc|
            tc_id = tc.respond_to?(:id) ? tc.id : nil
            next if tc_id && emitted_tool_call_ids.include?(tc_id)

            out << "event: tool-call\ndata: #{Legion::JSON.generate({
                                                                      toolCallId: tc_id,
                                                                      toolName:   tc.respond_to?(:name) ? tc.name : tc.to_s,
                                                                      args:       tc.respond_to?(:arguments) ? tc.arguments : {}
                                                                    })}\n\n"
          end
        end

        # Emit any model-fallback warnings collected post-hoc
        Array(pipeline_response.warnings).each do |w|
          next unless w.is_a?(Hash) && w[:type] == :provider_fallback

          fallback = w[:fallback].to_s
          provider, model = fallback.split(':', 2)
          resolved_model = (model || provider).to_s.strip
          next if resolved_model.empty?

          out << "event: model-fallback\ndata: #{Legion::JSON.generate({
                                                                         fromModel:  pipeline_response.routing&.dig(:model),
                                                                         toModel:    resolved_model,
                                                                         toModelKey: resolved_model,
                                                                         error:      w[:original_error] || 'Provider unavailable',
                                                                         reason:     'provider_fallback'
                                                                       })}\n\n"
        end

        enrichments = pipeline_response.enrichments
        out << "event: enrichment\ndata: #{Legion::JSON.generate(enrichments)}\n\n" if enrichments.is_a?(Hash) && !enrichments.empty?

        tokens = pipeline_response.tokens
        out << "event: done\ndata: #{Legion::JSON.generate({
          content:            full_text,
          model:              pipeline_response.routing&.dig(:model),
          conversation_id:    pipeline_response.conversation_id,
          stop_reason:        pipeline_response.stop&.dig(:reason)&.to_s,
          input_tokens:       tokens.respond_to?(:input_tokens)        ? tokens.input_tokens        : nil,
          output_tokens:      tokens.respond_to?(:output_tokens)       ? tokens.output_tokens       : nil,
          cache_read_tokens:  tokens.respond_to?(:cache_read_tokens)   ? tokens.cache_read_tokens   : nil,
          cache_write_tokens: tokens.respond_to?(:cache_write_tokens)  ? tokens.cache_write_tokens  : nil
        }.compact)}\n\n"
      rescue StandardError => e
        Legion::Logging.log_exception(e, payload_summary: 'api/llm/inference stream failed', component_type: :api)
        out << "event: error\ndata: #{Legion::JSON.generate({ code: 'stream_error', message: e.message })}\n\n"
      end
    else
      pipeline_response = executor.call
      tokens = pipeline_response.tokens

      json_response({
                      content:       pipeline_response.message&.dig(:content),
                      tool_calls:    extract_tool_calls(pipeline_response),
                      stop_reason:   pipeline_response.stop&.dig(:reason),
                      model:         pipeline_response.routing&.dig(:model) || model,
                      input_tokens:  tokens.respond_to?(:input_tokens) ? tokens.input_tokens : nil,
                      output_tokens: tokens.respond_to?(:output_tokens) ? tokens.output_tokens : nil
                    }, status_code: 200)
    end
  rescue Legion::LLM::AuthError => e
    Legion::Logging.log_exception(e, payload_summary: 'api/llm/inference auth failed', component_type: :api)
    json_response({ error: { code: 'auth_error', message: e.message } }, status_code: 401)
  rescue Legion::LLM::RateLimitError => e
    Legion::Logging.log_exception(e, payload_summary: 'api/llm/inference rate limited', component_type: :api)
    json_response({ error: { code: 'rate_limit', message: e.message } }, status_code: 429)
  rescue Legion::LLM::TokenBudgetExceeded => e
    Legion::Logging.log_exception(e, payload_summary: 'api/llm/inference token budget exceeded', component_type: :api)
    json_response({ error: { code: 'token_budget_exceeded', message: e.message } }, status_code: 413)
  rescue Legion::LLM::ProviderDown, Legion::LLM::ProviderError => e
    Legion::Logging.log_exception(e, payload_summary: 'api/llm/inference provider error', component_type: :api)
    json_response({ error: { code: 'provider_error', message: e.message } }, status_code: 502)
  rescue StandardError => e
    Legion::Logging.log_exception(e, payload_summary: 'api/llm/inference failed', component_type: :api)
    json_response({ error: { code: 'inference_error', message: e.message } }, status_code: 500)
  end
end

.register_providers(app) ⇒ Object

# File 'lib/legion/api/llm.rb', line 469

def self.register_providers(app)
  app.get '/api/llm/providers' do
    require_llm!

    json_response({
                    providers: provider_health_report,
                    summary:   provider_circuit_summary
                  })
  end

  app.get '/api/llm/providers/:name' do
    require_llm!

    json_response(provider_detail(params[:name]))
  end
end

.registered(app) ⇒ Object

# File 'lib/legion/api/llm.rb', line 9

def self.registered(app)
  app.helpers do
    define_method(:require_llm!) do
      return if defined?(Legion::LLM) &&
                Legion::LLM.respond_to?(:started?) &&
                Legion::LLM.started?

      halt 503, { 'Content-Type' => 'application/json' },
           Legion::JSON.generate({ error: { code:    'llm_unavailable',
                                            message: 'LLM subsystem is not available' } })
    end

    define_method(:cache_available?) do
      defined?(Legion::Cache) &&
        Legion::Cache.respond_to?(:connected?) &&
        Legion::Cache.connected?
    end

    define_method(:gateway_available?) do
      defined?(Legion::Extensions::Llm::Gateway::Runners::Inference)
    end

    define_method(:native_provider_stats_available?) do
      defined?(Legion::LLM::Inventory) && Legion::LLM::Inventory.respond_to?(:providers)
    end

    define_method(:provider_health_report) do
      if native_provider_stats_available?
        groups = Legion::LLM::Inventory.providers
        return [] unless groups.respond_to?(:map)

        groups.map do |provider, offerings|
          provider_offerings = Array(offerings)
          health = provider_offerings.map { |offering| offering_value(offering, :health) }
                                     .find { |entry| entry.is_a?(Hash) } || {}
          circuit = health[:circuit_state] || health['circuit_state'] || 'unknown'
          {
            provider:   provider.to_s,
            circuit:    circuit,
            adjustment: health[:adjustment] || health['adjustment'] || 0,
            healthy:    circuit.to_s != 'open',
            offerings:  provider_offerings.size,
            models:     provider_offerings.map { |offering| offering_value(offering, :model) }.compact.uniq,
            types:      provider_offerings.map { |offering| offering_value(offering, :type) }.compact.uniq,
            instances:  provider_offerings.map do |offering|
              offering_value(offering, :provider_instance) || offering_value(offering, :instance_id)
            end.compact.uniq
          }
        end
      elsif defined?(Legion::Extensions::Llm::Gateway::Runners::ProviderStats)
        Legion::Extensions::Llm::Gateway::Runners::ProviderStats.health_report
      else
        []
      end
    end

    define_method(:provider_circuit_summary) do
      report = provider_health_report
      return Legion::Extensions::Llm::Gateway::Runners::ProviderStats.circuit_summary if
        report.empty? && defined?(Legion::Extensions::Llm::Gateway::Runners::ProviderStats)

      circuits = report.map { |entry| entry[:circuit].to_s }
      {
        total:     report.size,
        closed:    circuits.count('closed'),
        open:      circuits.count('open'),
        half_open: circuits.count('half_open')
      }
    end

    define_method(:provider_detail) do |provider|
      provider_name = provider.to_s
      if native_provider_stats_available?
        entry = provider_health_report.find { |candidate| candidate[:provider] == provider_name }
        halt 404, json_error('provider_not_found', "Provider '#{provider_name}' not found", status_code: 404) unless entry

        entry
      elsif defined?(Legion::Extensions::Llm::Gateway::Runners::ProviderStats)
        Legion::Extensions::Llm::Gateway::Runners::ProviderStats.provider_detail(provider: provider_name.to_sym)
      else
        halt 503, json_error('providers_unavailable', 'LLM provider inventory is not loaded', status_code: 503)
      end
    end

    define_method(:offering_value) do |offering, key|
      next unless offering.respond_to?(:[])

      offering[key] || offering[key.to_s]
    end

    define_method(:build_client_tool_class) do |tname, tdesc, tschema|
      require 'legion/llm/types/tool_definition' unless defined?(Legion::LLM::Types::ToolDefinition)

      Legion::LLM::Types::ToolDefinition.build(
        name:        tname,
        description: tdesc,
        parameters:  tschema || {},
        source:      { type: :client, executable: true }
      )
    rescue StandardError => e
      Legion::Logging.log_exception(e, payload_summary: "build_client_tool_class failed for #{tname}", component_type: :api)
      nil
    end

    define_method(:extract_tool_calls) do |pipeline_response|
      tools_data = pipeline_response.tools
      return nil unless tools_data.is_a?(Array) && !tools_data.empty?

      tools_data.map do |tc|
        {
          id:        tc.respond_to?(:id) ? tc.id : nil,
          name:      tc.respond_to?(:name) ? tc.name : tc.to_s,
          arguments: tc.respond_to?(:arguments) ? tc.arguments : {}
        }
      end
    end
  end

  register_chat(app)
  register_providers(app)
end