Module: RobotLab::Robot::BusMessaging

Defined in:
lib/robot_lab/robot/bus_messaging.rb

Overview

Inter-robot communication via TypedBus.

Expects the including class to provide:

@bus, @bus_poller, @bus_poller_group, @message_counter,
@outbox, @message_handler, @bus_subscriber_id, @name
and the `run` instance method.

Delivery Serialization

TypedBus delivers messages in concurrent Async fibers. Robots enqueue deliveries into a BusPoller rather than handling them inline. The BusPoller drains each group's queue sequentially on a dedicated OS thread, so robot.run() calls never interleave.

Owns: @bus, @bus_poller, @private_bus_poller, @bus_poller_group, @bus_subscriber_id, @message_counter, @outbox, @message_handler Reads: @name Contract: ivars initialized by initialize_runtime_state before first bus operation

Instance Method Summary collapse

Instance Method Details

#assign_bus_poller(poller, group: :default) ⇒ void

This method returns an undefined value.

Assign a shared BusPoller from a Network.

Stops any private poller this robot auto-created, then adopts the network's shared poller for the given group.

Parameters:

  • poller (BusPoller) —

    the network's shared poller

  • group (Symbol) (defaults to: :default) —

    poller group for this robot (default: :default)



175
176
177
178
179
180
 
# File 'lib/robot_lab/robot/bus_messaging.rb', line 175

def assign_bus_poller(poller, group: :default)
  @private_bus_poller&.stop
  @private_bus_poller = nil
  @bus_poller       = poller
  @bus_poller_group = group
end

#inherited_llm_settings ⇒ Hash

Model/provider a spawned robot inherits from its parent so a specialist runs on the SAME LLM as the robot that spawned it (e.g. a local Ollama model) instead of falling back to the global default. Caller-supplied opts override these.

Returns:

  • (Hash) 


141
142
143
144
145
146
 
# File 'lib/robot_lab/robot/bus_messaging.rb', line 141

def inherited_llm_settings
  settings = {}
  settings[:model]    = model    if model
  settings[:provider] = provider if provider
  settings
end

#on_message {|message| ... } ⇒ self

Register a custom handler for incoming bus messages.

Block arity controls delivery handling:

  • 1 argument |message|: auto-acks before calling, auto-nacks on exception
  • 2 arguments |delivery, message|: manual mode, you call ack!/nack!

Yields:

  • (message) —

    or [delivery, message]

Returns:

  • (self) 


71
72
73
74
 
# File 'lib/robot_lab/robot/bus_messaging.rb', line 71

def on_message(&block)
  @message_handler = block
  self
end

#respond_to_tasks(auto_reply: true) {|message| ... } ⇒ self

Automatically respond to inbound (non-reply) bus tasks: run responder to produce a reply, and send it back to the sender. This is the symmetric counterpart to how a Cyborg answers its human — one call makes any bus member a first-class responder instead of hand-wiring #on_message.

The responder runs on the poller drain thread, so deliveries to this member are handled one at a time (a long turn delays the next inbound message).

Parameters:

  • auto_reply (Boolean) (defaults to: true) —

    send the responder's result back to the sender

Yields:

  • (message) —

    the inbound task; return the reply content (nil => no reply)

Returns:

  • (self) 


87
88
89
90
91
92
93
94
95
 
# File 'lib/robot_lab/robot/bus_messaging.rb', line 87

def respond_to_tasks(auto_reply: true, &responder)
  on_message do |message|
    next if message.reply?

    reply = responder.call(message)
    send_reply(to: message.from, content: reply, in_reply_to: message.key) if auto_reply && reply
  end
  self
end

#send_message(to:, content:) ⇒ RobotMessage

Send a message to another robot via the bus.

Parameters:

  • to (String, Symbol) —

    target robot's channel name

  • content (String, Hash) —

    message payload

Returns:

Raises:

  • (BusError) —

    if no bus is configured



29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
 
