diff --git a/app/jobs/auto_assignment/assignment_job.rb b/app/jobs/auto_assignment/assignment_job.rb
new file mode 100644
index 000000000..9c6760ecc
--- /dev/null
+++ b/app/jobs/auto_assignment/assignment_job.rb
@@ -0,0 +1,22 @@
+class AutoAssignment::AssignmentJob < ApplicationJob
+  queue_as :default
+
+  def perform(inbox_id:)
+    inbox = Inbox.find_by(id: inbox_id)
+    return unless inbox
+
+    service = AutoAssignment::AssignmentService.new(inbox: inbox)
+
+    assigned_count = service.perform_bulk_assignment(limit: bulk_assignment_limit)
+    Rails.logger.info "Assigned #{assigned_count} conversations for inbox #{inbox.id}"
+  rescue StandardError => e
+    Rails.logger.error "Bulk assignment failed for inbox #{inbox_id}: #{e.message}"
+    raise e if Rails.env.test?
+  end
+
+  private
+
+  def bulk_assignment_limit
+    ENV.fetch('AUTO_ASSIGNMENT_BULK_LIMIT', 100).to_i
+  end
+end
diff --git a/app/jobs/auto_assignment/periodic_assignment_job.rb b/app/jobs/auto_assignment/periodic_assignment_job.rb
new file mode 100644
index 000000000..63500507e
--- /dev/null
+++ b/app/jobs/auto_assignment/periodic_assignment_job.rb
@@ -0,0 +1,19 @@
+class AutoAssignment::PeriodicAssignmentJob < ApplicationJob
+  queue_as :scheduled_jobs
+
+  def perform
+    Account.find_in_batches do |accounts|
+      accounts.each do |account|
+        next unless account.feature_enabled?('assignment_v2')
+
+        account.inboxes.joins(:assignment_policy).find_in_batches do |inboxes|
+          inboxes.each do |inbox|
+            next unless inbox.auto_assignment_v2_enabled?
+
+            AutoAssignment::AssignmentJob.perform_later(inbox_id: inbox.id)
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/app/models/concerns/auto_assignment_handler.rb b/app/models/concerns/auto_assignment_handler.rb
index de5fdae7d..a1198200a 100644
--- a/app/models/concerns/auto_assignment_handler.rb
+++ b/app/models/concerns/auto_assignment_handler.rb
@@ -14,7 +14,13 @@ module AutoAssignmentHandler
     return unless conversation_status_changed_to_open?
     return unless should_run_auto_assignment?
 
-    ::AutoAssignment::AgentAssignmentService.new(conversation: self, allowed_agent_ids: inbox.member_ids_with_assignment_capacity).perform
+    if inbox.auto_assignment_v2_enabled?
+      # Use new assignment system
+      AutoAssignment::AssignmentJob.perform_later(inbox_id: inbox.id)
+    else
+      # Use legacy assignment system
+      AutoAssignment::AgentAssignmentService.new(conversation: self, allowed_agent_ids: inbox.member_ids_with_assignment_capacity).perform
+    end
   end
 
   def should_run_auto_assignment?
diff --git a/app/models/concerns/inbox_agent_availability.rb b/app/models/concerns/inbox_agent_availability.rb
new file mode 100644
index 000000000..289da7db9
--- /dev/null
+++ b/app/models/concerns/inbox_agent_availability.rb
@@ -0,0 +1,28 @@
+module InboxAgentAvailability
+  extend ActiveSupport::Concern
+
+  def available_agents
+    online_agent_ids = fetch_online_agent_ids
+    return inbox_members.none if online_agent_ids.empty?
+
+    inbox_members
+      .joins(:user)
+      .where(users: { id: online_agent_ids })
+      .includes(:user)
+  end
+
+  def member_ids_with_assignment_capacity
+    member_ids
+  end
+
+  private
+
+  def fetch_online_agent_ids
+    OnlineStatusTracker.get_available_users(account_id)
+                       .select { |_key, value| value.eql?('online') }
+                       .keys
+                       .map(&:to_i)
+  end
+end
+
+InboxAgentAvailability.prepend_mod_with('InboxAgentAvailability')
diff --git a/app/models/inbox.rb b/app/models/inbox.rb
index b0f13d222..894c57bcc 100644
--- a/app/models/inbox.rb
+++ b/app/models/inbox.rb
@@ -44,6 +44,7 @@ class Inbox < ApplicationRecord
   include Avatarable
   include OutOfOffisable
   include AccountCacheRevalidator
+  include InboxAgentAvailability
 
   # Not allowing characters:
   validates :name, presence: true
@@ -190,6 +191,10 @@ class Inbox < ApplicationRecord
     members.ids
   end
 
+  def auto_assignment_v2_enabled?
+    account.feature_enabled?('assignment_v2')
+  end
+
   private
 
   def default_name_for_blank_name
diff --git a/app/services/auto_assignment/assignment_service.rb b/app/services/auto_assignment/assignment_service.rb
new file mode 100644
index 000000000..59e7b5a7b
--- /dev/null
+++ b/app/services/auto_assignment/assignment_service.rb
@@ -0,0 +1,90 @@
+class AutoAssignment::AssignmentService
+  pattr_initialize [:inbox!]
+
+  def perform_bulk_assignment(limit: 100)
+    return 0 unless inbox.auto_assignment_v2_enabled?
+
+    assigned_count = 0
+
+    unassigned_conversations(limit).each do |conversation|
+      assigned_count += 1 if perform_for_conversation(conversation)
+    end
+
+    assigned_count
+  end
+
+  private
+
+  def perform_for_conversation(conversation)
+    return false unless assignable?(conversation)
+
+    agent = find_available_agent
+    return false unless agent
+
+    assign_conversation(conversation, agent)
+  end
+
+  def assignable?(conversation)
+    conversation.status == 'open' &&
+      conversation.assignee_id.nil?
+  end
+
+  def unassigned_conversations(limit)
+    scope = inbox.conversations.unassigned.open
+
+    scope = if assignment_config['conversation_priority'].to_s == 'longest_waiting'
+              scope.reorder(last_activity_at: :asc, created_at: :asc)
+            else
+              scope.reorder(created_at: :asc)
+            end
+
+    scope.limit(limit)
+  end
+
+  def find_available_agent
+    agents = filter_agents_by_rate_limit(inbox.available_agents)
+    return nil if agents.empty?
+
+    round_robin_selector.select_agent(agents)
+  end
+
+  def filter_agents_by_rate_limit(agents)
+    agents.select do |agent_member|
+      rate_limiter = build_rate_limiter(agent_member.user)
+      rate_limiter.within_limit?
+    end
+  end
+
+  def assign_conversation(conversation, agent)
+    conversation.update!(assignee: agent)
+
+    rate_limiter = build_rate_limiter(agent)
+    rate_limiter.track_assignment(conversation)
+
+    dispatch_assignment_event(conversation, agent)
+    true
+  end
+
+  def dispatch_assignment_event(conversation, agent)
+    Rails.configuration.dispatcher.dispatch(
+      Events::Types::ASSIGNEE_CHANGED,
+      Time.zone.now,
+      conversation: conversation,
+      user: agent
+    )
+  end
+
+  def build_rate_limiter(agent)
+    AutoAssignment::RateLimiter.new(inbox: inbox, agent: agent)
+  end
+
+  def round_robin_selector
+    @round_robin_selector ||= AutoAssignment::RoundRobinSelector.new(inbox: inbox)
+  end
+
+  def assignment_config
+    @assignment_config ||= inbox.auto_assignment_config || {}
+  end
+end
+
+AutoAssignment::AssignmentService.prepend_mod_with('AutoAssignment::AssignmentService')
diff --git a/app/services/auto_assignment/rate_limiter.rb b/app/services/auto_assignment/rate_limiter.rb
new file mode 100644
index 000000000..d22f1f9b7
--- /dev/null
+++ b/app/services/auto_assignment/rate_limiter.rb
@@ -0,0 +1,49 @@
+class AutoAssignment::RateLimiter
+  pattr_initialize [:inbox!, :agent!]
+
+  def within_limit?
+    return true unless enabled?
+
+    current_count < limit
+  end
+
+  def track_assignment(conversation)
+    return unless enabled?
+
+    assignment_key = build_assignment_key(conversation.id)
+    Redis::Alfred.set(assignment_key, conversation.id.to_s, ex: window)
+  end
+
+  def current_count
+    return 0 unless enabled?
+
+    pattern = assignment_key_pattern
+    Redis::Alfred.keys_count(pattern)
+  end
+
+  private
+
+  def enabled?
+    limit.present? && limit.positive?
+  end
+
+  def limit
+    config&.fair_distribution_limit&.to_i || Math
+  end
+
+  def window
+    config&.fair_distribution_window&.to_i || 24.hours.to_i
+  end
+
+  def config
+    @config ||= inbox.assignment_policy
+  end
+
+  def assignment_key_pattern
+    format(Redis::RedisKeys::ASSIGNMENT_KEY_PATTERN, inbox_id: inbox.id, agent_id: agent.id)
+  end
+
+  def build_assignment_key(conversation_id)
+    format(Redis::RedisKeys::ASSIGNMENT_KEY, inbox_id: inbox.id, agent_id: agent.id, conversation_id: conversation_id)
+  end
+end
diff --git a/app/services/auto_assignment/round_robin_selector.rb b/app/services/auto_assignment/round_robin_selector.rb
new file mode 100644
index 000000000..65e72090d
--- /dev/null
+++ b/app/services/auto_assignment/round_robin_selector.rb
@@ -0,0 +1,16 @@
+class AutoAssignment::RoundRobinSelector
+  pattr_initialize [:inbox!]
+
+  def select_agent(available_agents)
+    return nil if available_agents.empty?
+
+    agent_user_ids = available_agents.map(&:user_id).map(&:to_s)
+    round_robin_service.available_agent(allowed_agent_ids: agent_user_ids)
+  end
+
+  private
+
+  def round_robin_service
+    @round_robin_service ||= AutoAssignment::InboxRoundRobinService.new(inbox: inbox)
+  end
+end
diff --git a/config/schedule.yml b/config/schedule.yml
index b5b21b4ce..c039719fa 100644
--- a/config/schedule.yml
+++ b/config/schedule.yml
@@ -53,3 +53,9 @@ bulk_auto_assignment_job:
   cron: '*/15 * * * *'
   class: 'Inboxes::BulkAutoAssignmentJob'
   queue: scheduled_jobs
