Class: Phronomy::WorkflowRunner Private

Inherits:
Object
  • Object
show all
Includes:
Runnable
Defined in:
lib/phronomy/workflow_runner.rb

Overview

This class is part of a private API. You should avoid using this class if possible, as it may be removed or be changed in the future.

Execution engine for compiled workflows. Manages state entry/exit action execution, phase transitions, halt/resume, and wait states. Instantiated by Phronomy::Workflow and used internally.

Design principle

State transitions are driven entirely by state_machines. The PhaseTracker holds a reference to the current WorkflowContext via attr_accessor :context, and guard lambdas evaluate m.context (the WorkflowContext) rather than the PhaseTracker itself. This ensures that "what happens next" is always determined by the declared state machine topology, never by Phronomy internals.

Entry and exit actions are registered as state_machines after_transition to: and before_transition from: callbacks respectively. Entry actions may either mutate the context in place or return a new context (e.g. via s.merge(...)). When an entry action returns a Phronomy::WorkflowContext, that value replaces the current context; otherwise the return value is ignored. Exit actions are always mutation-in-place; their return value is ignored.

The sole exception is the initial state: state_machines does not fire transition callbacks on initialization, so the entry action for the entry point is invoked directly by WorkflowRunner before the main execution loop begins.

Two transition categories registered in PhaseTracker

1. state_completed  — all auto-fire transitions (with or without guards).
                   Fired when an action state's action completes.
                   Guards are evaluated in declaration order; first match wins.
                   (declared with +transition from: :foo, to: :bar+ or
                    +transition from: :foo, guard: ..., to: :bar+)

2. <event_name>    — external events triggered by human input, originating
                   from wait states
                   (declared with +transition from: :awaiting, on: :approve, to: :run+)

Constant Summary collapse

FINISH =

This constant is part of a private API. You should avoid using this constant if possible, as it may be removed or be changed in the future.

Sentinel value for the terminal state of a workflow.

:__end__

Instance Method Summary collapse

Methods included from Runnable

#batch, #trace

Constructor Details

#initialize(state_class:, entry_actions:, declared_states:, auto_transitions:, external_events:, entry_point:, exit_actions: {}, wait_state_names: [], state_store: nil, action_timeouts: {}) ⇒ WorkflowRunner

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns a new instance of WorkflowRunner.



48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
 
# File 'lib/phronomy/workflow_runner.rb', line 48

def initialize(state_class:, entry_actions:, declared_states:, auto_transitions:, external_events:, entry_point:, exit_actions: {}, wait_state_names: [], state_store: nil, action_timeouts: {})
  @state_class = state_class
  @entry_actions = entry_actions   # { state_name => [callable, ...] }
  @declared_states = declared_states
  # Lookup set: states with at least one auto-fire transition declared.
  @auto_state_set = auto_transitions.each_with_object({}) { |t, h| h[t[:from]] = true }
  @external_events = external_events    # { name => [{from:, to:, guard:}, ...] }
  @entry_point = entry_point
  @wait_state_names = wait_state_names
  @state_store = state_store
  @action_timeouts = action_timeouts   # { state_name => seconds }
  @phase_machine_class = Workflow::PhaseMachineBuilder.new(
    entry_point: @entry_point,
    declared_states: @declared_states,
    wait_state_names: @wait_state_names,
    external_events: @external_events,
    entry_actions: @entry_actions,
    action_timeouts: @action_timeouts,
    auto_transitions: auto_transitions,
    exit_actions: exit_actions
  ).build
end

Instance Method Details

#invoke(input, config: {}) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Executes the workflow from the initial state.

Parameters:

  • input (Hash) —

    initial context field values

  • config (Hash) (defaults to: {}) —

    { thread_id:, recursion_limit:, user_id:, session_id:, state_store: }

Returns:

  • (Object) —

    final context (includes Phronomy::WorkflowContext)



76
77
78
79
80
81
82
83
84
85
86
87
 
# File 'lib/phronomy/workflow_runner.rb', line 76

def invoke(input, config: {})
  caller_meta = {}
  caller_meta[:user_id] = config[:user_id] if config[:user_id]
  caller_meta[:session_id] = config[:session_id] if config[:session_id]

  trace("workflow.invoke", input: input.inspect, **caller_meta) do |_span|
    state, thread_id, recursion_limit, store = _build_initial_context(input, config)
    result = run_via_event_loop(state, recursion_limit: recursion_limit)
    store&.save(thread_id, {fields: result.to_h, phase: result.phase.to_s}) if config[:thread_id]
    [result, nil]
  end
