Class: Phronomy::MultiAgent::FanOutInvocation

Inherits:
Object
  • Object
show all
Defined in:
lib/phronomy/multi_agent/fan_out_invocation.rb

Overview

Mutable FSM context for one fan-out operation.

Defined Under Namespace

Classes: Child

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(children:, max_concurrency:, on_error:) ⇒ FanOutInvocation

Returns a new instance of FanOutInvocation.



13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
 
# File 'lib/phronomy/multi_agent/fan_out_invocation.rb', line 13

def initialize(children:, max_concurrency:, on_error:)
  @id = SecureRandom.uuid.to_s
  @phase = nil
  @children = children
  @max_concurrency = max_concurrency || children.length
  @on_error = on_error
  @pending = children.dup
  @active = {}
  @results = Array.new(children.length)
  @child_errors = Array.new(children.length)  # indexed by input order
  @fatal_error = nil   # driver_failed / timeout / cancel
  @cancelled = false
  @timed_out = false
  @session_id = nil
end

Instance Attribute Details

#id ⇒ Object (readonly)

Returns the value of attribute id.



11
12
13
 
# File 'lib/phronomy/multi_agent/fan_out_invocation.rb', line 11

def id
  @id
end

#phase ⇒ Object (readonly)

Returns the value of attribute phase.



11
12
13
 
# File 'lib/phronomy/multi_agent/fan_out_invocation.rb', line 11

def phase
  @phase
end

#results ⇒ Object (readonly)

Returns the value of attribute results.



11
12
13
 
# File 'lib/phronomy/multi_agent/fan_out_invocation.rb', line 11

def results
  @results
end

Instance Method Details

#cancelled? ⇒ Boolean

Returns:

  • (Boolean) 


128
129
130
 
# File 'lib/phronomy/multi_agent/fan_out_invocation.rb', line 128

def cancelled?
  @cancelled
end

#completed? ⇒ Boolean

Returns:

  • (Boolean) 


111
112
113
 
# File 'lib/phronomy/multi_agent/fan_out_invocation.rb', line 111

def completed?
  !failed? && @pending.empty? && @active.empty?
end

#error ⇒ Object 



120
121
122
 
# File 'lib/phronomy/multi_agent/fan_out_invocation.rb', line 120

def error
  @fatal_error || @child_errors.find(&:itself)
end

#failed? ⇒ Boolean

Returns:

  • (Boolean) 


115
116
117
118
 
# File 'lib/phronomy/multi_agent/fan_out_invocation.rb', line 115

def failed?
  return true if @fatal_error
  @on_error == :raise && @pending.empty? && @active.empty? && @child_errors.any?(&:itself)
end

#handle_fsm_event(event) ⇒ Object 



34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/phronomy/multi_agent/fan_out_invocation.rb', line 34

def handle_fsm_event(event)
  case event.type
  when :child_completed
    payload = event.payload
    index = payload.fetch(:index)
    @active.delete(index)
    child_error = payload[:error]
    if child_error
      if @on_error == :raise
        @child_errors[index] = child_error
        # Signal cancellation so other children can stop early, but continue
        # waiting for all active children to respond so we can return the
        # first error in INPUT ORDER (not arrival order).
        cancel_active_children! unless @child_errors.any?(&:itself)
      end
    else
      @results[index] = payload[:result]
    end
    true
  when :driver_failed
    @fatal_error ||= event.payload.fetch(:error)
    cancel_active_children!
    true
  when :timeout
    @timed_out = true
    @fatal_error ||= Phronomy::TimeoutError.new(event.payload.fetch(:message))
    cancel_active_children!
    true
  when :cancel
    @cancelled = true
    @fatal_error ||= Phronomy::CancellationError.new("fan-out cancelled")
    cancel_active_children!
    true
  else
    false
  end
end

#set_graph_metadata(thread_id: nil, phase: nil) ⇒ Object 



29
30
31
32
 
# File 'lib/phronomy/multi_agent/fan_out_invocation.rb', line 29

def set_graph_metadata(thread_id: nil, phase: nil)
  @session_id = thread_id if thread_id
  @phase = phase
end

#start_available!(runtime) ⇒ Object 



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
 
# File 'lib/phronomy/multi_agent/fan_out_invocation.rb', line 72

def start_available!(runtime)
  return self if @fatal_error

  while @active.length < @max_concurrency && (child = @pending.shift)
    child_config = build_child_config(child.config)
    handle = child.agent.invoke_async(
      child.input,
      config: child_config,
      thread_id: child.thread_id
    )
    @active[child.index] = {handle: handle, token: child_config[:cancellation_token]}
    # [child.index].each creates a block parameter with a unique binding
    # per iteration, avoiding the while-loop variable capture problem.
    [child.index].each do |captured_index|
      handle.on_complete do |result, error|
        runtime.event_loop.post_to_session(
          Phronomy::Event.new(
            type: :child_completed,
            target_id: @id,
            payload: {index: captured_index, result: result, error: error}
          )
        )
      end
    end
  end
  self
rescue => caught
  @fatal_error ||= caught
  cancel_active_children!
  runtime.event_loop.post_to_session(
    Phronomy::Event.new(
      type: :driver_failed,
      target_id: @id,
      payload: {error: caught}
    )
  )
  self
end

#timed_out? ⇒ Boolean

Returns:

  • (Boolean) 


124
125
126
 
# File 'lib/phronomy/multi_agent/fan_out_invocation.rb', line 124

def timed_out?
  @timed_out
end