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



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

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 = []
end

Instance Attribute Details

#declarations ⇒ Object (readonly)

Returns the value of attribute declarations.



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

def declarations
  @declarations
end

#store ⇒ Object (readonly)

Returns the value of attribute store.



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

def store
  @store
end

Instance Method Details

#checkpoint_each!(index) ⇒ Object 



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

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

#resume(context, input:) ⇒ Object 



57
58
59
60
 
# File 'lib/ask/graph/runner.rb', line 57

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

#resume_index_for(items) ⇒ Object 



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

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

#run(context) ⇒ Object 



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

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