Class: Ask::Graph::Runner

Inherits:
Object
  • Object
show all
Defined in:
lib/ask/graph/runner.rb

Overview

Executes the declared steps of a graph, managing checkpointing, condition evaluation, parallel execution, timeouts, retries, lifecycle hooks, and human-in-the-loop pauses.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(declarations, storage: nil, hooks: { before_step: [], after_step: [], on_failure: [] }, graph_instance: nil, default_timeout: nil) ⇒ Runner

Returns a new instance of Runner.

Parameters:

  • declarations (Array<Hash>) —

    step declarations

  • storage (#set, #get, #delete) (defaults to: nil) —

    key-value store for checkpoint data

  • hooks (Hash) (defaults to: { before_step: [], after_step: [], on_failure: [] }) —

    lifecycle hook method names

  • graph_instance (Object) (defaults to: nil) —

    the graph instance (for hooks)

  • default_timeout (Integer, nil) (defaults to: nil) —

    fallback timeout for steps with no explicit timeout



28
29
30
31
32
33
34
35
36
37
38
 
# File 'lib/ask/graph/runner.rb', line 28

def initialize(declarations, storage: nil,
               hooks: { before_step: [], after_step: [], on_failure: [] },
               graph_instance: nil,
               default_timeout: nil)
  @declarations = declarations
  @store = storage
  @hooks = hooks
  @graph = graph_instance
  @default_timeout = default_timeout
  @completed_steps = []
end

Instance Attribute Details

#declarations ⇒ Object (readonly)

Returns the value of attribute declarations.



21
22
23
 
# File 'lib/ask/graph/runner.rb', line 21

def declarations
  @declarations
end

#store ⇒ Object (readonly)

Returns the value of attribute store.



21
22
23
 
# File 'lib/ask/graph/runner.rb', line 21

def store
  @store
end

Instance Method Details

#checkpoint_each!(index) ⇒ Object 



73
74
75
76
 
# File 'lib/ask/graph/runner.rb', line 73

def checkpoint_each!(index)
  key = each_checkpoint_key
  @store.set(key, index.to_s)
end

#resume(context, input:) ⇒ Object 



62
63
64
65
 
# File 'lib/ask/graph/runner.rb', line 62

def resume(context, input:)
  context.resume_input = input
  run(context)
end

#resume_index_for(items) ⇒ Object 



67
68
69
70
71
 
# File 'lib/ask/graph/runner.rb', line 67

def resume_index_for(items)
  key = each_checkpoint_key
  raw = @store.get(key)
  raw ? raw.to_i : nil
end

#run(context) ⇒ Object 



40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
 
# File 'lib/ask/graph/runner.rb', line 40

def run(context)
  resume_index = load_checkpoint(context)

  @declarations.each_with_index do |decl, idx|
    next if resume_index && idx <= resume_index

    unless condition_met?(decl, context)
      record_completion(idx, :skipped)
      next
    end

    run_hooks(:before_step, decl, context)
    execute_with_retry(decl, context)
    run_hooks(:after_step, decl, context)

    record_completion(idx, :completed)
    save_checkpoint(context, idx)
  end

  context
end