Class: Ocak::PipelineRunner

Inherits:
Object
  • Object
show all
Defined in:
lib/ocak/pipeline_runner.rb

Instance Method Summary collapse

Constructor Details

#initialize(config:, options: {}) ⇒ PipelineRunner

Returns a new instance of PipelineRunner.



7
8
9
10
11
12
13
14
15
 
# File 'lib/ocak/pipeline_runner.rb', line 7

def initialize(config:, options: {})
  @config = config
  @options = options
  @watch_formatter = options[:watch] ? WatchFormatter.new : nil
  @shutting_down = false
  @active_issues = []
  @active_mutex = Mutex.new
  @executor = PipelineExecutor.new(config: config)
end

Instance Method Details

#run ⇒ Object 



17
18
19
 
# File 'lib/ocak/pipeline_runner.rb', line 17

def run
  @options[:single] ? run_single(@options[:single]) : run_loop
end

#run_pipeline(issue_number, logger:, claude:, chdir: nil, skip_steps: [], complexity: 'full') ⇒ Object 



21
22
23
24
 
# File 'lib/ocak/pipeline_runner.rb', line 21

def run_pipeline(issue_number, logger:, claude:, chdir: nil, skip_steps: [], complexity: 'full')
  @executor.run_pipeline(issue_number, logger: logger, claude: claude, chdir: chdir,
                                       skip_steps: skip_steps, complexity: complexity)
end

#shutdown! ⇒ Object 



26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
 
# File 'lib/ocak/pipeline_runner.rb', line 26

def shutdown!
  @shutting_down = true
  logger = build_logger
  logger.info('Graceful shutdown initiated...')

  issues = IssueFetcher.new(config: @config, logger: logger)
  @active_mutex.synchronize do
    @active_issues.each do |issue_number|
      logger.info("Returning issue ##{issue_number} to ready queue")
      issues.transition(issue_number, from: @config.label_in_progress, to: @config.label_ready)
    rescue StandardError => e
      logger.warn("Failed to reset issue ##{issue_number}: #{e.message}")
    end
  end
end