Class: Aidp::Watch::RoundRobinScheduler

Inherits:
Object
  • Object
show all
Defined in:
lib/aidp/watch/round_robin_scheduler.rb

Overview

Schedules work items using a round-robin strategy across all tagged issues and PRs. Processes one iteration per work item before rotating to the next, ensuring fair distribution of attention.

Priority handling:

  • aidp-plan items receive higher priority in rotation
  • Within the same priority, items are processed in queue order

Paused items (aidp-needs-input label) are skipped during their turn but remain in the queue for later processing.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(state_store:) ⇒ RoundRobinScheduler

Returns a new instance of RoundRobinScheduler.



20
21
22
23
24
25
26
27
 
# File 'lib/aidp/watch/round_robin_scheduler.rb', line 20

def initialize(state_store:)
  @state_store = state_store
  @queue = []
  @last_processed_key = state_store.round_robin_last_key

  Aidp.log_debug("round_robin_scheduler", "initialized",
    last_key: @last_processed_key)
end

Instance Attribute Details

#last_processed_key ⇒ Object (readonly)

Returns the value of attribute last_processed_key.



18
19
20
 
# File 'lib/aidp/watch/round_robin_scheduler.rb', line 18

def last_processed_key
  @last_processed_key
end

#queue ⇒ Object (readonly)

Returns the value of attribute queue.



18
19
20
 
# File 'lib/aidp/watch/round_robin_scheduler.rb', line 18

def queue
  @queue
end

Instance Method Details

#mark_processed(item) ⇒ Object

Mark a work item as processed and persist the rotation state.

Parameters:

  • item (WorkItem) —

    The item that was just processed



89
90
91
92
93
94
95
96
97
98
 
# File 'lib/aidp/watch/round_robin_scheduler.rb', line 89

def mark_processed(item)
  @last_processed_key = item.key
  @state_store.record_round_robin_position(
    last_key: item.key,
    processed_at: Time.now.utc.iso8601
  )

  Aidp.log_debug("round_robin_scheduler", "mark_processed",
    key: item.key, number: item.number)
end

#next_item(paused_numbers: []) ⇒ WorkItem?

Get the next work item to process using round-robin rotation. Skips paused items but keeps them in queue.

Parameters:

  • paused_numbers (Array<Integer>) (defaults to: []) —

    Issue/PR numbers that are paused

Returns:

  • (WorkItem, nil) —

    Next item to process, or nil if none available



52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/aidp/watch/round_robin_scheduler.rb', line 52

def next_item(paused_numbers: [])
  return nil if @queue.empty?

  Aidp.log_debug("round_robin_scheduler", "next_item.start",
    queue_size: @queue.size,
    paused_count: paused_numbers.size,
    last_key: @last_processed_key)

  # Find starting position based on last processed key
  start_index = find_start_index

  # Iterate through queue starting from rotation point
  @queue.size.times do |offset|
    index = (start_index + offset) % @queue.size
    item = @queue[index]

    # Skip paused items
    if paused_numbers.include?(item.number)
      Aidp.log_debug("round_robin_scheduler", "next_item.skip_paused",
        key: item.key, number: item.number)
      next
    end

    Aidp.log_debug("round_robin_scheduler", "next_item.selected",
      key: item.key, number: item.number, processor_type: item.processor_type)

    return item
  end

  Aidp.log_debug("round_robin_scheduler", "next_item.all_paused",
    queue_size: @queue.size)
  nil
end

#refresh_queue(work_items) ⇒ Object

Refresh the queue with current work items from GitHub. Maintains position in rotation when possible.

Parameters:

  • work_items (Array<WorkItem>) —

    Current work items from all processors



33
34
35
36
37
38
39
40
41
42
43
44
45
 
# File 'lib/aidp/watch/round_robin_scheduler.rb', line 33

def refresh_queue(work_items)
  Aidp.log_debug("round_robin_scheduler", "refresh_queue.start",
    incoming_count: work_items.size,
    current_queue_size: @queue.size)

  # Sort by priority (primary) and key (secondary) for deterministic ordering
  # This ensures consistent queue order across refreshes
  @queue = work_items.sort_by { |item| [item.priority, item.key] }

  Aidp.log_debug("round_robin_scheduler", "refresh_queue.complete",
    queue_size: @queue.size,
    priorities: @queue.map(&:priority).uniq.sort)
end

#stats ⇒ Hash

Get queue statistics for debugging/monitoring.

Returns:

  • (Hash) —

    Queue statistics



111
112
113
114
115
116
117
118
119
120
121
 
# File 'lib/aidp/watch/round_robin_scheduler.rb', line 111

def stats
  by_processor = @queue.group_by(&:processor_type)
  by_priority = @queue.group_by(&:priority)

  {
    total: @queue.size,
    by_processor: by_processor.transform_values(&:size),
    by_priority: by_priority.transform_values(&:size),
    last_processed_key: @last_processed_key
  }
end

#work?(paused_numbers: []) ⇒ Boolean

Check if there are any non-paused items in the queue.

Parameters:

  • paused_numbers (Array<Integer>) (defaults to: []) —

    Issue/PR numbers that are paused

Returns:

  • (Boolean) —

    True if there's work to do



104
105
106
 
# File 'lib/aidp/watch/round_robin_scheduler.rb', line 104

def work?(paused_numbers: [])
  @queue.any? { |item| !paused_numbers.include?(item.number) }
end