diff --git a/app/controllers/api/v1/jobs_controller.rb b/app/controllers/api/v1/jobs_controller.rb
index 03ff424d926..38f39d70c61 100644
--- a/app/controllers/api/v1/jobs_controller.rb
+++ b/app/controllers/api/v1/jobs_controller.rb
@@ -17,6 +17,7 @@ class Api::V1::JobsController < Api::ApplicationController
     "nightly_syncs" => NightlySyncsJob,
     "out_of_service_reminder" => OutOfServiceReminderJob,
     "prepare_establish_claim" => PrepareEstablishClaimTasksJob,
+    "push_priority_appeals_to_judges" => PushPriorityAppealsToJudgesJob,
     "reassign_old_tasks" => ReassignOldTasksJob,
     "retrieve_documents_for_reader" => RetrieveDocumentsForReaderJob,
     "set_appeal_age_aod" => SetAppealAgeAodJob,
diff --git a/app/jobs/push_priority_appeals_to_judges_job.rb b/app/jobs/push_priority_appeals_to_judges_job.rb
new file mode 100644
index 00000000000..7f5fbf43e57
--- /dev/null
+++ b/app/jobs/push_priority_appeals_to_judges_job.rb
@@ -0,0 +1,150 @@
+# frozen_string_literal: true
+
+# Job that pushes priority cases to a judge rather than waiting for them to request cases. This will distribute cases
+# to all judges whose teams that have `accepts_priority_pushed_cases` enabled. The first step distributes all priority
+# cases tied to a judge without limit. The second step distributes remaining general population cases (cases not tied to
+# an active judge) while attempting to even out the number of priority cases all judges have received over one month.
+class PushPriorityAppealsToJudgesJob < CaseflowJob
+  queue_with_priority :low_priority
+  application_attr :queue
+
+  include AutomaticCaseDistribution
+
+  def perform
+    @tied_distributions = distribute_non_genpop_priority_appeals
+    @genpop_distributions = distribute_genpop_priority_appeals
+    send_job_report
+  end
+
+  def send_job_report
+    slack_service.send_notification(slack_report.join("\n"), self.class.name, "#appeals-job-alerts")
+    datadog_report_runtime(metric_group_name: "priority_appeal_push_job")
+  end
+
+  def slack_report
+    report = []
+    report << "*Number of cases tied to judges distributed*: " \
+              "#{@tied_distributions.map { |distribution| distribution.statistics['batch_size'] }.sum}"
+    report << "*Number of general population cases distributed*: " \
+              "#{@genpop_distributions.map { |distribution| distribution.statistics['batch_size'] }.sum}"
+
+    appeals_not_distributed = docket_coordinator.dockets.map do |docket_type, docket|
+      report << "*Age of oldest #{docket_type} case*: #{docket.oldest_priority_appeal_days_waiting} days"
+      [docket_type, docket.ready_priority_appeal_ids]
+    end.to_h
+
+    report << "*Number of appeals _not_ distributed*: #{appeals_not_distributed.values.flatten.count}"
+
+    report << ""
+    report << "*Debugging information*"
+    report << "Priority Target: #{priority_target}"
+    report << "Previous monthly distributions: #{priority_distributions_this_month_for_eligible_judges}"
+
+    if appeals_not_distributed.values.flatten.any?
+      add_stuck_appeals_to_report(report, appeals_not_distributed)
+    end
+
+    report
+  end
+
+  def add_stuck_appeals_to_report(report, appeals)
+    report.unshift("[WARN]")
+    report << "Legacy appeals not distributed: `LegacyAppeal.where(vacols_id: #{appeals[:legacy]})`"
+    report << "AMA appeals not distributed: `Appeal.where(uuid: #{appeals.values.drop(1).flatten})`"
+    report << COPY::PRIORITY_PUSH_WARNING_MESSAGE
+  end
+
+  # Distribute all priority cases tied to a judge without limit
+  def distribute_non_genpop_priority_appeals
+    eligible_judges.map do |judge|
+      Distribution.create!(judge: User.find(judge.id), priority_push: true).tap(&:distribute!)
+    end
+  end
+
+  # Distribute remaining general population cases while attempting to even out the number of priority cases all judges
+  # have received over one month
+  def distribute_genpop_priority_appeals
+    eligible_judge_target_distributions_with_leftovers.map do |judge_id, target|
+      Distribution.create!(
+        judge: User.find(judge_id),
+        priority_push: true
+      ).tap { |distribution| distribution.distribute!(target) }
+    end
+  end
+
+  # Give any leftover cases to judges with the lowest distribution targets. Remove judges with 0 cases to be distributed
+  # as these are the final counts to distribute remaining ready priority cases
+  def eligible_judge_target_distributions_with_leftovers
+    leftover_cases = leftover_cases_count
+    target_distributions_for_eligible_judges.sort_by(&:last).map do |judge_id, target|
+      if leftover_cases > 0
+        leftover_cases -= 1
+        target += 1
+      end
+      (target > 0) ? [judge_id, target] : nil
+    end.compact.to_h
+  end
+
+  # Because we cannot distribute fractional cases, there can be cases leftover after taking the priority target
+  # into account. This number will always be less than the number of judges that need distribution because division
+  def leftover_cases_count
+    ready_priority_appeals_count - target_distributions_for_eligible_judges.values.sum
+  end
+
+  # Calculate the number of cases a judge should receive based on the priority target. Don't toss out judges with 0 as
+  # they could receive some of the leftover cases (if any)
+  def target_distributions_for_eligible_judges
+    priority_distributions_this_month_for_eligible_judges.map do |judge_id, distributions_this_month|
+      target = priority_target - distributions_this_month
+      (target >= 0) ? [judge_id, target] : nil
+    end.compact.to_h
+  end
+
+  # Calculates a target that will distribute all ready appeals so the remaining counts for each judge will produce
+  # even case counts over a full month (or as close as we can get to it)
+  def priority_target
+    @priority_target ||= begin
+      distribution_counts = priority_distributions_this_month_for_eligible_judges.values
+      target = (distribution_counts.sum + ready_priority_appeals_count) / distribution_counts.count
+
+      # If there are any judges that have previous distributions that are MORE than the currently calculated priority
+      # target, no target will be large enough to get all other judges up to their number of cases. Remove them from
+      # consideration and recalculate the target for all other judges.
+      while distribution_counts.any? { |distribution_count| distribution_count > target }
+        distribution_counts = distribution_counts.reject { |distribution_count| distribution_count > target }
+        target = (distribution_counts.sum + ready_priority_appeals_count) / distribution_counts.count
+      end
+
+      target
+    end
+  end
+
+  def docket_coordinator
+    @docket_coordinator ||= DocketCoordinator.new
+  end
+
+  def ready_priority_appeals_count
+    @ready_priority_appeals_count ||= docket_coordinator.priority_count
+  end
+
+  # Number of priority distributions every eligible judge has received in the last month
+  def priority_distributions_this_month_for_eligible_judges
+    eligible_judges.map { |judge| [judge.id, priority_distributions_this_month_for_all_judges[judge.id] || 0] }.to_h
+  end
+
+  def eligible_judges
+    @eligible_judges ||= JudgeTeam.pushed_priority_cases_allowed.map(&:judge)
+  end
+
+  # Produces a hash of judge_id and the number of cases distributed to them in the last month
+  def priority_distributions_this_month_for_all_judges
+    @priority_distributions_this_month_for_all_judges ||= priority_distributions_this_month
+      .pluck(:judge_id, :statistics)
+      .group_by(&:first)
+      .map { |judge_id, arr| [judge_id, arr.flat_map(&:last).map { |stats| stats["batch_size"] }.sum] }.to_h
+  end
+
+  def priority_distributions_this_month
+    Distribution.priority_pushed.completed.where(completed_at: 30.days.ago..Time.zone.now)
+  end
+end
diff --git a/app/models/concerns/ama_case_distribution.rb b/app/models/concerns/automatic_case_distribution.rb
similarity index 70%
rename from app/models/concerns/ama_case_distribution.rb
rename to app/models/concerns/automatic_case_distribution.rb
index ca0d05c30aa..2641fa6db63 100644
--- a/app/models/concerns/ama_case_distribution.rb
+++ b/app/models/concerns/automatic_case_distribution.rb
@@ -1,6 +1,6 @@
 # frozen_string_literal: true
 
-module AmaCaseDistribution
+module AutomaticCaseDistribution
   extend ActiveSupport::Concern
 
   delegate :dockets,
@@ -17,7 +17,21 @@ def docket_coordinator
     @docket_coordinator ||= DocketCoordinator.new
   end
 
-  def ama_distribution
+  def priority_push_distribution(limit = nil)
+    @appeals = []
+    @rem = 0
+
+    if limit.nil?
+      # Distribute priority appeals that are tied to judges (not genpop) with no limit.
+      distribute_appeals(:legacy, nil, priority: true, genpop: "not_genpop")
+      distribute_appeals(:hearing, nil, priority: true, genpop: "not_genpop")
+    else
+      # Distribute <limit> number of cases, regardless of docket type, oldest first.
+      distribute_limited_priority_appeals_from_all_dockets(limit)
+    end
+  end
+
+  def requested_distribution
     @appeals = []
     @rem = batch_size
     @remaining_docket_proportions = docket_proportions.clone
@@ -40,9 +54,7 @@ def ama_distribution
 
     # If we haven't yet met the priority target, distribute additional priority appeals.
     priority_rem = (priority_target - @appeals.count(&:priority)).clamp(0, @rem)
-    oldest_priority_appeals_by_docket(priority_rem).each do |docket, n|
-      distribute_appeals(docket, n, priority: true)
-    end
+    distribute_limited_priority_appeals_from_all_dockets(priority_rem)
 
     # As we may have already distributed nonpriority legacy and hearing docket cases, we adjust the docket proportions.
     deduct_distributed_actuals_from_remaining_docket_proportions(:legacy, :hearing)
@@ -51,15 +63,20 @@ def ama_distribution
     # If a docket runs out of available appeals, we reallocate its cases to the other dockets.
     until @rem == 0 || @remaining_docket_proportions.all_zero?
       distribute_appeals_according_to_remaining_docket_proportions
-      @nonpriority_iterations += 1
     end
 
     @appeals
   end
 
+  def distribute_limited_priority_appeals_from_all_dockets(limit)
+    num_oldest_priority_appeals_by_docket(limit).each do |docket, number_of_appeals_to_distribute|
+      distribute_appeals(docket, number_of_appeals_to_distribute, priority: true)
+    end
+  end
+
   def ama_statistics
     {
-      batch_size: batch_size,
+      batch_size: @appeals.count,
       total_batch_size: total_batch_size,
       priority_count: priority_count,
       direct_review_due_count: direct_review_due_count,
@@ -74,14 +91,17 @@ def ama_statistics
     }
   end
 
-  def distribute_appeals(docket, num, priority: false, genpop: "any", range: nil, bust_backlog: false)
-    return [] unless num > 0
+  # Handles the distribution of appeals from any docket while tracking appeals distributed and the remaining number of
+  # appeals to distribute. A nil limit will distribute an infinate number of appeals, only to be used for non_genpop
+  # distributions (distributions tied to a judge)
+  def distribute_appeals(docket, limit = nil, priority: false, genpop: "any", range: nil, bust_backlog: false)
+    return [] unless limit.nil? || limit > 0
 
     if range.nil? && !bust_backlog
-      appeals = dockets[docket].distribute_appeals(self, priority: priority, genpop: genpop, limit: num)
+      appeals = dockets[docket].distribute_appeals(self, priority: priority, genpop: genpop, limit: limit)
     elsif docket == :legacy && priority == false
       appeals = dockets[docket].distribute_nonpriority_appeals(
-        self, genpop: genpop, range: range, limit: num, bust_backlog: bust_backlog
+        self, genpop: genpop, range: range, limit: limit, bust_backlog: bust_backlog
       )
     else
       fail "'range' and 'bust_backlog' are only valid arguments when distributing nonpriority, legacy appeals"
@@ -106,12 +126,13 @@ def deduct_distributed_actuals_from_remaining_docket_proportions(*dockets)
   end
 
   def distribute_appeals_according_to_remaining_docket_proportions
+    @nonpriority_iterations += 1
     @remaining_docket_proportions
       .normalize!
       .stochastic_allocation(@rem)
-      .each do |docket, n|
-        appeals = distribute_appeals(docket, n, priority: false)
-        @remaining_docket_proportions[docket] = 0 if appeals.count < n
+      .each do |docket, number_of_appeals_to_distribute|
+        appeals = distribute_appeals(docket, number_of_appeals_to_distribute, priority: false)
+        @remaining_docket_proportions[docket] = 0 if appeals.count < number_of_appeals_to_distribute
       end
   end
 
@@ -128,13 +149,14 @@ def legacy_docket_range
     (docket_margin_net_of_priority * docket_proportions[:legacy]).round
   end
 
-  def oldest_priority_appeals_by_docket(num)
+  def num_oldest_priority_appeals_by_docket(num)
     return {} unless num > 0
 
     dockets
       .flat_map { |sym, docket| docket.age_of_n_oldest_priority_appeals(num).map { |age| [age, sym] } }
-      .sort_by { |a| a[0] }
+      .sort_by { |age, _| age }
       .first(num)
-      .each_with_object(Hash.new(0)) { |a, counts| counts[a[1]] += 1 }
+      .group_by { |_, sym| sym }
+      .transform_values(&:count)
   end
 end
diff --git a/app/models/distribution.rb b/app/models/distribution.rb
index 7a892cc2df9..465626e4d15 100644
--- a/app/models/distribution.rb
+++ b/app/models/distribution.rb
@@ -2,7 +2,7 @@
 
 class Distribution < CaseflowRecord
   include ActiveModel::Serializers::JSON
-  include AmaCaseDistribution
+  include AutomaticCaseDistribution
 
   has_many :distributed_cases
   belongs_to :judge, class_name: "User"
@@ -20,13 +20,15 @@ class Distribution < CaseflowRecord
   CASES_PER_ATTORNEY = 3
   ALTERNATIVE_BATCH_SIZE = 15
 
+  scope :priority_pushed, -> { where(priority_push: true) }
+
   class << self
     def pending_for_judge(judge)
       where(status: %w[pending started], judge: judge)
     end
   end
 
-  def distribute!
+  def distribute!(limit = nil)
     return unless %w[pending error].include? status
 
     if status == "error"
@@ -41,7 +43,7 @@ def distribute!
     multi_transaction do
       ActiveRecord::Base.connection.execute "SET LOCAL statement_timeout = #{transaction_time_out}"
 
-      ama_distribution
+      priority_push? ? priority_push_distribution(limit) : requested_distribution
 
       update!(status: "completed", completed_at: Time.zone.now, statistics: ama_statistics)
     end
diff --git a/app/models/docket.rb b/app/models/docket.rb
index 824e84f1b26..2eeaa1439a4 100644
--- a/app/models/docket.rb
+++ b/app/models/docket.rb
@@ -38,6 +38,17 @@ def age_of_n_oldest_priority_appeals(num)
     appeals(priority: true, ready: true).limit(num).map(&:ready_for_distribution_at)
   end
 
+  def oldest_priority_appeal_days_waiting
+    oldest_appeal_ready_at = age_of_n_oldest_priority_appeals(1).first
+    return 0 if oldest_appeal_ready_at.nil?
+
+    (Time.zone.now.to_date - oldest_appeal_ready_at.to_date).to_i
+  end
+
+  def ready_priority_appeal_ids
+    appeals(priority: true, ready: true).pluck(:uuid)
+  end
+
   # rubocop:disable Lint/UnusedMethodArgument
   def distribute_appeals(distribution, priority: false, genpop: nil, limit: 1)
     Distribution.transaction do
diff --git a/app/models/dockets/hearing_request_docket.rb b/app/models/dockets/hearing_request_docket.rb
index 2efa1ab1061..f8c6101c83c 100644
--- a/app/models/dockets/hearing_request_docket.rb
+++ b/app/models/dockets/hearing_request_docket.rb
@@ -20,8 +20,18 @@ def distribute_appeals(distribution, priority: false, genpop: "any", limit: 1)
       base_relation: base_relation, genpop: genpop, judge: distribution.judge
     ).call
 
+    appeals = self.class.limit_genpop_appeals(appeals, limit) if genpop.eql? "any"
+
     HearingRequestCaseDistributor.new(
       appeals: appeals, genpop: genpop, distribution: distribution, priority: priority
     ).call
   end
