Class: RobotLab::Cyborg

Inherits:
Object
  • Object
show all
Includes:
Robot::BusMessaging
Defined in:
lib/robot_lab/cyborg.rb,
lib/robot_lab/cyborg/channel.rb,
lib/robot_lab/cyborg/version.rb,
lib/robot_lab/cyborg/interviewer.rb,
lib/robot_lab/cyborg/conversation.rb,
sig/robot_lab/cyborg.rbs

Overview

RobotLab::Cyborg is a class (a human peer worker), so its VERSION and the nested helper classes hang off the class itself rather than a module.

Defined Under Namespace

Classes: Channel, ChannelMessage, Conversation, Error, Interviewer, Question

Constant Summary collapse

VERSION =

Returns:

  • (String) 
"0.2.7"

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(name:, bus: nil, channel: nil, interviewer: nil, auto_reply: true, memory: nil, ask_timeout: nil) ⇒ Cyborg

Create a human peer.

Parameters:

  • name (String) —

    unique name (and bus channel name) for this peer

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

    shared bus to join immediately

  • channel (Channel, nil) (defaults to: nil) —

    means of reaching the human (default: a terminal channel on $stdin/$stdout)

  • interviewer (Interviewer, nil) (defaults to: nil) —

    the interaction process (default: a fresh Interviewer over channel)

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

    reply to inbound bus tasks automatically

  • memory (RobotLab::Memory, nil) (defaults to: nil) —

    standalone memory (default: fresh)

  • ask_timeout (Numeric, nil) (defaults to: nil) —

    seconds to wait for the human before giving up on an answer (nil = wait indefinitely)

  • name: (String)
  • bus: (Object) (defaults to: nil)
  • interviewer: (Cyborg::Interviewer, nil) (defaults to: nil)
  • input: (Object)
  • output: (Object)
  • auto_reply: (Boolean) (defaults to: true)
  • memory: (Object) (defaults to: nil) 


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
 
# File 'lib/robot_lab/cyborg.rb', line 75

def initialize(name:, bus: nil, channel: nil, interviewer: nil,
               auto_reply: true, memory: nil, ask_timeout: nil)
  @name = name.to_s

  # ivars the BusMessaging mixin expects to find already initialized
  @bus               = bus
  @bus_poller        = nil
  @private_bus_poller = nil
  @bus_poller_group  = :default
  @bus_subscriber_id = nil
  @message_counter   = 0
  @outbox            = {}
  @bus_mutex         = Mutex.new
  @message_handler   = method(:handle_incoming)

  @auto_reply    = auto_reply
  @ask_timeout   = ask_timeout
  @presence      = :online
  @on_task       = nil
  @on_human      = nil
  @inbox         = []
  @inbox_mutex   = Mutex.new
  @state_mutex   = Mutex.new
  @shared_memory = nil
  @memory        = memory || Memory.new
  @channel       = channel || Channel::Terminal.new(name: @name)
  @interviewer   = interviewer || Interviewer.new(channel: @channel, default_timeout: @ask_timeout)
  @interviewer.on_initiative { |message| handle_human_initiative(message) }

  setup_bus_channel if @bus
end

Instance Attribute Details

#bus ⇒ TypedBus::MessageBus? (readonly)

Returns the shared bus, if any.

Returns:

  • (TypedBus::MessageBus, nil) —

    the shared bus, if any



49
50
51
 
# File 'lib/robot_lab/cyborg.rb', line 49

def bus
  @bus
end

#channel ⇒ Channel (readonly)

Returns the injectable means by which this peer reaches its human.

Returns:

  • (Channel) —

    the injectable means by which this peer reaches its human



55
56
57
 
# File 'lib/robot_lab/cyborg.rb', line 55

def channel
  @channel
end

#interviewer ⇒ Interviewer (readonly)

Returns the process conducting this peer's human interaction.

Returns:

  • (Interviewer) —

    the process conducting this peer's human interaction



58
59
60
 
# File 'lib/robot_lab/cyborg.rb', line 58

def interviewer
  @interviewer
end

#memory ⇒ RobotLab::Memory (readonly)

Returns the peer's own (standalone) memory.

Returns:

  • (RobotLab::Memory) —

    the peer's own (standalone) memory



61
62
63
 
# File 'lib/robot_lab/cyborg.rb', line 61

def memory
  @memory
end

#name ⇒ String (readonly)

