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, step_timeout: nil, workflow_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)

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

    fallback timeout for steps with no explicit timeout

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

    total runtime cap for the whole run



34
35
36
37
38
39
40
41
42
43
44
45
46
47
 
# File 'lib/ask/graph/runner.rb', line 34

def initialize(declarations, storage: nil,
               hooks: { before_step: [], after_step: [], on_failure: [] },
               graph_instance: nil,
               step_timeout: nil,
               workflow_timeout: nil)
  @declarations = declarations
  @store = storage
  @hooks = hooks
  @graph = graph_instance
  @step_timeout = step_timeout
  @workflow_timeout = workflow_timeout
  @completed_steps = []
  @run_id = SecureRandom.hex(8)
end

Instance Attribute Details

#declarations ⇒ Object (readonly)

Returns the value of attribute declarations.



26
27
28
 
# File 'lib/ask/graph/runner.rb', line 26

def declarations
  @declarations
end

#store ⇒ Object (readonly)

Returns the value of attribute store.



26
27
28
 
# File 'lib/ask/graph/runner.rb', line 26

def store
  @store
end

Instance Method Details

#checkpoint_each!(index) ⇒ Object 



70
71
72
73
 
# File 'lib/ask/graph/runner.rb', line 70

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

#resume(context, input:) ⇒ Object 



59
60
61
62
 
# File 'lib/ask/graph/runner.rb', line 59

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

#resume_index_for(items) ⇒ Object 



64
65
66
67
68
 
# File 'lib/ask/graph/runner.rb', line 64

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

#run(context) ⇒ Object 



49
50
51
52
53
54
55
56
57
 
# File 'lib/ask/graph/runner.rb', line 49

def run(context)
  if @workflow_timeout
    ::Timeout.timeout(@workflow_timeout) { run_steps(context) }
  else
    run_steps(context)
  end
rescue ::Timeout::Error
  raise WorkflowTimeout, "workflow timed out after #{@workflow_timeout}s"
end