+
+  def self.limit_genpop_appeals(appeals_array, limit)
+    # genpop 'any' returns 2 arrays of the limited base relation. This means if we only request 2 cases, appeals is a
+    # 2x2 array containing 4 cases overall and we will end up distributing 4 cases rather than 2. Instead, reinstate the
+    # limit here by filtering out the newest cases
+    appeals_to_reject = appeals_array.flatten.sort_by(&:ready_for_distribution_at).drop(limit)
+    appeals_array.map { |appeals| appeals - appeals_to_reject }
+  end
 end
diff --git a/app/models/dockets/legacy_docket.rb b/app/models/dockets/legacy_docket.rb
index 53c29e5a19c..38954c1baf7 100644
--- a/app/models/dockets/legacy_docket.rb
+++ b/app/models/dockets/legacy_docket.rb
@@ -26,6 +26,17 @@ def weight
     count(priority: false) + nod_count * NOD_ADJUSTMENT
   end
 
+  def ready_priority_appeal_ids
+    VACOLS::CaseDocket.priority_ready_appeal_vacols_ids
+  end
+
+  def oldest_priority_appeal_days_waiting
+    oldest_appeal_ready_at = age_of_n_oldest_priority_appeals(1).first
+    return 0 if oldest_appeal_ready_at.nil?
+
+    (Time.zone.now.to_date - oldest_appeal_ready_at.to_date).to_i
+  end
+
   def age_of_n_oldest_priority_appeals(num)
     LegacyAppeal.repository.age_of_n_oldest_priority_appeals(num)
   end
diff --git a/app/models/organizations/judge_team.rb b/app/models/organizations/judge_team.rb
index e0ef20e2206..f8c668ddcf6 100644
--- a/app/models/organizations/judge_team.rb
+++ b/app/models/organizations/judge_team.rb
@@ -1,7 +1,7 @@
 # frozen_string_literal: true
 
 class JudgeTeam < Organization
-  scope :pushed_priority_cases_allowed, -> { where(accepts_priority_pushed_cases: true) }
+  scope :pushed_priority_cases_allowed, -> { active.where(accepts_priority_pushed_cases: true) }
 
   class << self
     def for_judge(user)
diff --git a/app/models/vacols/case_docket.rb b/app/models/vacols/case_docket.rb
index 099c9987e24..1ce39c19392 100644
--- a/app/models/vacols/case_docket.rb
+++ b/app/models/vacols/case_docket.rb
@@ -117,8 +117,6 @@ class DocketNumberCentennialLoop < StandardError; end
         #{JOIN_ASSOCIATED_VLJS_BY_HEARINGS}
         #{JOIN_ASSOCIATED_VLJS_BY_PRIOR_DECISIONS}
       )
-      where ((VLJ = ? and 1 = ?) or (VLJ is null and 1 = ?))
-        and rownum <= ?
   "
 
   SELECT_NONPRIORITY_APPEALS = "
@@ -133,9 +131,6 @@ class DocketNumberCentennialLoop < StandardError; end
       ) BRIEFF
       #{JOIN_ASSOCIATED_VLJS_BY_HEARINGS}
     )
-    where ((VLJ = ? and 1 = ?) or (VLJ is null and 1 = ?))
-      and (DOCKET_INDEX <= ? or 1 = ?)
-      and rownum <= ?
   "
 
   # rubocop:disable Metrics/MethodLength
@@ -244,18 +239,7 @@ def self.age_of_n_oldest_priority_appeals(num)
     conn = connection
 
     query = <<-SQL
-      select BFKEY, BFDLOOUT, VLJ
-      from (
-        select BFKEY, BFDLOOUT,
-          case when BFHINES is null or BFHINES <> 'GP' then nvl(VLJ_HEARINGS.VLJ, VLJ_PRIORDEC.VLJ) end VLJ
-        from (
-          #{SELECT_READY_APPEALS}
-            and (BFAC = '7' or AOD = '1')
-          order by BFDLOOUT
-        ) BRIEFF
-        #{JOIN_ASSOCIATED_VLJS_BY_HEARINGS}
-        #{JOIN_ASSOCIATED_VLJS_BY_PRIOR_DECISIONS}
-      )
+      #{SELECT_PRIORITY_APPEALS}
       where VLJ is null and rownum <= ?
     SQL
 
@@ -276,15 +260,7 @@ def self.nonpriority_decisions_per_year
 
   def self.nonpriority_hearing_cases_for_judge_count(judge)
     query = <<-SQL
-      select BFKEY
-      from (
-        select BFKEY, case when BFHINES is null or BFHINES <> 'GP' then VLJ_HEARINGS.VLJ end VLJ
-        from (
-          #{SELECT_READY_APPEALS}
-            and BFAC <> '7' and AOD = '0'
-        ) BRIEFF
-        #{JOIN_ASSOCIATED_VLJS_BY_HEARINGS}
-      )
+      #{SELECT_NONPRIORITY_APPEALS}
       where (VLJ = ?)
     SQL
 
@@ -292,6 +268,10 @@ def self.nonpriority_hearing_cases_for_judge_count(judge)
     connection.exec_query(fmtd_query).count
   end
 
+  def self.priority_ready_appeal_vacols_ids
+    connection.exec_query(SELECT_PRIORITY_APPEALS).to_hash.map { |appeal| appeal["bfkey"] }
+  end
+
   def self.distribute_nonpriority_appeals(judge, genpop, range, limit, bust_backlog, dry_run = false)
     fail(DocketNumberCentennialLoop, COPY::MAX_LEGACY_DOCKET_NUMBER_ERROR_MESSAGE) if Time.zone.now.year >= 2030
 
