Class: Phronomy::FSMSession

Inherits:
Object
  • Object
show all
Defined in:
lib/phronomy/engine/fsm_session.rb

Overview

Event-driven execution wrapper for a single FSM session.

Used by both WorkflowRunner (for Workflow) and Agent::InvocationSession (for Agent invoke). Not Workflow-specific.

Created by a runner and registered with EventLoop. All public methods are called from the EventLoop thread — FSMSession is NOT thread-safe and must not be accessed concurrently from multiple threads.

Constant Summary collapse

FINISH = 
WorkflowRunner::FINISH

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(id:, context:, entry_point:, entry_actions:, auto_state_set:, declared_states:, wait_state_names:, external_events:, phase_machine_class:, recursion_limit:, action_timeouts: {}, resume_event: nil, resume_phase: nil) ⇒ FSMSession

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 FSMSession.

Parameters:

  • id (String)
  • context (Object) —

    includes Phronomy::WorkflowContext

  • entry_point (Symbol) —

    initial state name

  • entry_actions (Hash) —

    { state_name => [callable, ...] }

  • auto_state_set (Hash) —

    { state_name => true }

  • declared_states (Array<Symbol>) —

    all action state names

  • wait_state_names (Array<Symbol>)
  • external_events (Hash) —

    { event_name => [to:, guard:] }

  • phase_machine_class (Class) —

    state_machines-backed phase tracker class

  • recursion_limit (Integer)
  • action_timeouts (Hash) (defaults to: {}) —

    { state_name => seconds }

  • resume_event (Symbol, nil) (defaults to: nil) —

    external event to fire when resuming

  • resume_phase (Symbol, nil) (defaults to: nil) —

    wait state name to resume from



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
 
# File 'lib/phronomy/engine/fsm_session.rb', line 33

def initialize(id:, context:, entry_point:, entry_actions:, auto_state_set:,
  declared_states:, wait_state_names:, external_events:, phase_machine_class:,
  recursion_limit:, action_timeouts: {}, resume_event: nil, resume_phase: nil)
  @id = id
  @ctx = context
  @entry_point = entry_point
  @entry_actions = entry_actions
  @auto_state_set = auto_state_set
  @declared_states = declared_states
  @wait_state_names = wait_state_names
  @external_events = external_events
  @phase_machine_class = phase_machine_class
  @recursion_limit = recursion_limit
  @action_timeouts = action_timeouts
  @resume_event = resume_event
  @resume_phase = resume_phase
  @step = 0
  @done = false
  @current_state = nil
  @tracker = nil
end

Instance Attribute Details

#id ⇒ String (readonly)

Returns workflow thread_id (matches WorkflowContext#thread_id).

Returns:

  • (String) —

    workflow thread_id (matches WorkflowContext#thread_id)



17
18
19
 
# File 'lib/phronomy/engine/fsm_session.rb', line 17

def id
  @id
end

Instance Method Details

#handle(event) ⇒ 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.

Processes an event dispatched from EventLoop. Called for :state_completed, :action_completed, and all user-defined external events.

Parameters:



124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
 
# File 'lib/phronomy/engine/fsm_session.rb', line 124

def handle(event)
  return if @done

  if event.type == :action_completed
    # An awaitable entry action completed: update context and advance.
    @ctx = event.payload if _fsm_context?(event.payload)
    @tracker.context = @ctx
    @tracker.async_pending = false  # Reset flag set by start or fire_and_advance!
    advance_or_halt
    return
  end

  # When :state_completed arrives from an async Task (non-WorkflowContext result),
  # async_pending may still be true from the spawn. Clear it before advancing.
  @tracker.async_pending = false if event.type == :state_completed && @tracker.async_pending

  fire_and_advance!(event.type)
rescue => e
  finish_with_error(e)
end

#start ⇒ Object

Begins workflow execution. Called by EventLoop on :start event.



56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
 
# File 'lib/phronomy/engine/fsm_session.rb', line 56

def start
  if @resume_event
    # Resume from wait state: position tracker at the wait state, then fire the
    # external event. state_machines fires before_transition (exit) and
    # after_transition (entry) callbacks, so both actions execute here.
    @current_state = @resume_phase
    @tracker = build_tracker(@current_state)
    @tracker.context = @ctx
    @tracker.session_id = @id if @tracker.respond_to?(:session_id=)
    fire_and_advance!(@resume_event)
  else
    # Fresh start: state_machines does not fire callbacks on initialization,
    # so we invoke the entry action for the initial state manually.
    @current_state = @entry_point
    @tracker = build_tracker(@current_state)
    @tracker.context = @ctx
    @tracker.session_id = @id if @tracker.respond_to?(:session_id=)
    (@entry_actions[@current_state] || []).each do |c|
      result = c.call(@ctx)
      if result.is_a?(Phronomy::Task)
        # Awaitable action: resume via on_complete without blocking EventLoop.
        @tracker.async_pending = true
        session_id = @id
        current_state_name = @current_state
        timeout_secs = @action_timeouts[current_state_name]
        if timeout_secs
          Phronomy::Runtime.instance.timer_queue.schedule(seconds: timeout_secs) do
            next if result.done?

            event_loop.post(
              Event.new(
                type: :error,
                target_id: Phronomy::EventLoop::SYSTEM_CHANNEL_ID,
                payload: {session_id: session_id, result: Phronomy::ActionTimeoutError.new(
                  "Action in state #{current_state_name.inspect} timed out after #{timeout_secs}s"
                )}
              )
            )
          end
        end
        result.on_complete do |task_result, error|
          if error
            event_loop.post(Event.new(type: :error, target_id: Phronomy::EventLoop::SYSTEM_CHANNEL_ID, payload: {session_id: session_id, result: error}))
            next
          end
          if _fsm_context?(task_result)
            event_loop.post(Event.new(type: :action_completed, target_id: session_id, payload: task_result))
          else
            event_loop.post(Event.new(type: :state_completed, target_id: session_id, payload: nil))
          end
        end
        break # Only one async action at a time per state
      elsif _fsm_context?(result)
        @ctx = result
      end
    end
    @tracker.context = @ctx
    advance_or_halt unless @tracker.async_pending
  end
rescue => e
  finish_with_error(e)
end