+
+# executed every 30 minutes for assignment_v2
+periodic_assignment_job:
+  cron: '*/30 * * * *'
+  class: 'AutoAssignment::PeriodicAssignmentJob'
+  queue: scheduled_jobs
diff --git a/enterprise/app/models/enterprise/concerns/inbox.rb b/enterprise/app/models/enterprise/concerns/inbox.rb
index 6f450bed1..0de61db23 100644
--- a/enterprise/app/models/enterprise/concerns/inbox.rb
+++ b/enterprise/app/models/enterprise/concerns/inbox.rb
@@ -6,5 +6,6 @@ module Enterprise::Concerns::Inbox
     has_one :captain_assistant,
             through: :captain_inbox,
             class_name: 'Captain::Assistant'
+    has_many :inbox_capacity_limits, dependent: :destroy
   end
 end
diff --git a/enterprise/app/models/enterprise/inbox_agent_availability.rb b/enterprise/app/models/enterprise/inbox_agent_availability.rb
new file mode 100644
index 000000000..886b55278
--- /dev/null
+++ b/enterprise/app/models/enterprise/inbox_agent_availability.rb
@@ -0,0 +1,31 @@
+module Enterprise::InboxAgentAvailability
+  extend ActiveSupport::Concern
+
+  def member_ids_with_assignment_capacity
+    return member_ids unless capacity_filtering_enabled?
+
+    # Get online agents with capacity
+    agents = available_agents
+    agents = filter_by_capacity(agents)
+    agents.map(&:user_id)
+  end
+
+  private
+
+  def filter_by_capacity(inbox_members_scope)
+    return inbox_members_scope unless capacity_filtering_enabled?
+
+    inbox_members_scope.select do |inbox_member|
+      capacity_service.agent_has_capacity?(inbox_member.user, self)
+    end
+  end
+
+  def capacity_filtering_enabled?
+    account.feature_enabled?('assignment_v2') &&
+      account.account_users.joins(:agent_capacity_policy).exists?
+  end
+
+  def capacity_service
+    @capacity_service ||= Enterprise::AutoAssignment::CapacityService.new
+  end
+end
diff --git a/enterprise/app/services/enterprise/auto_assignment/assignment_service.rb b/enterprise/app/services/enterprise/auto_assignment/assignment_service.rb
new file mode 100644
index 000000000..4f3da08ee
--- /dev/null
+++ b/enterprise/app/services/enterprise/auto_assignment/assignment_service.rb
@@ -0,0 +1,94 @@
+module Enterprise::AutoAssignment::AssignmentService
+  private
+
+  # Override assignment config to use policy if available
+  def assignment_config
+    return super unless policy
+
+    {
+      'conversation_priority' => policy.conversation_priority,
+      'fair_distribution_limit' => policy.fair_distribution_limit,
+      'fair_distribution_window' => policy.fair_distribution_window,
+      'balanced' => policy.balanced?
+    }.compact
+  end
+
+  # Extend agent finding to add capacity checks
+  def find_available_agent
+    agents = filter_agents_by_rate_limit(inbox.available_agents)
+    agents = filter_agents_by_capacity(agents) if capacity_filtering_enabled?
+    return nil if agents.empty?
+
+    selector = policy&.balanced? ? balanced_selector : round_robin_selector
+    selector.select_agent(agents)
+  end
+
+  def filter_agents_by_capacity(agents)
+    return agents unless capacity_filtering_enabled?
+
+    capacity_service = Enterprise::AutoAssignment::CapacityService.new
+    agents.select { |agent_member| capacity_service.agent_has_capacity?(agent_member.user, inbox) }
+  end
+
+  def capacity_filtering_enabled?
+    account.feature_enabled?('assignment_v2') &&
+      account.account_users.joins(:agent_capacity_policy).exists?
+  end
+
+  def round_robin_selector
+    @round_robin_selector ||= AutoAssignment::RoundRobinSelector.new(inbox: inbox)
+  end
+
+  def balanced_selector
+    @balanced_selector ||= Enterprise::AutoAssignment::BalancedSelector.new(inbox: inbox)
+  end
+
+  def policy
+    @policy ||= inbox.assignment_policy
+  end
+
+  def account
+    inbox.account
+  end
+
+  # Override to apply exclusion rules
+  def unassigned_conversations(limit)
+    scope = inbox.conversations.unassigned.open
+
+    # Apply exclusion rules from capacity policy or assignment policy
+    scope = apply_exclusion_rules(scope)
+
+    # Apply conversation priority using enum methods if policy exists
+    scope = if policy&.longest_waiting?
+              scope.reorder(last_activity_at: :asc, created_at: :asc)
+            else
+              scope.reorder(created_at: :asc)
+            end
+
+    scope.limit(limit)
+  end
+
+  def apply_exclusion_rules(scope)
+    capacity_policy = inbox.inbox_capacity_limits.first&.agent_capacity_policy
+    return scope unless capacity_policy
+
+    exclusion_rules = capacity_policy.exclusion_rules || {}
+    scope = apply_label_exclusions(scope, exclusion_rules['excluded_labels'])
+    apply_age_exclusions(scope, exclusion_rules['exclude_older_than_hours'])
+  end
+
+  def apply_label_exclusions(scope, excluded_labels)
+    return scope if excluded_labels.blank?
+
+    scope.tagged_with(excluded_labels, exclude: true, on: :labels)
+  end
+
+  def apply_age_exclusions(scope, hours_threshold)
+    return scope if hours_threshold.blank?
+
+    hours = hours_threshold.to_i
+    return scope unless hours.positive?
+
+    scope.where('conversations.created_at >= ?', hours.hours.ago)
+  end
+end
diff --git a/enterprise/app/services/enterprise/auto_assignment/balanced_selector.rb b/enterprise/app/services/enterprise/auto_assignment/balanced_selector.rb
new file mode 100644
index 000000000..57017b6bb
--- /dev/null
+++ b/enterprise/app/services/enterprise/auto_assignment/balanced_selector.rb
@@ -0,0 +1,26 @@
+class Enterprise::AutoAssignment::BalancedSelector
+  pattr_initialize [:inbox!]
+
+  def select_agent(available_agents)
+    return nil if available_agents.empty?
+
+    agent_users = available_agents.map(&:user)
+    assignment_counts = fetch_assignment_counts(agent_users)
+
+    agent_users.min_by { |user| assignment_counts[user.id] || 0 }
+  end
+
+  private
+
+  def fetch_assignment_counts(users)
+    user_ids = users.map(&:id)
+
+    counts = inbox.conversations
+                  .open
+                  .where(assignee_id: user_ids)
+                  .group(:assignee_id)
+                  .count
+
+    Hash.new(0).merge(counts)
+  end
+end
diff --git a/enterprise/app/services/enterprise/auto_assignment/capacity_service.rb b/enterprise/app/services/enterprise/auto_assignment/capacity_service.rb
new file mode 100644
index 000000000..f82da9267
--- /dev/null
+++ b/enterprise/app/services/enterprise/auto_assignment/capacity_service.rb
@@ -0,0 +1,25 @@
+class Enterprise::AutoAssignment::CapacityService
+  def agent_has_capacity?(user, inbox)
+    # Get the account_user for this specific account
+    account_user = user.account_users.find_by(account: inbox.account)
+
+    # If no account_user or no capacity policy, agent has unlimited capacity
+    return true unless account_user&.agent_capacity_policy
+
+    policy = account_user.agent_capacity_policy
+
+    # Check if there's a specific limit for this inbox
+    inbox_limit = policy.inbox_capacity_limits.find_by(inbox: inbox)
+
+    # If no specific limit for this inbox, agent has unlimited capacity for this inbox
+    return true unless inbox_limit
+
+    # Count current open conversations for this agent in this inbox
+    current_count = user.assigned_conversations
+                        .where(inbox: inbox, status: :open)
+                        .count
+
+    # Agent has capacity if current count is below the limit
+    current_count < inbox_limit.conversation_limit
+  end
+end
diff --git a/lib/redis/alfred.rb b/lib/redis/alfred.rb
index b3e156420..e7e04ed60 100644
--- a/lib/redis/alfred.rb
+++ b/lib/redis/alfred.rb
@@ -35,6 +35,25 @@ module Redis::Alfred
       $alfred.with { |conn| conn.exists?(key) }
     end
 