@@ -308,8 +288,15 @@ def self.distribute_nonpriority_appeals(judge, genpop, range, limit, bust_backlo
       limit = (number_of_hearings_over_limit > 0) ? [number_of_hearings_over_limit, limit].min : 0
     end
 
+    query = <<-SQL
+      #{SELECT_NONPRIORITY_APPEALS}
+      where ((VLJ = ? and 1 = ?) or (VLJ is null and 1 = ?))
+      and (DOCKET_INDEX <= ? or 1 = ?)
+      and rownum <= ?
+    SQL
+
     fmtd_query = sanitize_sql_array([
-                                      SELECT_NONPRIORITY_APPEALS,
+                                      query,
                                       judge.vacols_attorney_id,
                                       (genpop == "any" || genpop == "not_genpop") ? 1 : 0,
                                       (genpop == "any" || genpop == "only_genpop") ? 1 : 0,
@@ -322,12 +309,19 @@ def self.distribute_nonpriority_appeals(judge, genpop, range, limit, bust_backlo
   end
 
   def self.distribute_priority_appeals(judge, genpop, limit, dry_run = false)
+    query = <<-SQL
+      #{SELECT_PRIORITY_APPEALS}
+      where ((VLJ = ? and 1 = ?) or (VLJ is null and 1 = ?))
+      and (rownum <= ? or 1 = ?)
+    SQL
+
     fmtd_query = sanitize_sql_array([
-                                      SELECT_PRIORITY_APPEALS,
+                                      query,
                                       judge.vacols_attorney_id,
                                       (genpop == "any" || genpop == "not_genpop") ? 1 : 0,
                                       (genpop == "any" || genpop == "only_genpop") ? 1 : 0,
-                                      limit
+                                      limit,
+                                      limit.nil? ? 1 : 0
                                     ])
 
     distribute_appeals(fmtd_query, judge, dry_run)
diff --git a/client/COPY.json b/client/COPY.json
index 123aac04031..6ce73de91ce 100644
--- a/client/COPY.json
+++ b/client/COPY.json
@@ -822,5 +822,7 @@
   "VLJ_VIRTUAL_HEARINGS_LINK_TEXT": "Start Virtual Hearing",
   "GUEST_VIRTUAL_HEARINGS_LINK_TEXT": "Join Virtual Hearing",
 
-  "BULK_REASSIGN_INSTRUCTIONS": "This task has been %s due to the reassignment of all tasks previously assigned to %s."
+  "BULK_REASSIGN_INSTRUCTIONS": "This task has been %s due to the reassignment of all tasks previously assigned to %s.",
+
+  "PRIORITY_PUSH_WARNING_MESSAGE": "Some cases are ready to distribute but could not be. This could be due to the tied judge being an acting judge or an attorney (`!user.judge_in_vacols?`). These will be handled manually by the board. The tied judge could also have priority distribution turned off and should not be pushed cases (`JudgeTeam.for_judge(user).accepts_priority_pushed_cases?`). There could also be a duplicate distributed case blocking distribution (`DistributedCase.find_by(case_id: case_id`) or our validations for redistributing legacy appeals failed (https://github.com/department-of-veterans-affairs/caseflow/blob/master/app/services/redistributed_case.rb#L26)"
 }
diff --git a/db/seeds/priority_distributions.rb b/db/seeds/priority_distributions.rb
new file mode 100644
index 00000000000..0704ee0157b
--- /dev/null
+++ b/db/seeds/priority_distributions.rb
@@ -0,0 +1,391 @@
+# frozen_string_literal: true
+
+# Create distribution seeds to test priority push case distribution job.
+# Mocks previously distributed cases for judges. Sets up cases ready to be distributed.
+# Invoke with:
+# RequestStore[:current_user] = User.system_user
+# Dir[Rails.root.join("db/seeds/*.rb")].sort.each { |f| require f }
+# Seeds::PriorityDistributions.new.seed!
+
+module Seeds
+  class PriorityDistributions < Base
+    # :nocov:
+    def seed!
+      organize_judges
+      create_previous_distribtions
+      create_cases_tied_to_judges
+      create_genpop_cases
+      create_errorable_cases
+    end
+
+    private
+
+    def organize_judges
+      JudgeTeam.unscoped.find_by(name: "BVAACTING").inactive!
+      JudgeTeam.find_by(name: "BVAAWAKEFIELD").update!(accepts_priority_pushed_cases: false)
+    end
+
+    def create_previous_distribtions
+      judges_with_previous_distributions.each do |judge|
+        2.times { create_priority_distribution_this_month(judge) }
+        create_priority_distribution_last_month(judge)
+        create_nonpriority_distribution_this_month(judge)
+      end
+    end
+
+    def create_cases_tied_to_judges
+      judges_with_tied_cases.each do |judge|
+        create_legacy_cases_tied_to_judge(judge)
+        create_hearing_cases_tied_to_judge(judge)
+      end
+    end
+
+    def create_genpop_cases
+      create_legacy_genpop_cases
+      create_ama_hearing_genpop_cases
+      create_direct_review_genpop_cases
+      create_evidence_submission_genpop_cases
+    end
+
+    def create_errorable_cases
+      create_legacy_appeal_with_previous_distribution
+    end
+
+    def create_legacy_cases_tied_to_judge(judge)
+      create_legacy_ready_priority_cases_tied_to_judge(judge)
+      create_legacy_nonready_priority_cases_tied_to_judge(judge)
+      create_legacy_ready_nonpriority_cases_tied_to_judge(judge)
+    end
+
+    def create_hearing_cases_tied_to_judge(judge)
+      create_hearing_ready_priority_cases_tied_to_judge(judge)
+      create_hearing_nonready_priority_cases_tied_to_judge(judge)
+      create_hearing_ready_nonpriority_cases_tied_to_judge(judge)
+    end
+
+    def create_legacy_genpop_cases
+      create_legacy_ready_priority_genpop_cases
+      create_legacy_ready_nonpriority_genpop_cases
+      create_legacy_nonready_priority_genpop_cases
+    end
+
+    def create_ama_hearing_genpop_cases
+      create_ama_hearing_ready_priority_genpop_cases
+      create_ama_hearing_ready_nonpriority_genpop_cases
+      create_ama_hearing_nonready_priority_genpop_cases
+    end
+
+    def create_direct_review_genpop_cases
+      create_direct_review_ready_priority_genpop_cases
+      create_direct_review_ready_nonpriority_genpop_cases
+      create_direct_review_nonready_priority_genpop_cases
+    end
+
+    def create_evidence_submission_genpop_cases
+      create_evidence_submission_ready_priority_genpop_cases
+      create_evidence_submission_ready_nonpriority_genpop_cases
+      create_evidence_submission_nonready_priority_genpop_cases
+    end
+
+    def create_priority_distribution_this_month(judge)
+      create_unvalidated_completed_distribution(
+        traits: [:priority, :this_month],
+        judge: judge,
+        statistics: { "batch_size" => rand(10) }
+      )
+    end
+
+    def create_priority_distribution_last_month(judge)
+      create_unvalidated_completed_distribution(
+        traits: [:priority, :last_month],
+        judge: judge,
+        statistics: { "batch_size" => rand(10) }
+      )
+    end
+
+    def create_nonpriority_distribution_this_month(judge)
+      create_unvalidated_completed_distribution(
+        traits: [:this_month],
+        judge: judge,
+        statistics: { "batch_size" => rand(10) }
+      )
+    end
+
+    def create_legacy_ready_priority_cases_tied_to_judge(judge)
+      rand(5).times do
+        create(
+          :case,
+          :aod,
+          :ready_for_distribution,
+          :tied_to_judge,
+          tied_judge: judge,
+          bfkey: random_key,
+          correspondent: create(:correspondent, stafkey: random_key)
+        )
+      end
+    end
+
+    def create_legacy_nonready_priority_cases_tied_to_judge(judge)
+      rand(5).times do
+        create(
+          :case,
+          :aod,
+          :tied_to_judge,
+          tied_judge: judge,
+          bfkey: random_key,
+          correspondent: create(:correspondent, stafkey: random_key)
+        )
+      end
+    end
+
+    def create_legacy_ready_nonpriority_cases_tied_to_judge(judge)
+      rand(5).times do
+        create(
+          :case,
+          :ready_for_distribution,
+          :tied_to_judge,
+          tied_judge: judge,
+          bfkey: random_key,
+          correspondent: create(:correspondent, stafkey: random_key)
+        )
+      end
+    end
+
+    def create_hearing_ready_priority_cases_tied_to_judge(judge)
+      rand(5).times do
+        create(
+          :appeal,
+          :hearing_docket,
+          :ready_for_distribution,
+          :advanced_on_docket_due_to_age,
+          :held_hearing,
+          :tied_to_judge,
+          tied_judge: judge,
+          adding_user: User.first
+        )
+      end
+    end
+
+    def create_hearing_nonready_priority_cases_tied_to_judge(judge)
+      rand(5).times do
+        create(
+          :appeal,
+          :hearing_docket,
+          :with_post_intake_tasks,
+          :advanced_on_docket_due_to_age,
+          :held_hearing,
+          :tied_to_judge,
+          tied_judge: judge,
+          adding_user: User.first
+        )
+      end
+    end
+
+    def create_hearing_ready_nonpriority_cases_tied_to_judge(judge)
+      rand(5).times do
+        create(
+          :appeal,
+          :hearing_docket,
+          :ready_for_distribution,
+          :held_hearing,
+          :tied_to_judge,
+          tied_judge: judge,
+          adding_user: User.first
+        )
+      end
+    end
+
+    def create_legacy_ready_priority_genpop_cases
+      rand(50).times do
+        create(
+          :case,
+          :aod,
+          :ready_for_distribution,
+          bfkey: random_key,
+          correspondent: create(:correspondent, stafkey: random_key)
+        )
+      end
+    end
+
+    def create_legacy_nonready_priority_genpop_cases
+      rand(5).times do
+        create(
+          :case,
+          :aod,
+          bfkey: random_key,
+          correspondent: create(:correspondent, stafkey: random_key)
+        )
+      end
+    end
+
+    def create_legacy_ready_nonpriority_genpop_cases
+      rand(5).times do
+        create(
+          :case,
+          :ready_for_distribution,
+          bfkey: random_key,
+          correspondent: create(:correspondent, stafkey: random_key)
+        )
+      end
+    end
+
+    def create_ama_hearing_ready_priority_genpop_cases
+      rand(50).times do
+        create(
+          :appeal,
+          :hearing_docket,
+          :ready_for_distribution,
+          :advanced_on_docket_due_to_age,
+          :held_hearing,
+          adding_user: User.first
+        )
+      end
+    end
+
+    def create_ama_hearing_nonready_priority_genpop_cases
+      rand(5).times do
+        create(
+          :appeal,
+          :hearing_docket,
+          :with_post_intake_tasks,
+          :advanced_on_docket_due_to_age,
+          :held_hearing,
+          adding_user: User.first
+        )
+      end
+    end
+
+    def create_ama_hearing_ready_nonpriority_genpop_cases
+      rand(5).times do
+        create(
+          :appeal,
+          :hearing_docket,
+          :ready_for_distribution,
+          :held_hearing,
+          adding_user: User.first
+        )
+      end
+    end
+
+    def create_direct_review_ready_priority_genpop_cases
+      rand(50).times do
+        create(
+          :appeal,
+          :direct_review_docket,
+          :ready_for_distribution,
+          :advanced_on_docket_due_to_age
+        )
+      end
+    end
+
+    def create_direct_review_nonready_priority_genpop_cases
+      rand(5).times do
+        create(
+          :appeal,
+          :direct_review_docket,
+          :with_post_intake_tasks,
+          :advanced_on_docket_due_to_age
+        )
+      end
+    end
+
+    def create_direct_review_ready_nonpriority_genpop_cases
+      rand(5).times do
+        create(
+          :appeal,
+          :direct_review_docket,
+          :ready_for_distribution
+        )
+      end
+    end
+
+    def create_evidence_submission_ready_priority_genpop_cases
+      rand(10).times do
+        FactoryBot.create(
+          :appeal,
+          :evidence_submission_docket,
+          :ready_for_distribution,
+          :advanced_on_docket_due_to_age
+        )
+      end
+    end
+
+    def create_evidence_submission_nonready_priority_genpop_cases
+      rand(50).times do
+        create(
+          :appeal,
+          :evidence_submission_docket,
+          :with_post_intake_tasks,
+          :advanced_on_docket_due_to_age
+        )
+      end
+    end
+
+    def create_evidence_submission_ready_nonpriority_genpop_cases
+      rand(50).times do
+        create(
+          :appeal,
+          :evidence_submission_docket,
+          :ready_for_distribution
+        )
+      end
+    end
+
+    def create_legacy_appeal_with_previous_distribution
+      vacols_case = create(
+        :case,
+        :aod,
+        :ready_for_distribution,
+        bfkey: random_key,
+        correspondent: create(:correspondent, stafkey: random_key)
+      )
+      create(:legacy_appeal, :with_schedule_hearing_tasks, vacols_case: vacols_case)
+
+      create_distribution_for_case_id(vacols_case.bfkey)
+    end
+
+    def create_distribution_for_case_id(case_id)
+      create_unvalidated_completed_distribution(
+        traits: [:priority, :completed, :this_month],
+        judge: User.third,
+        statistics: { "batch_size" => rand(10) }
+      ).distributed_cases.create(
+        case_id: case_id,
+        priority: case_id,
+        docket: "legacy",
+        ready_at: Time.zone.now,
+        docket_index: "123",
+        genpop: false,
+        genpop_query: "any"
+      )
+    end
+
+    def create_unvalidated_completed_distribution(attrs = {})
+      distribution = FactoryBot.build(:distribution, *attrs.delete(:traits), attrs)
+      distribution.save!(validate: false)
+      distribution.completed!
+      distribution
+    end
+
+    def judges_with_tied_cases
+      @judges_with_tied_cases ||= begin
+        judges_with_judge_teams.unshift(User.find_by(full_name: "Steve Attorney_Cases Casper"))
+      end
+    end
+
+    def judges_with_previous_distributions
+      @judges_with_previous_distributions ||= begin
+        judges_with_judge_teams.unshift(User.find_by(full_name: "Keith Judge_CaseToAssign_NoTeam Keeling"))
+      end
+    end
+
+    def judges_with_judge_teams
+      JudgeTeam.unscoped.map(&:judge)
+    end
+
+    def random_key
+      rand.to_s[2..11]
+    end
+    # :nocov:
+  end
+end
diff --git a/db/seeds/tasks.rb b/db/seeds/tasks.rb
index daf388aa747..253c162a7fe 100644
--- a/db/seeds/tasks.rb
+++ b/db/seeds/tasks.rb
@@ -208,8 +208,7 @@ def create_task_at_bva_dispatch(seed = Faker::Number.number(digits: 3))
       )
 
       root_task = appeal.root_task
-      judge = create(:user, station_id: 101, full_name: "Apurva Judge_CaseAtDispatch Wakefield")
-      create(:staff, :judge_role, user: judge)
+      judge = User.find_by(css_id: "BVAAWAKEFIELD")
       judge_task = create(
         :ama_judge_decision_review_task,
         assigned_to: judge,
@@ -217,8 +216,7 @@ def create_task_at_bva_dispatch(seed = Faker::Number.number(digits: 3))
         parent: root_task
       )
 
-      atty = create(:user, station_id: 101, full_name: "Andy Attorney_CaseAtDispatch Belanger")
-      create(:staff, :attorney_role, user: atty)
+      atty = User.find_by(css_id: "BVAABELANGER")
       atty_task = create(
         :ama_attorney_task,
         :in_progress,
@@ -228,9 +226,6 @@ def create_task_at_bva_dispatch(seed = Faker::Number.number(digits: 3))
         appeal: appeal
       )
 
-      judge_team = JudgeTeam.create_for_judge(judge)
-      judge_team.add_user(atty)
-
       appeal.request_issues.each do |request_issue|
         create(
           :decision_issue,
diff --git a/db/seeds/users.rb b/db/seeds/users.rb
index 69a631fbbf1..f100b45c1f4 100644
--- a/db/seeds/users.rb
+++ b/db/seeds/users.rb
@@ -11,7 +11,8 @@ class Users < Base
       "BVAGSPORER" => { attorneys: %w[BVAOTRANTOW BVAGBOTSFORD BVAJWEHNER1] },
       "BVAEBECKER" => { attorneys: %w[BVAKBLOCK BVACMERTZ BVAHLUETTGEN] },
       "BVARERDMAN" => { attorneys: %w[BVASRITCHIE BVAJSCHIMMEL BVAKROHAN1] },
-      "BVAOSCHOWALT" => { attorneys: %w[BVASCASPER1 BVAOWEHNER BVASFUNK1] }
+      "BVAOSCHOWALT" => { attorneys: %w[BVASCASPER1 BVAOWEHNER BVASFUNK1] },
+      "BVAAWAKEFIELD" => { attorneys: %w[BVAABELANGER] }
     }.freeze
 
     DEVELOPMENT_DVC_TEAMS = {
@@ -31,6 +32,8 @@ def create_users
       User.create(css_id: "BVARERDMAN", station_id: 101, full_name: "Rachael JudgeHasAttorneys_Cases Erdman")
       User.create(css_id: "BVAEBECKER", station_id: 101, full_name: "Elizabeth Judge_CaseToAssign Becker")
       User.create(css_id: "BVAKKEELING", station_id: 101, full_name: "Keith Judge_CaseToAssign_NoTeam Keeling")
+      User.create(css_id: "BVAAWAKEFIELD", station_id: 101, full_name: "Apurva Judge_CaseAtDispatch Wakefield")
+      User.create(css_id: "BVAABELANGER", station_id: 101, full_name: "Andy Attorney_CaseAtDispatch Belanger")
       User.create(css_id: "BVATWARNER", station_id: 101, full_name: "Theresa BuildHearingSchedule Warner")
       User.create(css_id: "BVAGWHITE", station_id: 101, full_name: "George BVADispatchUser_Cases White")
       User.create(css_id: "BVAGGREY", station_id: 101, full_name: "Gina BVADispatchUser_NoCases Grey")
diff --git a/spec/factories/appeal.rb b/spec/factories/appeal.rb
index 911dd6ebaf9..a666057a955 100644
--- a/spec/factories/appeal.rb
+++ b/spec/factories/appeal.rb
@@ -104,6 +104,43 @@
       docket_type { Constants.AMA_DOCKETS.hearing }
     end
 
+    trait :evidence_submission_docket do
+      docket_type { Constants.AMA_DOCKETS.evidence_submission }
+    end
+
+    trait :direct_review_docket do
+      docket_type { Constants.AMA_DOCKETS.direct_review }
+    end
+
+    trait :held_hearing do
+      transient do
+        adding_user { nil }
+      end
+
+      after(:create) do |appeal, evaluator|
+        create(:hearing, judge: nil, disposition: "held", appeal: appeal, adding_user: evaluator.adding_user)
+      end
+    end
+
+    trait :tied_to_judge do
+      transient do
+        tied_judge { nil }
+      end
+
+      after(:create) do |appeal, evaluator|
+        hearing_day = create(
+          :hearing_day,
+          scheduled_for: 1.day.ago,
+          created_by: evaluator.tied_judge,
+          updated_by: evaluator.tied_judge
+        )
+        Hearing.find_by(disposition: Constants.HEARING_DISPOSITION_TYPES.held, appeal: appeal).update!(
+          judge: evaluator.tied_judge,
+          hearing_day: hearing_day
+        )
+      end
+    end
+
     trait :outcoded do
       after(:create) do |appeal, _evaluator|
         appeal.create_tasks_on_intake_success!
@@ -180,20 +217,16 @@
 
     ## Appeal with a realistic task tree
     ## The appeal is ready for distribution by the ACD
-    ## Leaves incorrectly open & incomplete Hearing / Evidence Window task branches
-    ## for those dockets
     trait :ready_for_distribution do
       with_post_intake_tasks
       after(:create) do |appeal, _evaluator|
         distribution_tasks = appeal.tasks.select { |task| task.is_a?(DistributionTask) }
-        distribution_tasks.each(&:ready_for_distribution!)
+        (distribution_tasks.flat_map(&:descendants) - distribution_tasks).each(&:completed!)
       end
     end
 
     ## Appeal with a realistic task tree
     ## The appeal would be ready for distribution by the ACD except there is a blocking mail task
-    ## Leaves incorrectly open & incomplete Hearing / Evidence Window task branches
-    ## for those dockets
     trait :mail_blocking_distribution do
       ready_for_distribution
       after(:create) do |appeal, _evaluator|
@@ -208,8 +241,7 @@
 
     ## Appeal with a realistic task tree
     ## The appeal is assigned to a Judge for a decision
-    ## Leaves incorrectly open & incomplete Hearing / Evidence Window task branches
-    ## for those dockets. Strongly suggest you provide a judge.
+    ## Strongly suggest you provide a judge.
     trait :assigned_to_judge do
       ready_for_distribution
       after(:create) do |appeal, evaluator|
@@ -223,8 +255,7 @@
 
     ## Appeal with a realistic task tree
     ## The appeal is assigned to an Attorney for decision drafting
-    ## Leaves incorrectly open & incomplete Hearing / Evidence Window task branches
-    ## for those dockets. Strongly suggest you provide a judge and attorney.
+    ## Strongly suggest you provide a judge and attorney.
     trait :at_attorney_drafting do
       assigned_to_judge
       after(:create) do |appeal, evaluator|
@@ -240,8 +271,7 @@
 
     ## Appeal with a realistic task tree
     ## The appeal is assigned to a judge at decision review
-    ## Leaves incorrectly open & incomplete Hearing / Evidence Window task branches
-    ## for those dockets. Strongly suggest you provide a judge and attorney.
+    ## Strongly suggest you provide a judge and attorney.
     trait :at_judge_review do
       at_attorney_drafting
       after(:create) do |appeal|
diff --git a/spec/factories/distribution.rb b/spec/factories/distribution.rb
index bce6643a5a0..04d38273fb0 100644
--- a/spec/factories/distribution.rb
+++ b/spec/factories/distribution.rb
@@ -3,5 +3,23 @@
 FactoryBot.define do
   factory :distribution do
     association :judge, factory: :user
+
+    trait :completed do
+      after(:create) do |distribution|
+        distribution.update(status: :completed)
+      end
+    end
+
+    trait :priority do
+      priority_push { true }
+    end
+
+    trait :this_month do
+      completed_at { 5.days.ago }
+    end
+
+    trait :last_month do
+      completed_at { 35.days.ago }
+    end
   end
 end
diff --git a/spec/factories/hearing.rb b/spec/factories/hearing.rb
index 2fe770848c4..f873dff808f 100644
--- a/spec/factories/hearing.rb
+++ b/spec/factories/hearing.rb
@@ -5,6 +5,7 @@
     transient do
       regional_office { nil }
       judge { create(:user, roles: ["Hearing Prep"]) }
+      adding_user { create(:user) }
     end
     appeal { create(:appeal, :hearing_docket) }
     uuid { SecureRandom.uuid }
@@ -13,7 +14,9 @@
              regional_office: regional_office,
              scheduled_for: Time.zone.today,
              judge: judge,
-             request_type: regional_office.nil? ? "C" : "V")
+             request_type: regional_office.nil? ? "C" : "V",
+             created_by: adding_user,
+             updated_by: adding_user)
     end
     hearing_location do
       if regional_office.present?
@@ -21,8 +24,8 @@
       end
     end
     scheduled_time { "8:30AM" }
-    created_by { create(:user) }
-    updated_by { create(:user) }
+    created_by { adding_user }
+    updated_by { adding_user }
     virtual_hearing { nil }
 
     trait :held do
diff --git a/spec/factories/vacols/case.rb b/spec/factories/vacols/case.rb
index bfd87f21a01..ff00be0f3f0 100644
--- a/spec/factories/vacols/case.rb
+++ b/spec/factories/vacols/case.rb
@@ -7,9 +7,10 @@
     sequence(:bfcorlid, 300_000_000) { |n| "#{n}S" }
 
     association :correspondent, factory: :correspondent
-    association :folder, factory: :folder, ticknum: :bfkey
+    folder { association :folder, ticknum: bfkey, tinum: bfkey }
 
     bfregoff { "RO18" }
+    bfdloout { Time.zone.now }
 
     trait :assigned do
       transient do
@@ -172,6 +173,23 @@
       end
     end
 
+    trait :tied_to_judge do
+      transient do
+        tied_judge { nil }
+      end
+
+      after(:create) do |vacols_case, evaluator|
+        VACOLS::Folder.find_by(tinum: vacols_case.bfkey).update!(titrnum: "123456789S")
+        create(
+          :case_hearing,
+          :disposition_held,
+          folder_nr: vacols_case.bfkey,
+          hearing_date: 5.days.ago.to_date,
+          user: evaluator.tied_judge
+        )
+      end
+    end
+
     trait :type_original do
       bfac { "1" }
     end
@@ -201,6 +219,12 @@
       bfmpro { "ACT" }
     end
 
+    trait :ready_for_distribution do
+      status_active
+      bfcurloc { "81" }
+      bfd19 { 1.year.ago.to_date }
+    end
+
     trait :status_remand do
       bfmpro { "REM" }
       bfdc { "3" }
@@ -293,7 +317,7 @@
 
     trait :docs_in_vbms do
       after(:build) do |vacols_case, _evaluator|
-        vacols_case.folder.tivbms = %w[Y 1 0].sample
+        vacols_case.folder.update!(tivbms: %w[Y 1 0].sample)
       end
     end
 
diff --git a/spec/factories/vacols/case_hearing.rb b/spec/factories/vacols/case_hearing.rb
index b813d3286db..25d75e8e547 100644
--- a/spec/factories/vacols/case_hearing.rb
+++ b/spec/factories/vacols/case_hearing.rb
@@ -42,11 +42,17 @@
       # Build Caseflow hearing day and associate with legacy hearing.
       if hearing.vdkey.nil?
         regional_office = (hearing.hearing_type == HearingDay::REQUEST_TYPES[:video]) ? "RO13" : nil
-        hearing_day = create(:hearing_day,
-                             scheduled_for: hearing.hearing_date,
-                             request_type: hearing.hearing_type,
-                             regional_office: regional_office)
+        attrs = {
+          scheduled_for: hearing.hearing_date,
+          request_type: hearing.hearing_type,
+          regional_office: regional_office
+        }
+        if evaluator.user
+          attrs[:created_by] = evaluator.user
+          attrs[:updated_by] = evaluator.user
+        end
 
+        hearing_day = create(:hearing_day, attrs)
         hearing.vdkey = hearing_day.id
       end
 
diff --git a/spec/jobs/push_priority_appeals_to_judges_job_spec.rb b/spec/jobs/push_priority_appeals_to_judges_job_spec.rb
new file mode 100644
index 00000000000..77a3d5fe574
--- /dev/null
+++ b/spec/jobs/push_priority_appeals_to_judges_job_spec.rb
@@ -0,0 +1,922 @@
+# frozen_string_literal: true
+
+describe PushPriorityAppealsToJudgesJob, :all_dbs do
+  def to_judge_hash(arr)
+    arr.each_with_index.map { |count, i| [i, count] }.to_h
+  end
+
+  context ".distribute_non_genpop_priority_appeals" do
+    before do
+      allow_any_instance_of(DirectReviewDocket)
+        .to receive(:nonpriority_receipts_per_year)
+        .and_return(100)
+
+      allow(Docket)
+        .to receive(:nonpriority_decisions_per_year)
+        .and_return(1000)
+      allow_any_instance_of(PushPriorityAppealsToJudgesJob).to receive(:eligible_judges).and_return(eligible_judges)
+    end
+
+    let(:ready_priority_bfkey) { "12345" }
+    let(:ready_priority_uuid) { "bece6907-3b6f-4c49-a580-6d5f2e1ca65c" }
+    let!(:judge_with_ready_priority_cases) do
+      create(:user).tap do |judge|
+        create(:staff, :judge_role, user: judge)
+        vacols_case = create(
+          :case,
+          :aod,
+          bfkey: ready_priority_bfkey,
+          bfd19: 1.year.ago,
+          bfac: "3",
+          bfmpro: "ACT",
+          bfcurloc: "81",
+          bfdloout: 3.days.ago,
+          bfbox: nil,
+          folder: build(:folder, tinum: "1801003", titrnum: "123456789S")
+        )
+        create(
+          :case_hearing,
+          :disposition_held,
+          folder_nr: vacols_case.bfkey,
+          hearing_date: 5.days.ago.to_date,
+          board_member: judge.vacols_attorney_id
+        )
+
+        appeal = create(
+          :appeal,
+          :ready_for_distribution,
+          :advanced_on_docket_due_to_age,
+          uuid: ready_priority_uuid,
+          docket_type: Constants.AMA_DOCKETS.hearing
+        )
+        most_recent = create(:hearing_day, scheduled_for: 1.day.ago)
+        hearing = create(:hearing, judge: nil, disposition: "held", appeal: appeal, hearing_day: most_recent)
+        hearing.update!(judge: judge)
+      end
+    end
+
+    let!(:judge_with_ready_nonpriority_cases) do
+      create(:user).tap do |judge|
+        create(:staff, :judge_role, user: judge)
+        vacols_case = create(
+          :case,
+          bfd19: 1.year.ago,
+          bfac: "3",
+          bfmpro: "ACT",
+          bfcurloc: "81",
+          bfdloout: 3.days.ago,
+          bfbox: nil,
+          folder: build(:folder, tinum: "1801002", titrnum: "123456782S")
+        )
+        create(
+          :case_hearing,
+          :disposition_held,
+          folder_nr: vacols_case.bfkey,
+          hearing_date: 5.days.ago.to_date,
+          board_member: judge.vacols_attorney_id
+        )
+
+        appeal = create(
+          :appeal,
+          :ready_for_distribution,
+          docket_type: Constants.AMA_DOCKETS.hearing
+        )
+        most_recent = create(:hearing_day, scheduled_for: 1.day.ago)
+        hearing = create(:hearing, judge: nil, disposition: "held", appeal: appeal, hearing_day: most_recent)
+        hearing.update!(judge: judge)
+      end
+    end
+
+    let!(:judge_with_nonready_priority_cases) do
+      create(:user).tap do |judge|
+        create(:staff, :judge_role, user: judge)
+        vacols_case = create(
+          :case,
+          :aod,
+          bfd19: 1.year.ago,
+          bfac: "3",
+          bfmpro: "ACT",
+          bfcurloc: "not ready",
+          bfdloout: 3.days.ago,
+          bfbox: nil,
+          folder: build(:folder, tinum: "1801003", titrnum: "123456783S")
+        )
+        create(
+          :case_hearing,
+          :disposition_held,
+          folder_nr: vacols_case.bfkey,
+          hearing_date: 5.days.ago.to_date,
+          board_member: judge.vacols_attorney_id
+        )
+
+        appeal = create(
+          :appeal,
+          :advanced_on_docket_due_to_age,
+          docket_type: Constants.AMA_DOCKETS.hearing
+        )
+        most_recent = create(:hearing_day, scheduled_for: 1.day.ago)
+        hearing = create(:hearing, judge: nil, disposition: "held", appeal: appeal, hearing_day: most_recent)
+        hearing.update!(judge: judge)
+      end
+    end
+    let(:eligible_judges) do
+      [
+        judge_with_ready_priority_cases,
+        judge_with_ready_nonpriority_cases,
+        judge_with_nonready_priority_cases
+      ]
+    end
+
+    subject { PushPriorityAppealsToJudgesJob.new.distribute_non_genpop_priority_appeals }
+
+    it "should only distribute the ready priority cases tied to a judge" do
+      expect(subject.count).to eq eligible_judges.count
+      expect(subject.map { |dist| dist.statistics["batch_size"] }).to match_array [2, 0, 0]
+
+      # Ensure we only distributed the 2 ready legacy and hearing priority cases that are tied to a judge
+      distributed_cases = DistributedCase.where(distribution: subject)
+      expect(distributed_cases.count).to eq 2
+      expect(distributed_cases.map(&:case_id)).to match_array [ready_priority_bfkey, ready_priority_uuid]
+      expect(distributed_cases.map(&:docket)).to match_array ["legacy", Constants.AMA_DOCKETS.hearing]
+      expect(distributed_cases.map(&:priority).uniq).to match_array [true]
+      expect(distributed_cases.map(&:genpop).uniq).to match_array [false]
+    end
+  end
+
+  context ".distribute_genpop_priority_appeals" do
+    before do
+      allow_any_instance_of(DirectReviewDocket)
+        .to receive(:nonpriority_receipts_per_year)
+        .and_return(100)
+
+      allow(Docket)
+        .to receive(:nonpriority_decisions_per_year)
+        .and_return(1000)
+
+      allow_any_instance_of(PushPriorityAppealsToJudgesJob)
+        .to receive(:priority_distributions_this_month_for_eligible_judges).and_return(
+          judges.each_with_index.map { |judge, i| [judge.id, judge_distributions_this_month[i]] }.to_h
+        )
+    end
+
+    subject { PushPriorityAppealsToJudgesJob.new.distribute_genpop_priority_appeals }
+
+    let(:judges) { create_list(:user, 5, :with_vacols_judge_record) }
+    let(:judge_distributions_this_month) { (0..4).to_a }
+    let!(:legacy_priority_cases) do
+      (1..5).map do |i|
+        vacols_case = create(
+          :case,
+          :aod,
+          bfd19: 1.year.ago,
+          bfac: "1",
+          bfmpro: "ACT",
+          bfcurloc: "81",
+          bfdloout: i.months.ago,
+          folder: build(
+            :folder,
+            tinum: "1801#{format('%<index>03d', index: i)}",
+            titrnum: "123456789S"
+          )
+        )
+        create(
+          :case_hearing,
+          :disposition_held,
+          folder_nr: vacols_case.bfkey,
+          hearing_date: 5.days.ago.to_date
+        )
+        vacols_case
+      end
+    end
+    let!(:ready_priority_hearing_cases) do
+      (1..5).map do |i|
+        appeal = create(:appeal,
+                        :advanced_on_docket_due_to_age,
+                        :ready_for_distribution,
+                        docket_type: Constants.AMA_DOCKETS.hearing)
+        appeal.tasks.find_by(type: DistributionTask.name).update(assigned_at: i.months.ago)
+        appeal.reload
+      end
+    end
+    let!(:ready_priority_evidence_cases) do
+      (1..5).map do |i|
+        appeal = create(:appeal,
+                        :with_post_intake_tasks,
+                        :advanced_on_docket_due_to_age,
+                        docket_type: Constants.AMA_DOCKETS.evidence_submission)
+        appeal.tasks.find_by(type: EvidenceSubmissionWindowTask.name).completed!
+        appeal.tasks.find_by(type: DistributionTask.name).update(assigned_at: i.month.ago)
+        appeal
+      end
+    end
+    let!(:ready_priority_direct_cases) do
+      (1..5).map do |i|
+        appeal = create(:appeal,
+                        :with_post_intake_tasks,
+                        :advanced_on_docket_due_to_age,
+                        docket_type: Constants.AMA_DOCKETS.direct_review,
+                        receipt_date: 1.month.ago)
+        appeal.tasks.find_by(type: DistributionTask.name).update(assigned_at: i.month.ago)
+        appeal
+      end
+    end
+
+    let(:priority_count) { Appeal.count + VACOLS::Case.count }
+    let(:priority_target) { (priority_count + judge_distributions_this_month.sum) / judges.count }
+
+    it "should distibute ready priortiy appeals to the judges" do
+      expect(subject.count).to eq judges.count
+
+      # Ensure we distributed all available ready cases from any docket that are not tied to a judge
+      distributed_cases = DistributedCase.where(distribution: subject)
+      expect(distributed_cases.count).to eq priority_count
+      expect(distributed_cases.map(&:priority).uniq.compact).to match_array [true]
+      expect(distributed_cases.map(&:genpop).uniq.compact).to match_array [true]
+      expect(distributed_cases.pluck(:docket).uniq).to match_array(Constants::AMA_DOCKETS.keys.unshift("legacy"))
+      expect(distributed_cases.group(:docket).count.values.uniq).to match_array [5]
+    end
+
+    it "distributes cases to each judge based on their priority target" do
+      judges.each_with_index do |judge, i|
+        target_distributions = priority_target - judge_distributions_this_month[i]
+        distribution = subject.detect { |dist| dist.judge_id == judge.id }
+        expect(distribution.statistics["batch_size"]).to eq target_distributions
+        distributed_cases = DistributedCase.where(distribution: distribution)
+        expect(distributed_cases.count).to eq target_distributions
+      end
+    end
+  end
+
+  context ".slack_report" do
+    let!(:job) { PushPriorityAppealsToJudgesJob.new }
+    let(:previous_distributions) { to_judge_hash([4, 3, 2, 1, 0]) }
+    let!(:legacy_priority_case) do
+      create(
+        :case,
+        :aod,
+        bfd19: 1.year.ago,
+        bfac: "1",
+        bfmpro: "ACT",
+        bfcurloc: "81",
+        bfdloout: 1.month.ago,
+        folder: build(
+          :folder,
+          tinum: "1801000",
+          titrnum: "123456789S"
+        )
+      )
+    end
+    let!(:ready_priority_hearing_case) do
+      appeal = FactoryBot.create(:appeal,
+                                 :advanced_on_docket_due_to_age,
+                                 :ready_for_distribution,
+                                 docket_type: Constants.AMA_DOCKETS.hearing)
+      appeal.tasks.find_by(type: DistributionTask.name).update(assigned_at: 2.months.ago)
+      appeal.reload
+    end
+    let!(:ready_priority_evidence_case) do
+      appeal = create(:appeal,
+                      :with_post_intake_tasks,
+                      :advanced_on_docket_due_to_age,
+                      docket_type: Constants.AMA_DOCKETS.evidence_submission)
+      appeal.tasks.find_by(type: EvidenceSubmissionWindowTask.name).completed!
+      appeal.tasks.find_by(type: DistributionTask.name).update(assigned_at: 3.months.ago)
+      appeal
+    end
+    let!(:ready_priority_direct_case) do
+      appeal = create(:appeal,
+                      :with_post_intake_tasks,
+                      :advanced_on_docket_due_to_age,
+                      docket_type: Constants.AMA_DOCKETS.direct_review,
+                      receipt_date: 1.month.ago)
+      appeal.tasks.find_by(type: DistributionTask.name).update(assigned_at: 4.months.ago)
+      appeal
+    end
+
+    let(:distributed_cases) do
+      (0...5).map do |count|
+        distribution = build(:distribution, statistics: { "batch_size" => count })
+        distribution.save(validate: false)
+        distribution
+      end
+    end
+
+    subject { job.slack_report }
+
+    before do
+      job.instance_variable_set(:@tied_distributions, distributed_cases)
+      job.instance_variable_set(:@genpop_distributions, distributed_cases)
+      allow_any_instance_of(PushPriorityAppealsToJudgesJob)
+        .to receive(:priority_distributions_this_month_for_eligible_judges).and_return(previous_distributions)
+      allow_any_instance_of(DocketCoordinator).to receive(:priority_count).and_return(20)
+    end
+
+    it "returns ids and age of ready priority appeals not distributed" do
+      expect(subject.second).to eq "*Number of cases tied to judges distributed*: 10"
+      expect(subject.third).to eq "*Number of general population cases distributed*: 10"
+
+      today = Time.zone.now.to_date
+      legacy_days_waiting = (today - legacy_priority_case.bfdloout.to_date).to_i
+      expect(subject[3]).to eq "*Age of oldest legacy case*: #{legacy_days_waiting} days"
+      direct_review_days_waiting = (today - ready_priority_direct_case.ready_for_distribution_at.to_date).to_i
+      expect(subject[4]).to eq "*Age of oldest direct_review case*: #{direct_review_days_waiting} days"
+      evidence_submission_days_waiting = (today - ready_priority_evidence_case.ready_for_distribution_at.to_date).to_i
+      expect(subject[5]).to eq "*Age of oldest evidence_submission case*: #{evidence_submission_days_waiting} days"
+      hearing_days_waiting = (today - ready_priority_hearing_case.ready_for_distribution_at.to_date).to_i
+      expect(subject[6]).to eq "*Age of oldest hearing case*: #{hearing_days_waiting} days"
+
+      expect(subject[7]).to eq "*Number of appeals _not_ distributed*: 4"
+
+      expect(subject[10]).to eq "Priority Target: 6"
+      expect(subject[11]).to eq "Previous monthly distributions: #{previous_distributions}"
+      expect(subject[12].include?(legacy_priority_case.bfkey)).to be true
+      expect(subject[13].include?(ready_priority_hearing_case.uuid)).to be true
+      expect(subject[13].include?(ready_priority_evidence_case.uuid)).to be true
+      expect(subject[13].include?(ready_priority_direct_case.uuid)).to be true
+
+      expect(subject.last).to eq COPY::PRIORITY_PUSH_WARNING_MESSAGE
+    end
+  end
+
+  context ".eligible_judge_target_distributions_with_leftovers" do
+    shared_examples "correct target distributions with leftovers" do
+      before do
+        allow_any_instance_of(PushPriorityAppealsToJudgesJob)
+          .to receive(:priority_distributions_this_month_for_eligible_judges)
+          .and_return(to_judge_hash(distribution_counts))
+        allow_any_instance_of(DocketCoordinator).to receive(:priority_count).and_return(priority_count)
+      end
+
+      subject { PushPriorityAppealsToJudgesJob.new.eligible_judge_target_distributions_with_leftovers }
+
+      it "returns hash of how many cases should be distributed to each judge that is below the priority target " \
+        "including any leftover cases" do
+        expect(subject.values).to match_array expected_priority_targets_with_leftovers
+      end
+    end
+
+    context "when the distributions this month have been even" do
+      shared_examples "even distributions this month" do
+        context "when there are more eligible judges than cases to distribute" do
+          let(:eligible_judge_count) { 10 }
+          let(:priority_count) { 5 }
+          # Priority target will be 0, 5 cases are allotted to 5 judges
+          let(:expected_priority_targets_with_leftovers) { Array.new(priority_count, 1) }
+
+          it_behaves_like "correct target distributions with leftovers"
+        end
+
+        context "when there are more cases to distribute than eligible judges" do
+          let(:eligible_judge_count) { 5 }
+          let(:priority_count) { 10 }
+          # Priority target will be 2, evenly distributed amongst judges
+          let(:expected_priority_targets_with_leftovers) { Array.new(eligible_judge_count, 2) }
+
+          it_behaves_like "correct target distributions with leftovers"
+
+          context "when the cases cannot be evenly distributed" do
+            let(:priority_count) { 9 }
+            # Priority target rounds down, leftover 4 cases are allotted to judges with the fewest cases to distribute
+            let(:expected_priority_targets_with_leftovers) { [1, 2, 2, 2, 2] }
+
+            it_behaves_like "correct target distributions with leftovers"
+          end
+        end
+      end
+
+      let(:distribution_counts) { Array.new(eligible_judge_count, number_of_distributions) }
+
+      context "when there have been no distributions this month" do
+        let(:number_of_distributions) { 0 }
+
+        it_behaves_like "even distributions this month"
+      end
+
+      context "when there have been some distributions this month" do
+        let(:number_of_distributions) { eligible_judge_count * 2 }
+
+        it_behaves_like "even distributions this month"
+      end
+    end
+
+    context "when previous distributions counts are not greater than the priority target" do
+      let(:distribution_counts) { [10, 15, 20, 25, 30] }
+      let(:priority_count) { 75 }
+      # 75 should be able to be divided up to get all judges up to 35
+      let(:expected_priority_targets_with_leftovers) { [25, 20, 15, 10, 5] }
+
+      it_behaves_like "correct target distributions with leftovers"
+
+      context "when cases are not evenly distributable" do
+        let(:priority_count) { 79 }
+        # Priority target still is 35, the leftover 4 cases are allotted to judges with the fewest cases to distribute
+        let(:expected_priority_targets_with_leftovers) { [25, 21, 16, 11, 6] }
+
+        it_behaves_like "correct target distributions with leftovers"
+      end
+    end
+
+    context "when previous distributions counts are greater than the priority target" do
+      let(:distribution_counts) { [0, 0, 22, 28, 50] }
+      let(:priority_count) { 50 }
+      # The two judges with 0 cases should receive 24 cases, the one judge with 22 cases should recieve 2, leaving us
+      # with [24, 24, 24, 28, 50] to hopefully even out more in the next distribution.
+      let(:expected_priority_targets_with_leftovers) { [24, 24, 2] }
+
+      it_behaves_like "correct target distributions with leftovers"
+
+      context "when the distribution counts are even more disparate" do
+        let(:distribution_counts) { [0, 0, 5, 9, 26, 27, 55, 56, 89, 100] }
+        let(:priority_count) { 50 }
+        # Ending counts should be [16, 16, 16, 16, 26, 27, 55, 56, 89, 100], perfectly using up all 50 cases
+        let(:expected_priority_targets_with_leftovers) { [16, 16, 11, 7] }
+
+        it_behaves_like "correct target distributions with leftovers"
+
+        context "when there are leftover cases" do
+          let(:priority_count) { 53 }
+          # Ending counts should be [16, 17, 17, 17, 26, 27, 55, 56, 89, 100]
+          let(:expected_priority_targets_with_leftovers) { [16, 17, 12, 8] }
+
+          it_behaves_like "correct target distributions with leftovers"
+        end
+      end
+    end
+
+    context "tracking distributions over time" do
+      let(:number_judges) { rand(5..10) }
+      let(:priority_count) { rand(10..30) }
+
+      before do
+        # Mock cases already distributed this month
+        @distribution_counts = to_judge_hash(Array.new(number_judges).map { rand(12) })
+        allow_any_instance_of(PushPriorityAppealsToJudgesJob)
+          .to receive(:priority_distributions_this_month_for_eligible_judges).and_return(@distribution_counts)
+        allow_any_instance_of(PushPriorityAppealsToJudgesJob)
+          .to receive(:ready_priority_appeals_count).and_return(priority_count)
+      end
+
+      it "evens out over multiple calls" do
+        4.times do
+          # Mock distributing cases each week
+          target_distributions = PushPriorityAppealsToJudgesJob.new.eligible_judge_target_distributions_with_leftovers
+          @distribution_counts.merge!(target_distributions) { |_, prev_dists, target_dist| prev_dists + target_dist }
+        end
+        final_counts = @distribution_counts.values.uniq
+        # Expect no more than two distinct counts, no more than 1 apart
+        expect(final_counts.max - final_counts.min).to be <= 1
+        expect(final_counts.count).to be <= 2
+      end
+    end
+  end
+
+  context ".leftover_cases_count" do
+    before do
+      allow_any_instance_of(PushPriorityAppealsToJudgesJob)
+        .to receive(:target_distributions_for_eligible_judges).and_return(target_distributions)
+      allow_any_instance_of(DocketCoordinator).to receive(:priority_count).and_return(priority_count)
+    end
+
+    subject { PushPriorityAppealsToJudgesJob.new.leftover_cases_count }
+
+    context "when the number of cases to distribute is evenly divisible by the number of judges that need cases" do
+      let(:eligible_judge_count) { 4 }
+      let(:priority_count) { 100 }
+      let(:target_distributions) do
+        Array.new(eligible_judge_count, priority_count / eligible_judge_count)
+          .each_with_index
+          .map { |count, i| [i, count] }.to_h
+      end
+
+      it "returns no leftover cases" do
+        expect(subject).to eq 0
+      end
+    end
+
+    context "when the number of cases can be distributed to all judges evenly" do
+      let(:priority_count) { target_distributions.values.sum }
+      let(:target_distributions) { to_judge_hash([5, 10, 15, 20, 25]) }
+
+      it "returns no leftover cases" do
+        expect(subject).to eq 0
+      end
+    end
+
+    context "when the number of cases are not evenly distributable bewteen all judges" do
+      let(:leftover_cases_count) { target_distributions.count - 1 }
+      let(:priority_count) { target_distributions.values.sum + leftover_cases_count }
+      let(:target_distributions) { to_judge_hash([5, 10, 15, 20, 25]) }
+
+      it "returns the correct number of leftover cases" do
+        expect(subject).to eq leftover_cases_count
+      end
+    end
+  end
+
+  context ".target_distributions_for_eligible_judges" do
+    shared_examples "correct target distributions" do
+      before do
+        allow_any_instance_of(PushPriorityAppealsToJudgesJob)
+          .to receive(:priority_distributions_this_month_for_eligible_judges)
+          .and_return(to_judge_hash(distribution_counts))
+        allow_any_instance_of(DocketCoordinator).to receive(:priority_count).and_return(priority_count)
+      end
+
+      subject { PushPriorityAppealsToJudgesJob.new.target_distributions_for_eligible_judges }
+
+      it "returns hash of how many cases should be distributed to each judge that is below the priority target " \
+        "excluding any leftover cases" do
+        expect(subject).to eq to_judge_hash(expected_priority_targets)
+      end
+    end
+
+    context "when the distributions this month have been even" do
+      let(:distribution_counts) { Array.new(eligible_judge_count, number_of_distributions) }
+
+      context "when there have been no distributions this month" do
+        let(:number_of_distributions) { 0 }
+
+        context "when there are more eligible judges than cases to distribute" do
+          let(:eligible_judge_count) { 10 }
+          let(:priority_count) { 5 }
+          # Priority target will be 0, cases will be allotted later from the leftover cases
+          let(:expected_priority_targets) { Array.new(eligible_judge_count, 0) }
+
+          it_behaves_like "correct target distributions"
+        end
+
+        context "when there are more cases to distribute than eligible judges" do
+          let(:eligible_judge_count) { 5 }
+          let(:priority_count) { 10 }
+          # Priority target will be 2, evenly distributed amongst judges
+          let(:expected_priority_targets) { Array.new(eligible_judge_count, 2) }
+
+          it_behaves_like "correct target distributions"
+
+          context "when the cases cannot be evenly distributed" do
+            let(:priority_count) { 9 }
+            # Priority target rounds down, the leftover 4 cases will be allotted later in the algorithm
+            let(:expected_priority_targets) { Array.new(eligible_judge_count, 1) }
+
+            it_behaves_like "correct target distributions"
+          end
+        end
+      end
+
+      context "when there have been some distributions this month" do
+        let(:number_of_distributions) { eligible_judge_count * 2 }
+
+        context "when there are more eligible judges than cases to distribute" do
+          let(:eligible_judge_count) { 10 }
+          let(:priority_count) { 5 }
+          # Priority target is equal to how many cases have already been distributed. Cases will be allotted from the
+          # leftover cases. 5 judges will each be distributed 1 case
+          let(:expected_priority_targets) { Array.new(eligible_judge_count, 0) }
+
+          it_behaves_like "correct target distributions"
+        end
+
+        context "when there are more cases to distribute than eligible judges" do
+          let(:eligible_judge_count) { 5 }
+          let(:priority_count) { 10 }
+          # Priority target will be evenly distributed amongst judges
+          let(:expected_priority_targets) { Array.new(eligible_judge_count, 2) }
+
+          it_behaves_like "correct target distributions"
+
+          context "when the cases cannot be evenly distributed" do
+            let(:priority_count) { 9 }
+            # Priority target rounds down, the leftover 4 cases will be allotted later in the algorithm
+            let(:expected_priority_targets) { Array.new(eligible_judge_count, 1) }
+
+            it_behaves_like "correct target distributions"
+          end
+        end
+      end
+    end
+
+    context "when previous distributions counts are not greater than the priority target" do
+      let(:distribution_counts) { [10, 15, 20, 25, 30] }
+      let(:priority_count) { 75 }
+      # 75 should be able to be divided up to get all judges up to 35
+      let(:expected_priority_targets) { [25, 20, 15, 10, 5] }
+
+      it_behaves_like "correct target distributions"
+
+      context "when cases are not evenly distributable" do
+        let(:priority_count) { 79 }
+        # Priority target still is 35, the leftover 4 cases will be allotted later in the algorithm
+
+        it_behaves_like "correct target distributions"
+      end
+    end
+
+    context "when previous distributions counts are greater than the priority target" do
+      let(:distribution_counts) { [0, 0, 22, 28, 50] }
+      let(:priority_count) { 50 }
+      # The two judges with 0 cases should receive 24 cases, the one judge with 22 cases should recieve 2, leaving us
+      # with [24, 24, 24, 28, 50] to hopefully even out more in the next distribution.
+      let(:expected_priority_targets) { [24, 24, 2] }
+
+      it_behaves_like "correct target distributions"
+
+      context "when the dirstibution counts are even more disparate" do
+        let(:distribution_counts) { [0, 0, 5, 9, 26, 27, 55, 56, 89, 100] }
+        let(:priority_count) { 50 }
+        # Ending counts should be [16, 16, 16, 16, 26, 27, 55, 56, 89, 100], perfectly using up all 50 cases
+        let(:expected_priority_targets) { [16, 16, 11, 7] }
+
+        it_behaves_like "correct target distributions"
+      end
+    end
+  end
+
+  context ".priority_target" do
+    shared_examples "correct target" do
+      before do
+        allow_any_instance_of(PushPriorityAppealsToJudgesJob)
+          .to receive(:priority_distributions_this_month_for_eligible_judges)
+          .and_return(to_judge_hash(distribution_counts))
+        allow_any_instance_of(DocketCoordinator).to receive(:priority_count).and_return(priority_count)
+      end
+
+      subject { PushPriorityAppealsToJudgesJob.new.priority_target }
+
+      it "calculates a target that distributes cases evenly over one month" do
+        expect(subject).to eq expected_priority_target
+      end
+    end
+
+    context "when the distributions this month have been even" do
+      let(:distribution_counts) { Array.new(eligible_judge_count, number_of_distributions) }
+
+      context "when there have been no distributions this month" do
+        let(:number_of_distributions) { 0 }
+
+        context "when there are more eligible judges than cases to distribute" do
+          let(:eligible_judge_count) { 10 }
+          let(:priority_count) { 5 }
+          # Priority target will be 0, cases will be allotted from the leftover cases. judges will be distributed 1 case
+          let(:expected_priority_target) { 0 }
+
+          it_behaves_like "correct target"
+        end
+
+        context "when there are more cases to distribute than eligible judges" do
+          let(:eligible_judge_count) { 5 }
+          let(:priority_count) { 10 }
+          # Priority target will be evenly distributed amongst judges
+          let(:expected_priority_target) { 2 }
+
+          it_behaves_like "correct target"
+
+          context "when the cases cannot be evenly distributed" do
+            let(:priority_count) { 9 }
+            # Priority target rounds down, the leftover 4 cases will be allotted later in the algorithm
+            let(:expected_priority_target) { 1 }
+
+            it_behaves_like "correct target"
+          end
+        end
+      end
+
+      context "when there have been some distributions this month" do
+        let(:number_of_distributions) { 10 }
+
+        context "when there are more eligible judges than cases to distribute" do
+          let(:eligible_judge_count) { 10 }
+          let(:priority_count) { 5 }
+          # Priority target is equal to how many cases have already been distributed. Cases will be allotted from the
+          # leftover cases. judges will each be distributed 1 case
+          let(:expected_priority_target) { 10 }
+
+          it_behaves_like "correct target"
+        end
+
+        context "when there are more cases to distribute than eligible judges" do
+          let(:eligible_judge_count) { 5 }
+          let(:priority_count) { 10 }
+          # Priority target will be evenly distributed amongst judges
+          let(:expected_priority_target) { 12 }
+
+          it_behaves_like "correct target"
+
+          context "when the cases cannot be evenly distributed" do
+            let(:priority_count) { 9 }
+            # Priority target rounds down, the leftover 4 cases will be allotted later in the algorithm
+            let(:expected_priority_target) { 11 }
+
+            it_behaves_like "correct target"
+          end
+        end
+      end
+    end
+
+    context "when previous distributions counts are not greater than the priority target" do
+      let(:distribution_counts) { [10, 15, 20, 25, 30] }
+      let(:priority_count) { 75 }
+      # 75 should be able to be divided up to get all judges up to 35
+      let(:expected_priority_target) { 35 }
+
+      it_behaves_like "correct target"
+
+      context "when cases are not evenly distributable" do
+        let(:priority_count) { 79 }
+        # Priority target still is 35, the leftover 4 cases will be allotted later in the algorithm
+
+        it_behaves_like "correct target"
+      end
+    end
+
+    context "when previous distributions counts are greater than the priority target" do
+      let(:distribution_counts) { [0, 0, 22, 28, 50] }
+      let(:priority_count) { 50 }
+      # The two judges with 0 cases should receive 24 cases, the one judge with 22 cases should recieve 2, leaving us
+      # with [24, 24, 24, 28, 50] to hopefully even out more in the next distribution.
+      let(:expected_priority_target) { 24 }
+
+      it_behaves_like "correct target"
+
+      context "when the dirstibution counts are even more disparate" do
+        let(:distribution_counts) { [0, 0, 5, 9, 26, 27, 55, 56, 89, 100] }
+        let(:priority_count) { 50 }
+        # Ending counts should be [16, 16, 16, 16, 26, 27, 55, 56, 89, 100], perfectly using up all 50 cases
+        let(:expected_priority_target) { 16 }
+
+        it_behaves_like "correct target"
+      end
+    end
+  end
+
+  context ".priority_distributions_this_month_for_eligible_judges" do
+    let!(:judge_without_team) { create(:user) }
+    let!(:judge_without_active_team) { create(:user).tap { |judge| JudgeTeam.create_for_judge(judge).inactive! } }
+    let!(:judge_without_priority_push_team) do
+      create(:user).tap { |judge| JudgeTeam.create_for_judge(judge).update(accepts_priority_pushed_cases: false) }
+    end
+    let!(:judge_with_org) { create(:user).tap { |judge| create(:organization).add_user(judge) } }
+    let!(:judge_with_team_and_distributions) { create(:user).tap { |judge| JudgeTeam.create_for_judge(judge) } }
+    let!(:judge_with_team_without_distributions) { create(:user).tap { |judge| JudgeTeam.create_for_judge(judge) } }
+
+    let!(:distributions_for_valid_judge) { 6 }
+
+    subject { PushPriorityAppealsToJudgesJob.new.priority_distributions_this_month_for_eligible_judges }
+
+    before do
+      allow_any_instance_of(PushPriorityAppealsToJudgesJob)
+        .to receive(:priority_distributions_this_month_for_all_judges)
+        .and_return(
+          judge_without_team.id => 5,
+          judge_without_active_team.id => 5,
+          judge_without_priority_push_team.id => 5,
+          judge_with_org.id => 5,
+          judge_with_team_and_distributions.id => distributions_for_valid_judge
+        )
+    end
+
+    it "only returns hash containing the distribution counts of judges that can be pushed priority appeals" do
+      [
+        judge_without_team,
+        judge_without_active_team,
+        judge_without_priority_push_team,
+        judge_with_org
+      ].each do |ineligible_judge|
+        expect(subject[ineligible_judge]).to be nil
+      end
+      expect(subject[judge_with_team_and_distributions.id]).to eq distributions_for_valid_judge
+      expect(subject[judge_with_team_without_distributions.id]).to eq 0
+    end
+  end
+
+  context ".eligible_judges" do
+    let!(:judge_without_team) { create(:user) }
+    let!(:judge_without_active_team) { create(:user).tap { |judge| JudgeTeam.create_for_judge(judge).inactive! } }
+    let!(:judge_without_priority_push_team) do
+      create(:user).tap { |judge| JudgeTeam.create_for_judge(judge).update(accepts_priority_pushed_cases: false) }
+    end
+    let!(:judge_with_org) { create(:user).tap { |judge| create(:organization).add_user(judge) } }
+    let!(:judge_with_team) { create(:user).tap { |judge| JudgeTeam.create_for_judge(judge) } }
+
+    subject { PushPriorityAppealsToJudgesJob.new.eligible_judges }
+
+    it "only returns judges of active judge teams" do
+      expect(subject).to match_array([judge_with_team])
+    end
+  end
+
+  context ".priority_distributions_this_month_for_all_judges" do
+    let(:batch_size) { 20 }
+    let!(:judge_with_no_priority_distributions) do
+      create(:user, :with_vacols_judge_record) do |judge|
+        create(
+          :distribution,
+          judge: judge,
+          priority_push: false,
+          completed_at: 1.day.ago,
+          statistics: { "batch_size": batch_size }
+        ).tap { |distribution| distribution.update!(status: :completed) }
+      end
+    end
+    let!(:judge_with_no_recent_distributions) do
+      create(:user, :with_vacols_judge_record) do |judge|
+        create(
+          :distribution,
+          judge: judge,
+          priority_push: true,
+          completed_at: 41.days.ago,
+          statistics: { "batch_size": batch_size }
+        ).tap { |distribution| distribution.update!(status: :completed) }
+      end
+    end
+    let!(:judge_with_no_completed_distributions) do
+      create(:user, :with_vacols_judge_record) do |judge|
+        create(
+          :distribution,
+          judge: judge,
+          priority_push: true,
+          statistics: { "batch_size": batch_size }
+        )
+      end
+    end
+    let!(:judge_with_a_valid_distribution) do
+      create(:user, :with_vacols_judge_record) do |judge|
+        create(
+          :distribution,
+          judge: judge,
+          priority_push: true,
+          completed_at: 1.day.ago,
+          statistics: { "batch_size": batch_size }
+        ).tap { |distribution| distribution.update!(status: :completed) }
+      end
+    end
+    let!(:judge_with_multiple_valid_distributions) do
+      create(:user, :with_vacols_judge_record) do |judge|
+        create(
+          :distribution,
+          judge: judge,
+          priority_push: true,
+          completed_at: 1.day.ago,
+          statistics: { "batch_size": batch_size }
+        ).tap { |distribution| distribution.update!(status: :completed) }
+        create(
+          :distribution,
+          judge: judge,
+          priority_push: true,
+          completed_at: 1.day.ago,
+          statistics: { "batch_size": batch_size }
+        ).tap { |distribution| distribution.update!(status: :completed) }
+      end
+    end
+
+    subject { PushPriorityAppealsToJudgesJob.new.priority_distributions_this_month_for_all_judges }
+
+    it "returns the sum of the batch sizes from all valid distributions for each judge" do
+      expect(subject.keys).to match_array(
+        [judge_with_a_valid_distribution.id, judge_with_multiple_valid_distributions.id]
+      )
+      expect(subject[judge_with_a_valid_distribution.id]).to eq batch_size
+      expect(subject[judge_with_multiple_valid_distributions.id]).to eq batch_size * 2
+    end
+  end
+
+  context ".priority_distributions_this_month" do
+    let!(:non_priority_distribution) do
+      create(
+        :distribution,
+        judge: create(:user, :with_vacols_judge_record),
+        priority_push: false,
+        completed_at: 1.day.ago
+      ).tap { |distribution| distribution.update!(status: :completed) }
+    end
+    let!(:pending_priority_distribution) do
+      create(
+        :distribution,
+        judge: create(:user, :with_vacols_judge_record),
+        priority_push: true
+      )
+    end
+    let!(:older_priority_distribution) do
+      create(
+        :distribution,
+        judge: create(:user, :with_vacols_judge_record),
+        priority_push: true,
+        completed_at: 41.days.ago
+      ).tap { |distribution| distribution.update!(status: :completed) }
+    end
+    let!(:recent_completed_priority_distribution) do
+      create(
+        :distribution,
+        judge: create(:user, :with_vacols_judge_record),
+        priority_push: true,
+        completed_at: 1.day.ago
+      ).tap { |distribution| distribution.update!(status: :completed) }
+    end
+
+    subject { PushPriorityAppealsToJudgesJob.new.priority_distributions_this_month }
+
+    it "only returns recently completed priority distributions" do
+      expect(subject.count).to eq 1
+      expect(subject.first).to eq recent_completed_priority_distribution
+    end
+  end
+end
diff --git a/spec/models/distribution_spec.rb b/spec/models/distribution_spec.rb
index 22a751a4310..257b579a041 100644
--- a/spec/models/distribution_spec.rb
+++ b/spec/models/distribution_spec.rb
@@ -30,9 +30,15 @@
   end
 
   context "#distribute!" do
-    subject { Distribution.create!(judge: judge) }
+    before do
+      allow_any_instance_of(DirectReviewDocket)
+        .to receive(:nonpriority_receipts_per_year)
+        .and_return(100)
 
-    let(:legacy_priority_count) { 14 }
+      allow(Docket)
+        .to receive(:nonpriority_decisions_per_year)
+        .and_return(1000)
+    end
 
     def create_legacy_case(index, traits = nil)
       create(
@@ -51,14 +57,6 @@ def create_legacy_case(index, traits = nil)
       )
     end
 
-    let!(:legacy_priority_cases) do
-      (1..legacy_priority_count).map { |i| create_legacy_case(i, :aod) }
-    end
-
-    let!(:legacy_nonpriority_cases) do
-      (15..100).map { |i| create_legacy_case(i) }
-    end
-
     def create_legacy_case_hearing_for(appeal, board_member: judge.vacols_attorney_id)
       create(:case_hearing,
              :disposition_held,
@@ -67,365 +65,542 @@ def create_legacy_case_hearing_for(appeal, board_member: judge.vacols_attorney_i
              board_member: board_member)
     end
 
-    let!(:same_judge_priority_hearings) do
-      legacy_priority_cases[0..1].map { |appeal| create_legacy_case_hearing_for(appeal) }
-    end
+    context "priority_push is false" do
+      before do
+        allow_any_instance_of(HearingRequestDocket)
+          .to receive(:age_of_n_oldest_priority_appeals)
+          .and_return([])
 
-    let!(:same_judge_nonpriority_hearings) do
-      legacy_nonpriority_cases[29..33].map { |appeal| create_legacy_case_hearing_for(appeal) }
-    end
+        allow_any_instance_of(HearingRequestDocket)
+          .to receive(:distribute_appeals)
+          .and_return([])
+      end
 
-    let!(:other_judge_hearings) do
-      legacy_nonpriority_cases[2..27].map { |appeal| create_legacy_case_hearing_for(appeal, board_member: "1234") }
-    end
+      subject { Distribution.create!(judge: judge) }
 
-    before do
-      allow_any_instance_of(DirectReviewDocket)
-        .to receive(:nonpriority_receipts_per_year)
-        .and_return(100)
+      let(:legacy_priority_count) { 14 }
 
-      allow(Docket)
-        .to receive(:nonpriority_decisions_per_year)
-        .and_return(1000)
+      let!(:legacy_priority_cases) do
+        (1..legacy_priority_count).map { |i| create_legacy_case(i, :aod) }
+      end
 
-      allow_any_instance_of(HearingRequestDocket)
-        .to receive(:age_of_n_oldest_priority_appeals)
-        .and_return([])
+      let!(:legacy_nonpriority_cases) do
+        (15..100).map { |i| create_legacy_case(i) }
+      end
 
-      allow_any_instance_of(HearingRequestDocket)
-        .to receive(:distribute_appeals)
-        .and_return([])
-    end
+      let!(:same_judge_priority_hearings) do
+        legacy_priority_cases[0..1].map { |appeal| create_legacy_case_hearing_for(appeal) }
+      end
 
-    let!(:due_direct_review_cases) do
-      (0...6).map do
-        create(:appeal,
-               :with_post_intake_tasks,
-               docket_type: Constants.AMA_DOCKETS.direct_review,
-               receipt_date: 11.months.ago,
-               target_decision_date: 1.month.from_now)
+      let!(:same_judge_nonpriority_hearings) do
+        legacy_nonpriority_cases[29..33].map { |appeal| create_legacy_case_hearing_for(appeal) }
       end
-    end
 
-    let!(:priority_direct_review_case) do
-      appeal = create(:appeal,
-                      :with_post_intake_tasks,
-                      :advanced_on_docket_due_to_age,
-                      docket_type: Constants.AMA_DOCKETS.direct_review,
-                      receipt_date: 1.month.ago)
-      appeal.tasks.find_by(type: DistributionTask.name).update(assigned_at: 1.month.ago)
-      appeal
-    end
+      let!(:other_judge_hearings) do
+        legacy_nonpriority_cases[2..27].map { |appeal| create_legacy_case_hearing_for(appeal, board_member: "1234") }
+      end
 
-    let!(:other_direct_review_cases) do
-      (0...20).map do
-        create(:appeal,
-               :with_post_intake_tasks,
-               docket_type: Constants.AMA_DOCKETS.direct_review,
-               receipt_date: 61.days.ago,
-               target_decision_date: 304.days.from_now)
+      let!(:due_direct_review_cases) do
+        (0...6).map do
+          create(:appeal,
+                 :with_post_intake_tasks,
+                 docket_type: Constants.AMA_DOCKETS.direct_review,
+                 receipt_date: 11.months.ago,
+                 target_decision_date: 1.month.from_now)
+        end
       end
-    end
 
-    let!(:evidence_submission_cases) do
-      (0...43).map do
-        create(:appeal,
-               :with_post_intake_tasks,
-               docket_type: Constants.AMA_DOCKETS.evidence_submission)
+      let!(:priority_direct_review_case) do
+        appeal = create(:appeal,
+                        :with_post_intake_tasks,
+                        :advanced_on_docket_due_to_age,
+                        docket_type: Constants.AMA_DOCKETS.direct_review,
+                        receipt_date: 1.month.ago)
+        appeal.tasks.find_by(type: DistributionTask.name).update(assigned_at: 1.month.ago)
+        appeal
       end
-    end
 
-    let!(:hearing_cases) do
-      (0...43).map do
-        create(:appeal,
-               :with_post_intake_tasks,
-               docket_type: Constants.AMA_DOCKETS.hearing)
+      let!(:other_direct_review_cases) do
+        (0...20).map do
+          create(:appeal,
+                 :with_post_intake_tasks,
+                 docket_type: Constants.AMA_DOCKETS.direct_review,
+                 receipt_date: 61.days.ago,
+                 target_decision_date: 304.days.from_now)
+        end
       end
-    end
 
-    it "correctly distributes cases" do
-      evidence_submission_cases[0...2].each do |appeal|
-        appeal.tasks
-          .find_by(type: EvidenceSubmissionWindowTask.name)
-          .update!(status: :completed)
-      end
-      subject.distribute!
-      expect(subject.valid?).to eq(true)
-      expect(subject.status).to eq("completed")
-      expect(subject.started_at).to eq(Time.zone.now) # time is frozen so appears zero time elapsed
-      expect(subject.errored_at).to be_nil
-      expect(subject.completed_at).to eq(Time.zone.now)
-      expect(subject.statistics["batch_size"]).to eq(15)
-      expect(subject.statistics["total_batch_size"]).to eq(45)
-      expect(subject.statistics["priority_count"]).to eq(legacy_priority_count + 1)
-      expect(subject.statistics["legacy_proportion"]).to eq(0.4)
-      expect(subject.statistics["legacy_hearing_backlog_count"]).to be <= 3
-      expect(subject.statistics["direct_review_proportion"]).to eq(0.2)
-      expect(subject.statistics["evidence_submission_proportion"]).to eq(0.2)
-      expect(subject.statistics["hearing_proportion"]).to eq(0.2)
-      expect(subject.statistics["pacesetting_direct_review_proportion"]).to eq(0.1)
-      expect(subject.statistics["interpolated_minimum_direct_review_proportion"]).to eq(0.067)
-      expect(subject.statistics["nonpriority_iterations"]).to be_between(2, 3)
-      expect(subject.distributed_cases.count).to eq(15)
-      expect(subject.distributed_cases.first.docket).to eq("legacy")
-      expect(subject.distributed_cases.first.ready_at).to eq(2.days.ago.beginning_of_day)
-      expect(subject.distributed_cases.where(priority: true).count).to eq(5)
-      expect(subject.distributed_cases.where(genpop: true).count).to eq(5)
-      expect(subject.distributed_cases.where(priority: true, genpop: false).count).to eq(2)
-      expect(subject.distributed_cases.where(priority: false, genpop_query: "not_genpop").count).to eq(0)
-      expect(subject.distributed_cases.where(priority: false, genpop_query: "any").map(&:docket_index).max).to eq(30)
-      expect(subject.distributed_cases.where(priority: true,
-                                             docket: Constants.AMA_DOCKETS.direct_review).count).to eq(1)
-      expect(subject.distributed_cases.where(docket: "legacy").count).to be >= 8
-      expect(subject.distributed_cases.where(docket: Constants.AMA_DOCKETS.direct_review).count).to be >= 3
-      expect(subject.distributed_cases.where(docket: Constants.AMA_DOCKETS.evidence_submission).count).to eq(2)
-    end
+      let!(:evidence_submission_cases) do
+        (0...43).map do
+          create(:appeal,
+                 :with_post_intake_tasks,
+                 docket_type: Constants.AMA_DOCKETS.evidence_submission)
+        end
+      end
 
-    context "with priority_acd on" do
-      before { FeatureToggle.enable!(:priority_acd, users: [judge.css_id]) }
-      after { FeatureToggle.disable!(:priority_acd) }
+      let!(:hearing_cases) do
+        (0...43).map do
+          create(:appeal,
+                 :with_post_intake_tasks,
+                 docket_type: Constants.AMA_DOCKETS.hearing)
+        end
+      end
 
-      BACKLOG_LIMIT = VACOLS::CaseDocket::HEARING_BACKLOG_LIMIT
+      it "correctly distributes cases" do
+        evidence_submission_cases[0...2].each do |appeal|
+          appeal.tasks
+            .find_by(type: EvidenceSubmissionWindowTask.name)
+            .update!(status: :completed)
+        end
+        subject.distribute!
+        expect(subject.valid?).to eq(true)
+        expect(subject.priority_push).to eq(false)
+        expect(subject.status).to eq("completed")
+        expect(subject.started_at).to eq(Time.zone.now) # time is frozen so appears zero time elapsed
+        expect(subject.errored_at).to be_nil
+        expect(subject.completed_at).to eq(Time.zone.now)
+        expect(subject.statistics["batch_size"]).to eq(15)
+        expect(subject.statistics["total_batch_size"]).to eq(45)
+        expect(subject.statistics["priority_count"]).to eq(legacy_priority_count + 1)
+        expect(subject.statistics["legacy_proportion"]).to eq(0.4)
+        expect(subject.statistics["legacy_hearing_backlog_count"]).to be <= 3
+        expect(subject.statistics["direct_review_proportion"]).to eq(0.2)
+        expect(subject.statistics["evidence_submission_proportion"]).to eq(0.2)
+        expect(subject.statistics["hearing_proportion"]).to eq(0.2)
+        expect(subject.statistics["pacesetting_direct_review_proportion"]).to eq(0.1)
+        expect(subject.statistics["interpolated_minimum_direct_review_proportion"]).to eq(0.067)
+        expect(subject.statistics["nonpriority_iterations"]).to be_between(2, 3)
+        expect(subject.distributed_cases.count).to eq(15)
+        expect(subject.distributed_cases.first.docket).to eq("legacy")
+        expect(subject.distributed_cases.first.ready_at).to eq(2.days.ago.beginning_of_day)
+        expect(subject.distributed_cases.where(priority: true).count).to eq(5)
+        expect(subject.distributed_cases.where(genpop: true).count).to eq(5)
+        expect(subject.distributed_cases.where(priority: true, genpop: false).count).to eq(2)
+        expect(subject.distributed_cases.where(priority: false, genpop_query: "not_genpop").count).to eq(0)
+        expect(subject.distributed_cases.where(priority: false, genpop_query: "any").map(&:docket_index).max).to eq(30)
+        expect(subject.distributed_cases.where(priority: true,
+                                               docket: Constants.AMA_DOCKETS.direct_review).count).to eq(1)
+        expect(subject.distributed_cases.where(docket: "legacy").count).to be >= 8
+        expect(subject.distributed_cases.where(docket: Constants.AMA_DOCKETS.direct_review).count).to be >= 3
+        expect(subject.distributed_cases.where(docket: Constants.AMA_DOCKETS.evidence_submission).count).to eq(2)
+      end
+
+      context "with priority_acd on" do
+        before { FeatureToggle.enable!(:priority_acd) }
+        after { FeatureToggle.disable!(:priority_acd) }
+
+        BACKLOG_LIMIT = VACOLS::CaseDocket::HEARING_BACKLOG_LIMIT
+
+        let!(:more_same_judge_nonpriority_hearings) do
+          to_add = total_tied_nonpriority_hearings - same_judge_nonpriority_hearings.count
+          legacy_nonpriority_cases[34..(34 + to_add - 1)].map { |appeal| create_legacy_case_hearing_for(appeal) }
+        end
 
-      let!(:more_same_judge_nonpriority_hearings) do
-        to_add = total_tied_nonpriority_hearings - same_judge_nonpriority_hearings.count
-        legacy_nonpriority_cases[34..(34 + to_add - 1)].map { |appeal| create_legacy_case_hearing_for(appeal) }
-      end
+        context "the judge's backlog has more than #{BACKLOG_LIMIT} legacy hearing non priority cases" do
+          let(:total_tied_nonpriority_hearings) { BACKLOG_LIMIT + 5 }
+
+          it "distributes legacy hearing non priority cases down to #{BACKLOG_LIMIT}" do
+            expect(VACOLS::CaseDocket.nonpriority_hearing_cases_for_judge_count(judge))
+              .to eq total_tied_nonpriority_hearings
+            subject.distribute!
+
+            expect(subject.valid?).to eq(true)
+            expect(subject.statistics["legacy_hearing_backlog_count"]).to eq(BACKLOG_LIMIT)
+            dcs_legacy = subject.distributed_cases.where(docket: "legacy")
+
+            # distributions reliant on BACKLOG_LIMIT
+            judge_tied_leg_distributions = total_tied_nonpriority_hearings - BACKLOG_LIMIT
+            expect(dcs_legacy.where(priority: false, genpop_query: "not_genpop").count).to eq(
+              judge_tied_leg_distributions
+            )
+
+            # distributions after handling BACKLOG_LIMIT
+            expect(dcs_legacy.where(priority: true, genpop_query: "not_genpop").count).to eq(2)
+            expect(dcs_legacy.where(priority: true, genpop_query: "any").count).to eq(2)
+            expect(dcs_legacy.count).to be >= 8
+          end
+        end
 
-      context "the judge's backlog has more than #{BACKLOG_LIMIT} legacy hearing non priority cases" do
-        let(:total_tied_nonpriority_hearings) { BACKLOG_LIMIT + 5 }
+        context "the judge's backlog has less than #{BACKLOG_LIMIT} legacy hearing non priority cases" do
+          let(:total_tied_nonpriority_hearings) { BACKLOG_LIMIT - 5 }
 
-        it "distributes legacy hearing non priority cases down to #{BACKLOG_LIMIT}" do
-          expect(VACOLS::CaseDocket.nonpriority_hearing_cases_for_judge_count(judge))
-            .to eq total_tied_nonpriority_hearings
-          subject.distribute!
+          it "distributes legacy hearing non priority cases down to #{BACKLOG_LIMIT}" do
+            expect(VACOLS::CaseDocket.nonpriority_hearing_cases_for_judge_count(judge))
+              .to eq total_tied_nonpriority_hearings
+            subject.distribute!
 
-          expect(subject.valid?).to eq(true)
-          expect(subject.statistics["legacy_hearing_backlog_count"]).to eq(BACKLOG_LIMIT)
-          dcs_legacy = subject.distributed_cases.where(docket: "legacy")
+            expect(subject.valid?).to eq(true)
+            dcs_legacy = subject.distributed_cases.where(docket: "legacy")
+
+            # distributions reliant on BACKLOG_LIMIT
+            expect(dcs_legacy.where(priority: false, genpop_query: "not_genpop").count).to eq(0)
 
-          # distributions reliant on BACKLOG_LIMIT
-          judge_tied_leg_distributions = total_tied_nonpriority_hearings - BACKLOG_LIMIT
-          expect(dcs_legacy.where(priority: false, genpop_query: "not_genpop").count).to eq judge_tied_leg_distributions
+            # distributions after handling BACKLOG_LIMIT
+            remaining_in_backlog = total_tied_nonpriority_hearings -
+                                   dcs_legacy.where(priority: false, genpop: false).count
+            expect(subject.statistics["legacy_hearing_backlog_count"]).to eq remaining_in_backlog
 
-          # distributions after handling BACKLOG_LIMIT
-          expect(dcs_legacy.where(priority: true, genpop_query: "not_genpop").count).to eq(2)
-          expect(dcs_legacy.where(priority: true, genpop_query: "any").count).to eq(2)
-          expect(dcs_legacy.count).to be >= 8
+            expect(dcs_legacy.where(priority: true, genpop_query: "not_genpop").count).to eq(2)
+            expect(dcs_legacy.where(priority: true, genpop_query: "any").count).to eq(2)
+            expect(dcs_legacy.count).to be >= 8
+          end
         end
       end
 
-      context "the judge's backlog has less than #{BACKLOG_LIMIT} legacy hearing non priority cases" do
-        let(:total_tied_nonpriority_hearings) { BACKLOG_LIMIT - 5 }
+      def create_nonpriority_distributed_case(distribution, case_id, ready_at)
+        distribution.distributed_cases.create(
+          case_id: case_id,
+          priority: false,
+          docket: "legacy",
+          ready_at: VacolsHelper.normalize_vacols_datetime(ready_at),
+          docket_index: "123",
+          genpop: false,
+          genpop_query: "any"
+        )
+      end
 
-        it "distributes legacy hearing non priority cases down to #{BACKLOG_LIMIT}" do
-          expect(VACOLS::CaseDocket.nonpriority_hearing_cases_for_judge_count(judge))
-            .to eq total_tied_nonpriority_hearings
-          subject.distribute!
+      def cancel_relevant_legacy_appeal_tasks
+        legacy_appeal.tasks.reject { |t| t.type == "RootTask" }.each do |task|
+          # update_columns to avoid triggers that will update VACOLS location dates and
+          # mess up ACD date logic.
+          task.update_columns(status: Constants.TASK_STATUSES.cancelled, closed_at: Time.zone.now)
+        end
+      end
 
+      context "when an illegit nonpriority legacy case re-distribution is attempted" do
+        let(:case_id) { legacy_case.bfkey }
+        let!(:previous_location) { legacy_case.bfcurloc }
+        let(:legacy_case) { legacy_nonpriority_cases.second }
+
+        before do
+          @raven_called = false
+          distribution = create(:distribution, judge: judge)
+          # illegit because appeal has completed hearing tasks
+          appeal = create(:legacy_appeal, :with_schedule_hearing_tasks, vacols_case: legacy_case)
+          appeal.tasks.open.where.not(type: RootTask.name).each(&:completed!)
+          create_nonpriority_distributed_case(distribution, case_id, legacy_case.bfdloout)
+          distribution.update!(status: "completed", completed_at: today)
+          allow(Raven).to receive(:capture_exception) { @raven_called = true }
+        end
+
+        it "does not create a duplicate distributed_case and sends alert" do
+          subject.distribute!
           expect(subject.valid?).to eq(true)
-          dcs_legacy = subject.distributed_cases.where(docket: "legacy")
+          expect(subject.error?).to eq(false)
+          expect(@raven_called).to eq(true)
+          expect(subject.distributed_cases.pluck(:case_id)).to_not include(case_id)
+          expect(legacy_case.reload.bfcurloc).to eq(previous_location)
+        end
+      end
 
-          # distributions reliant on BACKLOG_LIMIT
-          expect(dcs_legacy.where(priority: false, genpop_query: "not_genpop").count).to eq(0)
+      context "when a legit nonpriority legacy case re-distribution is attempted" do
+        let(:case_id) { legacy_case.bfkey }
+        let(:legacy_case) { legacy_nonpriority_cases.first }
+        let(:legacy_appeal) { create(:legacy_appeal, :with_schedule_hearing_tasks, vacols_case: legacy_case) }
 
-          # distributions after handling BACKLOG_LIMIT
-          remaining_in_backlog = total_tied_nonpriority_hearings -
-                                 dcs_legacy.where(priority: false, genpop: false).count
-          expect(subject.statistics["legacy_hearing_backlog_count"]).to eq remaining_in_backlog
+        before do
+          @raven_called = false
+          distribution = create(:distribution, judge: judge)
+          cancel_relevant_legacy_appeal_tasks
+          create_nonpriority_distributed_case(distribution, case_id, legacy_case.bfdloout)
+          distribution.update!(status: "completed", completed_at: today)
+          allow(Raven).to receive(:capture_exception) { @raven_called = true }
+        end
 
-          expect(dcs_legacy.where(priority: true, genpop_query: "not_genpop").count).to eq(2)
-          expect(dcs_legacy.where(priority: true, genpop_query: "any").count).to eq(2)
-          expect(dcs_legacy.count).to be >= 8
+        it "renames existing case_id and does not create a duplicate distributed_case" do
+          subject.distribute!
+          expect(subject.valid?).to eq(true)
+          expect(subject.error?).to eq(false)
+          expect(@raven_called).to eq(false)
+          expect(subject.distributed_cases.pluck(:case_id)).to include(case_id)
+          expect(DistributedCase.find_by(case_id: case_id)).to_not be_nil
+          expect(DistributedCase.find_by(case_id: original_distributed_case_id)).to_not be_nil
+          expect(legacy_case.reload.bfcurloc).to eq(judge.vacols_uniq_id)
         end
       end
-    end
-
-    def create_nonpriority_distributed_case(distribution, case_id, ready_at)
-      distribution.distributed_cases.create(
-        case_id: case_id,
-        priority: false,
-        docket: "legacy",
-        ready_at: VacolsHelper.normalize_vacols_datetime(ready_at),
-        docket_index: "123",
-        genpop: false,
-        genpop_query: "any"
-      )
-    end
 
-    def cancel_relevant_legacy_appeal_tasks
-      legacy_appeal.tasks.reject { |t| t.type == "RootTask" }.each do |task|
-        # update_columns to avoid triggers that will update VACOLS location dates and
-        # mess up ACD date logic.
-        task.update_columns(status: Constants.TASK_STATUSES.cancelled, closed_at: Time.zone.now)
+      def create_priority_distributed_case(distribution, case_id, ready_at)
+        distribution.distributed_cases.create(
+          case_id: case_id,
+          priority: true,
+          docket: "legacy",
+          ready_at: VacolsHelper.normalize_vacols_datetime(ready_at),
+          docket_index: "123",
+          genpop: false,
+          genpop_query: "any"
+        )
       end
-    end
 
-    context "when an illegit nonpriority legacy case re-distribution is attempted" do
-      let(:case_id) { legacy_case.bfkey }
-      let!(:previous_location) { legacy_case.bfcurloc }
-      let(:legacy_case) { legacy_nonpriority_cases.second }
+      context "when an illegit priority legacy case re-distribution is attempted" do
+        let(:case_id) { legacy_case.bfkey }
+        let(:legacy_case) { legacy_priority_cases.last }
 
-      before do
-        @raven_called = false
-        distribution = create(:distribution, judge: judge)
-        # illegit because appeal has completed hearing tasks
-        appeal = create(:legacy_appeal, :with_schedule_hearing_tasks, vacols_case: legacy_case)
-        appeal.tasks.open.where.not(type: RootTask.name).each(&:completed!)
-        create_nonpriority_distributed_case(distribution, case_id, legacy_case.bfdloout)
-        distribution.update!(status: "completed", completed_at: today)
-        allow(Raven).to receive(:capture_exception) { @raven_called = true }
-      end
+        before do
+          @raven_called = false
+          distribution = create(:distribution, judge: judge)
+          # illegit because appeal has open tasks
+          create(:legacy_appeal, :with_schedule_hearing_tasks, vacols_case: legacy_case)
+          create_priority_distributed_case(distribution, case_id, legacy_case.bfdloout)
+          distribution.update!(status: "completed", completed_at: today)
+          allow(Raven).to receive(:capture_exception) { @raven_called = true }
+        end
 
-      it "does not create a duplicate distributed_case and sends alert" do
-        subject.distribute!
-        expect(subject.valid?).to eq(true)
-        expect(subject.error?).to eq(false)
-        expect(@raven_called).to eq(true)
-        expect(subject.distributed_cases.pluck(:case_id)).to_not include(case_id)
-        expect(legacy_case.reload.bfcurloc).to eq(previous_location)
+        it "does not create a duplicate distributed_case and sends alert" do
+          subject.distribute!
+          expect(subject.valid?).to eq(true)
+          expect(subject.error?).to eq(false)
+          expect(@raven_called).to eq(false)
+          expect(subject.distributed_cases.pluck(:case_id)).to_not include(case_id)
+          expect(legacy_case.reload.bfcurloc).to eq(LegacyAppeal::LOCATION_CODES[:caseflow])
+        end
       end
-    end
 
-    context "when a legit nonpriority legacy case re-distribution is attempted" do
-      let(:case_id) { legacy_case.bfkey }
-      let(:legacy_case) { legacy_nonpriority_cases.first }
-      let(:legacy_appeal) { create(:legacy_appeal, :with_schedule_hearing_tasks, vacols_case: legacy_case) }
+      context "when a legit priority legacy case re-distribution is attempted" do
+        let(:case_id) { legacy_case.bfkey }
+        let(:legacy_case) { legacy_priority_cases.last }
+        let(:legacy_appeal) { create(:legacy_appeal, :with_schedule_hearing_tasks, vacols_case: legacy_case) }
 
-      before do
-        @raven_called = false
-        distribution = create(:distribution, judge: judge)
-        cancel_relevant_legacy_appeal_tasks
-        create_nonpriority_distributed_case(distribution, case_id, legacy_case.bfdloout)
-        distribution.update!(status: "completed", completed_at: today)
-        allow(Raven).to receive(:capture_exception) { @raven_called = true }
+        before do
+          @raven_called = false
+          distribution = create(:distribution, judge: judge)
+          cancel_relevant_legacy_appeal_tasks
+          create_priority_distributed_case(distribution, case_id, legacy_case.bfdloout)
+          distribution.update!(status: "completed", completed_at: today)
+          allow(Raven).to receive(:capture_exception) { @raven_called = true }
+        end
+
+        it "renames existing case_id and does not create a duplicate distributed_case" do
+          subject.distribute!
+          expect(subject.valid?).to eq(true)
+          expect(subject.error?).to eq(false)
+          expect(@raven_called).to eq(false)
+          expect(subject.distributed_cases.pluck(:case_id)).to include(case_id)
+          expect(DistributedCase.find_by(case_id: case_id)).to_not be_nil
+          expect(DistributedCase.find_by(case_id: original_distributed_case_id)).to_not be_nil
+          expect(legacy_case.reload.bfcurloc).to eq(judge.vacols_uniq_id)
+        end
       end
 
-      it "renames existing case_id and does not create a duplicate distributed_case" do
-        subject.distribute!
-        expect(subject.valid?).to eq(true)
-        expect(subject.error?).to eq(false)
-        expect(@raven_called).to eq(false)
-        expect(subject.distributed_cases.pluck(:case_id)).to include(case_id)
-        expect(DistributedCase.find_by(case_id: case_id)).to_not be_nil
-        expect(DistributedCase.find_by(case_id: original_distributed_case_id)).to_not be_nil
-        expect(legacy_case.reload.bfcurloc).to eq(judge.vacols_uniq_id)
+      context "when the job errors" do
+        it "marks the distribution as error" do
+          allow_any_instance_of(LegacyDocket).to receive(:distribute_nonpriority_appeals).and_raise(StandardError)
+          expect { subject.distribute! }.to raise_error(StandardError)
+          expect(subject.status).to eq("error")
+          expect(subject.distributed_cases.count).to eq(0)
+          expect(subject.errored_at).to eq(Time.zone.now)
+        end
       end
-    end
 
-    def create_priority_distributed_case(distribution, case_id, ready_at)
-      distribution.distributed_cases.create(
-        case_id: case_id,
-        priority: true,
-        docket: "legacy",
-        ready_at: VacolsHelper.normalize_vacols_datetime(ready_at),
-        docket_index: "123",
-        genpop: false,
-        genpop_query: "any"
-      )
-    end
+      context "when the judge has an empty team" do
+        let(:judge_wo_attorneys) { create(:user) }
+        let!(:vacols_judge_wo_attorneys) { create(:staff, :judge_role, sdomainid: judge_wo_attorneys.css_id) }
 
-    context "when an illegit priority legacy case re-distribution is attempted" do
-      let(:case_id) { legacy_case.bfkey }
-      let(:legacy_case) { legacy_priority_cases.last }
+        subject { Distribution.create(judge: judge_wo_attorneys) }
 
-      before do
-        @raven_called = false
-        distribution = create(:distribution, judge: judge)
-        # illegit because appeal has open tasks
-        create(:legacy_appeal, :with_schedule_hearing_tasks, vacols_case: legacy_case)
-        create_priority_distributed_case(distribution, case_id, legacy_case.bfdloout)
-        distribution.update!(status: "completed", completed_at: today)
-        allow(Raven).to receive(:capture_exception) { @raven_called = true }
+        it "uses the alternative batch size" do
+          subject.distribute!
+          expect(subject.valid?).to eq(true)
+          expect(subject.status).to eq("completed")
+          expect(subject.statistics["batch_size"]).to eq(15)
+          expect(subject.distributed_cases.count).to eq(15)
+        end
       end
 
-      it "does not create a duplicate distributed_case and sends alert" do
-        subject.distribute!
-        expect(subject.valid?).to eq(true)
-        expect(subject.error?).to eq(false)
-        expect(@raven_called).to eq(false)
-        expect(subject.distributed_cases.pluck(:case_id)).to_not include(case_id)
-        expect(legacy_case.reload.bfcurloc).to eq(LegacyAppeal::LOCATION_CODES[:caseflow])
+      context "when there are zero legacy cases eligible" do
+        let!(:legacy_priority_cases) { [] }
+        let!(:legacy_nonpriority_cases) { [] }
+        let!(:same_judge_nonpriority_hearings) { [] }
+        let!(:other_judge_hearings) { [] }
+
+        it "fills the AMA dockets" do
+          evidence_submission_cases[0...2].each do |appeal|
+            appeal.tasks
+              .find_by(type: EvidenceSubmissionWindowTask.name)
+              .update!(status: :completed)
+          end
+          subject.distribute!
+          expect(subject.valid?).to eq(true)
+          expect(subject.status).to eq("completed")
+          expect(subject.statistics["batch_size"]).to eq(15)
+          expect(subject.statistics["total_batch_size"]).to eq(45)
+          expect(subject.statistics["priority_count"]).to eq(1)
+          expect(subject.statistics["legacy_proportion"]).to eq(0.0)
+          expect(subject.statistics["direct_review_proportion"]).to be_within(0.01).of(0.13)
+          expect(subject.statistics["evidence_submission_proportion"]).to be_within(0.01).of(0.43)
+          expect(subject.statistics["hearing_proportion"]).to be_within(0.01).of(0.43)
+          expect(subject.statistics["pacesetting_direct_review_proportion"]).to eq(0.1)
+          expect(subject.statistics["interpolated_minimum_direct_review_proportion"]).to eq(0.067)
+          expect(subject.statistics["nonpriority_iterations"]).to be_between(2, 3)
+          expect(subject.distributed_cases.count).to eq(15)
+        end
       end
     end
 
-    context "when a legit priority legacy case re-distribution is attempted" do
-      let(:case_id) { legacy_case.bfkey }
-      let(:legacy_case) { legacy_priority_cases.last }
-      let(:legacy_appeal) { create(:legacy_appeal, :with_schedule_hearing_tasks, vacols_case: legacy_case) }
+    context "priority_push is true" do
+      subject { Distribution.create!(judge: judge, priority_push: true) }
+
+      let!(:legacy_priority_cases) do
+        (1..4).map do |i|
+          create(
+            :case,
+            :aod,
+            bfd19: 1.year.ago,
+            bfac: "1",
+            bfmpro: "ACT",
+            bfcurloc: "81",
+            bfdloout: i.months.ago,
+            folder: build(
+              :folder,
+              tinum: "1801#{format('%<index>03d', index: i)}",
+              titrnum: "123456789S"
+            )
+          )
+        end
+      end
 
-      before do
-        @raven_called = false
-        distribution = create(:distribution, judge: judge)
-        cancel_relevant_legacy_appeal_tasks
-        create_priority_distributed_case(distribution, case_id, legacy_case.bfdloout)
-        distribution.update!(status: "completed", completed_at: today)
-        allow(Raven).to receive(:capture_exception) { @raven_called = true }
+      let!(:legacy_nonpriority_cases) do
+        (5..8).map do |i|
+          create(
+            :case,
+            bfd19: 1.year.ago,
+            bfac: "1",
+            bfmpro: "ACT",
+            bfcurloc: "81",
+            bfdloout: i.months.ago,
+            folder: build(
+              :folder,
+              tinum: "1801#{format('%<index>03d', index: i)}",
+              titrnum: "123456789S"
+            )
+          )
+        end
       end
 
-      it "renames existing case_id and does not create a duplicate distributed_case" do
-        subject.distribute!
-        expect(subject.valid?).to eq(true)
-        expect(subject.error?).to eq(false)
-        expect(@raven_called).to eq(false)
-        expect(subject.distributed_cases.pluck(:case_id)).to include(case_id)
-        expect(DistributedCase.find_by(case_id: case_id)).to_not be_nil
-        expect(DistributedCase.find_by(case_id: original_distributed_case_id)).to_not be_nil
-        expect(legacy_case.reload.bfcurloc).to eq(judge.vacols_uniq_id)
+      let!(:priority_legacy_hearings_not_tied_to_judge) do
+        legacy_priority_cases[0..1].map do |appeal|
+          create(:case_hearing,
+                 :disposition_held,
+                 folder_nr: appeal.bfkey,
+                 hearing_date: 1.month.ago)
+        end
       end
-    end
 
-    context "when the job errors" do
-      it "marks the distribution as error" do
-        allow_any_instance_of(LegacyDocket).to receive(:distribute_nonpriority_appeals).and_raise(StandardError)
-        expect { subject.distribute! }.to raise_error(StandardError)
-        expect(subject.status).to eq("error")
-        expect(subject.distributed_cases.count).to eq(0)
-        expect(subject.errored_at).to eq(Time.zone.now)
+      let!(:priority_legacy_hearings_tied_to_judge) do
+        legacy_priority_cases[2..3].map do |appeal|
+          create(:case_hearing,
+                 :disposition_held,
+                 folder_nr: appeal.bfkey,
+                 hearing_date: 1.month.ago,
+                 board_member: judge.vacols_attorney_id)
+        end
       end
-    end
 
-    context "when the judge has an empty team" do
-      let(:judge_wo_attorneys) { create(:user) }
-      let!(:vacols_judge_wo_attorneys) { create(:staff, :judge_role, sdomainid: judge_wo_attorneys.css_id) }
+      let!(:priority_ama_hearings_tied_to_judge) do
+        (1...5).map do
+          appeal = create(:appeal,
+                          :ready_for_distribution,
+                          :advanced_on_docket_due_to_motion,
+                          docket_type: Constants.AMA_DOCKETS.hearing)
+          most_recent = create(:hearing_day, scheduled_for: 1.day.ago)
+          hearing = create(:hearing, judge: nil, disposition: "held", appeal: appeal, hearing_day: most_recent)
+          hearing.update(judge: judge)
+          appeal
+        end
+      end
 
-      subject { Distribution.create(judge: judge_wo_attorneys) }
+      let!(:priority_ama_hearings_not_tied_to_judge) do
+        (1...3).map do |i|
+          appeal = create(:appeal,
+                          :advanced_on_docket_due_to_age,
+                          :ready_for_distribution,
+                          docket_type: Constants.AMA_DOCKETS.hearing)
+          appeal.tasks.find_by(type: DistributionTask.name).update(assigned_at: i.months.ago)
+          appeal.reload
+        end
+      end
 
-      it "uses the alternative batch size" do
-        subject.distribute!
-        expect(subject.valid?).to eq(true)
-        expect(subject.status).to eq("completed")
-        expect(subject.statistics["batch_size"]).to eq(15)
-        expect(subject.distributed_cases.count).to eq(15)
+      let!(:priority_direct_review_cases) do
+        (1...3).map do |i|
+          appeal = create(:appeal,
+                          :with_post_intake_tasks,
+                          :advanced_on_docket_due_to_age,
+                          docket_type: Constants.AMA_DOCKETS.direct_review,
+                          receipt_date: 1.month.ago)
+          appeal.tasks.find_by(type: DistributionTask.name).update(assigned_at: i.month.ago)
+          appeal
+        end
       end
-    end
 
-    context "when there are zero legacy cases eligible" do
-      let!(:legacy_priority_cases) { [] }
-      let!(:legacy_nonpriority_cases) { [] }
-      let!(:same_judge_nonpriority_hearings) { [] }
-      let!(:other_judge_hearings) { [] }
+      let!(:evidence_submission_cases) do
+        (1...3).map do |i|
+          appeal = create(:appeal,
+                          :with_post_intake_tasks,
+                          :advanced_on_docket_due_to_age,
+                          docket_type: Constants.AMA_DOCKETS.evidence_submission)
+          appeal.tasks.find_by(type: EvidenceSubmissionWindowTask.name).completed!
+          appeal.tasks.find_by(type: DistributionTask.name).update(assigned_at: i.month.ago)
+          appeal
+        end
+      end
 
-      it "fills the AMA dockets" do
-        evidence_submission_cases[0...2].each do |appeal|
-          appeal.tasks
-            .find_by(type: EvidenceSubmissionWindowTask.name)
-            .update!(status: :completed)
+      context "when there is no limit specified" do
+        it "distributes all priority cases associated with the judge" do
+          distributed_case_ids = priority_legacy_hearings_tied_to_judge.map(&:folder_nr)
+            .concat(priority_ama_hearings_tied_to_judge.map(&:uuid))
+          subject.distribute!
+          expect(subject.valid?).to eq(true)
+          expect(subject.priority_push).to eq(true)
+          expect(subject.status).to eq("completed")
+          expect(subject.started_at).to eq(Time.zone.now) # time is frozen so appears zero time elapsed
+          expect(subject.errored_at).to be_nil
+          expect(subject.completed_at).to eq(Time.zone.now)
+          expect(subject.statistics["batch_size"]).to eq(distributed_case_ids.count)
+          expect(subject.distributed_cases.count).to eq(distributed_case_ids.count)
+          expect(subject.distributed_cases.where(priority: true).count).to eq(distributed_case_ids.count)
+          expect(subject.distributed_cases.where(priority: false).count).to eq(0)
+          expect(subject.distributed_cases.where(genpop_query: "not_genpop").count).to eq(distributed_case_ids.count)
+          expect(subject.distributed_cases.where(genpop: true).count).to eq(0)
+          expect(subject.distributed_cases.where(docket: "legacy").count).to eq(
+            priority_legacy_hearings_tied_to_judge.count
+          )
+          expect(subject.distributed_cases.where(docket: Constants.AMA_DOCKETS.hearing).count).to eq(
+            priority_ama_hearings_tied_to_judge.count
+          )
+          expect(subject.distributed_cases.where(docket: Constants.AMA_DOCKETS.direct_review).count).to eq 0
+          expect(subject.distributed_cases.where(docket: Constants.AMA_DOCKETS.evidence_submission).count).to eq 0
+          expect(subject.distributed_cases.pluck(:case_id)).to match_array distributed_case_ids
+        end
+      end
+
+      context "when a limit is specified" do
+        let(:limit) { 4 }
+
+        it "distributes that number of priority cases from all dockets, based on docket age" do
+          oldest_case_from_each_docket = [
+            legacy_priority_cases.last.bfkey,
+            priority_ama_hearings_not_tied_to_judge.last.uuid,
+            priority_direct_review_cases.last.uuid,
+            evidence_submission_cases.last.uuid
+          ]
+
+          subject.distribute!(limit)
+          expect(subject.valid?).to eq(true)
+          expect(subject.priority_push).to eq(true)
+          expect(subject.status).to eq("completed")
+          expect(subject.started_at).to eq(Time.zone.now) # time is frozen so appears zero time elapsed
+          expect(subject.errored_at).to be_nil
+          expect(subject.completed_at).to eq(Time.zone.now)
+          expect(subject.statistics["batch_size"]).to eq(limit)
+          expect(subject.distributed_cases.count).to eq(limit)
+          expect(subject.distributed_cases.where(priority: true).count).to eq(limit)
+          expect(subject.distributed_cases.where(priority: false).count).to eq(0)
+          expect(subject.distributed_cases.where(genpop_query: "not_genpop").count).to eq(0)
+          expect(subject.distributed_cases.where(docket: "legacy").count).to eq(1)
+          expect(subject.distributed_cases.where(docket: Constants.AMA_DOCKETS.hearing).count).to eq(1)
+          expect(subject.distributed_cases.where(docket: Constants.AMA_DOCKETS.direct_review).count).to eq(1)
+          expect(subject.distributed_cases.where(docket: Constants.AMA_DOCKETS.evidence_submission).count).to eq(1)
+          expect(subject.distributed_cases.pluck(:case_id)).to match_array oldest_case_from_each_docket
         end
-        subject.distribute!
-        expect(subject.valid?).to eq(true)
-        expect(subject.status).to eq("completed")
-        expect(subject.statistics["batch_size"]).to eq(15)
-        expect(subject.statistics["total_batch_size"]).to eq(45)
-        expect(subject.statistics["priority_count"]).to eq(1)
-        expect(subject.statistics["legacy_proportion"]).to eq(0.0)
-        expect(subject.statistics["direct_review_proportion"]).to be_within(0.01).of(0.13)
-        expect(subject.statistics["evidence_submission_proportion"]).to be_within(0.01).of(0.43)
-        expect(subject.statistics["hearing_proportion"]).to be_within(0.01).of(0.43)
-        expect(subject.statistics["pacesetting_direct_review_proportion"]).to eq(0.1)
-        expect(subject.statistics["interpolated_minimum_direct_review_proportion"]).to eq(0.067)
-        expect(subject.statistics["nonpriority_iterations"]).to be_between(2, 3)
-        expect(subject.distributed_cases.count).to eq(15)
       end
     end
   end
diff --git a/spec/models/dockets/hearing_request_docket_spec.rb b/spec/models/dockets/hearing_request_docket_spec.rb
index 7769224df0f..3dc144aa4dd 100644
--- a/spec/models/dockets/hearing_request_docket_spec.rb
+++ b/spec/models/dockets/hearing_request_docket_spec.rb
@@ -92,9 +92,11 @@
     end
 
     context "priority appeals and genpop 'any'" do
+      let(:limit) { 10 }
+
       subject do
         HearingRequestDocket.new.distribute_appeals(
-          distribution, priority: true, limit: 10, genpop: "any"
+          distribution, priority: true, limit: limit, genpop: "any"
         )
       end
 
@@ -117,6 +119,26 @@
         expect(distribution.distributed_cases.length).to eq(2)
         expect(distribution_judge.reload.tasks.map(&:appeal)).to match_array([tied, not_tied])
       end
+
+      context "when the limit is one" do
+        let(:limit) { 1 }
+
+        it "only distributes priority, distributable, hearing docket cases
+          that are either genpop or not genpop" do
+          not_tied = create_priority_distributable_hearing_appeal_not_tied_to_any_judge
+          not_tied.tasks.find_by(type: DistributionTask.name).update(assigned_at: 1.month.ago)
+          not_tied.reload
+          matching_all_base_conditions_with_most_recent_held_hearing_tied_to_distribution_judge
+          tasks = subject
+
+          expect(tasks.length).to eq(1)
+          expect(tasks.first.class).to eq(DistributedCase)
+          expect(tasks.first.genpop).to eq true
+          expect(tasks.first.genpop_query).to eq "any"
+          expect(distribution.distributed_cases.length).to eq(1)
+          expect(distribution_judge.reload.tasks.map(&:appeal)).to match_array([not_tied])
+        end
+      end
     end
 
     context "nonpriority appeals and genpop 'any'" do
diff --git a/spec/models/vacols/case_docket_spec.rb b/spec/models/vacols/case_docket_spec.rb
index ec06dc986f9..d2d23b439c8 100644
--- a/spec/models/vacols/case_docket_spec.rb
+++ b/spec/models/vacols/case_docket_spec.rb
@@ -468,6 +468,7 @@
     context "when a case is tied to a judge by a hearing on a prior appeal" do
       let(:original_docket_number) { aod_ready_case_docket_number }
       let(:hearing_judge) { judge.vacols_attorney_id }
+      let(:another_hearing_judge) { another_judge.vacols_attorney_id }
       let!(:hearing) do
         create(:case_hearing,
                :disposition_held,
@@ -481,7 +482,7 @@
                :disposition_held,
                folder_nr: postcavc_ready_case.bfkey,
                hearing_date: 5.days.ago.to_date,
-               board_member: another_judge.vacols_attorney_id)
+               board_member: another_hearing_judge)
       end
 
       context "when genpop is no" do
@@ -491,6 +492,17 @@
           expect(aod_ready_case.reload.bfcurloc).to eq(judge.vacols_uniq_id)
           expect(postcavc_ready_case.reload.bfcurloc).to eq("83")
         end
+
+        context "when limit is nil" do
+          let(:limit) { nil }
+          let(:another_hearing_judge) { judge.vacols_attorney_id }
+
+          it "distributes all cases tied to the judge" do
+            expect(subject.count).to eq(2)
+            expect(aod_ready_case.reload.bfcurloc).to eq(judge.vacols_uniq_id)
+            expect(postcavc_ready_case.reload.bfcurloc).to eq(judge.vacols_uniq_id)
+          end
+        end
       end
 
       context "when genpop is any" do
diff --git a/spec/seeds/users_spec.rb b/spec/seeds/users_spec.rb
index 47fbcd5d071..8d205993b35 100644
--- a/spec/seeds/users_spec.rb
+++ b/spec/seeds/users_spec.rb
@@ -6,8 +6,8 @@
 
     it "creates all kinds of users and organizations" do
       expect { subject }.to_not raise_error
-      expect(User.count).to eq(88)
-      expect(Organization.count).to eq(26)
+      expect(User.count).to eq(78)
+      expect(Organization.count).to eq(27)
     end
   end
 end