end

#invoke_deferred(input, config: {}) ⇒ Phronomy::Task

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Registers the workflow with the EventLoop and returns a Task immediately without blocking the caller. The task resolves with the final context when the workflow finishes.

This is the EventLoop-driven equivalent of spawning a thread around #invoke. No extra OS thread is created; the EventLoop's existing thread drives the execution.

Parameters:

  • input (Hash) —

    initial context field values

  • config (Hash) (defaults to: {})

Returns:



101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
 
# File 'lib/phronomy/workflow_runner.rb', line 101

def invoke_deferred(input, config: {})
  state, thread_id, recursion_limit, store = _build_initial_context(input, config)
  result_task = Phronomy::Task.deferred(name: "workflow-async:#{thread_id}")
  Phronomy::EventLoop.instance.start
  session = build_session_for(context: state, recursion_limit: recursion_limit)
  if store && config[:thread_id]
    # Wrap so that state is persisted when the task resolves.
    persist_task = Phronomy::Task.deferred(name: "workflow-async-persist:#{thread_id}")
    Phronomy::EventLoop.instance.register(session, completion: persist_task)
    persist_task.on_complete do |result, error|
      store.save(thread_id, {fields: result.to_h, phase: result.phase.to_s}) unless error
      if error
        result_task.backend.unblock(nil, error)
        result_task.transition!(:failed, error: error)
      else
        result_task.backend.unblock(result, nil)
        result_task.transition!(:completed, value: result)
      end
    end
  else
    Phronomy::EventLoop.instance.register(session, completion: result_task)
  end
  result_task
end

#resume(state:, input: nil) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Generic resume. Equivalent to send_event(state:, event: :resume, input:).

Parameters:

  • state (Object) —

    halted context

  • input (Hash, nil) (defaults to: nil) —

    optional field updates to merge before resuming

Returns:

  • (Object) —

    final context



131
132
133
 
# File 'lib/phronomy/workflow_runner.rb', line 131

def resume(state:, input: nil)
  send_event(state: state, event: :resume, input: input)
end

#send_event(state:, event:, input: nil) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Fires a named event to advance a halted workflow.

The special event :resume selects the first external event registered for the current wait state and fires it.

Parameters:

  • state (Object) —

    halted context

  • event (Symbol) —

    named event or :resume for generic resumption

  • input (Hash, nil) (defaults to: nil) —

    optional field updates to merge before resuming

Returns:

  • (Object) —

    final context



145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
 
# File 'lib/phronomy/workflow_runner.rb', line 145

def send_event(state:, event:, input: nil)
  state = state.merge(input) if input
  event = event.to_sym
  current_phase = state.phase

  ev_to_fire = if event == :resume
    # Find the first external event that can originate from the current wait state.
    name, = @external_events.find { |_, ts| ts.any? { |t| t[:from] == current_phase } }
    unless name
      raise ArgumentError,
        "No external event registered for wait state #{current_phase.inspect}"
    end
    name
  else
    unless @external_events.key?(event)
      raise ArgumentError,
        "Unknown event #{event.inspect}. Valid events: #{@external_events.keys.inspect}"
    end
    event
  end

  run_via_event_loop(state,
    recursion_limit: Phronomy.configuration.recursion_limit,
    resume_event: ev_to_fire, resume_phase: current_phase)
end

#stream(input, config: {}) {|Hash| ... } ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Streaming execution. Yields { state: Symbol, context: Object } after each state action completes.

Parameters:

  • input (Hash)
  • config (Hash) (defaults to: {})

Yields:

  • (Hash)

Returns:

  • (Object) —

    final context



177
178
179
180
181
182
183
 
# File 'lib/phronomy/workflow_runner.rb', line 177

def stream(input, config: {}, &block)
  thread_id = config[:thread_id] || SecureRandom.uuid
  recursion_limit = config.fetch(:recursion_limit, Phronomy.configuration.recursion_limit)
  state = @state_class.new(**input)
  state.set_graph_metadata(thread_id: thread_id)
  run_workflow(state, recursion_limit: recursion_limit, &block)
end