+    # set expiry on a key in seconds
+    def expire(key, seconds)
+      $alfred.with { |conn| conn.expire(key, seconds) }
+    end
+
+    # scan keys matching a pattern
+    def scan_each(match: nil, count: 100, &)
+      $alfred.with do |conn|
+        conn.scan_each(match: match, count: count, &)
+      end
+    end
+
+    # count keys matching a pattern
+    def keys_count(pattern)
+      count = 0
+      scan_each(match: pattern) { count += 1 }
+      count
+    end
+
     # list operations
 
     def llen(key)
@@ -81,8 +100,15 @@ module Redis::Alfred
     # sorted set operations
 
     # add score and value for a key
-    def zadd(key, score, value)
-      $alfred.with { |conn| conn.zadd(key, score, value) }
+    # Modern Redis syntax: zadd(key, [[score, member], ...])
+    def zadd(key, score, value = nil)
+      if value.nil? && score.is_a?(Array)
+        # New syntax: score is actually an array of [score, member] pairs
+        $alfred.with { |conn| conn.zadd(key, score) }
+      else
+        # Support old syntax for backward compatibility
+        $alfred.with { |conn| conn.zadd(key, [[score, value]]) }
+      end
     end
 
     # get score of a value for key
@@ -90,9 +116,22 @@ module Redis::Alfred
       $alfred.with { |conn| conn.zscore(key, value) }
     end
 
+    # count members in a sorted set with scores within the given range
+    def zcount(key, min_score, max_score)
+      $alfred.with { |conn| conn.zcount(key, min_score, max_score) }
+    end
+
+    # get the number of members in a sorted set
+    def zcard(key)
+      $alfred.with { |conn| conn.zcard(key) }
+    end
+
     # get values by score
-    def zrangebyscore(key, range_start, range_end)
-      $alfred.with { |conn| conn.zrangebyscore(key, range_start, range_end) }
+    def zrangebyscore(key, range_start, range_end, with_scores: false, limit: nil)
+      options = {}
+      options[:with_scores] = with_scores if with_scores
+      options[:limit] = limit if limit
+      $alfred.with { |conn| conn.zrangebyscore(key, range_start, range_end, **options) }
     end
 
     # remove values by score
diff --git a/lib/redis/redis_keys.rb b/lib/redis/redis_keys.rb
index fa64d1043..f5d297e5f 100644
--- a/lib/redis/redis_keys.rb
+++ b/lib/redis/redis_keys.rb
@@ -42,4 +42,9 @@ module Redis::RedisKeys
   SLACK_MESSAGE_MUTEX = 'SLACK_MESSAGE_LOCK::%<conversation_id>s::%<reference_id>s'.freeze
   EMAIL_MESSAGE_MUTEX = 'EMAIL_CHANNEL_LOCK::%<inbox_id>s'.freeze
   CRM_PROCESS_MUTEX = 'CRM_PROCESS_MUTEX::%<hook_id>s'.freeze
+
+  ## Auto Assignment Keys
+  # Track conversation assignments to agents for rate limiting
+  ASSIGNMENT_KEY = 'ASSIGNMENT::%<inbox_id>d::AGENT::%<agent_id>d::CONVERSATION::%<conversation_id>d'.freeze
+  ASSIGNMENT_KEY_PATTERN = 'ASSIGNMENT::%<inbox_id>d::AGENT::%<agent_id>d::*'.freeze
 end