# File 'lib/robot_lab/robot/bus_messaging.rb', line 29

def send_message(to:, content:)
  raise BusError, "No bus configured on robot '#{@name}'" unless @bus

  # Counter + outbox are shared with the poller thread (reply correlation)
  # and with other senders; mutate them under the bus mutex. Publish (which
  # does I/O) stays outside the lock.
  message = @bus_mutex.synchronize do
    @message_counter += 1
    msg = RobotMessage.build(id: @message_counter, from: @name, content: content)
    @outbox[msg.key] = { message: msg, status: :sent, replies: [] }
    msg
  end
  publish_to_bus(to.to_sym, message)
  message
end

#send_reply(to:, content:, in_reply_to:) ⇒ RobotMessage

Send a reply to a specific message via the bus.

Parameters:

  • to (String, Symbol) —

    target robot's channel name

  • content (String, Hash) —

    reply payload

  • in_reply_to (String) —

    composite key of the message being replied to

Returns:

Raises:

  • (BusError) —

    if no bus is configured



52
53
54
55
56
57
58
59
60
61
 
# File 'lib/robot_lab/robot/bus_messaging.rb', line 52

def send_reply(to:, content:, in_reply_to:)
  raise BusError, "No bus configured on robot '#{@name}'" unless @bus

  reply = @bus_mutex.synchronize do
    @message_counter += 1
    RobotMessage.build(id: @message_counter, from: @name, content: content, in_reply_to: in_reply_to)
  end
  publish_to_bus(to.to_sym, reply)
  reply
end

#serve(auto_reply: true) ⇒ self

Serve inbound bus tasks by running each through this member's #run and replying with the result — the one-call way to make a Robot cooperate on the bus the way a Cyborg already does out of the box.

Parameters:

  • auto_reply (Boolean) (defaults to: true)

Returns:

  • (self) 


103
104
105
 
# File 'lib/robot_lab/robot/bus_messaging.rb', line 103

def serve(auto_reply: true)
  respond_to_tasks(auto_reply: auto_reply) { |message| run(bus_task_content(message)).reply }
end

#spawn(name: "robot", system_prompt: nil, template: nil, local_tools: []) ⇒ Robot

Spawn a new robot on a shared bus.

Creates a new Robot instance that shares this robot's bus, allowing it to immediately send and receive messages with all other robots on the bus. If no bus exists yet, one is created automatically and the parent robot is connected to it.

Parameters:

  • name (String) (defaults to: "robot") —

    unique name for the new robot

  • system_prompt (String, nil) (defaults to: nil) —

    inline system prompt

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

    prompt_manager template

  • local_tools (Array) (defaults to: []) —

    tools for the new robot

  • options (Hash) —

    additional options passed to RobotLab.build

Returns:

  • (Robot) —

    the newly created robot



121
122
123
124
125
126
127
128
129
130
131
132
133
 
# File 'lib/robot_lab/robot/bus_messaging.rb', line 121

def spawn(name: "robot", system_prompt: nil, template: nil, local_tools: [], **)
  ensure_bus

  RobotLab.build(
    name: name,
    system_prompt: system_prompt,
    template: template,
    local_tools: local_tools,
    bus: @bus,
    **inherited_llm_settings,
    **
  )
end

#with_bus(bus = nil) ⇒ self

Connect this robot to a message bus.

If a bus is provided, the robot joins it. If no bus is provided and the robot doesn't already have one, a new bus is created. No-op if the robot is already on the given bus.

Parameters:

  • bus (TypedBus::MessageBus, nil) (defaults to: nil) —

    bus to join (creates one if nil)

Returns:

  • (self) 


157
158
159
160
161
162
163
164
 
# File 'lib/robot_lab/robot/bus_messaging.rb', line 157

def with_bus(bus = nil)
  return self if bus && @bus == bus

  teardown_bus_channel if @bus
  @bus = bus || @bus || TypedBus::MessageBus.new
  setup_bus_channel
  self
end