Class: VSM::Coordination

Inherits:
Object
  • Object
show all
Defined in:
lib/vsm/roles/coordination.rb

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Coordination

Returns a new instance of Coordination.



4
5
6
7
8
 
# File 'lib/vsm/roles/coordination.rb', line 4

def initialize
  @queue = []
  @floor_by_session = nil
  @turn_waiters = {} # session_id => Async::Queue
end

Instance Method Details

#drain(bus) ⇒ Object 



17
18
19
20
21
22
23
24
25
26
 
# File 'lib/vsm/roles/coordination.rb', line 17

def drain(bus)
  return if @queue.empty?
  @queue.sort_by! { order(_1) }
  @queue.shift(@queue.size).each do |msg|
    yield msg
    if msg.kind == :assistant && (sid = msg.meta&.dig(:session_id)) && @turn_waiters[sid]
      @turn_waiters[sid].enqueue(:done)
    end
  end
end

#grant_floor!(session_id) ⇒ Object 



28
 
# File 'lib/vsm/roles/coordination.rb', line 28

def grant_floor!(session_id) = (@floor_by_session = session_id)

#observe(bus) ⇒ Object 



10
11
12
13
 
# File 'lib/vsm/roles/coordination.rb', line 10

def observe(bus)
  # Note: staging is handled by the capsule loop, not by subscription
  # This method exists for consistency but doesn't auto-stage messages
end

#order(m) ⇒ Object 



35
36
37
38
39
40
41
42
43
44
45
46
47
 
# File 'lib/vsm/roles/coordination.rb', line 35

def order(m)
  base =
    case m.kind
    when :user            then 0
    when :tool_result     then 1
    when :plan            then 2
    when :assistant_delta then 3
    when :assistant       then 4
    else 9
    end
  sid = m.meta&.dig(:session_id)
  sid == @floor_by_session ? base - 1 : base
end

#stage(message) ⇒ Object 



15
 
# File 'lib/vsm/roles/coordination.rb', line 15

def stage(message) = (@queue << message)

#wait_for_turn_end(session_id) ⇒ Object 



30
31
32
33
 
# File 'lib/vsm/roles/coordination.rb', line 30

def wait_for_turn_end(session_id)
  q = (@turn_waiters[session_id] ||= Async::Queue.new)
  q.dequeue
end