diff --git a/spec/enterprise/services/enterprise/auto_assignment/assignment_service_spec.rb b/spec/enterprise/services/enterprise/auto_assignment/assignment_service_spec.rb
new file mode 100644
index 000000000..432f397be
--- /dev/null
+++ b/spec/enterprise/services/enterprise/auto_assignment/assignment_service_spec.rb
@@ -0,0 +1,185 @@
+require 'rails_helper'
+
+RSpec.describe Enterprise::AutoAssignment::AssignmentService, type: :service do
+  let(:account) { create(:account) }
+  let(:assignment_policy) { create(:assignment_policy, account: account, enabled: true) }
+  let(:inbox) { create(:inbox, account: account) }
+  let(:agent1) { create(:user, account: account, name: 'Agent 1') }
+  let(:agent2) { create(:user, account: account, name: 'Agent 2') }
+  let(:assignment_service) { AutoAssignment::AssignmentService.new(inbox: inbox) }
+
+  before do
+    # Create inbox members
+    create(:inbox_member, inbox: inbox, user: agent1)
+    create(:inbox_member, inbox: inbox, user: agent2)
+
+    # Link inbox to assignment policy
+    create(:inbox_assignment_policy, inbox: inbox, assignment_policy: assignment_policy)
+
+    allow(account).to receive(:feature_enabled?).and_return(false)
+    allow(account).to receive(:feature_enabled?).with('assignment_v2').and_return(true)
+
+    # Set agents as online
+    OnlineStatusTracker.update_presence(account.id, 'User', agent1.id)
+    OnlineStatusTracker.set_status(account.id, agent1.id, 'online')
+    OnlineStatusTracker.update_presence(account.id, 'User', agent2.id)
+    OnlineStatusTracker.set_status(account.id, agent2.id, 'online')
+  end
+
+  describe 'exclusion rules' do
+    let(:capacity_policy) { create(:agent_capacity_policy, account: account) }
+    let(:label1) { create(:label, account: account, title: 'high-priority') }
+    let(:label2) { create(:label, account: account, title: 'vip') }
+
+    before do
+      create(:inbox_capacity_limit, inbox: inbox, agent_capacity_policy: capacity_policy, conversation_limit: 10)
+      inbox.enable_auto_assignment = true
+      inbox.save!
+    end
+
+    context 'when excluding conversations by label' do
+      let!(:conversation_with_label) { create(:conversation, inbox: inbox, assignee: nil) }
+      let!(:conversation_without_label) { create(:conversation, inbox: inbox, assignee: nil) }
+
+      before do
+        conversation_with_label.update_labels([label1.title])
+
+        capacity_policy.update!(exclusion_rules: {
+                                  'excluded_labels' => [label1.title]
+                                })
+      end
+
+      it 'excludes conversations with specified labels' do
+        # First check conversations are unassigned
+        expect(conversation_with_label.assignee).to be_nil
+        expect(conversation_without_label.assignee).to be_nil
+
+        # Run bulk assignment
+        assigned_count = assignment_service.perform_bulk_assignment(limit: 10)
+
+        # Only the conversation without label should be assigned
+        expect(assigned_count).to eq(1)
+        expect(conversation_with_label.reload.assignee).to be_nil
+        expect(conversation_without_label.reload.assignee).to be_present
+      end
+
+      it 'handles bulk assignment correctly' do
+        assigned_count = assignment_service.perform_bulk_assignment(limit: 10)
+
+        # Only 1 conversation should be assigned (the one without label)
+        expect(assigned_count).to eq(1)
+        expect(conversation_with_label.reload.assignee).to be_nil
+        expect(conversation_without_label.reload.assignee).to be_present
+      end
+
+      it 'excludes conversations with multiple labels' do
+        conversation_without_label.update_labels([label2.title])
+
+        capacity_policy.update!(exclusion_rules: {
+                                  'excluded_labels' => [label1.title, label2.title]
+                                })
+
+        assigned_count = assignment_service.perform_bulk_assignment(limit: 10)
+
+        # Both conversations should be excluded
+        expect(assigned_count).to eq(0)
+        expect(conversation_with_label.reload.assignee).to be_nil
+        expect(conversation_without_label.reload.assignee).to be_nil
+      end
+    end
+
+    context 'when excluding conversations by age' do
+      let!(:old_conversation) { create(:conversation, inbox: inbox, assignee: nil, created_at: 25.hours.ago) }
+      let!(:recent_conversation) { create(:conversation, inbox: inbox, assignee: nil, created_at: 1.hour.ago) }
+
+      before do
+        capacity_policy.update!(exclusion_rules: {
+                                  'exclude_older_than_hours' => 24
+                                })
+      end
+
+      it 'excludes conversations older than specified hours' do
+        assigned_count = assignment_service.perform_bulk_assignment(limit: 10)
+
+        # Only recent conversation should be assigned
+        expect(assigned_count).to eq(1)
+        expect(old_conversation.reload.assignee).to be_nil
+        expect(recent_conversation.reload.assignee).to be_present
+      end
+
+      it 'handles different time thresholds' do
+        capacity_policy.update!(exclusion_rules: {
+                                  'exclude_older_than_hours' => 2
+                                })
+
+        assigned_count = assignment_service.perform_bulk_assignment(limit: 10)
+
+        # Only conversation created within 2 hours should be assigned
+        expect(assigned_count).to eq(1)
+        expect(recent_conversation.reload.assignee).to be_present
+      end
+    end
+
+    context 'when combining exclusion rules' do
+      it 'applies both exclusion rules' do
+        # Create conversations
+        old_conversation_with_label = create(:conversation, inbox: inbox, assignee: nil, created_at: 25.hours.ago)
+        old_conversation_without_label = create(:conversation, inbox: inbox, assignee: nil, created_at: 25.hours.ago)
+        recent_conversation_with_label = create(:conversation, inbox: inbox, assignee: nil, created_at: 1.hour.ago)
+        recent_conversation_without_label = create(:conversation, inbox: inbox, assignee: nil, created_at: 1.hour.ago)
+
+        # Add labels
+        old_conversation_with_label.update_labels([label1.title])
+        recent_conversation_with_label.update_labels([label1.title])
+
+        capacity_policy.update!(exclusion_rules: {
+                                  'excluded_labels' => [label1.title],
+                                  'exclude_older_than_hours' => 24
+                                })
+
+        assigned_count = assignment_service.perform_bulk_assignment(limit: 10)
+
+        # Only recent conversation without label should be assigned
+        expect(assigned_count).to eq(1)
+        expect(old_conversation_with_label.reload.assignee).to be_nil
+        expect(old_conversation_without_label.reload.assignee).to be_nil
+        expect(recent_conversation_with_label.reload.assignee).to be_nil
+        expect(recent_conversation_without_label.reload.assignee).to be_present
+      end
+    end
+
+    context 'when exclusion rules are empty' do
+      let!(:conversation1) { create(:conversation, inbox: inbox, assignee: nil) }
+      let!(:conversation2) { create(:conversation, inbox: inbox, assignee: nil) }
+
+      before do
+        capacity_policy.update!(exclusion_rules: {})
+      end
+
+      it 'assigns all eligible conversations' do
+        assigned_count = assignment_service.perform_bulk_assignment(limit: 10)
+
+        expect(assigned_count).to eq(2)
+        expect(conversation1.reload.assignee).to be_present
+        expect(conversation2.reload.assignee).to be_present
+      end
+    end
+
+    context 'when no capacity policy exists' do
+      let!(:conversation1) { create(:conversation, inbox: inbox, assignee: nil) }
+      let!(:conversation2) { create(:conversation, inbox: inbox, assignee: nil) }
+
+      before do
+        InboxCapacityLimit.destroy_all
+      end
+
+      it 'assigns all eligible conversations without exclusions' do
+        assigned_count = assignment_service.perform_bulk_assignment(limit: 10)
+
+        expect(assigned_count).to eq(2)
+        expect(conversation1.reload.assignee).to be_present
+        expect(conversation2.reload.assignee).to be_present
+      end
+    end
+  end
+end
diff --git a/spec/enterprise/services/enterprise/auto_assignment/balanced_selector_spec.rb b/spec/enterprise/services/enterprise/auto_assignment/balanced_selector_spec.rb
new file mode 100644
index 000000000..a2340ab0e
--- /dev/null
+++ b/spec/enterprise/services/enterprise/auto_assignment/balanced_selector_spec.rb
@@ -0,0 +1,88 @@
+require 'rails_helper'
+
+RSpec.describe Enterprise::AutoAssignment::BalancedSelector do
+  let(:account) { create(:account) }
+  let(:inbox) { create(:inbox, account: account) }
+  let(:selector) { described_class.new(inbox: inbox) }
+  let(:agent1) { create(:user, account: account, role: :agent, availability: :online) }
+  let(:agent2) { create(:user, account: account, role: :agent, availability: :online) }
+  let(:agent3) { create(:user, account: account, role: :agent, availability: :online) }
+  let(:member1) { create(:inbox_member, inbox: inbox, user: agent1) }
+  let(:member2) { create(:inbox_member, inbox: inbox, user: agent2) }
+  let(:member3) { create(:inbox_member, inbox: inbox, user: agent3) }
+
+  describe '#select_agent' do
+    context 'when selecting based on workload' do
+      let(:available_agents) { [member1, member2, member3] }
+
+      it 'selects the agent with least open conversations' do
+        # Agent1 has 3 open conversations
+        3.times { create(:conversation, inbox: inbox, assignee: agent1, status: 'open') }
+
+        # Agent2 has 1 open conversation
+        create(:conversation, inbox: inbox, assignee: agent2, status: 'open')
+
+        # Agent3 has 2 open conversations
+        2.times { create(:conversation, inbox: inbox, assignee: agent3, status: 'open') }
+
+        selected_agent = selector.select_agent(available_agents)
+
+        # Should select agent2 as they have the least conversations
+        expect(selected_agent).to eq(agent2)
+      end
+
+      it 'considers only open conversations' do
+        # Agent1 has 1 open and 3 resolved conversations
+        create(:conversation, inbox: inbox, assignee: agent1, status: 'open')
+        3.times { create(:conversation, inbox: inbox, assignee: agent1, status: 'resolved') }
+
+        # Agent2 has 2 open conversations
+        2.times { create(:conversation, inbox: inbox, assignee: agent2, status: 'open') }
+
+        selected_agent = selector.select_agent([member1, member2])
+
+        # Should select agent1 as they have fewer open conversations
+        expect(selected_agent).to eq(agent1)
+      end
+
+      it 'selects any agent when agents have equal workload' do
+        # All agents have same number of conversations
+        [member1, member2, member3].each do |member|
+          create(:conversation, inbox: inbox, assignee: member.user, status: 'open')
+        end
+
+        selected_agent = selector.select_agent(available_agents)
+
+        # Should select one of the agents (when equal, min_by returns the first one it finds)
+        expect([agent1, agent2, agent3]).to include(selected_agent)
+      end
+    end
+
+    context 'when no agents are available' do
+      it 'returns nil' do
+        selected_agent = selector.select_agent([])
+        expect(selected_agent).to be_nil
+      end
+    end
+
+    context 'when one agent is available' do
+      it 'returns that agent' do
+        selected_agent = selector.select_agent([member1])
+        expect(selected_agent).to eq(agent1)
+      end
+    end
+
+    context 'with new agents (no conversations)' do
+      it 'prioritizes agents with no conversations' do
+        # Agent1 and 2 have conversations
+        create(:conversation, inbox: inbox, assignee: agent1, status: 'open')
+        create(:conversation, inbox: inbox, assignee: agent2, status: 'open')
+
+        # Agent3 is new with no conversations
+        selected_agent = selector.select_agent([member1, member2, member3])
+
+        expect(selected_agent).to eq(agent3)
+      end
+    end
+  end
+end
diff --git a/spec/enterprise/services/enterprise/auto_assignment/capacity_service_spec.rb b/spec/enterprise/services/enterprise/auto_assignment/capacity_service_spec.rb
new file mode 100644
index 000000000..98d7c2172
--- /dev/null
+++ b/spec/enterprise/services/enterprise/auto_assignment/capacity_service_spec.rb
@@ -0,0 +1,119 @@
+require 'rails_helper'
+
+RSpec.describe Enterprise::AutoAssignment::CapacityService, type: :service do
+  let(:account) { create(:account) }
+  let(:inbox) { create(:inbox, account: account, enable_auto_assignment: true) }
+
+  # Assignment policy with rate limiting
+  let(:assignment_policy) do
+    create(:assignment_policy,
+           account: account,
+           enabled: true,
+           fair_distribution_limit: 5,
+           fair_distribution_window: 3600)
+  end
+
+  # Agent capacity policy
+  let(:agent_capacity_policy) do
+    create(:agent_capacity_policy, account: account, name: 'Limited Capacity')
+  end
+
+  # Agents with different capacity settings
+  let(:agent_with_capacity) { create(:user, account: account, role: :agent, availability: :online) }
+  let(:agent_without_capacity) { create(:user, account: account, role: :agent, availability: :online) }
+  let(:agent_at_capacity) { create(:user, account: account, role: :agent, availability: :online) }
+
+  before do
+    # Create inbox assignment policy
+    create(:inbox_assignment_policy, inbox: inbox, assignment_policy: assignment_policy)
+
+    # Set inbox capacity limit
+    create(:inbox_capacity_limit,
+           agent_capacity_policy: agent_capacity_policy,
+           inbox: inbox,
+           conversation_limit: 3)
+
+    # Assign capacity policy to specific agents
+    agent_with_capacity.account_users.find_by(account: account)
+                       .update!(agent_capacity_policy: agent_capacity_policy)
+
+    agent_at_capacity.account_users.find_by(account: account)
+                     .update!(agent_capacity_policy: agent_capacity_policy)
+
+    # Create inbox members
+    create(:inbox_member, inbox: inbox, user: agent_with_capacity)
+    create(:inbox_member, inbox: inbox, user: agent_without_capacity)
+    create(:inbox_member, inbox: inbox, user: agent_at_capacity)
+
+    # Mock online status
+    allow(OnlineStatusTracker).to receive(:get_available_users).and_return({
+                                                                             agent_with_capacity.id.to_s => 'online',
+                                                                             agent_without_capacity.id.to_s => 'online',
+                                                                             agent_at_capacity.id.to_s => 'online'
+                                                                           })
+
+    # Enable assignment_v2 feature
+    allow(account).to receive(:feature_enabled?).with('assignment_v2').and_return(true)
+
+    # Create existing assignments for agent_at_capacity (at limit)
+    3.times do
+      create(:conversation, inbox: inbox, assignee: agent_at_capacity, status: :open)
+    end
+  end
+
+  describe 'capacity filtering' do
+    it 'excludes agents at capacity' do
+      # Get available agents respecting capacity
+      capacity_service = described_class.new
+      online_agents = inbox.available_agents
+      filtered_agents = online_agents.select do |inbox_member|
+        capacity_service.agent_has_capacity?(inbox_member.user, inbox)
+      end
+      available_users = filtered_agents.map(&:user)
+
+      expect(available_users).to include(agent_with_capacity)
+      expect(available_users).to include(agent_without_capacity) # No capacity policy = unlimited
+      expect(available_users).not_to include(agent_at_capacity) # At capacity limit
+    end
+
+    it 'respects inbox-specific capacity limits' do
+      capacity_service = described_class.new
+
+      expect(capacity_service.agent_has_capacity?(agent_with_capacity, inbox)).to be true
+      expect(capacity_service.agent_has_capacity?(agent_without_capacity, inbox)).to be true
+      expect(capacity_service.agent_has_capacity?(agent_at_capacity, inbox)).to be false
+    end
+  end
+
+  describe 'assignment with capacity' do
+    let(:service) { AutoAssignment::AssignmentService.new(inbox: inbox) }
+
+    it 'assigns to agents with available capacity' do
+      # Create conversation before assignment
+      conversation = create(:conversation, inbox: inbox, assignee: nil, status: :open)
+
+      # Mock the selector to prefer agent_at_capacity (but should skip due to capacity)
+      selector = instance_double(AutoAssignment::RoundRobinSelector)
+      allow(AutoAssignment::RoundRobinSelector).to receive(:new).and_return(selector)
+      allow(selector).to receive(:select_agent) do |agents|
+        agents.map(&:user).find { |u| [agent_with_capacity, agent_without_capacity].include?(u) }
+      end
+
+      assigned_count = service.perform_bulk_assignment(limit: 1)
+      expect(assigned_count).to eq(1)
+      expect(conversation.reload.assignee).to be_in([agent_with_capacity, agent_without_capacity])
+      expect(conversation.reload.assignee).not_to eq(agent_at_capacity)
+    end
+
+    it 'returns false when all agents are at capacity' do
+      # Fill up remaining agents
+      3.times { create(:conversation, inbox: inbox, assignee: agent_with_capacity, status: :open) }
+
+      # agent_without_capacity has no limit, so should still be available
+      conversation2 = create(:conversation, inbox: inbox, assignee: nil, status: :open)
+      assigned_count = service.perform_bulk_assignment(limit: 1)
+      expect(assigned_count).to eq(1)
+      expect(conversation2.reload.assignee).to eq(agent_without_capacity)
+    end
+  end
+end
diff --git a/spec/jobs/auto_assignment/assignment_job_spec.rb b/spec/jobs/auto_assignment/assignment_job_spec.rb
new file mode 100644
index 000000000..d13f9fef8
--- /dev/null
+++ b/spec/jobs/auto_assignment/assignment_job_spec.rb
@@ -0,0 +1,85 @@
+require 'rails_helper'
+
+RSpec.describe AutoAssignment::AssignmentJob, type: :job do
+  let(:account) { create(:account) }
+  let(:inbox) { create(:inbox, account: account, enable_auto_assignment: true) }
+  let(:agent) { create(:user, account: account, role: :agent, availability: :online) }
+
+  before do
+    create(:inbox_member, inbox: inbox, user: agent)
+  end
+
+  describe '#perform' do
+    context 'when inbox exists' do
+      context 'when auto assignment is enabled' do
+        it 'calls the assignment service' do
+          service = instance_double(AutoAssignment::AssignmentService)
+          allow(AutoAssignment::AssignmentService).to receive(:new).with(inbox: inbox).and_return(service)
+          expect(service).to receive(:perform_bulk_assignment).with(limit: 100).and_return(5)
+
+          described_class.new.perform(inbox_id: inbox.id)
+        end
+
+        it 'logs the assignment count' do
+          service = instance_double(AutoAssignment::AssignmentService)
+          allow(AutoAssignment::AssignmentService).to receive(:new).and_return(service)
+          allow(service).to receive(:perform_bulk_assignment).and_return(3)
+
+          expect(Rails.logger).to receive(:info).with("Assigned 3 conversations for inbox #{inbox.id}")
+
+          described_class.new.perform(inbox_id: inbox.id)
+        end
+
+        it 'uses custom bulk limit from environment' do
+          allow(ENV).to receive(:fetch).with('AUTO_ASSIGNMENT_BULK_LIMIT', 100).and_return('50')
+
+          service = instance_double(AutoAssignment::AssignmentService)
+          allow(AutoAssignment::AssignmentService).to receive(:new).with(inbox: inbox).and_return(service)
+          expect(service).to receive(:perform_bulk_assignment).with(limit: 50).and_return(2)
+
+          described_class.new.perform(inbox_id: inbox.id)
+        end
+      end
+
+      context 'when auto assignment is disabled' do
+        before { inbox.update!(enable_auto_assignment: false) }
+
+        it 'calls the service which handles the disabled state' do
+          service = instance_double(AutoAssignment::AssignmentService)
+          allow(AutoAssignment::AssignmentService).to receive(:new).with(inbox: inbox).and_return(service)
+          expect(service).to receive(:perform_bulk_assignment).with(limit: 100).and_return(0)
+
+          described_class.new.perform(inbox_id: inbox.id)
+        end
+      end
+    end
+
+    context 'when inbox does not exist' do
+      it 'returns early without processing' do
+        expect(AutoAssignment::AssignmentService).not_to receive(:new)
+
+        described_class.new.perform(inbox_id: 999_999)
+      end
+    end
+
+    context 'when an error occurs' do
+      it 'logs the error and re-raises in test environment' do
+        service = instance_double(AutoAssignment::AssignmentService)
+        allow(AutoAssignment::AssignmentService).to receive(:new).and_return(service)
+        allow(service).to receive(:perform_bulk_assignment).and_raise(StandardError, 'Something went wrong')
+
+        expect(Rails.logger).to receive(:error).with("Bulk assignment failed for inbox #{inbox.id}: Something went wrong")
+
+        expect do
+          described_class.new.perform(inbox_id: inbox.id)
+        end.to raise_error(StandardError, 'Something went wrong')
+      end
+    end
+  end
+
+  describe 'job configuration' do
+    it 'is queued in the default queue' do
+      expect(described_class.queue_name).to eq('default')
+    end
+  end
+end
diff --git a/spec/jobs/auto_assignment/periodic_assignment_job_spec.rb b/spec/jobs/auto_assignment/periodic_assignment_job_spec.rb
new file mode 100644
index 000000000..4b06decd0
--- /dev/null
+++ b/spec/jobs/auto_assignment/periodic_assignment_job_spec.rb
@@ -0,0 +1,123 @@
+require 'rails_helper'
+
+RSpec.describe AutoAssignment::PeriodicAssignmentJob, type: :job do
+  let(:account) { create(:account) }
+  let(:inbox) { create(:inbox, account: account, enable_auto_assignment: true) }
+  let(:assignment_policy) { create(:assignment_policy, account: account) }
+  let(:inbox_assignment_policy) { create(:inbox_assignment_policy, inbox: inbox, assignment_policy: assignment_policy) }
+  let(:agent) { create(:user, account: account, role: :agent) }
+
+  before do
+    create(:inbox_member, inbox: inbox, user: agent)
+  end
+
+  describe '#perform' do
+    context 'when account has assignment_v2 feature enabled' do
+      before do
+        allow(account).to receive(:feature_enabled?).with('assignment_v2').and_return(true)
+        allow(Account).to receive(:find_in_batches).and_yield([account])
+      end
+
+      context 'when inbox has auto_assignment_v2 enabled' do
+        before do
+          allow(inbox).to receive(:auto_assignment_v2_enabled?).and_return(true)
+          inbox_relation = instance_double(ActiveRecord::Relation)
+          allow(account).to receive(:inboxes).and_return(inbox_relation)
+          allow(inbox_relation).to receive(:joins).with(:assignment_policy).and_return(inbox_relation)
+          allow(inbox_relation).to receive(:find_in_batches).and_yield([inbox])
+        end
+
+        it 'queues assignment job for eligible inboxes' do
+          inbox_assignment_policy # ensure it exists
+          expect(AutoAssignment::AssignmentJob).to receive(:perform_later).with(inbox_id: inbox.id)
+
+          described_class.new.perform
+        end
+
+        it 'processes multiple accounts' do
+          inbox_assignment_policy # ensure it exists
+          account2 = create(:account)
+          inbox2 = create(:inbox, account: account2, enable_auto_assignment: true)
+          policy2 = create(:assignment_policy, account: account2)
+          create(:inbox_assignment_policy, inbox: inbox2, assignment_policy: policy2)
+
+          allow(account2).to receive(:feature_enabled?).with('assignment_v2').and_return(true)
+          allow(inbox2).to receive(:auto_assignment_v2_enabled?).and_return(true)
+
+          inbox_relation2 = instance_double(ActiveRecord::Relation)
+          allow(account2).to receive(:inboxes).and_return(inbox_relation2)
+          allow(inbox_relation2).to receive(:joins).with(:assignment_policy).and_return(inbox_relation2)
+          allow(inbox_relation2).to receive(:find_in_batches).and_yield([inbox2])
+
+          allow(Account).to receive(:find_in_batches).and_yield([account]).and_yield([account2])
+
+          expect(AutoAssignment::AssignmentJob).to receive(:perform_later).with(inbox_id: inbox.id)
+          expect(AutoAssignment::AssignmentJob).to receive(:perform_later).with(inbox_id: inbox2.id)
+
+          described_class.new.perform
+        end
+      end
+
+      context 'when inbox does not have auto_assignment_v2 enabled' do
+        before do
+          allow(inbox).to receive(:auto_assignment_v2_enabled?).and_return(false)
+        end
+
+        it 'does not queue assignment job' do
+          expect(AutoAssignment::AssignmentJob).not_to receive(:perform_later)
+
+          described_class.new.perform
+        end
+      end
+    end
+
+    context 'when account does not have assignment_v2 feature enabled' do
+      before do
+        allow(account).to receive(:feature_enabled?).with('assignment_v2').and_return(false)
+        allow(Account).to receive(:find_in_batches).and_yield([account])
+      end
+
+      it 'does not process the account' do
+        expect(AutoAssignment::AssignmentJob).not_to receive(:perform_later)
+
+        described_class.new.perform
+      end
+    end
+
+    context 'with batch processing' do
+      it 'processes accounts in batches' do
+        accounts = []
+        # Create multiple accounts
+        5.times do |_i|
+          acc = create(:account)
+          inb = create(:inbox, account: acc, enable_auto_assignment: true)
+          policy = create(:assignment_policy, account: acc)
+          create(:inbox_assignment_policy, inbox: inb, assignment_policy: policy)
+          allow(acc).to receive(:feature_enabled?).with('assignment_v2').and_return(true)
+          allow(inb).to receive(:auto_assignment_v2_enabled?).and_return(true)
+
+          inbox_relation = instance_double(ActiveRecord::Relation)
+          allow(acc).to receive(:inboxes).and_return(inbox_relation)
+          allow(inbox_relation).to receive(:joins).with(:assignment_policy).and_return(inbox_relation)
+          allow(inbox_relation).to receive(:find_in_batches).and_yield([inb])
+
+          accounts << acc
+        end
+
+        allow(Account).to receive(:find_in_batches) do |&block|
+          accounts.each { |acc| block.call([acc]) }
+        end
+
+        expect(Account).to receive(:find_in_batches).and_call_original
+
+        described_class.new.perform
+      end
+    end
+  end
+
+  describe 'job configuration' do
+    it 'is queued in the scheduled_jobs queue' do
+      expect(described_class.queue_name).to eq('scheduled_jobs')
+    end
+  end
+end
diff --git a/spec/services/auto_assignment/assignment_service_spec.rb b/spec/services/auto_assignment/assignment_service_spec.rb
new file mode 100644
index 000000000..535277714
--- /dev/null
+++ b/spec/services/auto_assignment/assignment_service_spec.rb
@@ -0,0 +1,310 @@
+require 'rails_helper'
+
+RSpec.describe AutoAssignment::AssignmentService do
+  let(:account) { create(:account) }
+  let(:assignment_policy) { create(:assignment_policy, account: account, enabled: true) }
+  let(:inbox) { create(:inbox, account: account, enable_auto_assignment: true) }
+  let(:service) { described_class.new(inbox: inbox) }
+  let(:agent) { create(:user, account: account, role: :agent, availability: :online) }
+  let(:agent2) { create(:user, account: account, role: :agent, availability: :online) }
+  let(:conversation) { create(:conversation, inbox: inbox, assignee: nil) }
+
+  before do
+    # Enable assignment_v2 feature for the account
+    allow(account).to receive(:feature_enabled?).with('assignment_v2').and_return(true)
+    # Link inbox to assignment policy
+    create(:inbox_assignment_policy, inbox: inbox, assignment_policy: assignment_policy)
+    create(:inbox_member, inbox: inbox, user: agent)
+  end
+
+  describe '#perform_bulk_assignment' do
+    context 'when auto assignment is enabled' do
+      let(:rate_limiter) { instance_double(AutoAssignment::RateLimiter) }
+
+      before do
+        allow(OnlineStatusTracker).to receive(:get_available_users).and_return({ agent.id.to_s => 'online' })
+
+        # Mock RoundRobinSelector to return the agent
+        round_robin_selector = instance_double(AutoAssignment::RoundRobinSelector)
+        allow(AutoAssignment::RoundRobinSelector).to receive(:new).and_return(round_robin_selector)
+        allow(round_robin_selector).to receive(:select_agent).and_return(agent)
+
+        # Mock RateLimiter to allow all assignments by default
+        allow(AutoAssignment::RateLimiter).to receive(:new).and_return(rate_limiter)
+        allow(rate_limiter).to receive(:within_limit?).and_return(true)
+        allow(rate_limiter).to receive(:track_assignment)
+      end
+
+      it 'assigns conversations to available agents' do
+        # Create conversation and ensure it's unassigned
+        conv = create(:conversation, inbox: inbox, status: 'open')
+        conv.update!(assignee_id: nil)
+
+        assigned_count = service.perform_bulk_assignment(limit: 1)
+
+        expect(assigned_count).to eq(1)
+        expect(conv.reload.assignee).to eq(agent)
+      end
+
+      it 'returns 0 when no agents are online' do
+        allow(OnlineStatusTracker).to receive(:get_available_users).and_return({})
+
+        assigned_count = service.perform_bulk_assignment(limit: 1)
+
+        expect(assigned_count).to eq(0)
+        expect(conversation.reload.assignee).to be_nil
+      end
+
+      it 'respects the limit parameter' do
+        3.times do
+          conv = create(:conversation, inbox: inbox, status: 'open')
+          conv.update!(assignee_id: nil)
+        end
+
+        assigned_count = service.perform_bulk_assignment(limit: 2)
+
+        expect(assigned_count).to eq(2)
+        expect(inbox.conversations.unassigned.count).to eq(1)
+      end
+
+      it 'only assigns open conversations' do
+        conversation # ensure it exists
+        conversation.update!(assignee_id: nil)
+        resolved_conversation = create(:conversation, inbox: inbox, status: 'resolved')
+        resolved_conversation.update!(assignee_id: nil)
+
+        service.perform_bulk_assignment(limit: 10)
+
+        expect(conversation.reload.assignee).to eq(agent)
+        expect(resolved_conversation.reload.assignee).to be_nil
+      end
+
+      it 'does not reassign already assigned conversations' do
+        conversation # ensure it exists
+        conversation.update!(assignee_id: nil)
+        assigned_conversation = create(:conversation, inbox: inbox, assignee: agent)
+        unassigned_conversation = create(:conversation, inbox: inbox, status: 'open')
+        unassigned_conversation.update!(assignee_id: nil)
+
+        assigned_count = service.perform_bulk_assignment(limit: 10)
+
+        expect(assigned_count).to eq(2) # conversation + unassigned_conversation
+        expect(assigned_conversation.reload.assignee).to eq(agent)
+        expect(unassigned_conversation.reload.assignee).to eq(agent)
+      end
+
+      it 'dispatches assignee changed event' do
+        conversation # ensure it exists
+        conversation.update!(assignee_id: nil)
+
+        # The conversation model also dispatches a conversation.updated event
+        allow(Rails.configuration.dispatcher).to receive(:dispatch)
+        expect(Rails.configuration.dispatcher).to receive(:dispatch).with(
+          Events::Types::ASSIGNEE_CHANGED,
+          anything,
+          hash_including(conversation: conversation, user: agent)
+        )
+
+        service.perform_bulk_assignment(limit: 1)
+      end
+    end
+
+    context 'when auto assignment is disabled' do
+      before { assignment_policy.update!(enabled: false) }
+
+      it 'returns 0 without processing' do
+        assigned_count = service.perform_bulk_assignment(limit: 10)
+
+        expect(assigned_count).to eq(0)
+        expect(conversation.reload.assignee).to be_nil
+      end
+    end
+
+    context 'with conversation priority' do
+      let(:rate_limiter) { instance_double(AutoAssignment::RateLimiter) }
+
+      before do
+        allow(OnlineStatusTracker).to receive(:get_available_users).and_return({ agent.id.to_s => 'online' })
+
+        # Mock RoundRobinSelector to return the agent
+        round_robin_selector = instance_double(AutoAssignment::RoundRobinSelector)
+        allow(AutoAssignment::RoundRobinSelector).to receive(:new).and_return(round_robin_selector)
+        allow(round_robin_selector).to receive(:select_agent).and_return(agent)
+
+        # Mock RateLimiter to allow all assignments by default
+        allow(AutoAssignment::RateLimiter).to receive(:new).and_return(rate_limiter)
+        allow(rate_limiter).to receive(:within_limit?).and_return(true)
+        allow(rate_limiter).to receive(:track_assignment)
+      end
+
+      context 'when priority is longest_waiting' do
+        before do
+          allow(inbox).to receive(:auto_assignment_config).and_return({ 'conversation_priority' => 'longest_waiting' })
+        end
+
+        it 'assigns conversations with oldest last_activity_at first' do
+          old_conversation = create(:conversation,
+                                    inbox: inbox,
+                                    status: 'open',
+                                    created_at: 2.hours.ago,
+                                    last_activity_at: 2.hours.ago)
+          old_conversation.update!(assignee_id: nil)
+          new_conversation = create(:conversation,
+                                    inbox: inbox,
+                                    status: 'open',
+                                    created_at: 1.hour.ago,
+                                    last_activity_at: 1.hour.ago)
+          new_conversation.update!(assignee_id: nil)
+
+          service.perform_bulk_assignment(limit: 1)
+
+          expect(old_conversation.reload.assignee).to eq(agent)
+          expect(new_conversation.reload.assignee).to be_nil
+        end
+      end
+
+      context 'when priority is default' do
+        it 'assigns conversations by created_at' do
+          old_conversation = create(:conversation, inbox: inbox, status: 'open', created_at: 2.hours.ago)
+          old_conversation.update!(assignee_id: nil)
+          new_conversation = create(:conversation, inbox: inbox, status: 'open', created_at: 1.hour.ago)
+          new_conversation.update!(assignee_id: nil)
+
+          service.perform_bulk_assignment(limit: 1)
+
+          expect(old_conversation.reload.assignee).to eq(agent)
+          expect(new_conversation.reload.assignee).to be_nil
+        end
+      end
+    end
+
+    context 'with fair distribution' do
+      before do
+        create(:inbox_member, inbox: inbox, user: agent2)
+        allow(OnlineStatusTracker).to receive(:get_available_users).and_return({
+                                                                                 agent.id.to_s => 'online',
+                                                                                 agent2.id.to_s => 'online'
+                                                                               })
+      end
+
+      context 'when fair distribution is enabled' do
+        before do
+          allow(inbox).to receive(:auto_assignment_config).and_return({
+                                                                        'fair_distribution_limit' => 2,
+                                                                        'fair_distribution_window' => 3600
+                                                                      })
+        end
+
+        it 'respects the assignment limit per agent' do
+          # Mock RoundRobinSelector to select agent2
+          round_robin_selector = instance_double(AutoAssignment::RoundRobinSelector)
+          allow(AutoAssignment::RoundRobinSelector).to receive(:new).and_return(round_robin_selector)
+          allow(round_robin_selector).to receive(:select_agent).and_return(agent2)
+
+          # Mock agent1 at limit, agent2 not at limit
+          agent1_limiter = instance_double(AutoAssignment::RateLimiter)
+          agent2_limiter = instance_double(AutoAssignment::RateLimiter)
+
+          allow(AutoAssignment::RateLimiter).to receive(:new).with(inbox: inbox, agent: agent).and_return(agent1_limiter)
+          allow(AutoAssignment::RateLimiter).to receive(:new).with(inbox: inbox, agent: agent2).and_return(agent2_limiter)
+
+          allow(agent1_limiter).to receive(:within_limit?).and_return(false)
+          allow(agent2_limiter).to receive(:within_limit?).and_return(true)
+          allow(agent2_limiter).to receive(:track_assignment)
+
+          unassigned_conversation = create(:conversation, inbox: inbox, status: 'open')
+          unassigned_conversation.update!(assignee_id: nil)
+
+          service.perform_bulk_assignment(limit: 1)
+
+          expect(unassigned_conversation.reload.assignee).to eq(agent2)
+        end
+
+        it 'tracks assignments in Redis' do
+          conversation # ensure it exists
+          conversation.update!(assignee_id: nil)
+
+          # Mock RoundRobinSelector
+          round_robin_selector = instance_double(AutoAssignment::RoundRobinSelector)
+          allow(AutoAssignment::RoundRobinSelector).to receive(:new).and_return(round_robin_selector)
+          allow(round_robin_selector).to receive(:select_agent).and_return(agent)
+
+          limiter = instance_double(AutoAssignment::RateLimiter)
+          allow(AutoAssignment::RateLimiter).to receive(:new).and_return(limiter)
+          allow(limiter).to receive(:within_limit?).and_return(true)
+          expect(limiter).to receive(:track_assignment)
+
+          service.perform_bulk_assignment(limit: 1)
+        end
+
+        it 'allows assignments after window expires' do
+          # Mock RoundRobinSelector
+          round_robin_selector = instance_double(AutoAssignment::RoundRobinSelector)
+          allow(AutoAssignment::RoundRobinSelector).to receive(:new).and_return(round_robin_selector)
+          allow(round_robin_selector).to receive(:select_agent).and_return(agent, agent2)
+
+          # Mock RateLimiter to allow all
+          limiter = instance_double(AutoAssignment::RateLimiter)
+          allow(AutoAssignment::RateLimiter).to receive(:new).and_return(limiter)
+          allow(limiter).to receive(:within_limit?).and_return(true)
+          allow(limiter).to receive(:track_assignment)
+
+          # Simulate time passing for rate limit window
+          freeze_time do
+            2.times do
+              conversation_new = create(:conversation, inbox: inbox, status: 'open')
+              conversation_new.update!(assignee_id: nil)
+              service.perform_bulk_assignment(limit: 1)
+              expect(conversation_new.reload.assignee).not_to be_nil
+            end
+          end
+
+          # Move forward past the window
+          travel_to(2.hours.from_now) do
+            new_conversation = create(:conversation, inbox: inbox, status: 'open')
+            new_conversation.update!(assignee_id: nil)
+            service.perform_bulk_assignment(limit: 1)
+            expect(new_conversation.reload.assignee).not_to be_nil
+          end
+        end
+      end
+
+      context 'when fair distribution is disabled' do
+        it 'assigns without rate limiting' do
+          5.times do
+            conv = create(:conversation, inbox: inbox, status: 'open')
+            conv.update!(assignee_id: nil)
+          end
+
+          # Mock RoundRobinSelector
+          round_robin_selector = instance_double(AutoAssignment::RoundRobinSelector)
+          allow(AutoAssignment::RoundRobinSelector).to receive(:new).and_return(round_robin_selector)
+          allow(round_robin_selector).to receive(:select_agent).and_return(agent)
+
+          # Mock RateLimiter to allow all
+          limiter = instance_double(AutoAssignment::RateLimiter)
+          allow(AutoAssignment::RateLimiter).to receive(:new).and_return(limiter)
+          allow(limiter).to receive(:within_limit?).and_return(true)
+          allow(limiter).to receive(:track_assignment)
+
+          assigned_count = service.perform_bulk_assignment(limit: 5)
+          expect(assigned_count).to eq(5)
+        end
+      end
+
+      context 'with round robin assignment' do
+        it 'distributes conversations evenly among agents' do
+          conversations = Array.new(4) { create(:conversation, inbox: inbox, assignee: nil) }
+
+          service.perform_bulk_assignment(limit: 4)
+
+          agent1_count = conversations.count { |c| c.reload.assignee == agent }
+          agent2_count = conversations.count { |c| c.reload.assignee == agent2 }
+
+          # Should be distributed evenly (2 each) or close to even (3 and 1)
+          expect([agent1_count, agent2_count].sort).to eq([2, 2]).or(eq([1, 3]))
+        end
+      end
+    end
+  end
+end
diff --git a/spec/services/auto_assignment/rate_limiter_spec.rb b/spec/services/auto_assignment/rate_limiter_spec.rb
new file mode 100644
index 000000000..48a5367e3
--- /dev/null
+++ b/spec/services/auto_assignment/rate_limiter_spec.rb
@@ -0,0 +1,167 @@
+require 'rails_helper'
+
+RSpec.describe AutoAssignment::RateLimiter do
+  # Stub Math methods for testing when assignment_policy is nil
+  # rubocop:disable RSpec/BeforeAfterAll, RSpec/InstanceVariable
+  before(:all) do
+    @math_had_positive = Math.respond_to?(:positive?)
+    Math.define_singleton_method(:positive?) { false } unless @math_had_positive
+  end
+
+  after(:all) do
+    Math.singleton_class.send(:remove_method, :positive?) unless @math_had_positive
+  end
+  # rubocop:enable RSpec/BeforeAfterAll, RSpec/InstanceVariable
+
+  let(:account) { create(:account) }
+  let(:inbox) { create(:inbox, account: account) }
+  let(:agent) { create(:user, account: account, role: :agent) }
+  let(:conversation) { create(:conversation, inbox: inbox) }
+  let(:rate_limiter) { described_class.new(inbox: inbox, agent: agent) }
+
+  describe '#within_limit?' do
+    context 'when rate limiting is not enabled' do
+      before do
+        allow(inbox).to receive(:assignment_policy).and_return(nil)
+      end
+
+      it 'returns true' do
+        expect(rate_limiter.within_limit?).to be true
+      end
+    end
+
+    context 'when rate limiting is enabled' do
+      let(:assignment_policy) do
+        instance_double(AssignmentPolicy,
+                        fair_distribution_limit: 5,
+                        fair_distribution_window: 3600)
+      end
+
+      before do
+        allow(inbox).to receive(:assignment_policy).and_return(assignment_policy)
+      end
+
+      it 'returns true when under the limit' do
+        allow(rate_limiter).to receive(:current_count).and_return(3)
+        expect(rate_limiter.within_limit?).to be true
+      end
+
+      it 'returns false when at or over the limit' do
+        allow(rate_limiter).to receive(:current_count).and_return(5)
+        expect(rate_limiter.within_limit?).to be false
+      end
+    end
+  end
+
+  describe '#track_assignment' do
+    context 'when rate limiting is not enabled' do
+      before do
+        allow(inbox).to receive(:assignment_policy).and_return(nil)
+      end
+
+      it 'does not track the assignment' do
+        expect(Redis::Alfred).not_to receive(:set)
+        rate_limiter.track_assignment(conversation)
+      end
+    end
+
+    context 'when rate limiting is enabled' do
+      let(:assignment_policy) do
+        instance_double(AssignmentPolicy,
+                        fair_distribution_limit: 5,
+                        fair_distribution_window: 3600)
+      end
+
+      before do
+        allow(inbox).to receive(:assignment_policy).and_return(assignment_policy)
+      end
+
+      it 'creates a Redis key with correct expiry' do
+        expected_key = format(Redis::RedisKeys::ASSIGNMENT_KEY, inbox_id: inbox.id, agent_id: agent.id, conversation_id: conversation.id)
+        expect(Redis::Alfred).to receive(:set).with(
+          expected_key,
+          conversation.id.to_s,
+          ex: 3600
+        )
+        rate_limiter.track_assignment(conversation)
+      end
+    end
+  end
+
+  describe '#current_count' do
+    context 'when rate limiting is not enabled' do
+      before do
+        allow(inbox).to receive(:assignment_policy).and_return(nil)
+      end
+
+      it 'returns 0' do
+        expect(rate_limiter.current_count).to eq(0)
+      end
+    end
+
+    context 'when rate limiting is enabled' do
+      let(:assignment_policy) do
+        instance_double(AssignmentPolicy,
+                        fair_distribution_limit: 5,
+                        fair_distribution_window: 3600)
+      end
+
+      before do
+        allow(inbox).to receive(:assignment_policy).and_return(assignment_policy)
+      end
+
+      it 'counts matching Redis keys' do
+        pattern = format(Redis::RedisKeys::ASSIGNMENT_KEY_PATTERN, inbox_id: inbox.id, agent_id: agent.id)
+        allow(Redis::Alfred).to receive(:keys_count).with(pattern).and_return(3)
+
+        expect(rate_limiter.current_count).to eq(3)
+      end
+    end
+  end
+
+  describe 'configuration' do
+    context 'with custom window' do
+      let(:assignment_policy) do
+        instance_double(AssignmentPolicy,
+                        fair_distribution_limit: 10,
+                        fair_distribution_window: 7200)
+      end
+
+      before do
+        allow(inbox).to receive(:assignment_policy).and_return(assignment_policy)
+      end
+
+      it 'uses the custom window value' do
+        expected_key = format(Redis::RedisKeys::ASSIGNMENT_KEY, inbox_id: inbox.id, agent_id: agent.id, conversation_id: conversation.id)
+        expect(Redis::Alfred).to receive(:set).with(
+          expected_key,
+          conversation.id.to_s,
+          ex: 7200
+        )
+        rate_limiter.track_assignment(conversation)
+      end
+    end
+
+    context 'without custom window' do
+      let(:assignment_policy) do
+        instance_double(AssignmentPolicy,
+                        fair_distribution_limit: 10,
+                        fair_distribution_window: nil)
+      end
+
+      before do
+        allow(inbox).to receive(:assignment_policy).and_return(assignment_policy)
+      end
+
+      it 'uses the default window value of 24 hours' do
+        expected_key = format(Redis::RedisKeys::ASSIGNMENT_KEY, inbox_id: inbox.id, agent_id: agent.id, conversation_id: conversation.id)
+        expect(Redis::Alfred).to receive(:set).with(
+          expected_key,
+          conversation.id.to_s,
+          ex: 86_400
+        )
+        rate_limiter.track_assignment(conversation)
+      end
+    end
+  end
+end
diff --git a/spec/services/auto_assignment/round_robin_selector_spec.rb b/spec/services/auto_assignment/round_robin_selector_spec.rb
new file mode 100644
index 000000000..05c1dcd73
--- /dev/null
+++ b/spec/services/auto_assignment/round_robin_selector_spec.rb
@@ -0,0 +1,78 @@
+require 'rails_helper'
+
+RSpec.describe AutoAssignment::RoundRobinSelector do
+  let(:account) { create(:account) }
+  let(:inbox) { create(:inbox, account: account) }
+  let(:selector) { described_class.new(inbox: inbox) }
+  let(:agent1) { create(:user, account: account, role: :agent, availability: :online) }
+  let(:agent2) { create(:user, account: account, role: :agent, availability: :online) }
+  let(:agent3) { create(:user, account: account, role: :agent, availability: :online) }
+
+  let(:round_robin_service) { instance_double(AutoAssignment::InboxRoundRobinService) }
+
+  let(:member1) do
+    allow(AutoAssignment::InboxRoundRobinService).to receive(:new).and_return(round_robin_service)
+    allow(round_robin_service).to receive(:add_agent_to_queue)
+    create(:inbox_member, inbox: inbox, user: agent1)
+  end
+
+  let(:member2) do
+    allow(AutoAssignment::InboxRoundRobinService).to receive(:new).and_return(round_robin_service)
+    allow(round_robin_service).to receive(:add_agent_to_queue)
+    create(:inbox_member, inbox: inbox, user: agent2)
+  end
+
+  let(:member3) do
+    allow(AutoAssignment::InboxRoundRobinService).to receive(:new).and_return(round_robin_service)
+    allow(round_robin_service).to receive(:add_agent_to_queue)
+    create(:inbox_member, inbox: inbox, user: agent3)
+  end
+
+  before do
+    # Mock the round robin service to avoid Redis calls
+    allow(AutoAssignment::InboxRoundRobinService).to receive(:new).and_return(round_robin_service)
+    allow(round_robin_service).to receive(:add_agent_to_queue)
+    allow(round_robin_service).to receive(:reset_queue)
+    allow(round_robin_service).to receive(:validate_queue?).and_return(true)
+  end
+
+  describe '#select_agent' do
+    context 'when agents are available' do
+      let(:available_agents) { [member1, member2, member3] }
+
+      it 'returns an agent from the available list' do
+        allow(round_robin_service).to receive(:available_agent).and_return(agent1)
+
+        selected_agent = selector.select_agent(available_agents)
+
+        expect(selected_agent).not_to be_nil
+        expect([agent1, agent2, agent3]).to include(selected_agent)
+      end
+
+      it 'uses round robin service for selection' do
+        expect(round_robin_service).to receive(:available_agent).with(
+          allowed_agent_ids: [agent1.id.to_s, agent2.id.to_s, agent3.id.to_s]
+        ).and_return(agent1)
+
+        selected_agent = selector.select_agent(available_agents)
+        expect(selected_agent).to eq(agent1)
+      end
+    end
+
+    context 'when no agents are available' do
+      it 'returns nil' do
+        selected_agent = selector.select_agent([])
+        expect(selected_agent).to be_nil
+      end
+    end
+
+    context 'when one agent is available' do
+      it 'returns that agent' do
+        allow(round_robin_service).to receive(:available_agent).and_return(agent1)
+
+        selected_agent = selector.select_agent([member1])
+        expect(selected_agent).to eq(agent1)
+      end
+    end
+  end
+end