Module: Legion::Extensions::Mesh::Runners::TaskRequest

Includes:
Mesh
Included in:
Client
Defined in:
lib/legion/extensions/mesh/runners/task_request.rb

Overview

rubocop:disable Legion/Extension/RunnerIncludeHelpers

Constant Summary collapse

DEFAULT_TIMEOUT = 
30

Instance Method Summary collapse

Methods included from Mesh

#dispatch_gossip_message, #expire_silent_agents, #find_agents, #heartbeat, #merge_gossip, #mesh_status, #publish_gossip, #register, #send_message, #unregister

Instance Method Details

#expire_pending_tasks ⇒ Object 



59
60
61
62
63
 
# File 'lib/legion/extensions/mesh/runners/task_request.rb', line 59

def expire_pending_tasks(**)
  expired = task_pending.expire
  log.debug "[mesh-task] expired #{expired.size} pending tasks" unless expired.empty?
  { success: true, expired_count: expired.size, expired_ids: expired }
end

#handle_task_reply(correlation_id:, result:) ⇒ Object 



45
46
47
48
49
50
51
52
53
 
# File 'lib/legion/extensions/mesh/runners/task_request.rb', line 45

def handle_task_reply(correlation_id:, result:, **)
  resolved = task_pending.resolve(correlation_id: correlation_id, result: result)
  if resolved
    log.debug "[mesh-task] reply resolved: cid=#{correlation_id[0..11]}"
    { success: true, resolved: true, correlation_id: correlation_id }
  else
    { success: false, reason: :not_found, correlation_id: correlation_id }
  end
end

#pending_task_stats ⇒ Object 



55
56
57
 
# File 'lib/legion/extensions/mesh/runners/task_request.rb', line 55

def pending_task_stats(**)
  { success: true, pending_count: task_pending.pending_count }
end

#request_task(from:, to:, task:, payload:, timeout: DEFAULT_TIMEOUT, consent_level: :execute) ⇒ Object 



18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
 
# File 'lib/legion/extensions/mesh/runners/task_request.rb', line 18

def request_task(from:, to:, task:, payload:, timeout: DEFAULT_TIMEOUT, consent_level: :execute, **)
  target_id = resolve_target(to)
  return { success: false, reason: :no_agent_found, requested: to } unless target_id

  correlation_id = "task-#{SecureRandom.uuid}"

  delegation = delegation_tracker.create(
    from: from, to: target_id,
    task_context: task, consent_level: consent_level
  )
  return { success: false, reason: delegation[:error] } if delegation[:error]

  task_pending.register(
    correlation_id: correlation_id,
    callback:       nil,
    ttl:            timeout
  )

  send_message(from: from, to: target_id, pattern: :unicast,
               payload: { type: :task_request, correlation_id: correlation_id,
                          task: task, payload: payload, reply_to: from })

  log.info "[mesh-task] request: from=#{from} to=#{target_id} task=#{task} cid=#{correlation_id[0..11]}"
  { success: true, correlation_id: correlation_id, delegation_id: delegation[:delegation_id],
    target_agent: target_id }
end