Function bodies 405 total

      def list
        Turbofan::CLI::Resources.list
      end

      def new(name = nil)
        if name.nil?
          name = Turbofan::CLI::Prompt.ask("Step name (snake_case)")
          raise Thor::Error, "Step name is required" if name.nil? || name.empty?
        end

      def router(step_name)
        Turbofan::CLI::AddRouter.call(step_name)
      end

    def self.project_root
      dir = Dir.pwd
      loop do
        return dir if Dir.exist?(File.join(dir, "turbofans"))
        return dir if File.exist?(File.join(dir, "Gemfile")) && !Dir.glob(File.join(dir, "*.gemspec")).any?
        parent = File.dirname(dir)
        return Dir.pwd if parent == dir
        dir = parent
      end

  class CLI < Thor
    module Resources
      def self.deploy(stage:)
        require "aws-sdk-cloudformation"

        resources = Turbofan::Resource.discover.select { |r| !r.turbofan_consumable.nil? }

        if resources.empty?
          puts "No resources defined."
          return
        end

        cf_client = Aws::CloudFormation::Client.new
        stack_name = "turbofan-resources-#{stage}"
        template_body = generate_template(resources, stage)

        Turbofan::Deploy::StackManager.deploy(
          cf_client,
          stack_name: stack_name,
          template_body: template_body
        )
      end

      def self.list
        resources = Turbofan::Resource.discover

        if resources.empty?
          puts "No resources defined."
          return
        end

        resources.each do |resource_class|
          key = resource_class.turbofan_key
          consumable = resource_class.turbofan_consumable
          type = resource_class.respond_to?(:turbofan_resource

      def self.deploy(stage:)
        require "aws-sdk-cloudformation"

        resources = Turbofan::Resource.discover.select { |r| !r.turbofan_consumable.nil? }

        if resources.empty?
          puts "No resources defined."
          return
        end

      def self.list
        resources = Turbofan::Resource.discover

        if resources.empty?
          puts "No resources defined."
          return
        end

      def self.generate_template(resources, stage)
        cfn_resources = {}

        resources.each do |resource_class|
          key = resource_class.turbofan_key
          logical_id = "ConsumableResource#{Turbofan::Naming.pascal_case(key)}"

          cfn_resources[logical_id] = {
            "Type" => "AWS::Batch::ConsumableResource",
            "Properties" => {
              "ConsumableResourceName" => "turbofan-#{key}-#{stage}",
              "TotalQuantity" => resource_class.turbofan_consumable,
              "ResourceType" => "REPLENISHABLE",
              "Tags" => {
                "turbofan:managed" => "true",
                "turbofan:resource" => key.to_s,
                "turbofan:stage" => stage
              }
            }
          }
        end

  class CLI < Thor
    module Rollback
      def self.call(pipeline_name:, stage:)
        cf_client = Aws::CloudFormation::Client.new
        # Note: uses directory name directly. Assumes it matches pipeline's turbofan_name.
        stack_name = Turbofan::Naming.stack_name(pipeline_name, stage)
        state = Turbofan::Deploy::StackManager.detect_state(cf_client, stack_name)

        case state
        when :does_not_exist
          raise "Stack does not exist: #{stack_name}"
        when :create_complete
          raise "No previous deployment to rollback to for #{stack_name}"
        when :in_progress
          raise "Another operation is in progress on #{stack_name}"
        when :update_complete, :update_rollback_complete
          # Reapplies the previous template version. Effective for reverting
          # image tag changes since tags are baked into the CF template.
          cf_client.update_stack(
            stack_name: stack_name,
            use_previous_template: true,
 

      def self.call(pipeline_name:, stage:)
        cf_client = Aws::CloudFormation::Client.new
        # Note: uses directory name directly. Assumes it matches pipeline's turbofan_name.
        stack_name = Turbofan::Naming.stack_name(pipeline_name, stage)
        state = Turbofan::Deploy::StackManager.detect_state(cf_client, stack_name)

        case state
        when :does_not_exist
          raise "Stack does not exist: #{stack_name}"
        when :create_complete
          raise "No previous deployment to rollback to for #{stack_name}"
        when :in_progress
          raise "Another operation is in progress on #{stack_name}"
        when :update_complete, :update_rollback_complete
          # Reapplies the previous template version. Effective for reverting
          # image tag changes since tags are baked into the CF template.
          cf_client.update_stack(
            stack_name: stack_name,
            use_previous_template: true,
            capabilities: ["CAPABILITY_N

  class CLI < Thor
    module Run
      def self.call(pipeline_name:, stage:, input: nil, input_file: nil, dry_run: false)
        cf = Aws::CloudFormation::Client.new
        sfn = Aws::States::Client.new
        stack_name = Turbofan::Naming.stack_name(pipeline_name, stage)

        sm_arn = Turbofan::Deploy::StackManager.stack_output(cf, stack_name, "StateMachineArn")

        exec_input = input || (input_file && File.read(input_file)) || "{}"

        if dry_run
          puts "Dry run: Validation checks passed"
          puts "Pipeline: #{pipeline_name}"
          puts "Stage: #{stage}"
          puts "Steps would execute with input: #{exec_input}"
          puts "State machine: #{sm_arn}"
          puts "Dry run complete. Use without --dry-run to execute."
          return
        end

        execution_arn = Turbofan::Deploy::Execution.start(
          sfn,
          state_machine_arn: sm_arn,
          input: exec_input
        )

        puts "Execution started: #{execution_ar

      def self.call(pipeline_name:, stage:, input: nil, input_file: nil, dry_run: false)
        cf = Aws::CloudFormation::Client.new
        sfn = Aws::States::Client.new
        stack_name = Turbofan::Naming.stack_name(pipeline_name, stage)

        sm_arn = Turbofan::Deploy::StackManager.stack_output(cf, stack_name, "StateMachineArn")

        exec_input = input || (input_file && File.read(input_file)) || "{}"

        if dry_run
          puts "Dry run: Validation checks passed"
          puts "Pipeline: #{pipeline_name}"
          puts "Stage: #{stage}"
          puts "Steps would execute with input: #{exec_input}"
          puts "State machine: #{sm_arn}"
          puts "Dry run complete. Use without --dry-run to execute."
          return
        end

  class CLI < Thor
    module Status
      STATUS_INDICATORS = {
        "SUCCEEDED" => "✓",
        "RUNNING" => "⟳",
        "FAILED" => "✗",
        "PENDING" => "·"
      }.freeze

      def self.call(pipeline_name:, stage:, watch: false)
        cf = Aws::CloudFormation::Client.new
        sfn = Aws::States::Client.new
        stack_name = Turbofan::Naming.stack_name(pipeline_name, stage)
        sm_arn = Turbofan::Deploy::StackManager.stack_output(cf, stack_name, "StateMachineArn")

        steps = load_pipeline_steps(pipeline_name)

        loop do
          executions = sfn.list_executions(
            state_machine_arn: sm_arn,
            status_filter: "RUNNING"
          ).executions

          if executions.empty?
            $stdout.puts "No active executions for #{stack_name}."
            break
          end

          $stdout.puts "Active executions for #{stack_name}:"
          $stdout.puts ""

          has_fetch = false
          executions.each do |exec|
          

      def self.call(pipeline_name:, stage:, watch: false)
        cf = Aws::CloudFormation::Client.new
        sfn = Aws::States::Client.new
        stack_name = Turbofan::Naming.stack_name(pipeline_name, stage)
        sm_arn = Turbofan::Deploy::StackManager.stack_output(cf, stack_name, "StateMachineArn")

        steps = load_pipeline_steps(pipeline_name)

        loop do
          executions = sfn.list_executions(
            state_machine_arn: sm_arn,
            status_filter: "RUNNING"
          ).executions

          if executions.empty?
            $stdout.puts "No active executions for #{stack_name}."
            break
          end

      def self.print_fetch_status(status)
        started = status[:started_at] ? Time.parse(status[:started_at]) : nil
        started_ago = time_ago(started)
        $stdout.puts "#{status[:execution_id]} (#{status[:status]}, started #{started_ago})"

        status[:steps].each do |step|
          indicator = STATUS_INDICATORS.fetch(step[:status], "?")
          jobs = step[:jobs]
          total = jobs.values.sum
          parts = []
          parts << "#{jobs[:succeeded]}/#{total} succeeded" if jobs[:succeeded] > 0 || step[:status] == "SUCCEEDED"
          parts << "#{jobs[:running]} running" if jobs[:running] > 0
          parts << "#{jobs[:failed]} failed" if jobs[:failed] > 0
          parts << "#{jobs[:pending]} pending" if jobs[:pending] > 0
          job_str = parts.empty? ? "" : "  #{parts.join("  ")}"
          $stdout.puts "  #{indicator} #{step[:name]} #{step[:status]}#{job_str}"
        end

      def self.time_ago(time)
        return "just now" unless time
        seconds = (Time.now - time).to_i
        if seconds < 60
          "#{seconds}s ago"
        elsif seconds < 3600
          "#{seconds / 60}m ago"
        else
          "#{seconds / 3600}h ago"
        end

      def self.load_pipeline_steps(pipeline_name)
        turbofans_root = "turbofans"
        pipeline_file = File.join(turbofans_root, "pipelines", "#{pipeline_name}.rb")
        return [] unless File.exist?(pipeline_file)

        load_result = Turbofan::Deploy::PipelineLoader.load(pipeline_file, turbofans_root: turbofans_root)
        load_result.steps
      rescue StandardError => e
        warn("[Turbofan] WARNING: Could not load pipeline steps: #{e.message}")
        []
      end

    def self.included(base)
      base.extend(ClassMethods)
      base.instance_variable_set(:@turbofan_instance_types, ["optimal"])
      base.instance_variable_set(:@turbofan_max_vcpus, 256)
      base.instance_variable_set(:@turbofan_min_vcpus, 0)
      base.instance_variable_set(:@turbofan_allocation_strategy, "SPOT_PRICE_CAPACITY_OPTIMIZED")
      base.instance_variable_set(:@turbofan_subnets, nil)
      base.instance_variable_set(:@turbofan_security_groups, nil)
    end

      def instance_types(types)
        @turbofan_instance_types = Array(types)
      end

      def max_vcpus(value)
        @turbofan_max_vcpus = value
      end

      def min_vcpus(value)
        @turbofan_min_vcpus = value
      end

      def allocation_strategy(value)
        @turbofan_allocation_strategy = value
      end

      def subnets(value)
        @turbofan_subnets = Array(value)
      end

      def security_groups(value)
        @turbofan_security_groups = Array(value)
      end

      def resolved_subnets
        @turbofan_subnets || Turbofan.config.subnets
      end

      def resolved_security_groups
        @turbofan_security_groups || Turbofan.config.security_groups
      end

      def stack_name(stage)
        slug = name.split("::").last
          .gsub(/([a-z])([A-Z])/, '\1_\2').downcase.tr("_", "-")
        "turbofan-ce-#{slug}-#{stage}"
      end

      def export_name(stage)
        "#{stack_name(stage)}-arn"
      end

      def generate_template(stage:)
        account_id = Turbofan.config.aws_account_id
        raise "Turbofan.config.aws_account_id is required for CE template generation" unless account_id

        slug = name.split("::").last
          .gsub(/([a-z])([A-Z])/, '\1_\2').downcase.tr("_", "-")

        subnet_list = resolved_subnets
        sg_list = resolved_security_groups
        raise "No subnets configured. Set subnets on the CE or in Turbofan.config.subnets" if subnet_list.empty?
        raise "No security_groups configured. Set security_groups on the CE or in Turbofan.config.security_groups" if sg_list.empty?

        instance_types_yaml = @turbofan_instance_types.map { |t| "            - #{t}" }.join("\n")
        subnets_yaml = subnet_list.map { |s| "            - #{s}" }.join("\n")
        sgs_yaml = sg_list.map { |s| "            - #{s}" }.join("\n")

        # Indent UserData for YAML embedding (10 spaces for Fn::Base64 value position)
        userdata_yaml = NVME_USERDATA.

    def self.discover
      ObjectSpace.each_object(Class).select { |c|
        next false unless c < self
        class_name = Turbofan::GET_CLASS_NAME.bind_call(c)
        next false unless class_name
        live = begin
          Object.const_get(class_name)
        rescue NameError
          nil
        end

    def self.resolve(sym)
      class_name = "ComputeEnvironments::#{Turbofan::Naming.pascal_case(sym)}"
      klass = Object.const_get(class_name)
      unless klass.include?(Turbofan::ComputeEnvironment)
        raise ArgumentError, "#{class_name} does not include Turbofan::ComputeEnvironment"
      end

  class Configuration
    attr_accessor :bucket, :schemas_path, :default_region,
      :log_retention_days, :notification_topic_arn, :docker_registry,
      :duckdb_version, :aws_account_id, :subnets, :security_groups

    def initialize
      @bucket = nil
      @schemas_path = nil
      @default_region = nil
      @log_retention_days = 30
      @notification_topic_arn = nil
      @docker_registry = nil
      @duckdb_version = "1.4.3"
      @aws_account_id = nil
      @subnets = []
      @security_groups = []
    end
  end

    def initialize
      @bucket = nil
      @schemas_path = nil
      @default_region = nil
      @log_retention_days = 30
      @notification_topic_arn = nil
      @docker_registry = nil
      @duckdb_version = "1.4.3"
      @aws_account_id = nil
      @subnets = []
      @security_groups = []
    end

  def self.config
    @config ||= Configuration.new
  end

  def self.configure
    yield config
  end

    def initialize(name, fan_out: false, batch_size: nil, **rest)
      raise ArgumentError, "unknown keyword: group (use batch_size: instead)" if rest.key?(:group)
      raise ArgumentError, "unknown keyword: concurrency (use batch_size: instead)" if rest.key?(:concurrency)
      raise ArgumentError, "unknown keyword(s): #{rest.keys.join(", ")}" if rest.any?
      if batch_size
        raise ArgumentError, "batch_size must be a positive integer" unless batch_size.is_a?(Integer) && batch_size > 0
      end

    def fan_out?
      fan_out
    end

  class Dag
    include TSort

    attr_reader :steps, :edges

    def initialize
      @steps = []
      @edges = []
      @predecessors = Hash.new { |h, k| h[k] = [] }
      @nodes = Set.new
      @frozen = false
    end

    def add_step(name, **kwargs)
      raise "DAG is frozen; cannot add steps after construction" if @frozen

      step = DagStep.new(name, **kwargs)
      @steps << step
      @nodes << name
      step
    end

    def add_edge(from:, to:)
      raise "DAG is frozen; cannot add edges after construction" if @frozen

      @edges << {from: from, to: to}
      @nodes << from << to
      @predecessors[to] << from
    end

    def freeze!
      @edges.freeze
      @steps.freeze
      @frozen = true
      self
    end

    def children_of(step_name)
      @edges.select { |e| e[:from] == step_name }.map { |e| e[:to] }
    end

    def parents_of(step_name)
      @predecessors[step_name]
    end

    def sorted_steps
      detect_self_cycles!
      step_map = @steps.each_

    def initialize
      @steps = []
      @edges = []
      @predecessors = Hash.new { |h, k| h[k] = [] }
      @nodes = Set.new
      @frozen = false
    end

    def add_step(name, **kwargs)
      raise "DAG is frozen; cannot add steps after construction" if @frozen

      step = DagStep.new(name, **kwargs)
      @steps << step
      @nodes << name
      step
    end

    def add_edge(from:, to:)
      raise "DAG is frozen; cannot add edges after construction" if @frozen

      @edges << {from: from, to: to}
      @nodes << from << to
      @predecessors[to] << from
    end

    def freeze!
      @edges.freeze
      @steps.freeze
      @frozen = true
      self
    end

    def children_of(step_name)
      @edges.select { |e| e[:from] == step_name }.map { |e| e[:to] }
    end

    def parents_of(step_name)
      @predecessors[step_name]
    end

    def sorted_steps
      detect_self_cycles!
      step_map = @steps.each_with_object({}) { |s, h| h[s.name] = s }
      tsort.filter_map { |name| step_map[name] }
    end

    def detect_self_cycles!
      @edges.each do |edge|
        if edge[:from] == edge[:to]
          raise TSort::Cyclic, "topological sort failed: #{edge[:from].inspect}"
        end

    def tsort_each_node(&block)
      @nodes.each(&block)
    end

    def tsort_each_child(node, &block)
      @predecessors[node].each(&block)
    end

  class DagProxy
    attr_reader :step_name, :schema

    def initialize(step_name, schema: nil)
      @step_name = step_name
      @schema = schema
    end
  end

    def initialize(step_name, schema: nil)
      @step_name = step_name
      @schema = schema
    end