Returns the peer's unique name — also its bus channel name.

Returns:

  • (String) —

    the peer's unique name — also its bus channel name



46
47
48
 
# File 'lib/robot_lab/cyborg.rb', line 46

def name
  @name
end

#outbox ⇒ Hash (readonly)

Returns outbox of messages this peer has sent, keyed by message key.

Returns:

  • (Hash) —

    outbox of messages this peer has sent, keyed by message key



52
53
54
 
# File 'lib/robot_lab/cyborg.rb', line 52

def outbox
  @outbox
end

#presence ⇒ Symbol (readonly)

Returns :online, :away, or :offline.

Returns:

  • (Symbol) —

    :online, :away, or :offline



264
265
266
 
# File 'lib/robot_lab/cyborg.rb', line 264

def presence
  @presence
end

Instance Method Details

#ask(question, choices: nil, default: nil, timeout: @ask_timeout, validate: nil, retries: 2) ⇒ Object?

Ask this peer's human a question and block for the answer. This is the synchronous boundary the network relies on (pipeline steps, bus tasks); the underlying interview is asynchronous. Returns the default (nil when none) if no answer arrives within timeout.

Parameters:

  • question (String)
  • choices (Array<String>, nil) (defaults to: nil)
  • default (String, nil) (defaults to: nil)
  • timeout (Numeric, nil) (defaults to: @ask_timeout) —

    seconds to wait (default: this peer's ask_timeout)

  • validate (#call, nil) (defaults to: nil) —

    a validator: return a coerced value when the answer is acceptable, or nil to reject and re-ask

  • retries (Integer) (defaults to: 2) —

    extra attempts allowed when validation rejects

  • choices: (Array[String], nil) (defaults to: nil)
  • default: (String, nil) (defaults to: nil)

Returns:

  • (Object, nil) —

    the human's answer (coerced when validated)



162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
 
# File 'lib/robot_lab/cyborg.rb', line 162

def ask(question, choices: nil, default: nil, timeout: @ask_timeout, validate: nil, retries: 2)
  attempt = 0
  loop do
    answer = @interviewer.ask_and_wait(question, choices: choices, default: default, timeout: timeout)
    return answer if validate.nil? || answer.nil?

    value = validate.call(answer)
    return value unless value.nil?

    attempt += 1
    return default if attempt > retries

    tell(%(Sorry, I couldn't use "#{answer}". Please try again.))
  end
end

#ask_async(question, choices: nil, default: nil) ⇒ Question

Ask this peer's human a question without blocking, returning the pending Question so the caller can wait on it later, on its own terms.

Parameters:

  • question (String)
  • choices (Array<String>, nil) (defaults to: nil)
  • default (String, nil) (defaults to: nil)

Returns:



197
198
199
 
# File 'lib/robot_lab/cyborg.rb', line 197

def ask_async(question, choices: nil, default: nil)
  @interviewer.ask(question, choices: choices, default: default)
end

#ask_confirm(question) ⇒ Boolean?

Ask a yes/no question, returning true/false (nil if never answered).

Returns:

  • (Boolean, nil) 


186
187
188
 
# File 'lib/robot_lab/cyborg.rb', line 186

def ask_confirm(question, **)
  ask(question, choices: %w[yes no], validate: method(:parse_bool), **)
end

#ask_int(question) ⇒ Integer?

Ask for an integer, re-asking until the human gives a parseable number.

Returns:

  • (Integer, nil) 


180
181
182
 
# File 'lib/robot_lab/cyborg.rb', line 180

def ask_int(question, **)
  ask(question, validate: ->(a) { Integer(a.to_s.strip, exception: false) }, **)
end

#assign(to:, task:) ⇒ RobotMessage

Issue a task to another member over the bus (fire-and-forget; the reply, if any, is correlated into #outbox). Alias for the robot idiom send_message, named for how a human hands off work.

Parameters:

  • to (String, Symbol) —

    target member's name/channel

  • task (String, Hash) —

    the task payload

  • to: (String, Symbol)
  • task: (Object)

Returns:

  • (RobotMessage) —

    the sent message



306
307
308
 
# File 'lib/robot_lab/cyborg.rb', line 306

def assign(to:, task:)
  send_message(to: to, content: task)
end

#attach_memory(mem) ⇒ self

Attach this peer to a shared memory — what #remember/#recall target. A network run attaches automatically; call #detach_memory to return to this peer's own standalone memory (so it doesn't keep writing to a finished network's memory).

Parameters:

  • mem (RobotLab::Memory)

Returns:

  • (self) 


367
368
369
370
 
# File 'lib/robot_lab/cyborg.rb', line 367

def attach_memory(mem)
  @state_mutex.synchronize { @shared_memory = mem }
  self
end

#available? ⇒ Boolean

Whether this human peer will take work now. The network can check this before delegating and route around or escalate for an absent human.

Returns:

  • (Boolean) 


270
271
272
 
# File 'lib/robot_lab/cyborg.rb', line 270

def available?
  @state_mutex.synchronize { @presence != :offline }
end

#away! ⇒ self

Mark the human present but slow to respond (still asked; caller should use a generous timeout).

Returns:

  • (self) 


284
285
286
287
 
# File 'lib/robot_lab/cyborg.rb', line 284

def away!
  @state_mutex.synchronize { @presence = :away }
  self
end

#call(result) ⇒ SimpleFlow::Result

SimpleFlow step interface. The network calls this when the pipeline reaches the human; the human performs the step and the result flows downstream just like any robot's RobotResult.

Parameters:

  • result (SimpleFlow::Result) —

    incoming pipeline result

Returns:

  • (SimpleFlow::Result) 


115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
 
# File 'lib/robot_lab/cyborg.rb', line 115

def call(result)
  run_context = extract_run_context(result)
  started = clock
  robot_result = run(run_context[:message], network_memory: run_context[:network_memory])
  robot_result.duration = clock - started

  result
    .with_context(@name.to_sym, robot_result)
    .continue(robot_result)
rescue StandardError => e
  error_result = build_result("Error: #{e.class}: #{e.message}")
  result
    .with_context(@name.to_sym, error_result)
    .continue(error_result)
end

#converse(peers: []) ⇒ Conversation

Start an interactive Conversation: the human addresses peers by @mention (no mention broadcasts to all), replies come back on the channel. Returns the started Conversation.

Parameters:

  • peers (Array<String, Symbol>) (defaults to: []) —

    addressable member names

Returns:



240
241
242
 
# File 'lib/robot_lab/cyborg.rb', line 240

def converse(peers: [])
  Conversation.new(cyborg: self, peers: peers).start
end

#delegate(to:, task:, async: false) ⇒ RobotResult, DelegationFuture

Delegate a task to another member and get a RobotResult back, synchronously or asynchronously. Works against robots and cyborgs alike, because both respond to run.

Parameters:

  • to (#run) —

    the member to delegate to (Robot or Cyborg)

  • task (String) —

    the task message

  • async (Boolean) (defaults to: false) —

    when true, returns a DelegationFuture immediately

  • to: (Object)
  • task: (String)
  • async: (Boolean) (defaults to: false)
  • (Object)

Returns:

  • (RobotResult, DelegationFuture) 


318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
 
# File 'lib/robot_lab/cyborg.rb', line 318

def delegate(to:, task:, async: false, **)
  if async
    future = DelegationFuture.new(robot_name: to.name, delegated_by: @name)
    delegator = @name
    Thread.new do
      result = to.run(task, **)
      result.delegated_by = delegator
      future.resolve!(result)
    rescue StandardError => e
      future.reject!(e)
    end
    future
  else
    result = to.run(task, **)
    result.delegated_by = @name
    result
  end
end

#detach_memory ⇒ self

Detach from any shared memory, returning to standalone memory.

Returns:

  • (self) 


374
375
376
377
 
# File 'lib/robot_lab/cyborg.rb', line 374

def detach_memory
  @state_mutex.synchronize { @shared_memory = nil }
  self
end

#inbox ⇒ Array<RobotMessage>

Inbound bus messages this peer has received, oldest first.

Returns:

  • (Array<RobotMessage>) 


384
385
386
 
# File 'lib/robot_lab/cyborg.rb', line 384

def inbox
  @inbox_mutex.synchronize { @inbox.dup }
end

#listen ⇒ self

Start always-on listening: keep reading the channel even when no question is outstanding, so the human can speak to the network unprompted at any time. Their input arrives via #on_human. Idempotent.

Returns:

  • (self) 


249
250
251
252
 
# File 'lib/robot_lab/cyborg.rb', line 249

def listen
  @interviewer.listen
  self
end

#offline! ⇒ self

Mark the human unavailable. Inbound bus tasks are declined immediately instead of waiting on a human who isn't there.

Returns:

  • (self) 


292
293
294
295
 
# File 'lib/robot_lab/cyborg.rb', line 292

def offline!
  @state_mutex.synchronize { @presence = :offline }
  self
end

#on_human {|ChannelMessage| ... } ⇒ self

Register a callback fired when the human sends something unprompted — a message over the channel that answers no outstanding question. This is the human-initiates-into-the-network path; a Conversation turns these into addressed bus messages, or handle them yourself here.

Yields:

Returns:

  • (self) 


217
218
219
220
 
# File 'lib/robot_lab/cyborg.rb', line 217

def on_human(&block)
  @on_human = block
  self
end

#on_task {|message, answer| ... } ⇒ self

Register a callback fired after the human answers an inbound bus task.

Yields:

  • (message, answer) —

    the inbound RobotMessage and the human's answer

Yield Parameters:

  • arg0 (Object)
  • arg1 (String)

Yield Returns:

  • (void)

Returns:

  • (self) 


205
206
207
208
 
# File 'lib/robot_lab/cyborg.rb', line 205

def on_task(&block)
  @on_task = block
  self
end

#online! ⇒ self

Mark the human present and taking work.

Returns:

  • (self) 


276
277
278
279
 
# File 'lib/robot_lab/cyborg.rb', line 276

def online!
  @state_mutex.synchronize { @presence = :online }
  self
end

#recall(key, wait: false) ⇒ Object?

Read from the active memory, optionally blocking until another member writes the key.

Parameters:

  • key (Object)
  • wait (Boolean, Numeric) (defaults to: false) —

    false, true, or seconds to wait

  • wait: (Object) (defaults to: false)

Returns:

  • (Object, nil) 


356
357
358
 
# File 'lib/robot_lab/cyborg.rb', line 356

def recall(key, wait: false)
  current_memory.get(key, wait: wait)
end

#remember(key, value) ⇒ Object

Write to the active memory (the network's shared memory when in a network, otherwise this peer's own memory). Other members see it immediately.

Parameters:

  • key (Object)
  • value (Object)

Returns:

  • (Object) —

    value



345
346
347
348
 
# File 'lib/robot_lab/cyborg.rb', line 345

def remember(key, value)
  current_memory.set(key, value)
  value
end

#run(message = nil, network_memory: nil, memory: nil, **_kwargs) ⇒ RobotResult

Perform one unit of work by asking the human, and return a RobotResult so the human is interchangeable with a robot everywhere (pipeline steps, Robot#delegate, etc.).

Parameters:

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

    the task / prompt for the human

  • network_memory (RobotLab::Memory, nil) (defaults to: nil) —

    shared memory when in a network

  • memory (RobotLab::Memory, nil) (defaults to: nil) —

    explicit memory override

  • network_memory: (Object) (defaults to: nil)
  • memory: (Object) (defaults to: nil)
  • (Object)

Returns:

  • (RobotResult) 


139
140
141
142
143
144
145
 
# File 'lib/robot_lab/cyborg.rb', line 139

def run(message = nil, network_memory: nil, memory: nil, **_kwargs)
  active = memory || network_memory || @memory
  attach_memory(network_memory) if network_memory

  answer = with_writer(active) { ask(message.to_s) }
  build_result(answer)
end

#tell(text, kind: :notice) ⇒ self

Say something to the human over the channel (network -> human), unprompted — the output half of the duplex. Use this to surface notices, or let a Conversation route peer replies here automatically.

Parameters:

  • text (String)
  • kind (Symbol) (defaults to: :notice) —

    :notice | :message | :question

Returns:

  • (self) 


229
230
231
232
 
# File 'lib/robot_lab/cyborg.rb', line 229

def tell(text, kind: :notice)
  @channel.deliver(ChannelMessage.new(content: text.to_s, kind: kind))
  self
end

#to_h ⇒ Hash

Returns:

  • (Hash) 


389
390
391
392
393
394
395
396
 
# File 'lib/robot_lab/cyborg.rb', line 389

def to_h
  {
    name: @name,
    kind: :cyborg,
    bus: @bus ? true : nil,
    channel: @channel.class.name
  }.compact
end

#unlisten ⇒ self

Stop always-on listening.

Returns:

  • (self) 


256
257
258
259
 
# File 'lib/robot_lab/cyborg.rb', line 256

def unlisten
  @interviewer.unlisten
  self
end