diff --git a/app/jobs/assignment_v2/assignment_job.rb b/app/jobs/assignment_v2/assignment_job.rb
new file mode 100644
index 000000000..78119862b
--- /dev/null
+++ b/app/jobs/assignment_v2/assignment_job.rb
@@ -0,0 +1,36 @@
+# frozen_string_literal: true
+
+class AssignmentV2::AssignmentJob < ApplicationJob
+  queue_as :low
+
+  def perform(inbox_id: nil, conversation_id: nil)
+    if conversation_id
+      assign_single_conversation(conversation_id)
+    elsif inbox_id
+      assign_inbox_conversations(inbox_id)
+    else
+      Rails.logger.error 'AssignmentV2::AssignmentJob: No inbox_id or conversation_id provided'
+    end
+  end
+
+  private
+
+  def assign_single_conversation(conversation_id)
+    conversation = Conversation.find_by(id: conversation_id)
+    return unless conversation
+
+    service = AssignmentV2::AssignmentService.new(inbox: conversation.inbox)
+    service.perform_for_conversation(conversation)
+  end
+
+  def assign_inbox_conversations(inbox_id)
+    inbox = Inbox.find_by(id: inbox_id)
+    return unless inbox
+    return unless inbox.assignment_v2_enabled?
+
+    service = AssignmentV2::AssignmentService.new(inbox: inbox)
+    assigned_count = service.perform_bulk_assignment
+
+    Rails.logger.info "AssignmentV2::AssignmentJob: Assigned #{assigned_count} conversations for inbox #{inbox_id}"
+  end
+end
\ No newline at end of file
diff --git a/app/models/account.rb b/app/models/account.rb
index f8eb998f0..96c076ff0 100644
--- a/app/models/account.rb
+++ b/app/models/account.rb
@@ -28,6 +28,7 @@ class Account < ApplicationRecord
   include Reportable
   include Featurable
   include CacheKeys
+  include AssignmentV2FeatureFlag
 
   SETTINGS_PARAMS_SCHEMA = {
     'type': 'object',
@@ -97,6 +98,9 @@ class Account < ApplicationRecord
   has_many :webhooks, dependent: :destroy_async
   has_many :whatsapp_channels, dependent: :destroy_async, class_name: '::Channel::Whatsapp'
   has_many :working_hours, dependent: :destroy_async
+  
+  # Assignment V2 associations
+  has_many :assignment_policies, dependent: :destroy_async
 
   has_one_attached :contacts_export
 
@@ -158,6 +162,7 @@ class Account < ApplicationRecord
     ISO_639.find(account_locale)&.english_name&.downcase || 'english'
   end
 
+
   private
 
   def notify_creation
diff --git a/app/models/account_user.rb b/app/models/account_user.rb
index d559ab5b4..060e5b38d 100644
--- a/app/models/account_user.rb
+++ b/app/models/account_user.rb
@@ -28,6 +28,7 @@ class AccountUser < ApplicationRecord
   belongs_to :account
   belongs_to :user
   belongs_to :inviter, class_name: 'User', optional: true
+  belongs_to :agent_capacity_policy, class_name: 'Enterprise::AgentCapacityPolicy', optional: true
 
   enum role: { agent: 0, administrator: 1 }
   enum availability: { online: 0, offline: 1, busy: 2 }
diff --git a/app/models/assignment_policy.rb b/app/models/assignment_policy.rb
new file mode 100644
index 000000000..b070c075b
--- /dev/null
+++ b/app/models/assignment_policy.rb
@@ -0,0 +1,106 @@
+# frozen_string_literal: true
+
+# == Schema Information
+#
+# Table name: assignment_policies
+#
+#  id                        :bigint           not null, primary key
+#  account_id                :bigint           not null
+#  name                      :string(255)      not null
+#  description               :text
+#  assignment_order          :integer          not null, default: 0
+#  conversation_priority     :integer          not null, default: 0
+#  fair_distribution_limit   :integer          not null, default: 10
+#  fair_distribution_window  :integer          not null, default: 3600
+#  enabled                   :boolean          not null, default: true
+#  created_at                :datetime         not null
+#  updated_at                :datetime         not null
+#
+# Indexes
+#
+#  index_assignment_policies_on_account_id                          (account_id)
+#  unique_assignment_policy_name_per_account                        (account_id,name) UNIQUE
+#
+# Foreign Keys
+#
+#  fk_rails_...  (account_id => accounts.id)
+#
+
+class AssignmentPolicy < ApplicationRecord
+  include AccountCacheRevalidator
+
+  # Enums
+  enum assignment_order: { round_robin: 0, balanced: 1 }
+  enum conversation_priority: { earliest_created: 0, longest_waiting: 1 }
+
+  # Associations
+  belongs_to :account
+  has_many :inbox_assignment_policies, dependent: :destroy
+  has_many :inboxes, through: :inbox_assignment_policies
+
+  # Validations
+  validates :name, presence: true, uniqueness: { scope: :account_id }
+  validates :name, length: { maximum: 255 }
+  validates :description, length: { maximum: 1000 }
+  validates :fair_distribution_limit, presence: true, numericality: { greater_than: 0, less_than_or_equal_to: 100 }
+  validates :fair_distribution_window, presence: true, numericality: { greater_than: 60, less_than_or_equal_to: 86_400 }
+  validates :assignment_order, inclusion: { in: assignment_orders.keys }
+  validates :conversation_priority, inclusion: { in: conversation_priorities.keys }
+
+  # Validate balanced assignment is only available for enterprise
+  validate :validate_balanced_assignment_enterprise_only
+  
+  # Server-side validation to prevent bypass
+  before_save :enforce_enterprise_features
+
+  # Scopes
+  scope :enabled, -> { where(enabled: true) }
+  scope :disabled, -> { where(enabled: false) }
+
+  # Callbacks
+  after_update_commit :clear_assignment_caches
+  after_destroy :clear_assignment_caches
+
+  def can_use_balanced_assignment?
+    account.feature_enabled?(:enterprise_agent_capacity) if account.respond_to?(:feature_enabled?)
+  end
+
+  def webhook_data
+    {
+      id: id,
+      name: name,
+      description: description,
+      assignment_order: assignment_order,
+      conversation_priority: conversation_priority,
+      fair_distribution_limit: fair_distribution_limit,
+      fair_distribution_window: fair_distribution_window,
+      enabled: enabled
+    }
+  end
+
+  private
+
+  def validate_balanced_assignment_enterprise_only
+    return unless balanced?
+    return if can_use_balanced_assignment?
+
+    errors.add(:assignment_order, 'Balanced assignment is only available for enterprise accounts')
+  end
+
+  def enforce_enterprise_features
+    # Server-side enforcement to prevent API bypass
+    if balanced? && !can_use_balanced_assignment?
+      # Force to round_robin if enterprise not available
+      self.assignment_order = 'round_robin'
+      Rails.logger.warn("Assignment V2: Forced assignment_order to round_robin for non-enterprise account #{account_id}")
+    end
+  end
+
+  def clear_assignment_caches
+    # Clear Redis caches when policy is updated
+    Rails.cache.delete("assignment_v2:policy:#{id}")
+    inboxes.find_each do |inbox|
+      Rails.cache.delete("assignment_v2:inbox_policy:#{inbox.id}")
+    end
+  end
+end
\ No newline at end of file
diff --git a/app/models/concerns/auto_assignment_handler.rb b/app/models/concerns/auto_assignment_handler.rb
index de5fdae7d..1cd2c2c37 100644
--- a/app/models/concerns/auto_assignment_handler.rb
+++ b/app/models/concerns/auto_assignment_handler.rb
@@ -14,11 +14,18 @@ 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.assignment_v2_enabled?
+      # Use Assignment V2 system
+      AssignmentV2::AssignmentJob.perform_later(conversation_id: 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?
-    return false unless inbox.enable_auto_assignment?
+    # Check auto assignment is enabled (either legacy or v2)
+    return false unless inbox.auto_assignment_enabled?
 
     # run only if assignee is blank or doesn't have access to inbox
     assignee.blank? || inbox.members.exclude?(assignee)
diff --git a/app/models/inbox.rb b/app/models/inbox.rb
index 1c898ba7f..2b12572bb 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 AssignmentV2FeatureFlag
 
   # Not allowing characters:
   validates :name, presence: true
@@ -71,6 +72,10 @@ class Inbox < ApplicationRecord
   has_one :agent_bot, through: :agent_bot_inbox
   has_many :webhooks, dependent: :destroy_async
   has_many :hooks, dependent: :destroy_async, class_name: 'Integrations::Hook'
+  
+  # Assignment V2 associations
+  has_one :inbox_assignment_policy, dependent: :destroy
+  has_one :assignment_policy, through: :inbox_assignment_policy
 
   enum sender_name_type: { friendly: 0, professional: 1 }
 
@@ -184,8 +189,136 @@ class Inbox < ApplicationRecord
     members.ids
   end
 
+  # Assignment V2 methods
+  def assignment_v2_enabled?
+    account.assignment_v2_enabled? && assignment_policy.present? && assignment_policy.enabled?
+  end
+
+  def auto_assignment_enabled?
+    if assignment_v2_enabled?
+      assignment_policy.present? && assignment_policy.enabled?
+    else
+      enable_auto_assignment?
+    end
+  end
+
+  # Returns inbox members who are available for assignment
+  # This method performs all filtering upfront at the database level for optimal performance
+  # 
+  # Filters applied:
+  # 1. Online status - Only agents marked as 'online' in OnlineStatusTracker
+  # 2. Capacity limits (Enterprise) - Agents who haven't reached their conversation limit
+  # 3. Rate limiting - Agents who haven't exceeded rate limits (when implemented)
+  # 4. User exclusions - Specific users can be excluded (e.g., for reassignment)
+  #
+  # @param options [Hash] Additional filter options
+  # @option options [Boolean] :check_capacity (true) Whether to check capacity limits
+  # @option options [Boolean] :check_rate_limits (false) Whether to check rate limits
+  # @option options [Array<Integer>] :exclude_user_ids Users to exclude from results
+  # 
+  # @return [ActiveRecord::Relation<InboxMember>] Available inbox members with preloaded users
+  #
+  # @example Get all available agents
+  #   inbox.available_agents
+  #
+  # @example Get available agents excluding specific users
+  #   inbox.available_agents(exclude_user_ids: [1, 2, 3])
+  #
+  # @example Get available agents without capacity check (faster but less accurate)
+  #   inbox.available_agents(check_capacity: false)
+  def available_agents(options = {})
+    options = { check_capacity: true }.merge(options)
+    
+    # Get online agent IDs
+    online_agent_ids = fetch_online_agent_ids
+    return inbox_members.none if online_agent_ids.empty?
+
+    # Base query - only online agents
+    scope = inbox_members
+              .joins(:user)
+              .where(users: { id: online_agent_ids })
+              .includes(:user)
+
+    # Exclude specific users if requested
+    if options[:exclude_user_ids].present?
+      scope = scope.where.not(users: { id: options[:exclude_user_ids] })
+    end
+
+    # Apply capacity filtering for enterprise accounts
+    if options[:check_capacity] && enterprise_capacity_enabled?
+      scope = filter_by_capacity(scope)
+    end
+
+    # Apply rate limiting if implemented
+    if options[:check_rate_limits] && defined?(AssignmentV2::RateLimiter)
+      scope = filter_by_rate_limits(scope)
+    end
+
+    scope
+  end
+
+
   private
 
+  def fetch_online_agent_ids
+    OnlineStatusTracker.get_available_users(account_id)
+                      .select { |_key, value| value.eql?('online') }
+                      .keys
+                      .map(&:to_i)
+  end
+
+  def enterprise_capacity_enabled?
+    defined?(Enterprise) && 
+      account.custom_attributes&.dig('enterprise_features', 'capacity_management').present?
+  end
+
+  def filter_by_capacity(inbox_members_scope)
+    return inbox_members_scope unless defined?(Enterprise::InboxCapacityLimit)
+
+    # For simple cases without capacity policies, return all agents
+    if !account.account_users.joins(:agent_capacity_policy).exists?
+      return inbox_members_scope
+    end
+
+    # Get current assignment counts for all agents
+    assignment_counts = conversations
+                       .where(status: :open)
+                       .where.not(assignee_id: nil)
+                       .group(:assignee_id)
+                       .count
+
+    # Filter agents based on capacity
+    inbox_members_scope.select do |inbox_member|
+      user = inbox_member.user
+      account_user = account.account_users.find_by(user: user)
+      
+      # If no capacity policy, allow assignment
+      next true unless account_user&.agent_capacity_policy_id
+
+      # Check if there's a limit for this inbox
+      capacity_limit = Enterprise::InboxCapacityLimit
+                      .where(agent_capacity_policy_id: account_user.agent_capacity_policy_id)
+                      .find_by(inbox_id: id)
+      
+      # If no limit defined for this inbox, allow assignment
+      next true unless capacity_limit&.conversation_limit
+
+      # Check current assignments against limit
+      current_count = assignment_counts[user.id] || 0
+      current_count < capacity_limit.conversation_limit
+    end
+  end
+
+  def filter_by_rate_limits(inbox_members_scope)
+    # Filter out agents who have exceeded rate limits
+    return inbox_members_scope unless assignment_policy&.enabled?
+    
+    inbox_members_scope.select do |inbox_member|
+      rate_limiter = AssignmentV2::RateLimiter.new(inbox: self, user: inbox_member.user)
+      rate_limiter.within_limits?
+    end
+  end
+
   def default_name_for_blank_name
     email? ? display_name_from_email : ''
   end
diff --git a/app/models/inbox_assignment_policy.rb b/app/models/inbox_assignment_policy.rb
new file mode 100644
index 000000000..88440ed20
--- /dev/null
+++ b/app/models/inbox_assignment_policy.rb
@@ -0,0 +1,74 @@
+# frozen_string_literal: true
+
+# == Schema Information
+#
+# Table name: inbox_assignment_policies
+#
+#  id                   :bigint           not null, primary key
+#  inbox_id             :bigint           not null
+#  assignment_policy_id :bigint           not null
+#  created_at           :datetime         not null
+#  updated_at           :datetime         not null
+#
+# Indexes
+#
+#  index_inbox_assignment_policies_on_assignment_policy_id  (assignment_policy_id)
+#  index_inbox_assignment_policies_on_inbox_id             (inbox_id)
+#  unique_inbox_assignment_policy                           (inbox_id) UNIQUE
+#
+# Foreign Keys
+#
+#  fk_rails_...  (assignment_policy_id => assignment_policies.id)
+#  fk_rails_...  (inbox_id => inboxes.id)
+#
+
+class InboxAssignmentPolicy < ApplicationRecord
+  include AccountCacheRevalidator
+
+  # Associations
+  belongs_to :inbox
+  belongs_to :assignment_policy
+
+  # Validations
+  validates :inbox_id, uniqueness: true
+  validate :inbox_belongs_to_same_account
+
+  # Delegations
+  delegate :account, to: :inbox
+  delegate :name, :description, :assignment_order, :conversation_priority, 
+           :fair_distribution_limit, :fair_distribution_window, :enabled?,
+           to: :assignment_policy, prefix: :policy
+
+  # Callbacks  
+  after_create_commit :clear_inbox_cache
+  after_update_commit :clear_inbox_cache
+  after_destroy_commit :clear_inbox_cache
+
+  # Scopes
+  scope :enabled, -> { joins(:assignment_policy).where(assignment_policies: { enabled: true }) }
+  scope :disabled, -> { joins(:assignment_policy).where(assignment_policies: { enabled: false }) }
+
+  def webhook_data
+    {
+      id: id,
+      inbox_id: inbox_id,
+      assignment_policy_id: assignment_policy_id,
+      policy: assignment_policy.webhook_data
+    }
+  end
+
+  private
+
+  def inbox_belongs_to_same_account
+    return unless inbox && assignment_policy
+    
+    if inbox.account_id != assignment_policy.account_id
+      errors.add(:inbox, 'must belong to the same account as the assignment policy')
+    end
+  end
+
+  def clear_inbox_cache
+    Rails.cache.delete("assignment_v2:inbox_policy:#{inbox_id}")
+    update_account_cache
+  end
+end
\ No newline at end of file
diff --git a/app/models/user.rb b/app/models/user.rb
index d1907362a..08a3c1bef 100644
--- a/app/models/user.rb
+++ b/app/models/user.rb
@@ -100,6 +100,10 @@ class User < ApplicationRecord
   has_many :macros, foreign_key: 'created_by_id', inverse_of: :created_by
   # rubocop:enable Rails/HasManyOrHasOneDependent
 
+  # Assignment V2 Enterprise associations
+  has_one :agent_capacity_policy_user, dependent: :destroy, class_name: 'Enterprise::AgentCapacityPolicyUser'
+  has_one :agent_capacity_policy, through: :agent_capacity_policy_user, class_name: 'Enterprise::AgentCapacityPolicy'
+
   before_validation :set_password_and_uid, on: :create
   after_destroy :remove_macros
 
diff --git a/app/services/assignment_v2/assignment_service.rb b/app/services/assignment_v2/assignment_service.rb
new file mode 100644
index 000000000..c39ce3fbc
--- /dev/null
+++ b/app/services/assignment_v2/assignment_service.rb
@@ -0,0 +1,122 @@
+# frozen_string_literal: true
+
+class AssignmentV2::AssignmentService
+  pattr_initialize [:inbox!]
+
+  def perform_for_conversation(conversation)
+    return false unless can_assign?(conversation)
+
+    agent = find_agent_for_conversation(conversation)
+    return false unless agent
+
+    assign_conversation_to_agent(conversation, agent)
+  end
+
+  def perform_bulk_assignment(limit: 50)
+    return 0 unless assignment_enabled?
+
+    conversations = unassigned_conversations(limit)
+    assigned_count = 0
+
+    conversations.find_each do |conversation|
+      assigned_count += 1 if perform_for_conversation(conversation)
+    end
+
+    assigned_count
+  end
+
+  private
+
+  def policy
+    @policy ||= inbox.assignment_policy
+  end
+
+  def assignment_enabled?
+    policy&.enabled?
+  end
+
+  def can_assign?(conversation)
+    assignment_enabled? && 
+      conversation.status == 'open' && 
+      conversation.assignee_id.nil?
+  end
+
+  def find_agent_for_conversation(conversation)
+    available_agents = inbox.available_agents(check_rate_limits: true)
+    
+    if available_agents.empty?
+      log_no_agents_available
+      return nil
+    end
+
+    selector_service.select_agent(available_agents)
+  end
+
+  def selector_service
+    @selector_service ||= case policy.assignment_order
+                          when 'round_robin'
+                            AssignmentV2::RoundRobinSelector.new(inbox: inbox)
+                          when 'balanced'
+                            if enterprise_enabled? && policy.can_use_balanced_assignment?
+                              Enterprise::AssignmentV2::BalancedSelector.new(inbox: inbox)
+                            else
+                              AssignmentV2::RoundRobinSelector.new(inbox: inbox)
+                            end
+                          else
+                            AssignmentV2::RoundRobinSelector.new(inbox: inbox)
+                          end
+  end
+
+
+  def unassigned_conversations(limit)
+    scope = inbox.conversations
+                 .unassigned
+                 .open
+
+    # Apply conversation priority ordering
+    scope = case policy.conversation_priority
+            when 'earliest_created'
+              scope.order(created_at: :asc)
+            when 'longest_waiting'
+              scope.order(last_activity_at: :asc, created_at: :asc)
+            else
+              scope.order(created_at: :asc)
+            end
+
+    scope.limit(limit)
+  end
+
+  def assign_conversation_to_agent(conversation, agent)
+    conversation.update!(assignee: agent)
+    create_assignment_activity(conversation, agent)
+    record_assignment_in_rate_limiter(conversation, agent)
+    true
+  rescue ActiveRecord::RecordInvalid => e
+    Rails.logger.error "AssignmentV2: Failed to assign conversation #{conversation.id}: #{e.message}"
+    false
+  end
+
+  def create_assignment_activity(conversation, agent)
+    Rails.configuration.dispatcher.dispatch(
+      ASSIGNEE_CHANGED,
+      Time.zone.now,
+      conversation: conversation,
+      user: agent
+    )
+  end
+
+  def enterprise_enabled?
+    @enterprise_enabled ||= defined?(Enterprise)
+  end
+
+  def log_no_agents_available
+    Rails.logger.warn("AssignmentV2: No agents available for inbox #{inbox.id}")
+  end
+
+  def record_assignment_in_rate_limiter(conversation, agent)
+    rate_limiter = AssignmentV2::RateLimiter.new(inbox: inbox, user: agent)
+    rate_limiter.record_assignment(conversation)
+  rescue StandardError => e
+    Rails.logger.error "AssignmentV2: Failed to record assignment in rate limiter: #{e.message}"
+  end
+end
\ No newline at end of file
diff --git a/app/services/assignment_v2/rate_limiter.rb b/app/services/assignment_v2/rate_limiter.rb
new file mode 100644
index 000000000..eda8a7812
--- /dev/null
+++ b/app/services/assignment_v2/rate_limiter.rb
@@ -0,0 +1,85 @@
+# frozen_string_literal: true
+
+# Rate limiter for assignment operations
+# Uses Redis to track assignment counts per agent per time window
+# based on assignment policy's fair_distribution_limit and fair_distribution_window
+class AssignmentV2::RateLimiter
+  pattr_initialize [:inbox!, :user!]
+
+  # Check if the user has exceeded rate limits
+  # @return [Boolean] true if within limits, false if exceeded
+  def within_limits?
+    return true unless policy_exists?
+
+    current_count < rate_limit
+  end
+
+  # Record an assignment for rate limiting purposes
+  # @param conversation [Conversation] The conversation being assigned
+  def record_assignment(conversation)
+    return unless policy_exists?
+
+    key = rate_limit_key
+    $alfred.with do |redis|
+      redis.multi do |multi|
+        multi.incr(key)
+        multi.expire(key, time_window)
+      end
+    end
+  end
+
+  # Get current rate limit status for the user
+  # @return [Hash] Rate limit status information
+  def status
+    if policy_exists?
+      {
+        within_limits: within_limits?,
+        current_count: current_count,
+        limit: rate_limit,
+        reset_at: Time.at(next_window_start)
+      }
+    else
+      {
+        within_limits: true,
+        current_count: 0,
+        limit: Float::INFINITY,
+        reset_at: nil
+      }
+    end
+  end
+
+  private
+
+  def policy
+    @policy ||= inbox.assignment_policy
+  end
+
+  def policy_exists?
+    policy.present? && policy.enabled?
+  end
+
+  def current_count
+    key = rate_limit_key
+    $alfred.with { |redis| redis.get(key).to_i }
+  end
+
+  def rate_limit
+    policy&.fair_distribution_limit || 10
+  end
+
+  def time_window
+    policy&.fair_distribution_window || 3600
+  end
+
+  def rate_limit_key
+    "assignment_v2:rate_limit:#{user.id}:#{current_window}"
+  end
+
+  def current_window
+    (Time.current.to_i / time_window) * time_window
+  end
+
+  def next_window_start
+    current_window + time_window
+  end
+end
\ No newline at end of file
diff --git a/app/services/assignment_v2/round_robin_selector.rb b/app/services/assignment_v2/round_robin_selector.rb
new file mode 100644
index 000000000..ecbbda5f4
--- /dev/null
+++ b/app/services/assignment_v2/round_robin_selector.rb
@@ -0,0 +1,37 @@
+# frozen_string_literal: true
+
+class AssignmentV2::RoundRobinSelector
+  pattr_initialize [:inbox!]
+
+  def select_agent(available_agents)
+    return nil if available_agents.empty?
+
+    # Extract user IDs from inbox members
+    agent_user_ids = available_agents.map { |inbox_member| inbox_member.user_id }.map(&:to_s)
+    
+    # Use Redis queue for round robin
+    selected_user_id = round_robin_service.available_agent(allowed_agent_ids: agent_user_ids)
+    return nil unless selected_user_id
+
+    # Return the user object
+    available_agents.find { |inbox_member| inbox_member.user_id.to_s == selected_user_id }&.user
+  end
+
+  def add_agent_to_queue(user_id)
+    round_robin_service.add_agent_to_queue(user_id)
+  end
+
+  def remove_agent_from_queue(user_id)
+    round_robin_service.remove_agent_from_queue(user_id)
+  end
+
+  def reset_queue
+    round_robin_service.reset_queue
+  end
+
+  private
+
+  def round_robin_service
+    @round_robin_service ||= AutoAssignment::InboxRoundRobinService.new(inbox: inbox)
+  end
+end
\ No newline at end of file
diff --git a/db/migrate/20250729140000_create_assignment_policies.rb b/db/migrate/20250729140000_create_assignment_policies.rb
new file mode 100644
index 000000000..4bc8a6884
--- /dev/null
+++ b/db/migrate/20250729140000_create_assignment_policies.rb
@@ -0,0 +1,21 @@
+# frozen_string_literal: true
+
+class CreateAssignmentPolicies < ActiveRecord::Migration[7.1]
+  def change
+    create_table :assignment_policies do |t|
+      t.references :account, null: false, index: true
+      t.string :name, null: false, limit: 255
+      t.text :description
+      t.integer :assignment_order, null: false, default: 0 # 0: round_robin, 1: balanced
+      t.integer :conversation_priority, null: false, default: 0 # 0: earliest_created, 1: longest_waiting
+      t.integer :fair_distribution_limit, null: false, default: 10
+      t.integer :fair_distribution_window, null: false, default: 3600 # seconds
+      t.boolean :enabled, null: false, default: true
+
+      t.timestamps
+    end
+
+    add_index :assignment_policies, [:account_id, :name], unique: true, name: 'unique_assignment_policy_name_per_account'
+    add_index :assignment_policies, :enabled, name: 'index_assignment_policies_on_enabled'
+  end
+end
\ No newline at end of file
diff --git a/db/migrate/20250729140001_create_inbox_assignment_policies.rb b/db/migrate/20250729140001_create_inbox_assignment_policies.rb
new file mode 100644
index 000000000..372440fff
--- /dev/null
+++ b/db/migrate/20250729140001_create_inbox_assignment_policies.rb
@@ -0,0 +1,14 @@
+# frozen_string_literal: true
+
+class CreateInboxAssignmentPolicies < ActiveRecord::Migration[7.1]
+  def change
+    create_table :inbox_assignment_policies do |t|
+      t.references :inbox, null: false, index: true
+      t.references :assignment_policy, null: false, index: true
+
+      t.timestamps
+    end
+
+    add_index :inbox_assignment_policies, :inbox_id, unique: true, name: 'unique_inbox_assignment_policy'
+  end
+end
\ No newline at end of file
diff --git a/db/migrate/20250729140002_create_enterprise_agent_capacity_policies.rb b/db/migrate/20250729140002_create_enterprise_agent_capacity_policies.rb
new file mode 100644
index 000000000..a14556f36
--- /dev/null
+++ b/db/migrate/20250729140002_create_enterprise_agent_capacity_policies.rb
@@ -0,0 +1,17 @@
+# frozen_string_literal: true
+
+class CreateEnterpriseAgentCapacityPolicies < ActiveRecord::Migration[7.1]
+  def change
+    create_table :enterprise_agent_capacity_policies do |t|
+      t.references :account, null: false, index: true
+      t.string :name, null: false, limit: 255
+      t.text :description
+      t.jsonb :exclusion_rules, default: {}, null: false
+
+      t.timestamps
+    end
+
+    add_index :enterprise_agent_capacity_policies, [:account_id, :name], unique: true, name: 'unique_capacity_policy_name_per_account'
+    add_index :enterprise_agent_capacity_policies, :exclusion_rules, using: :gin, name: 'index_capacity_policies_on_exclusion_rules'
+  end
+end
\ No newline at end of file
diff --git a/db/migrate/20250729140003_create_enterprise_inbox_capacity_limits.rb b/db/migrate/20250729140003_create_enterprise_inbox_capacity_limits.rb
new file mode 100644
index 000000000..067ef3b31
--- /dev/null
+++ b/db/migrate/20250729140003_create_enterprise_inbox_capacity_limits.rb
@@ -0,0 +1,15 @@
+# frozen_string_literal: true
+
+class CreateEnterpriseInboxCapacityLimits < ActiveRecord::Migration[7.1]
+  def change
+    create_table :enterprise_inbox_capacity_limits do |t|
+      t.references :agent_capacity_policy, null: false, index: { name: 'index_inbox_limits_on_capacity_policy' }
+      t.references :inbox, null: false, index: true
+      t.integer :conversation_limit, null: false
+
+      t.timestamps
+    end
+
+    add_index :enterprise_inbox_capacity_limits, [:agent_capacity_policy_id, :inbox_id], unique: true, name: 'unique_policy_inbox_limit'
+  end
+end
\ No newline at end of file
diff --git a/db/migrate/20250729140004_add_agent_capacity_policy_to_account_users.rb b/db/migrate/20250729140004_add_agent_capacity_policy_to_account_users.rb
new file mode 100644
index 000000000..57c2c4dac
--- /dev/null
+++ b/db/migrate/20250729140004_add_agent_capacity_policy_to_account_users.rb
@@ -0,0 +1,7 @@
+# frozen_string_literal: true
+
+class AddAgentCapacityPolicyToAccountUsers < ActiveRecord::Migration[7.1]
+  def change
+    add_reference :account_users, :agent_capacity_policy, null: true, index: true
+  end
+end
\ No newline at end of file
diff --git a/enterprise/app/models/enterprise/agent_capacity_policy.rb b/enterprise/app/models/enterprise/agent_capacity_policy.rb
new file mode 100644
index 000000000..128244150
--- /dev/null
+++ b/enterprise/app/models/enterprise/agent_capacity_policy.rb
@@ -0,0 +1,133 @@
+# frozen_string_literal: true
+
+# == Schema Information
+#
+# Table name: enterprise_agent_capacity_policies
+#
+#  id              :bigint           not null, primary key
+#  account_id      :bigint           not null
+#  name            :string(255)      not null
+#  description     :text
+#  exclusion_rules :jsonb            default: {}
+#  created_at      :datetime         not null
+#  updated_at      :datetime         not null
+#
+# Indexes
+#
+#  index_enterprise_agent_capacity_policies_on_account_id          (account_id)
+#  unique_capacity_policy_name_per_account                         (account_id,name) UNIQUE
+#
+# Foreign Keys
+#
+#  fk_rails_...  (account_id => accounts.id)
+#
+
+module Enterprise
+  class AgentCapacityPolicy < ApplicationRecord
+    include AccountCacheRevalidator
+
+    self.table_name = 'enterprise_agent_capacity_policies'
+
+    # Associations
+    belongs_to :account
+    has_many :account_users, dependent: :nullify, foreign_key: :agent_capacity_policy_id
+    has_many :users, through: :account_users
+    has_many :inbox_capacity_limits, dependent: :destroy, class_name: 'Enterprise::InboxCapacityLimit'
+    has_many :inboxes, through: :inbox_capacity_limits
+
+    # Validations
+    validates :name, presence: true, uniqueness: { scope: :account_id }
+    validates :name, length: { maximum: 255 }
+    validates :description, length: { maximum: 1000 }
+    validates :exclusion_rules, json: { schema: exclusion_rules_schema }
+
+    # Callbacks
+    before_save :validate_inbox_access
+    after_update_commit :invalidate_capacity_caches
+    after_destroy :invalidate_capacity_caches
+
+    # Scopes
+    scope :with_users, -> { joins(:account_users) }
+    scope :for_inbox, ->(inbox) { joins(:inbox_capacity_limits).where(enterprise_inbox_capacity_limits: { inbox: inbox }) }
+
+    def add_user(user)
+      # Find the account_user for this account and user
+      account_user = account.account_users.find_by!(user: user)
+      
+      # Update the capacity policy reference
+      account_user.update!(agent_capacity_policy_id: id)
+      invalidate_user_capacity_cache(user)
+    end
+
+    def remove_user(user)
+      account_user = account.account_users.find_by(user: user)
+      account_user&.update!(agent_capacity_policy_id: nil)
+      invalidate_user_capacity_cache(user)
+    end
+
+    def set_inbox_limit(inbox, limit)
+      inbox_capacity_limit = inbox_capacity_limits.find_or_initialize_by(inbox: inbox)
+      inbox_capacity_limit.conversation_limit = limit
+      inbox_capacity_limit.save!
+      
+      invalidate_inbox_capacity_cache(inbox)
+    end
+
+    def remove_inbox_limit(inbox)
+      inbox_capacity_limits.where(inbox: inbox).destroy_all
+      invalidate_inbox_capacity_cache(inbox)
+    end
+
+    def get_inbox_limit(inbox)
+      inbox_capacity_limits.find_by(inbox: inbox)&.conversation_limit
+    end
+
+    def webhook_data
+      {
+        id: id,
+        name: name,
+        description: description,
+        exclusion_rules: exclusion_rules,
+        users_count: users.count,
+        inboxes_count: inboxes.count
+      }
+    end
+
+    private
+
+    def validate_inbox_access
+      # Ensure all specified inboxes belong to the same account
+      invalid_inboxes = inbox_capacity_limits.joins(:inbox)
+                                           .where.not(inboxes: { account_id: account_id })
+
+      if invalid_inboxes.exists?
+        errors.add(:inbox_capacity_limits, 'contains inboxes from different accounts')
+        throw :abort
+      end
+    end
+
+    def invalidate_capacity_caches
+      users.find_each { |user| invalidate_user_capacity_cache(user) }
+      inboxes.find_each { |inbox| invalidate_inbox_capacity_cache(inbox) }
+    end
+
+    def invalidate_user_capacity_cache(user)
+      Rails.cache.delete_matched("assignment_v2:capacity:#{user.id}:*")
+    end
+
+    def invalidate_inbox_capacity_cache(inbox)
+      Rails.cache.delete_matched("assignment_v2:capacity:*:#{inbox.id}")
+    end
+
+    def self.exclusion_rules_schema
+      {
+        type: 'object',
+        properties: {
+          labels: { type: 'array', items: { type: 'string' } },
+          hours_threshold: { type: 'integer', minimum: 1, maximum: 168 }
+        },
+        additionalProperties: false
+      }.to_json
+    end
+  end
+end
\ No newline at end of file
diff --git a/enterprise/app/models/enterprise/inbox_capacity_limit.rb b/enterprise/app/models/enterprise/inbox_capacity_limit.rb
new file mode 100644
index 000000000..e3a1fc42c
--- /dev/null
+++ b/enterprise/app/models/enterprise/inbox_capacity_limit.rb
@@ -0,0 +1,70 @@
+# frozen_string_literal: true
+
+# == Schema Information
+#
+# Table name: enterprise_inbox_capacity_limits
+#
+#  id                        :bigint           not null, primary key
+#  agent_capacity_policy_id  :bigint           not null
+#  inbox_id                  :bigint           not null
+#  conversation_limit        :integer          not null
+#  created_at                :datetime         not null
+#  updated_at                :datetime         not null
+#
+# Indexes
+#
+#  index_inbox_limits_on_capacity_policy                            (agent_capacity_policy_id)
+#  index_enterprise_inbox_capacity_limits_on_inbox_id               (inbox_id)
+#  unique_policy_inbox_limit                                        (agent_capacity_policy_id,inbox_id) UNIQUE
+#
+# Foreign Keys
+#
+#  fk_rails_...  (agent_capacity_policy_id => enterprise_agent_capacity_policies.id)
+#  fk_rails_...  (inbox_id => inboxes.id)
+#
+
+module Enterprise
+  class InboxCapacityLimit < ApplicationRecord
+    include AccountCacheRevalidator
+
+    self.table_name = 'enterprise_inbox_capacity_limits'
+
+    # Associations
+    belongs_to :agent_capacity_policy, class_name: 'Enterprise::AgentCapacityPolicy'
+    belongs_to :inbox
+
+    # Validations
+    validates :agent_capacity_policy_id, uniqueness: { scope: :inbox_id }
+    validates :conversation_limit, presence: true, numericality: { greater_than: 0, less_than_or_equal_to: 1000 }
+
+    # Delegations
+    delegate :account, to: :agent_capacity_policy
+    delegate :name, :description, :exclusion_rules, to: :agent_capacity_policy, prefix: :policy
+
+    # Callbacks
+    after_create_commit :invalidate_inbox_cache
+    after_update_commit :invalidate_inbox_cache
+    after_destroy_commit :invalidate_inbox_cache
+
+    # Scopes
+    scope :for_inbox, ->(inbox) { where(inbox: inbox) }
+    scope :for_policy, ->(policy) { where(agent_capacity_policy: policy) }
+
+    def webhook_data
+      {
+        id: id,
+        inbox_id: inbox_id,
+        agent_capacity_policy_id: agent_capacity_policy_id,
+        conversation_limit: conversation_limit,
+        policy: agent_capacity_policy.webhook_data
+      }
+    end
+
+    private
+
+    def invalidate_inbox_cache
+      Rails.cache.delete_matched("assignment_v2:capacity:*:#{inbox_id}")
+      update_account_cache
+    end
+  end
+end
\ No newline at end of file
diff --git a/enterprise/app/services/assignment_v2/balanced_selector.rb b/enterprise/app/services/assignment_v2/balanced_selector.rb
new file mode 100644
index 000000000..f7cd656b1
--- /dev/null
+++ b/enterprise/app/services/assignment_v2/balanced_selector.rb
@@ -0,0 +1,42 @@
+# frozen_string_literal: true
+
+module Enterprise
+  class AssignmentV2::BalancedSelector
+    pattr_initialize [:inbox!]
+
+    def select_agent(available_agents)
+      return nil if available_agents.empty?
+
+      # Since agents are already filtered by capacity, we can compute workload distribution
+      agents_with_workload = compute_agent_workloads(available_agents)
+      return nil if agents_with_workload.empty?
+
+      # Select agent with lowest current workload
+      best_agent_data = agents_with_workload.min_by { |data| data[:current_assignments] }
+      best_agent_data[:agent]
+    rescue StandardError => e
+      Rails.logger.error "AssignmentV2: Balanced selection failed: #{e.message}"
+      # Fallback to simple selection
+      available_agents.first&.user
+    end
+
+    private
+
+    def compute_agent_workloads(available_agents)
+      available_agents.map do |inbox_member|
+        agent = inbox_member.user
+        
+        # Count current assignments
+        current_assignments = agent.assigned_conversations
+                                  .where(inbox: inbox)
+                                  .open
+                                  .count
+
+        {
+          agent: agent,
+          current_assignments: current_assignments
+        }
+      end
+    end
+  end
+end
\ No newline at end of file
diff --git a/enterprise/app/services/assignment_v2/capacity_service.rb b/enterprise/app/services/assignment_v2/capacity_service.rb
new file mode 100644
index 000000000..d8bfbbffd
--- /dev/null
+++ b/enterprise/app/services/assignment_v2/capacity_service.rb
@@ -0,0 +1,55 @@
+# frozen_string_literal: true
+
+module Enterprise
+  class AssignmentV2::CapacityService
+    pattr_initialize [:inbox!]
+
+    def filter_agents_by_capacity(inbox_members)
+      inbox_members.select do |inbox_member|
+        agent = inbox_member.user
+        has_available_capacity?(agent)
+      end
+    end
+
+    def get_agent_capacity(agent)
+      account_user = agent.account_users.find_by(account: inbox.account)
+      policy = account_user&.agent_capacity_policy
+
+      unless policy
+        return {
+          total_capacity: Float::INFINITY,
+          current_assignments: 0,
+          available_capacity: Float::INFINITY
+        }
+      end
+
+      inbox_limit = policy.inbox_capacity_limits.find_by(inbox: inbox)
+      
+      unless inbox_limit
+        return {
+          total_capacity: Float::INFINITY,
+          current_assignments: 0,
+          available_capacity: Float::INFINITY
+        }
+      end
+
+      current_count = agent.assigned_conversations
+                           .where(inbox: inbox)
+                           .open
+                           .count
+
+      {
+        total_capacity: inbox_limit.conversation_limit,
+        current_assignments: current_count,
+        available_capacity: inbox_limit.conversation_limit - current_count
+      }
+    end
+
+    private
+
+    def has_available_capacity?(agent)
+      capacity_data = get_agent_capacity(agent)
+      capacity_data[:available_capacity].positive?
+    end
+  end
+end
\ No newline at end of file
diff --git a/lib/tasks/assignment_v2.rake b/lib/tasks/assignment_v2.rake
new file mode 100644
index 000000000..9799575f2
--- /dev/null
+++ b/lib/tasks/assignment_v2.rake
@@ -0,0 +1,61 @@
+# frozen_string_literal: true
+
+namespace :assignment_v2 do
+  desc 'Enable Assignment V2 for an account with default policy'
+  task :enable, [:account_id] => :environment do |_task, args|
+    account = Account.find(args[:account_id])
+    
+    # Create default assignment policy if not exists
+    policy = account.assignment_policies.find_or_create_by!(name: 'Default Policy') do |p|
+      p.description = 'Default round-robin assignment policy'
+      p.assignment_order = 'round_robin'
+      p.conversation_priority = 'earliest_created'
+      p.enabled = true
+    end
+
+    puts "Assignment V2 enabled for account #{account.name} with policy #{policy.name}"
+  end
+
+  desc 'Disable Assignment V2 for an account'
+  task :disable, [:account_id] => :environment do |_task, args|
+    account = Account.find(args[:account_id])
+    
+    # Disable all assignment policies
+    account.assignment_policies.update_all(enabled: false)
+    
+    puts "Assignment V2 disabled for account #{account.name}"
+  end
+
+  desc 'Run assignment for all enabled inboxes'
+  task run_all: :environment do
+    count = 0
+    
+    Inbox.joins(:assignment_policy)
+         .where(assignment_policies: { enabled: true })
+         .find_each do |inbox|
+      AssignmentV2::AssignmentJob.perform_later(inbox_id: inbox.id)
+      count += 1
+    end
+    
+    puts "Queued assignment jobs for #{count} inboxes"
+  end
+
+  desc 'Show Assignment V2 status'
+  task status: :environment do
+    puts 'Assignment V2 Status'
+    puts '==================='
+    
+    total_policies = AssignmentPolicy.count
+    enabled_policies = AssignmentPolicy.enabled.count
+    
+    puts "Total Assignment Policies: #{total_policies}"
+    puts "Enabled Policies: #{enabled_policies}"
+    puts
+    
+    inboxes_with_v2 = Inbox.joins(:assignment_policy).count
+    total_inboxes = Inbox.count
+    
+    puts "Total Inboxes: #{total_inboxes}"
+    puts "Inboxes with Assignment V2: #{inboxes_with_v2}"
+  end
+end
\ No newline at end of file
diff --git a/spec/factories/assignment_policies.rb b/spec/factories/assignment_policies.rb
new file mode 100644
index 000000000..a1e439f12
--- /dev/null
+++ b/spec/factories/assignment_policies.rb
@@ -0,0 +1,34 @@
+# frozen_string_literal: true
+
+FactoryBot.define do
+  factory :assignment_policy do
+    account
+    sequence(:name) { |n| "Assignment Policy #{n}" }
+    description { 'Test assignment policy' }
+    assignment_order { :round_robin }
+    conversation_priority { :earliest_created }
+    fair_distribution_limit { 10 }
+    fair_distribution_window { 3600 }
+    enabled { true }
+
+    trait :balanced do
+      assignment_order { :balanced }
+    end
+
+    trait :disabled do
+      enabled { false }
+    end
+
+    trait :longest_waiting do
+      conversation_priority { :longest_waiting }
+    end
+
+    trait :with_high_limit do
+      fair_distribution_limit { 50 }
+    end
+
+    trait :with_short_window do
+      fair_distribution_window { 300 } # 5 minutes
+    end
+  end
+end
\ No newline at end of file
diff --git a/spec/factories/enterprise_agent_capacity_policies.rb b/spec/factories/enterprise_agent_capacity_policies.rb
new file mode 100644
index 000000000..2fde4f6ca
--- /dev/null
+++ b/spec/factories/enterprise_agent_capacity_policies.rb
@@ -0,0 +1,27 @@
+# frozen_string_literal: true
+
+FactoryBot.define do
+  factory :enterprise_agent_capacity_policy, class: 'Enterprise::AgentCapacityPolicy' do
+    account
+    sequence(:name) { |n| "Capacity Policy #{n}" }
+    description { 'Test capacity policy' }
+    exclusion_rules { {} }
+
+    trait :with_label_exclusion do
+      exclusion_rules { { 'labels' => ['vip', 'urgent'] } }
+    end
+
+    trait :with_time_exclusion do
+      exclusion_rules { { 'hours_threshold' => 24 } }
+    end
+
+    trait :with_combined_exclusions do
+      exclusion_rules do
+        {
+          'labels' => ['vip', 'urgent'],
+          'hours_threshold' => 48
+        }
+      end
+    end
+  end
+end
\ No newline at end of file
diff --git a/spec/factories/enterprise_inbox_capacity_limits.rb b/spec/factories/enterprise_inbox_capacity_limits.rb
new file mode 100644
index 000000000..97c263de2
--- /dev/null
+++ b/spec/factories/enterprise_inbox_capacity_limits.rb
@@ -0,0 +1,28 @@
+# frozen_string_literal: true
+
+FactoryBot.define do
+  factory :enterprise_inbox_capacity_limit, class: 'Enterprise::InboxCapacityLimit' do
+    association :agent_capacity_policy, factory: :enterprise_agent_capacity_policy
+    inbox
+    conversation_limit { 10 }
+
+    trait :low_limit do
+      conversation_limit { 3 }
+    end
+
+    trait :high_limit do
+      conversation_limit { 50 }
+    end
+
+    trait :max_limit do
+      conversation_limit { 1000 }
+    end
+
+    # Ensure inbox and policy belong to same account
+    after(:build) do |limit|
+      if limit.inbox && limit.agent_capacity_policy
+        limit.agent_capacity_policy.account = limit.inbox.account
+      end
+    end
+  end
+end
\ No newline at end of file
diff --git a/spec/factories/inbox_assignment_policies.rb b/spec/factories/inbox_assignment_policies.rb
new file mode 100644
index 000000000..1bf2c5980
--- /dev/null
+++ b/spec/factories/inbox_assignment_policies.rb
@@ -0,0 +1,15 @@
+# frozen_string_literal: true
+
+FactoryBot.define do
+  factory :inbox_assignment_policy do
+    inbox
+    assignment_policy
+
+    # Ensure inbox and policy belong to same account
+    after(:build) do |inbox_policy|
+      if inbox_policy.inbox && inbox_policy.assignment_policy
+        inbox_policy.assignment_policy.account = inbox_policy.inbox.account
+      end
+    end
+  end
+end
\ No newline at end of file
diff --git a/spec/integration/assignment_v2_flow_spec.rb b/spec/integration/assignment_v2_flow_spec.rb
new file mode 100644
index 000000000..0578037d4
--- /dev/null
+++ b/spec/integration/assignment_v2_flow_spec.rb
@@ -0,0 +1,369 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+RSpec.describe 'Assignment V2 Full Flow', type: :integration do
+  let(:account) { create(:account) }
+  let(:inbox) { create(:inbox, account: account) }
+  
+  # Create agents with different availability
+  let!(:agent1) { create(:user, account: account, name: 'Agent 1', role: :agent, availability: :online) }
+  let!(:agent2) { create(:user, account: account, name: 'Agent 2', role: :agent, availability: :online) }
+  let!(:agent3) { create(:user, account: account, name: 'Agent 3', role: :agent, availability: :busy) }
+  let!(:agent4) { create(:user, account: account, name: 'Agent 4', role: :agent, availability: :offline) }
+  
+  before do
+    # Make agents members of inbox
+    [agent1, agent2, agent3, agent4].each do |agent|
+      create(:inbox_member, inbox: inbox, user: agent)
+    end
+    
+    # Clear Redis to ensure clean state
+    Redis::Alfred.flushdb
+  end
+
+  describe 'Round Robin Assignment' do
+    let(:assignment_policy) do
+      create(:assignment_policy, 
+             account: account,
+             name: 'Round Robin Policy',
+             assignment_order: :round_robin,
+             conversation_priority: :earliest_created,
+             enabled: true)
+    end
+    
+    let!(:inbox_assignment_policy) do
+      create(:inbox_assignment_policy, inbox: inbox, assignment_policy: assignment_policy)
+    end
+
+    it 'assigns conversations in round-robin fashion to online agents only' do
+      # Create unassigned conversations
+      conversations = create_list(:conversation, 6, inbox: inbox, assignee: nil, status: :open)
+      
+      # Process assignments
+      service = AssignmentV2::AssignmentService.new(inbox)
+      assigned_count = service.assign_conversations
+      
+      expect(assigned_count).to eq(6)
+      
+      # Verify all conversations are assigned
+      conversations.each(&:reload)
+      expect(conversations.map(&:assignee).compact.count).to eq(6)
+      
+      # Verify only online agents received assignments
+      assigned_agents = conversations.map(&:assignee).uniq
+      expect(assigned_agents).to match_array([agent1, agent2])
+      
+      # Verify round-robin distribution
+      agent1_count = conversations.count { |c| c.assignee == agent1 }
+      agent2_count = conversations.count { |c| c.assignee == agent2 }
+      expect([agent1_count, agent2_count]).to match_array([3, 3])
+    end
+
+    it 'respects conversation priority order' do
+      # Create conversations with different creation times
+      old_conv = create(:conversation, inbox: inbox, assignee: nil, created_at: 2.hours.ago)
+      mid_conv = create(:conversation, inbox: inbox, assignee: nil, created_at: 1.hour.ago)
+      new_conv = create(:conversation, inbox: inbox, assignee: nil, created_at: 5.minutes.ago)
+      
+      # Assign only 2 conversations
+      service = AssignmentV2::AssignmentService.new(inbox)
+      service.assign_conversations(limit: 2)
+      
+      # Oldest conversations should be assigned first
+      expect(old_conv.reload.assignee).not_to be_nil
+      expect(mid_conv.reload.assignee).not_to be_nil
+      expect(new_conv.reload.assignee).to be_nil
+    end
+
+    it 'handles agent availability changes mid-assignment' do
+      conversations = create_list(:conversation, 4, inbox: inbox, assignee: nil)
+      
+      # Assign first batch
+      service = AssignmentV2::AssignmentService.new(inbox)
+      service.assign_conversations(limit: 2)
+      
+      # Make agent1 offline
+      agent1.update!(availability: :offline)
+      
+      # Assign remaining conversations
+      service.assign_conversations(limit: 2)
+      
+      # All remaining should go to agent2
+      remaining_assignments = conversations.reload.last(2).map(&:assignee)
+      expect(remaining_assignments).to all(eq(agent2))
+    end
+  end
+
+  describe 'Balanced Assignment' do
+    let(:assignment_policy) do
+      create(:assignment_policy,
+             account: account,
+             name: 'Balanced Policy',
+             assignment_order: :balanced,
+             enabled: true)
+    end
+    
+    let!(:inbox_assignment_policy) do
+      create(:inbox_assignment_policy, inbox: inbox, assignment_policy: assignment_policy)
+    end
+
+    before do
+      # Mock enterprise features
+      allow(account).to receive(:feature_enabled?).with(:enterprise_agent_capacity).and_return(true)
+    end
+
+    it 'assigns to agent with least conversations' do
+      # Create existing load imbalance
+      create_list(:conversation, 5, inbox: inbox, assignee: agent1, status: :open)
+      create_list(:conversation, 2, inbox: inbox, assignee: agent2, status: :open)
+      
+      # Create new conversations
+      new_conversations = create_list(:conversation, 3, inbox: inbox, assignee: nil)
+      
+      # Process assignments
+      service = AssignmentV2::AssignmentService.new(inbox)
+      service.assign_conversations
+      
+      # All should go to agent2 (less loaded)
+      new_conversations.each(&:reload)
+      expect(new_conversations.map(&:assignee)).to all(eq(agent2))
+      
+      # Final count should be more balanced
+      expect(agent1.assigned_conversations.open.where(inbox: inbox).count).to eq(5)
+      expect(agent2.assigned_conversations.open.where(inbox: inbox).count).to eq(5)
+    end
+
+    it 'only counts open conversations for balancing' do
+      # Agent1 has many resolved conversations (shouldn't count)
+      create_list(:conversation, 10, inbox: inbox, assignee: agent1, status: :resolved)
+      # Agent1 has 1 open conversation
+      create(:conversation, inbox: inbox, assignee: agent1, status: :open)
+      
+      # Agent2 has 3 open conversations
+      create_list(:conversation, 3, inbox: inbox, assignee: agent2, status: :open)
+      
+      # New conversation should go to agent1
+      new_conversation = create(:conversation, inbox: inbox, assignee: nil)
+      
+      service = AssignmentV2::AssignmentService.new(inbox)
+      service.assign_conversation(new_conversation)
+      
+      expect(new_conversation.reload.assignee).to eq(agent1)
+    end
+  end
+
+  describe 'Enterprise Capacity Management' do
+    let(:assignment_policy) do
+      create(:assignment_policy,
+             account: account,
+             assignment_order: :balanced,
+             enabled: true)
+    end
+    
+    let!(:inbox_assignment_policy) do
+      create(:inbox_assignment_policy, inbox: inbox, assignment_policy: assignment_policy)
+    end
+    
+    let(:capacity_policy) do
+      create(:enterprise_agent_capacity_policy, account: account, name: 'Limited Capacity')
+    end
+
+    before do
+      # Mock enterprise features
+      stub_const('Enterprise', Module.new)
+      stub_const('Enterprise::AgentCapacityPolicy', Class.new(ApplicationRecord))
+      stub_const('Enterprise::InboxCapacityLimit', Class.new(ApplicationRecord))
+      
+      allow(account).to receive(:feature_enabled?).with(:enterprise_agent_capacity).and_return(true)
+      
+      # Set up capacity limits
+      agent1.account_users.first.update!(agent_capacity_policy: capacity_policy)
+      agent2.account_users.first.update!(agent_capacity_policy: capacity_policy)
+      
+      create(:enterprise_inbox_capacity_limit,
+             agent_capacity_policy: capacity_policy,
+             inbox: inbox,
+             conversation_limit: 3)
+    end
+
+    it 'respects agent capacity limits' do
+      # Fill agent1 to capacity
+      create_list(:conversation, 3, inbox: inbox, assignee: agent1, status: :open)
+      
+      # Create new conversations
+      new_conversations = create_list(:conversation, 4, inbox: inbox, assignee: nil)
+      
+      # Mock capacity manager
+      capacity_manager = instance_double('Enterprise::AssignmentV2::CapacityManager')
+      allow(Enterprise::AssignmentV2::CapacityManager).to receive(:new).and_return(capacity_manager)
+      
+      # Agent1 at capacity, agent2 has room
+      allow(capacity_manager).to receive(:get_agent_capacity).with(agent1, inbox).and_return(
+        { available_capacity: 0, current_assignments: 3, total_capacity: 3 }
+      )
+      allow(capacity_manager).to receive(:get_agent_capacity).with(agent2, inbox).and_return(
+        { available_capacity: 3, current_assignments: 0, total_capacity: 3 }
+      )
+      
+      # Process assignments
+      service = AssignmentV2::AssignmentService.new(inbox)
+      assigned_count = service.assign_conversations
+      
+      # Only 3 should be assigned (agent2's capacity)
+      expect(assigned_count).to eq(3)
+      
+      # All should go to agent2
+      assigned_conversations = new_conversations.select { |c| c.reload.assignee.present? }
+      expect(assigned_conversations.map(&:assignee)).to all(eq(agent2))
+    end
+
+    it 'handles capacity policy with exclusion rules' do
+      # Update capacity policy with exclusion rules
+      capacity_policy.update!(
+        exclusion_rules: {
+          'labels' => ['urgent'],
+          'hours_threshold' => 24
+        }
+      )
+      
+      # Create urgent label
+      urgent_label = create(:label, account: account, title: 'urgent')
+      
+      # Create mixed conversations for agent1
+      regular_conv = create(:conversation, inbox: inbox, assignee: agent1, status: :open)
+      urgent_conv = create(:conversation, inbox: inbox, assignee: agent1, status: :open)
+      create(:conversation_label, conversation: urgent_conv, label: urgent_label)
+      old_conv = create(:conversation, inbox: inbox, assignee: agent1, status: :open, created_at: 2.days.ago)
+      
+      # Mock capacity calculation with exclusions
+      capacity_manager = instance_double('Enterprise::AssignmentV2::CapacityManager')
+      allow(Enterprise::AssignmentV2::CapacityManager).to receive(:new).and_return(capacity_manager)
+      
+      # Only regular conversation counts toward capacity
+      allow(capacity_manager).to receive(:get_agent_capacity).with(agent1, inbox).and_return(
+        { available_capacity: 2, current_assignments: 1, total_capacity: 3 }
+      )
+      
+      # New conversation should still be assignable
+      new_conversation = create(:conversation, inbox: inbox, assignee: nil)
+      
+      service = AssignmentV2::AssignmentService.new(inbox)
+      expect(service.assign_conversation(new_conversation)).to be true
+    end
+  end
+
+  describe 'Team-based Assignment' do
+    let(:team) { create(:team, account: account) }
+    let(:assignment_policy) { create(:assignment_policy, account: account, enabled: true) }
+    let!(:inbox_assignment_policy) { create(:inbox_assignment_policy, inbox: inbox, assignment_policy: assignment_policy) }
+    
+    before do
+      # Add only agent1 and agent2 to team
+      create(:team_member, team: team, user: agent1)
+      create(:team_member, team: team, user: agent2)
+    end
+
+    it 'assigns only to team members when conversation has team' do
+      # Create conversation with team
+      conversation = create(:conversation, inbox: inbox, assignee: nil, team: team)
+      
+      # Mock team filtering in service
+      service = AssignmentV2::AssignmentService.new(inbox)
+      
+      # Should only consider team members
+      100.times do
+        conversation.update!(assignee: nil)
+        service.assign_conversation(conversation)
+        expect(conversation.reload.assignee).to be_in([agent1, agent2])
+      end
+    end
+  end
+
+  describe 'Feature Flag Control' do
+    let(:assignment_policy) { create(:assignment_policy, account: account, enabled: true) }
+    let!(:inbox_assignment_policy) { create(:inbox_assignment_policy, inbox: inbox, assignment_policy: assignment_policy) }
+
+    before do
+      # Mock feature flag
+      allow(inbox).to receive(:assignment_v2_enabled?).and_return(false)
+    end
+
+    it 'falls back to legacy assignment when V2 is disabled' do
+      conversation = create(:conversation, inbox: inbox, assignee: nil)
+      
+      # Enable auto assignment
+      inbox.update!(enable_auto_assignment: true)
+      
+      # Should use legacy service
+      expect(::AutoAssignment::AgentAssignmentService).to receive(:new).and_call_original
+      
+      # Trigger assignment through model callback
+      conversation.update!(status: :open)
+    end
+  end
+
+  describe 'Concurrent Assignment Handling' do
+    let(:assignment_policy) { create(:assignment_policy, account: account, enabled: true) }
+    let!(:inbox_assignment_policy) { create(:inbox_assignment_policy, inbox: inbox, assignment_policy: assignment_policy) }
+
+    it 'handles multiple simultaneous assignment jobs' do
+      conversations = create_list(:conversation, 10, inbox: inbox, assignee: nil)
+      
+      # Simulate concurrent job execution
+      threads = []
+      
+      3.times do
+        threads << Thread.new do
+          AssignmentV2::AssignmentJob.new.perform(inbox_id: inbox.id)
+        end
+      end
+      
+      threads.each(&:join)
+      
+      # All conversations should be assigned without duplicates
+      conversations.each(&:reload)
+      assigned_count = conversations.count { |c| c.assignee.present? }
+      
+      expect(assigned_count).to eq(10)
+      
+      # No conversation should have been assigned multiple times
+      assignment_counts = conversations.group_by(&:assignee).transform_values(&:count)
+      expect(assignment_counts.values.sum).to eq(10)
+    end
+  end
+
+  describe 'Error Recovery' do
+    let(:assignment_policy) { create(:assignment_policy, account: account, enabled: true) }
+    let!(:inbox_assignment_policy) { create(:inbox_assignment_policy, inbox: inbox, assignment_policy: assignment_policy) }
+
+    it 'continues assignment after individual conversation failure' do
+      conversations = create_list(:conversation, 5, inbox: inbox, assignee: nil)
+      
+      # Make one conversation invalid
+      conversations[2].update_column(:status, 'invalid_status')
+      
+      service = AssignmentV2::AssignmentService.new(inbox)
+      assigned_count = service.assign_conversations
+      
+      # Should assign 4 out of 5
+      expect(assigned_count).to eq(4)
+      
+      # Invalid conversation remains unassigned
+      expect(conversations[2].reload.assignee).to be_nil
+    end
+
+    it 'recovers from Redis failures' do
+      # Simulate Redis connection failure
+      allow(Redis::Alfred).to receive(:lpop).and_raise(Redis::CannotConnectError)
+      
+      conversation = create(:conversation, inbox: inbox, assignee: nil)
+      
+      service = AssignmentV2::AssignmentService.new(inbox)
+      
+      # Should fall back to database-based assignment
+      expect(service.assign_conversation(conversation)).to be true
+      expect(conversation.reload.assignee).not_to be_nil
+    end
+  end
+end
\ No newline at end of file
diff --git a/spec/jobs/assignment_v2/assignment_job_spec.rb b/spec/jobs/assignment_v2/assignment_job_spec.rb
new file mode 100644
index 000000000..3039476d6
--- /dev/null
+++ b/spec/jobs/assignment_v2/assignment_job_spec.rb
@@ -0,0 +1,182 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+RSpec.describe AssignmentV2::AssignmentJob, type: :job do
+  let(:account) { create(:account) }
+  let(:inbox) { create(:inbox, account: account) }
+  let(:conversation) { create(:conversation, inbox: inbox, assignee: nil) }
+  let(:assignment_policy) { create(:assignment_policy, account: account, enabled: true) }
+  let!(:inbox_assignment_policy) { create(:inbox_assignment_policy, inbox: inbox, assignment_policy: assignment_policy) }
+
+  describe '#perform' do
+    context 'with conversation_id' do
+      it 'assigns a single conversation' do
+        service = instance_double(AssignmentV2::AssignmentService)
+        expect(AssignmentV2::AssignmentService).to receive(:new).with(inbox).and_return(service)
+        expect(service).to receive(:assign_conversation).with(conversation)
+
+        described_class.new.perform(conversation_id: conversation.id)
+      end
+
+      it 'handles non-existent conversation gracefully' do
+        expect(AssignmentV2::AssignmentService).not_to receive(:new)
+        
+        # Should not raise error
+        expect {
+          described_class.new.perform(conversation_id: 999999)
+        }.not_to raise_error
+      end
+    end
+
+    context 'with inbox_id' do
+      let!(:agent) { create(:user, account: account, role: :agent, availability: :online) }
+      let!(:conversations) { create_list(:conversation, 3, inbox: inbox, assignee: nil) }
+
+      before do
+        create(:inbox_member, inbox: inbox, user: agent)
+      end
+
+      it 'assigns multiple conversations for inbox' do
+        service = instance_double(AssignmentV2::AssignmentService)
+        expect(AssignmentV2::AssignmentService).to receive(:new).with(inbox).and_return(service)
+        expect(service).to receive(:assign_conversations).and_return(3)
+
+        described_class.new.perform(inbox_id: inbox.id)
+      end
+
+      it 'logs the number of assigned conversations' do
+        service = instance_double(AssignmentV2::AssignmentService)
+        allow(AssignmentV2::AssignmentService).to receive(:new).with(inbox).and_return(service)
+        allow(service).to receive(:assign_conversations).and_return(2)
+
+        expect(Rails.logger).to receive(:info).with("AssignmentJob: Assigned 2 conversations for inbox #{inbox.id}")
+
+        described_class.new.perform(inbox_id: inbox.id)
+      end
+
+      it 'skips assignment when inbox has no policy' do
+        inbox_assignment_policy.destroy!
+        
+        expect(AssignmentV2::AssignmentService).not_to receive(:new)
+        
+        described_class.new.perform(inbox_id: inbox.id)
+      end
+
+      it 'skips assignment when policy is disabled' do
+        assignment_policy.update!(enabled: false)
+        
+        expect(AssignmentV2::AssignmentService).not_to receive(:new)
+        
+        described_class.new.perform(inbox_id: inbox.id)
+      end
+
+      it 'handles non-existent inbox gracefully' do
+        expect(AssignmentV2::AssignmentService).not_to receive(:new)
+        
+        # Should not raise error
+        expect {
+          described_class.new.perform(inbox_id: 999999)
+        }.not_to raise_error
+      end
+    end
+
+    context 'without parameters' do
+      it 'logs error when no parameters provided' do
+        expect(Rails.logger).to receive(:error).with('AssignmentJob: No inbox_id or conversation_id provided')
+        
+        described_class.new.perform
+      end
+
+      it 'does not attempt assignment' do
+        expect(AssignmentV2::AssignmentService).not_to receive(:new)
+        
+        described_class.new.perform
+      end
+    end
+
+    context 'with both parameters' do
+      it 'prioritizes conversation_id over inbox_id' do
+        service = instance_double(AssignmentV2::AssignmentService)
+        expect(AssignmentV2::AssignmentService).to receive(:new).with(inbox).and_return(service)
+        expect(service).to receive(:assign_conversation).with(conversation)
+        expect(service).not_to receive(:assign_conversations)
+
+        described_class.new.perform(conversation_id: conversation.id, inbox_id: inbox.id)
+      end
+    end
+  end
+
+  describe 'job configuration' do
+    it 'uses the default queue' do
+      expect(described_class.new.queue_name).to eq('default')
+    end
+  end
+
+  describe 'error handling' do
+    context 'when assignment service raises error' do
+      it 'propagates the error for retry' do
+        service = instance_double(AssignmentV2::AssignmentService)
+        allow(AssignmentV2::AssignmentService).to receive(:new).and_return(service)
+        allow(service).to receive(:assign_conversation).and_raise(StandardError, 'Assignment failed')
+
+        expect {
+          described_class.new.perform(conversation_id: conversation.id)
+        }.to raise_error(StandardError, 'Assignment failed')
+      end
+    end
+
+    context 'when database connection fails' do
+      it 'raises error for retry' do
+        allow(Conversation).to receive(:find_by).and_raise(ActiveRecord::ConnectionNotEstablished)
+
+        expect {
+          described_class.new.perform(conversation_id: conversation.id)
+        }.to raise_error(ActiveRecord::ConnectionNotEstablished)
+      end
+    end
+  end
+
+  describe 'concurrency and idempotency' do
+    it 'handles concurrent job execution safely' do
+      # Create multiple jobs for same inbox
+      jobs = []
+      3.times { jobs << described_class.new }
+
+      # All should execute without issues
+      expect {
+        jobs.each { |job| job.perform(inbox_id: inbox.id) }
+      }.not_to raise_error
+    end
+
+    it 'is idempotent for conversation assignment' do
+      service = instance_double(AssignmentV2::AssignmentService)
+      allow(AssignmentV2::AssignmentService).to receive(:new).and_return(service)
+      
+      # First call assigns
+      allow(service).to receive(:assign_conversation).and_return(true)
+      described_class.new.perform(conversation_id: conversation.id)
+      
+      # Second call should handle already assigned conversation
+      allow(service).to receive(:assign_conversation).and_return(false)
+      described_class.new.perform(conversation_id: conversation.id)
+      
+      # No errors should occur
+    end
+  end
+
+  describe 'performance considerations' do
+    it 'processes large inbox assignments in batches' do
+      # Create many unassigned conversations
+      create_list(:conversation, 100, inbox: inbox, assignee: nil)
+      
+      service = instance_double(AssignmentV2::AssignmentService)
+      allow(AssignmentV2::AssignmentService).to receive(:new).and_return(service)
+      
+      # Service should be called with default limit
+      expect(service).to receive(:assign_conversations).with(no_args).and_return(50)
+      
+      described_class.new.perform(inbox_id: inbox.id)
+    end
+  end
+end
\ No newline at end of file
diff --git a/spec/jobs/assignment_v2/continuous_assignment_job_spec.rb b/spec/jobs/assignment_v2/continuous_assignment_job_spec.rb
new file mode 100644
index 000000000..82a7b7462
--- /dev/null
+++ b/spec/jobs/assignment_v2/continuous_assignment_job_spec.rb
@@ -0,0 +1,223 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+RSpec.describe AssignmentV2::ContinuousAssignmentJob, type: :job do
+  let(:account) { create(:account) }
+  let(:inbox) { create(:inbox, account: account) }
+  let(:assignment_policy) { create(:assignment_policy, account: account) }
+  let!(:inbox_assignment_policy) { create(:inbox_assignment_policy, inbox: inbox, assignment_policy: assignment_policy) }
+
+  before do
+    allow(inbox).to receive(:assignment_v2_enabled?).and_return(true)
+  end
+
+  describe '#perform' do
+    context 'when inbox exists and assignment should run' do
+      let!(:conversation1) { create(:conversation, inbox: inbox, assignee: nil, status: :open) }
+      let!(:conversation2) { create(:conversation, inbox: inbox, assignee: nil, status: :open) }
+
+      it 'runs assignment orchestrator' do
+        orchestrator_double = instance_double(AssignmentV2::AssignmentOrchestrator)
+        expect(AssignmentV2::AssignmentOrchestrator).to receive(:new).with(inbox).and_return(orchestrator_double)
+        expect(orchestrator_double).to receive(:assign_conversations).with(limit: 50).and_return(2)
+
+        described_class.new.perform(inbox.id)
+      end
+
+      it 'logs assignment start and completion' do
+        allow_any_instance_of(AssignmentV2::AssignmentOrchestrator).to receive(:assign_conversations).and_return(2)
+        
+        expect(Rails.logger).to receive(:info).with("Assignment V2: Running continuous assignment for inbox #{inbox.id}")
+        expect(Rails.logger).to receive(:info).with("Assignment V2: Completed continuous assignment for inbox #{inbox.id}, made 2 assignments")
+
+        described_class.new.perform(inbox.id)
+      end
+
+      context 'when more conversations need processing' do
+        it 'schedules next run when batch size equals assignments made' do
+          allow_any_instance_of(AssignmentV2::AssignmentOrchestrator).to receive(:assign_conversations).and_return(50)
+          allow(inbox.conversations.unassigned.open).to receive(:exists?).and_return(true)
+
+          expect(described_class).to receive(:set).with(wait: anything).and_return(described_class)
+          expect(described_class).to receive(:perform_later).with(inbox.id)
+
+          described_class.new.perform(inbox.id, batch_size: 50)
+        end
+
+        it 'does not schedule next run when fewer assignments made than batch size' do
+          allow_any_instance_of(AssignmentV2::AssignmentOrchestrator).to receive(:assign_conversations).and_return(25)
+
+          expect(described_class).not_to receive(:set)
+
+          described_class.new.perform(inbox.id, batch_size: 50)
+        end
+
+        it 'does not schedule next run when no more unassigned conversations' do
+          allow_any_instance_of(AssignmentV2::AssignmentOrchestrator).to receive(:assign_conversations).and_return(50)
+          allow(inbox.conversations.unassigned.open).to receive(:exists?).and_return(false)
+
+          expect(described_class).not_to receive(:set)
+
+          described_class.new.perform(inbox.id, batch_size: 50)
+        end
+      end
+    end
+
+    context 'when inbox does not exist' do
+      it 'logs error and does not raise exception' do
+        expect(Rails.logger).to receive(:error).with("Assignment V2: Inbox 999 not found")
+
+        expect { described_class.new.perform(999) }.not_to raise_error
+      end
+    end
+
+    context 'when assignment should not run' do
+      before do
+        allow(inbox).to receive(:assignment_v2_enabled?).and_return(false)
+      end
+
+      it 'returns early without running assignment' do
+        expect(AssignmentV2::AssignmentOrchestrator).not_to receive(:new)
+
+        described_class.new.perform(inbox.id)
+      end
+    end
+
+    context 'when assignment policy is disabled' do
+      before do
+        assignment_policy.update!(enabled: false)
+      end
+
+      it 'returns early without running assignment' do
+        expect(AssignmentV2::AssignmentOrchestrator).not_to receive(:new)
+
+        described_class.new.perform(inbox.id)
+      end
+    end
+
+    context 'when no unassigned conversations exist' do
+      it 'returns early without running assignment' do
+        expect(AssignmentV2::AssignmentOrchestrator).not_to receive(:new)
+
+        described_class.new.perform(inbox.id)
+      end
+    end
+
+    context 'when assignment fails with error' do
+      before do
+        create(:conversation, inbox: inbox, assignee: nil, status: :open)
+        allow_any_instance_of(AssignmentV2::AssignmentOrchestrator).to receive(:assign_conversations).and_raise(StandardError, 'Assignment failed')
+      end
+
+      it 'logs error and re-raises exception' do
+        expect(Rails.logger).to receive(:error).with("Assignment V2: Continuous assignment failed for inbox #{inbox.id}: Assignment failed")
+
+        expect { described_class.new.perform(inbox.id) }.to raise_error(StandardError, 'Assignment failed')
+      end
+    end
+  end
+
+  describe '.trigger' do
+    let(:conversation) { create(:conversation, inbox: inbox) }
+
+    context 'when inbox has assignment v2 enabled' do
+      before do
+        allow(inbox).to receive(:assignment_v2_enabled?).and_return(true)
+      end
+
+      it 'enqueues job for conversation inbox' do
+        expect(described_class).to receive(:perform_later).with(inbox.id, batch_size: 10)
+
+        described_class.trigger(conversation)
+      end
+    end
+
+    context 'when inbox does not have assignment v2 enabled' do
+      before do
+        allow(inbox).to receive(:assignment_v2_enabled?).and_return(false)
+      end
+
+      it 'does not enqueue job' do
+        expect(described_class).not_to receive(:perform_later)
+
+        described_class.trigger(conversation)
+      end
+    end
+  end
+
+  describe '.process_all_inboxes' do
+    let(:account2) { create(:account) }
+    let(:inbox2) { create(:inbox, account: account2) }
+    let(:policy2) { create(:assignment_policy, account: account2) }
+    let!(:inbox_policy2) { create(:inbox_assignment_policy, inbox: inbox2, assignment_policy: policy2) }
+
+    before do
+      allow(inbox).to receive(:assignment_v2_enabled?).and_return(true)
+      allow(inbox2).to receive(:assignment_v2_enabled?).and_return(true)
+
+      create(:conversation, inbox: inbox, assignee: nil, status: :open)
+      create(:conversation, inbox: inbox2, assignee: nil, status: :open)
+    end
+
+    it 'processes all inboxes with enabled assignment policies' do
+      expect(Rails.logger).to receive(:info).with(/Scheduling bulk assignment for inbox #{inbox.id}/)
+      expect(Rails.logger).to receive(:info).with(/Scheduling bulk assignment for inbox #{inbox2.id}/)
+
+      expect(described_class).to receive(:perform_later).with(inbox.id)
+      expect(described_class).to receive(:perform_later).with(inbox2.id)
+
+      described_class.process_all_inboxes
+    end
+
+    it 'skips inboxes without unassigned conversations' do
+      # Remove unassigned conversations
+      Conversation.update_all(status: :resolved)
+
+      expect(described_class).not_to receive(:perform_later)
+
+      described_class.process_all_inboxes
+    end
+
+    it 'skips inboxes without assignment v2 enabled' do
+      allow(inbox).to receive(:assignment_v2_enabled?).and_return(false)
+
+      expect(described_class).to receive(:perform_later).with(inbox2.id)
+      expect(described_class).not_to receive(:perform_later).with(inbox.id)
+
+      described_class.process_all_inboxes
+    end
+  end
+
+  describe 'delay calculation' do
+    let(:job_instance) { described_class.new }
+
+    it 'calculates delay with base time and jitter' do
+      delay = job_instance.send(:calculate_delay, inbox)
+      
+      expect(delay).to be >= 30.seconds
+      expect(delay).to be <= 40.seconds
+    end
+
+    it 'adds randomization to prevent thundering herd' do
+      delays = []
+      10.times do
+        delays << job_instance.send(:calculate_delay, inbox)
+      end
+
+      # All delays should be different due to randomization
+      expect(delays.uniq.size).to be > 1
+    end
+  end
+
+  describe 'retry configuration' do
+    it 'has proper retry configuration' do
+      expect(described_class.get_sidekiq_options['retry']).to eq(3)
+    end
+
+    it 'uses exponentially longer wait times' do
+      # This tests the retry_on configuration
+      expect(described_class.instance_variable_get(:@retry_callbacks)).to be_present
+    end
+  end
+end
\ No newline at end of file
diff --git a/spec/models/assignment_policy_spec.rb b/spec/models/assignment_policy_spec.rb
new file mode 100644
index 000000000..1c59861b8
--- /dev/null
+++ b/spec/models/assignment_policy_spec.rb
@@ -0,0 +1,124 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+RSpec.describe AssignmentPolicy, type: :model do
+  let(:account) { create(:account) }
+  let(:assignment_policy) { create(:assignment_policy, account: account) }
+
+  describe 'associations' do
+    it { is_expected.to belong_to(:account) }
+    it { is_expected.to have_many(:inbox_assignment_policies).dependent(:destroy) }
+    it { is_expected.to have_many(:inboxes).through(:inbox_assignment_policies) }
+  end
+
+  describe 'validations' do
+    it { is_expected.to validate_presence_of(:name) }
+    it { is_expected.to validate_uniqueness_of(:name).scoped_to(:account_id) }
+    it { is_expected.to validate_length_of(:name).is_at_most(255) }
+    it { is_expected.to validate_length_of(:description).is_at_most(1000) }
+    
+    it { is_expected.to validate_presence_of(:fair_distribution_limit) }
+    it { is_expected.to validate_numericality_of(:fair_distribution_limit).is_greater_than(0).is_less_than_or_equal_to(100) }
+    
+    it { is_expected.to validate_presence_of(:fair_distribution_window) }
+    it { is_expected.to validate_numericality_of(:fair_distribution_window).is_greater_than(60).is_less_than_or_equal_to(86_400) }
+
+    context 'balanced assignment validation' do
+      let(:enterprise_account) { create(:account) }
+      
+      before do
+        allow(enterprise_account).to receive(:feature_enabled?).with(:enterprise_agent_capacity).and_return(true)
+      end
+
+      it 'allows balanced assignment for enterprise accounts' do
+        policy = build(:assignment_policy, account: enterprise_account, assignment_order: :balanced)
+        expect(policy).to be_valid
+      end
+
+      it 'rejects balanced assignment for non-enterprise accounts' do
+        policy = build(:assignment_policy, account: account, assignment_order: :balanced)
+        expect(policy).not_to be_valid
+        expect(policy.errors[:assignment_order]).to include('Balanced assignment is only available for enterprise accounts')
+      end
+    end
+  end
+
+  describe 'enums' do
+    it { is_expected.to define_enum_for(:assignment_order).with_values(round_robin: 0, balanced: 1) }
+    it { is_expected.to define_enum_for(:conversation_priority).with_values(earliest_created: 0, longest_waiting: 1) }
+  end
+
+  describe 'scopes' do
+    let!(:enabled_policy) { create(:assignment_policy, account: account, enabled: true) }
+    let!(:disabled_policy) { create(:assignment_policy, account: account, enabled: false) }
+
+    it 'filters enabled policies' do
+      expect(AssignmentPolicy.enabled).to include(enabled_policy)
+      expect(AssignmentPolicy.enabled).not_to include(disabled_policy)
+    end
+
+    it 'filters disabled policies' do
+      expect(AssignmentPolicy.disabled).to include(disabled_policy)
+      expect(AssignmentPolicy.disabled).not_to include(enabled_policy)
+    end
+  end
+
+  describe '#can_use_balanced_assignment?' do
+    context 'when account has enterprise agent capacity feature' do
+      before do
+        allow(account).to receive(:feature_enabled?).with(:enterprise_agent_capacity).and_return(true)
+      end
+
+      it 'returns true' do
+        expect(assignment_policy.can_use_balanced_assignment?).to be true
+      end
+    end
+
+    context 'when account does not have enterprise features' do
+      before do
+        allow(account).to receive(:feature_enabled?).with(:enterprise_agent_capacity).and_return(false)
+      end
+
+      it 'returns false' do
+        expect(assignment_policy.can_use_balanced_assignment?).to be false
+      end
+    end
+  end
+
+  describe '#webhook_data' do
+    it 'returns correct data structure' do
+      data = assignment_policy.webhook_data
+      
+      expect(data).to include(
+        id: assignment_policy.id,
+        name: assignment_policy.name,
+        description: assignment_policy.description,
+        assignment_order: assignment_policy.assignment_order,
+        conversation_priority: assignment_policy.conversation_priority,
+        fair_distribution_limit: assignment_policy.fair_distribution_limit,
+        fair_distribution_window: assignment_policy.fair_distribution_window,
+        enabled: assignment_policy.enabled
+      )
+    end
+  end
+
+  describe 'cache invalidation' do
+    let(:inbox) { create(:inbox, account: account) }
+    let!(:inbox_policy) { create(:inbox_assignment_policy, inbox: inbox, assignment_policy: assignment_policy) }
+
+    it 'clears assignment caches on update' do
+      expect(Rails.cache).to receive(:delete).with("assignment_v2:policy:#{assignment_policy.id}")
+      expect(Rails.cache).to receive(:delete).with("assignment_v2:inbox_policy:#{inbox.id}")
+      
+      assignment_policy.update!(name: 'Updated Policy')
+    end
+
+    it 'clears assignment caches on destroy' do
+      expect(Rails.cache).to receive(:delete).with("assignment_v2:policy:#{assignment_policy.id}")
+      expect(Rails.cache).to receive(:delete).with("assignment_v2:inbox_policy:#{inbox.id}")
+      
+      assignment_policy.destroy!
+    end
+  end
+end
\ No newline at end of file
diff --git a/spec/models/inbox_assignment_policy_spec.rb b/spec/models/inbox_assignment_policy_spec.rb
new file mode 100644
index 000000000..462e580de
--- /dev/null
+++ b/spec/models/inbox_assignment_policy_spec.rb
@@ -0,0 +1,161 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+RSpec.describe InboxAssignmentPolicy, type: :model do
+  let(:account) { create(:account) }
+  let(:inbox) { create(:inbox, account: account) }
+  let(:assignment_policy) { create(:assignment_policy, account: account) }
+  let(:inbox_assignment_policy) { create(:inbox_assignment_policy, inbox: inbox, assignment_policy: assignment_policy) }
+
+  describe 'associations' do
+    it { is_expected.to belong_to(:inbox) }
+    it { is_expected.to belong_to(:assignment_policy) }
+  end
+
+  describe 'validations' do
+    subject { inbox_assignment_policy }
+    
+    it { is_expected.to validate_uniqueness_of(:inbox_id) }
+
+    context 'inbox and policy from different accounts' do
+      let(:other_account) { create(:account) }
+      let(:other_policy) { create(:assignment_policy, account: other_account) }
+
+      it 'validates inbox belongs to same account as policy' do
+        invalid_policy = build(:inbox_assignment_policy, inbox: inbox, assignment_policy: other_policy)
+        
+        expect(invalid_policy).not_to be_valid
+        expect(invalid_policy.errors[:inbox]).to include('must belong to the same account as the assignment policy')
+      end
+    end
+  end
+
+  describe 'delegations' do
+    it 'delegates account to inbox' do
+      expect(inbox_assignment_policy.account).to eq(account)
+    end
+
+    it 'delegates policy attributes' do
+      expect(inbox_assignment_policy.policy_name).to eq(assignment_policy.name)
+      expect(inbox_assignment_policy.policy_description).to eq(assignment_policy.description)
+      expect(inbox_assignment_policy.policy_assignment_order).to eq(assignment_policy.assignment_order)
+      expect(inbox_assignment_policy.policy_conversation_priority).to eq(assignment_policy.conversation_priority)
+      expect(inbox_assignment_policy.policy_fair_distribution_limit).to eq(assignment_policy.fair_distribution_limit)
+      expect(inbox_assignment_policy.policy_fair_distribution_window).to eq(assignment_policy.fair_distribution_window)
+      expect(inbox_assignment_policy.policy_enabled?).to eq(assignment_policy.enabled?)
+    end
+  end
+
+  describe 'scopes' do
+    let!(:enabled_policy) { create(:assignment_policy, account: account, enabled: true) }
+    let!(:disabled_policy) { create(:assignment_policy, account: account, enabled: false) }
+    let(:inbox2) { create(:inbox, account: account) }
+    let!(:enabled_inbox_policy) { create(:inbox_assignment_policy, inbox: inbox, assignment_policy: enabled_policy) }
+    let!(:disabled_inbox_policy) { create(:inbox_assignment_policy, inbox: inbox2, assignment_policy: disabled_policy) }
+
+    describe '.enabled' do
+      it 'returns only inbox policies with enabled assignment policies' do
+        expect(InboxAssignmentPolicy.enabled).to include(enabled_inbox_policy)
+        expect(InboxAssignmentPolicy.enabled).not_to include(disabled_inbox_policy)
+      end
+    end
+
+    describe '.disabled' do
+      it 'returns only inbox policies with disabled assignment policies' do
+        expect(InboxAssignmentPolicy.disabled).to include(disabled_inbox_policy)
+        expect(InboxAssignmentPolicy.disabled).not_to include(enabled_inbox_policy)
+      end
+    end
+  end
+
+  describe '#webhook_data' do
+    it 'returns correct data structure' do
+      data = inbox_assignment_policy.webhook_data
+      
+      expect(data).to include(
+        id: inbox_assignment_policy.id,
+        inbox_id: inbox.id,
+        assignment_policy_id: assignment_policy.id
+      )
+      
+      expect(data[:policy]).to eq(assignment_policy.webhook_data)
+    end
+  end
+
+  describe 'cache management' do
+    it 'clears inbox cache on create' do
+      expect(Rails.cache).to receive(:delete).with("assignment_v2:inbox_policy:#{inbox.id}")
+      
+      create(:inbox_assignment_policy, inbox: inbox, assignment_policy: assignment_policy)
+    end
+
+    it 'clears inbox cache on update' do
+      expect(Rails.cache).to receive(:delete).with("assignment_v2:inbox_policy:#{inbox.id}")
+      
+      inbox_assignment_policy.update!(updated_at: Time.current)
+    end
+
+    it 'clears inbox cache on destroy' do
+      expect(Rails.cache).to receive(:delete).with("assignment_v2:inbox_policy:#{inbox.id}")
+      
+      inbox_assignment_policy.destroy!
+    end
+
+    it 'updates account cache' do
+      # AccountCacheRevalidator concern should trigger cache update
+      expect(inbox_assignment_policy).to receive(:update_account_cache)
+      
+      inbox_assignment_policy.send(:clear_inbox_cache)
+    end
+  end
+
+  describe 'business logic constraints' do
+    it 'prevents multiple policies per inbox' do
+      policy2 = create(:assignment_policy, account: account)
+      
+      # First policy already exists
+      expect {
+        create(:inbox_assignment_policy, inbox: inbox, assignment_policy: policy2)
+      }.to raise_error(ActiveRecord::RecordInvalid)
+    end
+
+    it 'allows reassigning to different policy' do
+      policy2 = create(:assignment_policy, account: account)
+      
+      expect {
+        inbox_assignment_policy.update!(assignment_policy: policy2)
+      }.not_to raise_error
+      
+      expect(inbox_assignment_policy.reload.assignment_policy).to eq(policy2)
+    end
+  end
+
+  describe 'edge cases' do
+    it 'handles nil associations gracefully' do
+      # Build without saving to test nil handling
+      policy = build(:inbox_assignment_policy, inbox: nil, assignment_policy: nil)
+      
+      expect { policy.valid? }.not_to raise_error
+      expect(policy).not_to be_valid
+    end
+
+    it 'handles policy deletion cascade' do
+      inbox_policy_id = inbox_assignment_policy.id
+      
+      # Deleting policy should delete inbox assignment
+      assignment_policy.destroy!
+      
+      expect(InboxAssignmentPolicy.find_by(id: inbox_policy_id)).to be_nil
+    end
+
+    it 'handles inbox deletion cascade' do
+      inbox_policy_id = inbox_assignment_policy.id
+      
+      # Deleting inbox should delete inbox assignment
+      inbox.destroy!
+      
+      expect(InboxAssignmentPolicy.find_by(id: inbox_policy_id)).to be_nil
+    end
+  end
+end
\ No newline at end of file
diff --git a/spec/services/assignment_v2/assignment_orchestrator_spec.rb b/spec/services/assignment_v2/assignment_orchestrator_spec.rb
new file mode 100644
index 000000000..c8e947d2f
--- /dev/null
+++ b/spec/services/assignment_v2/assignment_orchestrator_spec.rb
@@ -0,0 +1,237 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+RSpec.describe AssignmentV2::AssignmentOrchestrator, type: :service 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(:agent1) { create(:user, account: account) }
+  let(:agent2) { create(:user, account: account) }
+  let(:orchestrator) { described_class.new(inbox) }
+
+  before do
+    create(:inbox_member, inbox: inbox, user: agent1)
+    create(:inbox_member, inbox: inbox, user: agent2)
+    allow(inbox).to receive(:assignment_v2_enabled?).and_return(true)
+  end
+
+  describe '#initialize' do
+    it 'sets up orchestrator with inbox and policy' do
+      expect(orchestrator.inbox).to eq(inbox)
+      expect(orchestrator.policy).to eq(assignment_policy)
+    end
+
+    it 'initializes rate limiter when policy exists' do
+      expect(orchestrator.instance_variable_get(:@rate_limiter)).to be_present
+    end
+
+    it 'initializes metrics tracker' do
+      expect(orchestrator.metrics).to be_a(described_class::AssignmentMetrics)
+    end
+  end
+
+  describe '#assign_conversations' do
+    let!(:conversation1) { create(:conversation, inbox: inbox, assignee: nil, status: :open) }
+    let!(:conversation2) { create(:conversation, inbox: inbox, assignee: nil, status: :open) }
+
+    context 'when assignment is possible' do
+      before do
+        allow_any_instance_of(AssignmentV2::RoundRobinSelector).to receive(:select_agent).and_return(agent1)
+        allow_any_instance_of(AssignmentV2::RateLimiter).to receive(:agent_within_limits?).and_return(true)
+      end
+
+      it 'assigns conversations to agents' do
+        expect(orchestrator.assign_conversations(limit: 2)).to eq(2)
+        
+        expect(conversation1.reload.assignee).to eq(agent1)
+        expect(conversation2.reload.assignee).to eq(agent1)
+      end
+
+      it 'creates audit logs for assignments' do
+        expect { orchestrator.assign_conversations(limit: 2) }.to change { conversation1.messages.activity.count }.by(1)
+      end
+
+      it 'triggers assignment notifications' do
+        expect(Rails.configuration.dispatcher).to receive(:dispatch).with(
+          'conversation.assigned',
+          anything,
+          hash_including(conversation: conversation1, assignee: agent1)
+        ).once
+
+        expect(Rails.configuration.dispatcher).to receive(:dispatch).with(
+          'conversation.assigned', 
+          anything,
+          hash_including(conversation: conversation2, assignee: agent1)
+        ).once
+
+        orchestrator.assign_conversations(limit: 2)
+      end
+
+      it 'records metrics for successful assignments' do
+        orchestrator.assign_conversations(limit: 2)
+        
+        metrics = orchestrator.metrics.instance_variable_get(:@assignments)
+        expect(metrics.size).to eq(2)
+        expect(metrics.first).to include(
+          conversation_id: conversation1.id,
+          agent_id: agent1.id,
+          policy_id: assignment_policy.id
+        )
+      end
+    end
+
+    context 'when no agent is available' do
+      before do
+        allow_any_instance_of(AssignmentV2::RoundRobinSelector).to receive(:select_agent).and_return(nil)
+      end
+
+      it 'does not assign conversations' do
+        expect(orchestrator.assign_conversations(limit: 2)).to eq(0)
+        
+        expect(conversation1.reload.assignee).to be_nil
+        expect(conversation2.reload.assignee).to be_nil
+      end
+    end
+
+    context 'when rate limiter blocks assignment' do
+      before do
+        allow_any_instance_of(AssignmentV2::RoundRobinSelector).to receive(:select_agent).and_return(agent1)
+        allow_any_instance_of(AssignmentV2::RateLimiter).to receive(:agent_within_limits?).and_return(false)
+      end
+
+      it 'does not perform assignment' do
+        expect(orchestrator.assign_conversations(limit: 2)).to eq(0)
+        
+        expect(conversation1.reload.assignee).to be_nil
+        expect(conversation2.reload.assignee).to be_nil
+      end
+    end
+
+    context 'when assignment fails due to database error' do
+      before do
+        allow_any_instance_of(AssignmentV2::RoundRobinSelector).to receive(:select_agent).and_return(agent1)
+        allow_any_instance_of(AssignmentV2::RateLimiter).to receive(:agent_within_limits?).and_return(true)
+        allow(conversation1).to receive(:update!).and_raise(ActiveRecord::RecordInvalid)
+      end
+
+      it 'continues with other conversations' do
+        expect(Rails.logger).to receive(:error).with(/Assignment failed/)
+        
+        result = orchestrator.assign_conversations(limit: 2)
+        expect(result).to eq(1) # Only conversation2 succeeds
+        expect(conversation2.reload.assignee).to eq(agent1)
+      end
+    end
+  end
+
+  describe '#assign_conversation' do
+    let(:conversation) { create(:conversation, inbox: inbox, assignee: nil, status: :open) }
+
+    context 'when assignment succeeds' do
+      before do
+        allow_any_instance_of(AssignmentV2::RoundRobinSelector).to receive(:select_agent).and_return(agent1)
+        allow_any_instance_of(AssignmentV2::RateLimiter).to receive(:agent_within_limits?).and_return(true)
+      end
+
+      it 'returns true and assigns conversation' do
+        expect(orchestrator.assign_conversation(conversation)).to be true
+        expect(conversation.reload.assignee).to eq(agent1)
+      end
+    end
+
+    context 'when conversation is already assigned' do
+      let(:conversation) { create(:conversation, inbox: inbox, assignee: agent2, status: :open) }
+
+      it 'returns false without changing assignment' do
+        expect(orchestrator.assign_conversation(conversation)).to be false
+        expect(conversation.reload.assignee).to eq(agent2)
+      end
+    end
+  end
+
+  describe 'enterprise balanced assignment' do
+    let(:enterprise_account) { create(:account) }
+    let(:enterprise_inbox) { create(:inbox, account: enterprise_account) }
+    let(:balanced_policy) { create(:assignment_policy, account: enterprise_account, assignment_order: :balanced) }
+    let!(:enterprise_inbox_policy) { create(:inbox_assignment_policy, inbox: enterprise_inbox, assignment_policy: balanced_policy) }
+    let(:enterprise_orchestrator) { described_class.new(enterprise_inbox) }
+
+    before do
+      allow(enterprise_inbox).to receive(:assignment_v2_enabled?).and_return(true)
+      allow(enterprise_account).to receive(:feature_enabled?).with(:enterprise_agent_capacity).and_return(true)
+      stub_const('Enterprise', Module.new)
+    end
+
+    it 'uses balanced selector for enterprise accounts' do
+      conversation = create(:conversation, inbox: enterprise_inbox, assignee: nil, status: :open)
+      
+      balanced_selector_double = instance_double('Enterprise::AssignmentV2::BalancedSelector')
+      expect(Enterprise::AssignmentV2::BalancedSelector).to receive(:new).with(enterprise_inbox, balanced_policy).and_return(balanced_selector_double)
+      expect(balanced_selector_double).to receive(:select_agent).and_return(agent1)
+      
+      allow_any_instance_of(AssignmentV2::RateLimiter).to receive(:agent_within_limits?).and_return(true)
+      
+      enterprise_orchestrator.assign_conversation(conversation)
+    end
+  end
+
+  describe '#can_assign?' do
+    it 'returns true when policy is enabled and inbox has auto assignment' do
+      expect(orchestrator.send(:can_assign?)).to be true
+    end
+
+    it 'returns false when policy is disabled' do
+      assignment_policy.update!(enabled: false)
+      expect(orchestrator.send(:can_assign?)).to be false
+    end
+
+    it 'returns false when inbox auto assignment is disabled' do
+      inbox.update!(enable_auto_assignment: false)
+      expect(orchestrator.send(:can_assign?)).to be false
+    end
+
+    it 'returns false when no policy exists' do
+      inbox_assignment_policy.destroy!
+      orchestrator_without_policy = described_class.new(inbox)
+      expect(orchestrator_without_policy.send(:can_assign?)).to be false
+    end
+  end
+
+  describe 'conversation prioritization' do
+    let!(:oldest_conversation) { create(:conversation, inbox: inbox, assignee: nil, status: :open, created_at: 2.hours.ago) }
+    let!(:newest_conversation) { create(:conversation, inbox: inbox, assignee: nil, status: :open, created_at: 1.hour.ago) }
+
+    context 'with earliest_created priority' do
+      before do
+        assignment_policy.update!(conversation_priority: :earliest_created)
+        allow_any_instance_of(AssignmentV2::RoundRobinSelector).to receive(:select_agent).and_return(agent1)
+        allow_any_instance_of(AssignmentV2::RateLimiter).to receive(:agent_within_limits?).and_return(true)
+      end
+
+      it 'processes oldest conversation first' do
+        orchestrator.assign_conversations(limit: 1)
+        expect(oldest_conversation.reload.assignee).to eq(agent1)
+        expect(newest_conversation.reload.assignee).to be_nil
+      end
+    end
+
+    context 'with longest_waiting priority' do
+      before do
+        assignment_policy.update!(conversation_priority: :longest_waiting)
+        oldest_conversation.update!(last_activity_at: 3.hours.ago)
+        newest_conversation.update!(last_activity_at: 30.minutes.ago)
+        
+        allow_any_instance_of(AssignmentV2::RoundRobinSelector).to receive(:select_agent).and_return(agent1)
+        allow_any_instance_of(AssignmentV2::RateLimiter).to receive(:agent_within_limits?).and_return(true)
+      end
+
+      it 'processes conversation with longest wait time first' do
+        orchestrator.assign_conversations(limit: 1) 
+        expect(oldest_conversation.reload.assignee).to eq(agent1)
+        expect(newest_conversation.reload.assignee).to be_nil
+      end
+    end
+  end
+end
\ No newline at end of file
diff --git a/spec/services/assignment_v2/assignment_service_spec.rb b/spec/services/assignment_v2/assignment_service_spec.rb
new file mode 100644
index 000000000..f8d1531e8
--- /dev/null
+++ b/spec/services/assignment_v2/assignment_service_spec.rb
@@ -0,0 +1,267 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+RSpec.describe AssignmentV2::AssignmentService do
+  let(:account) { create(:account) }
+  let(:inbox) { create(:inbox, account: account) }
+  let(:assignment_policy) { create(:assignment_policy, account: account, enabled: true) }
+  let!(:inbox_assignment_policy) { create(:inbox_assignment_policy, inbox: inbox, assignment_policy: assignment_policy) }
+  let(:service) { described_class.new(inbox) }
+
+  # Create agents
+  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: :offline) }
+  
+  # Make agents members of inbox
+  before do
+    create(:inbox_member, inbox: inbox, user: agent1)
+    create(:inbox_member, inbox: inbox, user: agent2)
+    create(:inbox_member, inbox: inbox, user: agent3)
+  end
+
+  describe '#assign_conversation' do
+    let(:conversation) { create(:conversation, inbox: inbox, assignee: nil) }
+
+    context 'when policy is enabled' do
+      it 'assigns conversation to an available agent' do
+        expect(service.assign_conversation(conversation)).to be true
+        expect(conversation.reload.assignee).to be_in([agent1, agent2])
+      end
+
+      it 'dispatches assignment event' do
+        expect(Rails.configuration.dispatcher).to receive(:dispatch).with(
+          'assignee.changed',
+          anything,
+          hash_including(conversation: conversation, user: anything)
+        )
+        service.assign_conversation(conversation)
+      end
+
+      it 'returns false when no agents are available' do
+        agent1.update!(availability: :offline)
+        agent2.update!(availability: :offline)
+        
+        expect(service.assign_conversation(conversation)).to be false
+        expect(conversation.reload.assignee).to be_nil
+      end
+    end
+
+    context 'when policy is disabled' do
+      before { assignment_policy.update!(enabled: false) }
+
+      it 'does not assign conversation' do
+        expect(service.assign_conversation(conversation)).to be false
+        expect(conversation.reload.assignee).to be_nil
+      end
+    end
+
+    context 'when conversation is already assigned' do
+      before { conversation.update!(assignee: agent1) }
+
+      it 'does not reassign conversation' do
+        expect(service.assign_conversation(conversation)).to be false
+        expect(conversation.reload.assignee).to eq(agent1)
+      end
+    end
+
+    context 'with round robin assignment' do
+      before { assignment_policy.update!(assignment_order: :round_robin) }
+
+      it 'assigns agents in rotation' do
+        conversations = create_list(:conversation, 4, inbox: inbox, assignee: nil)
+        
+        # Clear any existing round robin cache
+        Rails.cache.delete("assignment_v2:round_robin:#{inbox.id}")
+        
+        assignments = conversations.map do |conv|
+          service.assign_conversation(conv)
+          conv.reload.assignee
+        end
+
+        # Should rotate between available agents
+        expect(assignments[0]).to be_in([agent1, agent2])
+        expect(assignments[1]).to be_in([agent1, agent2])
+        expect(assignments[0]).not_to eq(assignments[1]) # Different agents
+        expect(assignments[2]).to eq(assignments[0]) # Back to first agent
+        expect(assignments[3]).to eq(assignments[1]) # Back to second agent
+      end
+    end
+
+    context 'with balanced assignment' do
+      before do
+        assignment_policy.update!(assignment_order: :balanced)
+        # Mock enterprise feature check
+        allow(inbox.account).to receive(:feature_enabled?).with(:enterprise_agent_capacity).and_return(true)
+      end
+
+      it 'assigns to agent with least conversations' do
+        # Create existing assignments
+        create_list(:conversation, 3, inbox: inbox, assignee: agent1, status: :open)
+        create(:conversation, inbox: inbox, assignee: agent2, status: :open)
+        
+        new_conversation = create(:conversation, inbox: inbox, assignee: nil)
+        
+        expect(service.assign_conversation(new_conversation)).to be true
+        expect(new_conversation.reload.assignee).to eq(agent2)
+      end
+
+      it 'only counts open and pending conversations' do
+        # Create resolved conversations (should not count)
+        create_list(:conversation, 5, inbox: inbox, assignee: agent1, status: :resolved)
+        
+        # Create open conversation
+        create(:conversation, inbox: inbox, assignee: agent2, status: :open)
+        
+        new_conversation = create(:conversation, inbox: inbox, assignee: nil)
+        
+        expect(service.assign_conversation(new_conversation)).to be true
+        expect(new_conversation.reload.assignee).to eq(agent1) # Less active conversations
+      end
+    end
+
+    context 'error handling' do
+      it 'returns false and logs error on assignment failure' do
+        allow_any_instance_of(Conversation).to receive(:update!).and_raise(ActiveRecord::RecordInvalid)
+        expect(Rails.logger).to receive(:error).with(/Assignment failed/)
+        
+        expect(service.assign_conversation(conversation)).to be false
+      end
+    end
+  end
+
+  describe '#assign_conversations' do
+    let!(:conversations) { create_list(:conversation, 5, inbox: inbox, assignee: nil, status: :open) }
+
+    context 'when policy is enabled' do
+      it 'assigns multiple conversations' do
+        assigned_count = service.assign_conversations(limit: 3)
+        
+        expect(assigned_count).to eq(3)
+        expect(inbox.conversations.unassigned.count).to eq(2)
+      end
+
+      it 'respects conversation priority order' do
+        # Create conversations with different timestamps
+        old_conversation = create(:conversation, inbox: inbox, assignee: nil, created_at: 1.hour.ago)
+        new_conversation = create(:conversation, inbox: inbox, assignee: nil, created_at: 1.minute.ago)
+        
+        assignment_policy.update!(conversation_priority: :earliest_created)
+        
+        service.assign_conversations(limit: 1)
+        
+        expect(old_conversation.reload.assignee).not_to be_nil
+        expect(new_conversation.reload.assignee).to be_nil
+      end
+
+      it 'handles longest_waiting priority' do
+        # Create conversations with different last activity
+        inactive_conversation = create(:conversation, inbox: inbox, assignee: nil, last_activity_at: 2.hours.ago)
+        active_conversation = create(:conversation, inbox: inbox, assignee: nil, last_activity_at: 5.minutes.ago)
+        
+        assignment_policy.update!(conversation_priority: :longest_waiting)
+        
+        service.assign_conversations(limit: 1)
+        
+        expect(inactive_conversation.reload.assignee).not_to be_nil
+        expect(active_conversation.reload.assignee).to be_nil
+      end
+
+      it 'returns 0 when no conversations to assign' do
+        Conversation.update_all(assignee_id: agent1.id)
+        
+        expect(service.assign_conversations).to eq(0)
+      end
+    end
+
+    context 'when policy is disabled' do
+      before { assignment_policy.update!(enabled: false) }
+
+      it 'does not assign any conversations' do
+        expect(service.assign_conversations).to eq(0)
+        expect(inbox.conversations.unassigned.count).to eq(5)
+      end
+    end
+  end
+
+  describe 'enterprise capacity features' do
+    let(:capacity_policy) { create(:enterprise_agent_capacity_policy, account: account) }
+    let(:conversation) { create(:conversation, inbox: inbox, assignee: nil) }
+
+    before do
+      # Mock enterprise availability
+      stub_const('Enterprise', Module.new)
+      stub_const('Enterprise::AssignmentV2::CapacityManager', Class.new)
+      
+      allow_any_instance_of(Enterprise::AssignmentV2::CapacityManager).to receive(:get_agent_capacity).and_return(
+        { available_capacity: 1 }
+      )
+      
+      allow(assignment_policy).to receive(:capacity_filtering_enabled?).and_return(true)
+      allow(inbox.account).to receive(:feature_enabled?).with(:enterprise_agent_capacity).and_return(true)
+    end
+
+    it 'applies capacity filters when available' do
+      # Mock capacity limits
+      allow_any_instance_of(Enterprise::AssignmentV2::CapacityManager).to receive(:get_agent_capacity)
+        .with(agent1, inbox).and_return({ available_capacity: 0 })
+      allow_any_instance_of(Enterprise::AssignmentV2::CapacityManager).to receive(:get_agent_capacity)
+        .with(agent2, inbox).and_return({ available_capacity: 5 })
+      
+      expect(service.assign_conversation(conversation)).to be true
+      expect(conversation.reload.assignee).to eq(agent2) # Only agent with capacity
+    end
+
+    it 'skips capacity filtering when enterprise not available' do
+      allow(assignment_policy).to receive(:capacity_filtering_enabled?).and_return(false)
+      
+      expect(service.assign_conversation(conversation)).to be true
+      expect(conversation.reload.assignee).to be_in([agent1, agent2])
+    end
+  end
+
+  describe 'cache management' do
+    it 'uses cache for round robin state' do
+      assignment_policy.update!(assignment_order: :round_robin)
+      cache_key = "assignment_v2:round_robin:#{inbox.id}"
+      
+      # First assignment
+      conversation1 = create(:conversation, inbox: inbox, assignee: nil)
+      service.assign_conversation(conversation1)
+      
+      # Check cache was written
+      expect(Rails.cache.read(cache_key)).not_to be_nil
+      
+      # Second assignment should use cached state
+      conversation2 = create(:conversation, inbox: inbox, assignee: nil)
+      expect(Rails.cache).to receive(:read).with(cache_key).and_call_original
+      
+      service.assign_conversation(conversation2)
+    end
+  end
+
+  describe 'edge cases' do
+    it 'handles inbox without policy gracefully' do
+      inbox_assignment_policy.destroy!
+      conversation = create(:conversation, inbox: inbox, assignee: nil)
+      
+      expect(service.assign_conversation(conversation)).to be false
+    end
+
+    it 'handles empty agent list' do
+      InboxMember.destroy_all
+      conversation = create(:conversation, inbox: inbox, assignee: nil)
+      
+      expect(service.assign_conversation(conversation)).to be false
+    end
+
+    it 'filters out agents without inbox membership' do
+      non_member_agent = create(:user, account: account, role: :agent, availability: :online)
+      conversation = create(:conversation, inbox: inbox, assignee: nil)
+      
+      expect(service.assign_conversation(conversation)).to be true
+      expect(conversation.reload.assignee).not_to eq(non_member_agent)
+    end
+  end
+end
\ No newline at end of file
diff --git a/spec/services/assignment_v2/capacity_manager_spec.rb b/spec/services/assignment_v2/capacity_manager_spec.rb
new file mode 100644
index 000000000..7b1a355b9
--- /dev/null
+++ b/spec/services/assignment_v2/capacity_manager_spec.rb
@@ -0,0 +1,260 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+RSpec.describe Enterprise::AssignmentV2::CapacityManager do
+  let(:manager) { described_class.new }
+  let(:account) { create(:account) }
+  let(:inbox) { create(:inbox, account: account) }
+  let(:agent) { create(:user, account: account, role: :agent) }
+  let(:capacity_policy) { create(:enterprise_agent_capacity_policy, account: account) }
+  let!(:account_user) { create(:account_user, account: account, user: agent, agent_capacity_policy: capacity_policy) }
+
+  before do
+    # Clear any existing cache
+    Rails.cache.clear
+  end
+
+  describe '#get_agent_capacity' do
+    context 'with capacity policy and limits' do
+      let!(:inbox_limit) { create(:enterprise_inbox_capacity_limit, agent_capacity_policy: capacity_policy, inbox: inbox, conversation_limit: 10) }
+
+      it 'returns correct capacity data' do
+        capacity = manager.get_agent_capacity(agent, inbox)
+        
+        expect(capacity).to include(
+          total_capacity: 10,
+          current_assignments: 0,
+          available_capacity: 10,
+          policy_id: capacity_policy.id,
+          has_policy: true
+        )
+      end
+
+      it 'counts only open conversations' do
+        # Create conversations with different statuses
+        create_list(:conversation, 3, inbox: inbox, assignee: agent, status: :open)
+        create_list(:conversation, 2, inbox: inbox, assignee: agent, status: :resolved)
+        create(:conversation, inbox: inbox, assignee: agent, status: :snoozed)
+        
+        capacity = manager.get_agent_capacity(agent, inbox)
+        
+        expect(capacity[:current_assignments]).to eq(3)
+        expect(capacity[:available_capacity]).to eq(7)
+      end
+
+      it 'caches capacity data' do
+        # First call
+        manager.get_agent_capacity(agent, inbox)
+        
+        # Second call should use cache
+        expect(Rails.cache).to receive(:fetch).and_call_original
+        manager.get_agent_capacity(agent, inbox)
+      end
+
+      it 'respects cache TTL' do
+        cache_key = "assignment_v2:capacity:#{agent.accounts.first.id}:#{agent.id}:#{inbox.id}"
+        
+        manager.get_agent_capacity(agent, inbox)
+        
+        # Check cache exists with TTL
+        expect(Rails.cache.exist?(cache_key)).to be true
+      end
+    end
+
+    context 'without capacity policy' do
+      before { account_user.update!(agent_capacity_policy: nil) }
+
+      it 'returns unlimited capacity' do
+        capacity = manager.get_agent_capacity(agent, inbox)
+        
+        expect(capacity).to include(
+          total_capacity: 999_999,
+          current_assignments: 0,
+          available_capacity: 999_999,
+          policy_id: nil,
+          has_policy: false
+        )
+      end
+    end
+
+    context 'without inbox limit' do
+      it 'returns unlimited capacity' do
+        capacity = manager.get_agent_capacity(agent, inbox)
+        
+        expect(capacity[:has_policy]).to be false
+        expect(capacity[:available_capacity]).to eq(999_999)
+      end
+    end
+
+    context 'with exclusion rules' do
+      let(:excluded_label) { create(:label, account: account, title: 'vip') }
+      let(:capacity_policy) do
+        create(:enterprise_agent_capacity_policy,
+               account: account,
+               exclusion_rules: {
+                 'labels' => ['vip'],
+                 'hours_threshold' => 24
+               })
+      end
+      let!(:inbox_limit) { create(:enterprise_inbox_capacity_limit, agent_capacity_policy: capacity_policy, inbox: inbox, conversation_limit: 10) }
+
+      it 'excludes conversations with specified labels' do
+        # Create regular conversations
+        create_list(:conversation, 3, inbox: inbox, assignee: agent, status: :open)
+        
+        # Create VIP conversations (should be excluded)
+        vip_conversations = create_list(:conversation, 2, inbox: inbox, assignee: agent, status: :open)
+        vip_conversations.each { |conv| create(:conversation_label, conversation: conv, label: excluded_label) }
+        
+        capacity = manager.get_agent_capacity(agent, inbox)
+        
+        expect(capacity[:current_assignments]).to eq(3) # Only non-VIP conversations
+      end
+
+      it 'excludes old conversations based on hours threshold' do
+        # Create recent conversations
+        create_list(:conversation, 2, inbox: inbox, assignee: agent, status: :open, created_at: 1.hour.ago)
+        
+        # Create old conversations (should be excluded)
+        create_list(:conversation, 3, inbox: inbox, assignee: agent, status: :open, created_at: 2.days.ago)
+        
+        capacity = manager.get_agent_capacity(agent, inbox)
+        
+        expect(capacity[:current_assignments]).to eq(2) # Only recent conversations
+      end
+    end
+
+    context 'error handling' do
+      it 'returns unlimited capacity on error' do
+        allow(Rails.cache).to receive(:fetch).and_raise(StandardError, 'Cache error')
+        expect(Rails.logger).to receive(:error).with(/Capacity manager failed/)
+        
+        capacity = manager.get_agent_capacity(agent, inbox)
+        
+        expect(capacity[:has_policy]).to be false
+        expect(capacity[:available_capacity]).to eq(999_999)
+      end
+    end
+  end
+
+  describe '#get_agents_capacity_status' do
+    let(:agent2) { create(:user, account: account, role: :agent) }
+    let!(:account_user2) { create(:account_user, account: account, user: agent2) }
+
+    it 'returns capacity status for multiple agents' do
+      agents = [agent, agent2]
+      
+      statuses = manager.get_agents_capacity_status(agents, inbox)
+      
+      expect(statuses.length).to eq(2)
+      expect(statuses[0][:agent]).to eq(agent)
+      expect(statuses[1][:agent]).to eq(agent2)
+    end
+  end
+
+  describe '#agent_has_capacity?' do
+    let!(:inbox_limit) { create(:enterprise_inbox_capacity_limit, agent_capacity_policy: capacity_policy, inbox: inbox, conversation_limit: 2) }
+
+    context 'when agent has capacity' do
+      it 'returns true' do
+        create(:conversation, inbox: inbox, assignee: agent, status: :open)
+        
+        expect(manager.agent_has_capacity?(agent, inbox)).to be true
+      end
+    end
+
+    context 'when agent is at capacity' do
+      it 'returns false' do
+        create_list(:conversation, 2, inbox: inbox, assignee: agent, status: :open)
+        
+        expect(manager.agent_has_capacity?(agent, inbox)).to be false
+      end
+    end
+  end
+
+  describe '#get_remaining_capacity' do
+    let!(:inbox_limit) { create(:enterprise_inbox_capacity_limit, agent_capacity_policy: capacity_policy, inbox: inbox, conversation_limit: 5) }
+
+    it 'returns correct remaining capacity' do
+      create_list(:conversation, 3, inbox: inbox, assignee: agent, status: :open)
+      
+      expect(manager.get_remaining_capacity(agent, inbox)).to eq(2)
+    end
+
+    it 'returns 0 when over capacity' do
+      create_list(:conversation, 6, inbox: inbox, assignee: agent, status: :open)
+      
+      expect(manager.get_remaining_capacity(agent, inbox)).to eq(0)
+    end
+  end
+
+  describe '#invalidate_agent_capacity_cache' do
+    it 'clears specific inbox cache when inbox provided' do
+      cache_key = "assignment_v2:capacity:#{agent.accounts.first.id}:#{agent.id}:#{inbox.id}"
+      Rails.cache.write(cache_key, 'test_data')
+      
+      manager.invalidate_agent_capacity_cache(agent, inbox)
+      
+      expect(Rails.cache.read(cache_key)).to be_nil
+    end
+
+    it 'clears all agent caches when inbox not provided' do
+      cache_key1 = "assignment_v2:capacity:#{agent.id}:inbox1"
+      cache_key2 = "assignment_v2:capacity:#{agent.id}:inbox2"
+      
+      Rails.cache.write(cache_key1, 'test_data')
+      Rails.cache.write(cache_key2, 'test_data')
+      
+      expect(Rails.cache).to receive(:delete_matched).with("assignment_v2:capacity:#{agent.id}:*")
+      
+      manager.invalidate_agent_capacity_cache(agent)
+    end
+  end
+
+  describe '#invalidate_inbox_capacity_cache' do
+    it 'clears all capacity caches for inbox' do
+      expect(Rails.cache).to receive(:delete_matched).with("assignment_v2:capacity:*:#{inbox.id}")
+      
+      manager.invalidate_inbox_capacity_cache(inbox)
+    end
+  end
+
+  describe 'edge cases' do
+    it 'handles agent without account correctly' do
+      agent_without_account = create(:user, role: :agent)
+      
+      capacity = manager.get_agent_capacity(agent_without_account, inbox)
+      
+      expect(capacity[:has_policy]).to be false
+    end
+
+    it 'handles multiple capacity policies gracefully' do
+      # Create another policy and try to assign it
+      another_policy = create(:enterprise_agent_capacity_policy, account: account)
+      
+      # Update account user
+      account_user.update!(agent_capacity_policy: another_policy)
+      
+      # Should use the new policy
+      capacity = manager.get_agent_capacity(agent, inbox)
+      expect(capacity[:policy_id]).to eq(another_policy.id)
+    end
+
+    it 'handles conversations from different inboxes' do
+      other_inbox = create(:inbox, account: account)
+      inbox_limit = create(:enterprise_inbox_capacity_limit, 
+                          agent_capacity_policy: capacity_policy, 
+                          inbox: inbox, 
+                          conversation_limit: 5)
+      
+      # Create conversations in different inboxes
+      create_list(:conversation, 3, inbox: inbox, assignee: agent, status: :open)
+      create_list(:conversation, 10, inbox: other_inbox, assignee: agent, status: :open)
+      
+      # Should only count conversations from specified inbox
+      capacity = manager.get_agent_capacity(agent, inbox)
+      expect(capacity[:current_assignments]).to eq(3)
+    end
+  end
+end
\ No newline at end of file
diff --git a/spec/services/assignment_v2/rate_limiter_spec.rb b/spec/services/assignment_v2/rate_limiter_spec.rb
new file mode 100644
index 000000000..52c709b0e
--- /dev/null
+++ b/spec/services/assignment_v2/rate_limiter_spec.rb
@@ -0,0 +1,260 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+RSpec.describe AssignmentV2::RateLimiter, type: :service do
+  let(:account) { create(:account) }
+  let(:policy) { create(:assignment_policy, account: account, fair_distribution_limit: 5, fair_distribution_window: 3600) }
+  let(:agent) { create(:user, account: account) }
+  let(:rate_limiter) { described_class.new(policy) }
+
+  before do
+    # Clear Redis state
+    Redis::Alfred.flushdb
+  end
+
+  describe '#initialize' do
+    it 'sets up rate limiter with policy parameters' do
+      expect(rate_limiter.instance_variable_get(:@limit)).to eq(5)
+      expect(rate_limiter.instance_variable_get(:@window_size)).to eq(3600)
+    end
+  end
+
+  describe '#agent_within_limits?' do
+    context 'when agent has no assignments in current window' do
+      it 'returns true' do
+        expect(rate_limiter.agent_within_limits?(agent)).to be true
+      end
+    end
+
+    context 'when agent is below limit' do
+      before do
+        # Simulate 3 assignments in current window
+        3.times { rate_limiter.increment_agent_assignments(agent) }
+      end
+
+      it 'returns true' do
+        expect(rate_limiter.agent_within_limits?(agent)).to be true
+      end
+    end
+
+    context 'when agent reaches limit' do
+      before do
+        # Simulate reaching the limit (5 assignments)
+        5.times { rate_limiter.increment_agent_assignments(agent) }
+      end
+
+      it 'returns false' do
+        expect(rate_limiter.agent_within_limits?(agent)).to be false
+      end
+    end
+
+    context 'when agent exceeds limit' do
+      before do
+        # Simulate exceeding the limit
+        6.times { rate_limiter.increment_agent_assignments(agent) }
+      end
+
+      it 'returns false' do
+        expect(rate_limiter.agent_within_limits?(agent)).to be false
+      end
+    end
+  end
+
+  describe '#increment_agent_assignments' do
+    it 'increments assignment count for agent' do
+      expect { rate_limiter.increment_agent_assignments(agent) }
+        .to change { rate_limiter.get_agent_assignment_count(agent) }.from(0).to(1)
+    end
+
+    it 'sets expiration on the key' do
+      rate_limiter.increment_agent_assignments(agent)
+      
+      current_window = Time.current.to_i / 3600
+      key = "assignment_v2:rate_limit:#{agent.id}:#{current_window}"
+      
+      ttl = Redis::Alfred.ttl(key)
+      expect(ttl).to be > 0
+      expect(ttl).to be <= 3600
+    end
+
+    context 'when Redis fails' do
+      before do
+        allow(Redis::Alfred).to receive(:multi).and_raise(Redis::ConnectionError)
+        allow(Rails.logger).to receive(:error)
+      end
+
+      it 'logs error and continues without raising' do
+        expect { rate_limiter.increment_agent_assignments(agent) }.not_to raise_error
+        expect(Rails.logger).to have_received(:error).with(/Rate limiter increment failed/)
+      end
+    end
+  end
+
+  describe '#get_agent_assignment_count' do
+    it 'returns 0 for agent with no assignments' do
+      expect(rate_limiter.get_agent_assignment_count(agent)).to eq(0)
+    end
+
+    it 'returns correct count after assignments' do
+      3.times { rate_limiter.increment_agent_assignments(agent) }
+      expect(rate_limiter.get_agent_assignment_count(agent)).to eq(3)
+    end
+
+    context 'when Redis fails' do
+      before do
+        allow(Redis::Alfred).to receive(:get).and_raise(Redis::ConnectionError)
+        allow(Rails.logger).to receive(:error)
+      end
+
+      it 'returns 0 and logs error' do
+        expect(rate_limiter.get_agent_assignment_count(agent)).to eq(0)
+        expect(Rails.logger).to have_received(:error).with(/Rate limiter get count failed/)
+      end
+    end
+  end
+
+  describe '#get_remaining_assignments' do
+    it 'returns full limit when no assignments made' do
+      expect(rate_limiter.get_remaining_assignments(agent)).to eq(5)
+    end
+
+    it 'returns correct remaining count' do
+      2.times { rate_limiter.increment_agent_assignments(agent) }
+      expect(rate_limiter.get_remaining_assignments(agent)).to eq(3)
+    end
+
+    it 'returns 0 when limit reached' do
+      5.times { rate_limiter.increment_agent_assignments(agent) }
+      expect(rate_limiter.get_remaining_assignments(agent)).to eq(0)
+    end
+
+    it 'returns 0 when limit exceeded' do
+      6.times { rate_limiter.increment_agent_assignments(agent) }
+      expect(rate_limiter.get_remaining_assignments(agent)).to eq(0)
+    end
+  end
+
+  describe '#can_assign_to_agent?' do
+    it 'returns true when agent has remaining capacity' do
+      2.times { rate_limiter.increment_agent_assignments(agent) }
+      expect(rate_limiter.can_assign_to_agent?(agent)).to be true
+    end
+
+    it 'returns false when agent has no capacity' do
+      5.times { rate_limiter.increment_agent_assignments(agent) }
+      expect(rate_limiter.can_assign_to_agent?(agent)).to be false
+    end
+
+    it 'checks for specific count requirement' do
+      3.times { rate_limiter.increment_agent_assignments(agent) }
+      
+      expect(rate_limiter.can_assign_to_agent?(agent, 1)).to be true
+      expect(rate_limiter.can_assign_to_agent?(agent, 2)).to be true
+      expect(rate_limiter.can_assign_to_agent?(agent, 3)).to be false
+    end
+  end
+
+  describe '#get_agents_assignment_status' do
+    let(:agent2) { create(:user, account: account) }
+    let(:agents) { [agent, agent2] }
+
+    before do
+      2.times { rate_limiter.increment_agent_assignments(agent) }
+      4.times { rate_limiter.increment_agent_assignments(agent2) }
+    end
+
+    it 'returns status for all agents' do
+      status = rate_limiter.get_agents_assignment_status(agents)
+      
+      expect(status).to be_an(Array)
+      expect(status.size).to eq(2)
+      
+      agent_status = status.find { |s| s[:agent] == agent }
+      expect(agent_status).to include(
+        agent: agent,
+        current_assignments: 2,
+        remaining_assignments: 3,
+        within_limits: true
+      )
+      
+      agent2_status = status.find { |s| s[:agent] == agent2 }
+      expect(agent2_status).to include(
+        agent: agent2,
+        current_assignments: 4,
+        remaining_assignments: 1,
+        within_limits: true
+      )
+    end
+  end
+
+  describe '#reset_agent_limits' do
+    before do
+      3.times { rate_limiter.increment_agent_assignments(agent) }
+    end
+
+    it 'resets agent assignment count to 0' do
+      expect { rate_limiter.reset_agent_limits(agent) }
+        .to change { rate_limiter.get_agent_assignment_count(agent) }.from(3).to(0)
+    end
+
+    context 'when Redis fails' do
+      before do
+        allow(Redis::Alfred).to receive(:del).and_raise(Redis::ConnectionError)
+        allow(Rails.logger).to receive(:error)
+      end
+
+      it 'logs error and continues' do
+        expect { rate_limiter.reset_agent_limits(agent) }.not_to raise_error
+        expect(Rails.logger).to have_received(:error).with(/Rate limiter reset failed/)
+      end
+    end
+  end
+
+  describe '#time_until_next_window' do
+    it 'returns time until next window boundary' do
+      # Mock current time to make test predictable
+      travel_to(Time.zone.parse('2024-01-01 10:30:00')) do
+        time_until = rate_limiter.time_until_next_window
+        expect(time_until).to be > 0
+        expect(time_until).to be <= 3600
+      end
+    end
+  end
+
+  describe 'window boundaries' do
+    it 'resets count in new window' do
+      # Set up assignment in current window
+      2.times { rate_limiter.increment_agent_assignments(agent) }
+      expect(rate_limiter.get_agent_assignment_count(agent)).to eq(2)
+      
+      # Travel to next window (advance by window size)
+      travel(3601.seconds) do
+        expect(rate_limiter.get_agent_assignment_count(agent)).to eq(0)
+        expect(rate_limiter.agent_within_limits?(agent)).to be true
+      end
+    end
+  end
+
+  describe 'concurrent access' do
+    it 'handles concurrent increments correctly' do
+      threads = []
+      results = []
+      
+      # Simulate concurrent assignment requests
+      5.times do
+        threads << Thread.new do
+          results << rate_limiter.agent_within_limits?(agent)
+          rate_limiter.increment_agent_assignments(agent) if results.last
+        end
+      end
+      
+      threads.each(&:join)
+      
+      # Final count should not exceed the limit
+      final_count = rate_limiter.get_agent_assignment_count(agent)
+      expect(final_count).to be <= 5
+      expect(results.count(true)).to eq(final_count)
+    end
+  end
+end
\ No newline at end of file
diff --git a/spec/services/assignment_v2/round_robin_selector_spec.rb b/spec/services/assignment_v2/round_robin_selector_spec.rb
new file mode 100644
index 000000000..ca85f6042
--- /dev/null
+++ b/spec/services/assignment_v2/round_robin_selector_spec.rb
@@ -0,0 +1,170 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+RSpec.describe AssignmentV2::RoundRobinSelector, type: :service do
+  let(:account) { create(:account) }
+  let(:inbox) { create(:inbox, account: account) }
+  let(:policy) { create(:assignment_policy, account: account) }
+  let(:user1) { create(:user, account: account, availability: User::AVAILABILITY_STATUSES['online']) }
+  let(:user2) { create(:user, account: account, availability: User::AVAILABILITY_STATUSES['online']) }
+  let(:user3) { create(:user, account: account, availability: User::AVAILABILITY_STATUSES['offline']) }
+
+  before do
+    create(:inbox_member, inbox: inbox, user: user1)
+    create(:inbox_member, inbox: inbox, user: user2)
+    create(:inbox_member, inbox: inbox, user: user3)
+    create(:account_user, account: account, user: user1, role: 'agent')
+    create(:account_user, account: account, user: user2, role: 'agent')
+    create(:account_user, account: account, user: user3, role: 'agent')
+  end
+
+  describe '#select_agent' do
+    let(:selector) { described_class.new(inbox, policy) }
+
+    context 'when Redis is available' do
+      before do
+        allow(Redis::Alfred).to receive(:set).and_return(true)
+        allow(Redis::Alfred).to receive(:del)
+        allow(Redis::Alfred).to receive(:lpop).and_return(user1.id.to_s)
+        allow(Redis::Alfred).to receive(:rpush)
+        allow(Redis::Alfred).to receive(:multi).and_yield(double(del: nil, rpush: nil, expire: nil))
+      end
+
+      it 'returns an online agent' do
+        result = selector.select_agent
+        expect(result).to be_a(User)
+        expect([user1.id, user2.id]).to include(result.id)
+      end
+
+      it 'excludes offline agents' do
+        allow(selector).to receive(:compute_eligible_agents).and_return([user1.id, user2.id])
+        result = selector.select_agent
+        expect(result&.id).not_to eq(user3.id)
+      end
+
+      it 'handles Redis lock contention gracefully' do
+        allow(Redis::Alfred).to receive(:set).and_return(false)
+        result = selector.select_agent
+        expect(result).to be_nil
+      end
+    end
+
+    context 'when Redis fails' do
+      before do
+        allow(Redis::Alfred).to receive(:set).and_raise(Redis::CannotConnectError)
+      end
+
+      it 'falls back to database selection' do
+        result = selector.select_agent
+        expect(result).to be_a(User)
+        expect([user1.id, user2.id]).to include(result.id)
+      end
+    end
+
+    context 'with rate limiting' do
+      let(:rate_limiter) { instance_double(AssignmentV2::RateLimiter) }
+
+      before do
+        allow(AssignmentV2::RateLimiter).to receive(:new).and_return(rate_limiter)
+      end
+
+      it 'filters agents by rate limits' do
+        allow(rate_limiter).to receive(:agent_within_limits?).with(user1).and_return(true)
+        allow(rate_limiter).to receive(:agent_within_limits?).with(user2).and_return(false)
+        
+        # Mock Redis operations
+        allow(Redis::Alfred).to receive(:set).and_return(true)
+        allow(Redis::Alfred).to receive(:del)
+        allow(Redis::Alfred).to receive(:lpop).and_return(user1.id.to_s)
+        allow(Redis::Alfred).to receive(:rpush)
+        allow(Redis::Alfred).to receive(:multi).and_yield(double(del: nil, rpush: nil, expire: nil))
+
+        result = selector.select_agent
+        expect(result&.id).to eq(user1.id)
+      end
+    end
+
+    context 'with enterprise capacity' do
+      before do
+        stub_const('Enterprise', Module.new)
+        allow(inbox.account).to receive(:feature_enabled?).with(:enterprise_agent_capacity).and_return(true)
+      end
+
+      it 'attempts to filter by capacity when enterprise is available' do
+        capacity_manager = instance_double('Enterprise::AssignmentV2::CapacityManager')
+        stub_const('Enterprise::AssignmentV2::CapacityManager', class_double('Enterprise::AssignmentV2::CapacityManager', new: capacity_manager))
+        
+        allow(capacity_manager).to receive(:get_agent_capacity).and_return({ available_capacity: 5 })
+        
+        # Mock Redis operations
+        allow(Redis::Alfred).to receive(:set).and_return(true)
+        allow(Redis::Alfred).to receive(:del)
+        allow(Redis::Alfred).to receive(:lpop).and_return(user1.id.to_s)
+        allow(Redis::Alfred).to receive(:rpush)
+        allow(Redis::Alfred).to receive(:multi).and_yield(double(del: nil, rpush: nil, expire: nil))
+
+        result = selector.select_agent
+        expect(result).to be_a(User)
+      end
+    end
+  end
+
+  describe '#refresh_queue!' do
+    let(:selector) { described_class.new(inbox, policy) }
+
+    it 'refreshes the Redis queue with eligible agents' do
+      expect(Redis::Alfred).to receive(:multi).and_yield(double(del: nil, rpush: nil, expire: nil))
+      selector.refresh_queue!
+    end
+  end
+
+  describe 'race condition safety' do
+    let(:selector) { described_class.new(inbox, policy) }
+
+    it 'handles concurrent access with Redis locks' do
+      # Simulate lock contention
+      call_count = 0
+      allow(Redis::Alfred).to receive(:set) do |key, value, options|
+        call_count += 1
+        call_count == 1 ? true : false  # First call succeeds, second fails
+      end
+
+      allow(Redis::Alfred).to receive(:del)
+      allow(Redis::Alfred).to receive(:lpop).and_return(user1.id.to_s)
+      allow(Redis::Alfred).to receive(:rpush)
+
+      # Multiple concurrent calls
+      results = []
+      threads = []
+      
+      3.times do
+        threads << Thread.new do
+          results << selector.select_agent
+        end
+      end
+      
+      threads.each(&:join)
+      
+      # At least one should succeed, others should be nil due to lock contention
+      expect(results.compact.length).to be >= 1
+    end
+  end
+
+  describe 'memory and performance' do
+    let(:selector) { described_class.new(inbox, policy) }
+
+    it 'cleans up Redis keys with TTL' do
+      expect(Redis::Alfred).to receive(:multi).and_yield(
+        double(del: nil, expire: receive(:expire).with(anything, AssignmentV2::RoundRobinSelector::QUEUE_TTL.to_i))
+      )
+      
+      allow(Redis::Alfred).to receive(:set).and_return(true)
+      allow(Redis::Alfred).to receive(:del)
+      allow(Redis::Alfred).to receive(:lpop).and_return(user1.id.to_s)
+      allow(Redis::Alfred).to receive(:rpush)
+      
+      selector.select_agent
+    end
+  end
+end
\ No newline at end of file
diff --git a/spec/services/assignment_v2/security_spec.rb b/spec/services/assignment_v2/security_spec.rb
new file mode 100644
index 000000000..72866dbe7
--- /dev/null
+++ b/spec/services/assignment_v2/security_spec.rb
@@ -0,0 +1,199 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+RSpec.describe 'Assignment V2 Security', type: :service do
+  let(:account1) { create(:account) }
+  let(:account2) { create(:account) }
+  let(:inbox1) { create(:inbox, account: account1) }
+  let(:inbox2) { create(:inbox, account: account2) }
+  let(:user1) { create(:user, account: account1) }
+  let(:user2) { create(:user, account: account2) }
+
+  describe 'Cross-account data access prevention' do
+    let(:policy1) { create(:assignment_policy, account: account1) }
+    let(:policy2) { create(:assignment_policy, account: account2) }
+
+    context 'AssignmentPolicy' do
+      it 'prevents cross-account policy assignment' do
+        expect {
+          InboxAssignmentPolicy.create!(
+            inbox: inbox1,
+            assignment_policy: policy2  # Different account policy
+          )
+        }.to raise_error(ActiveRecord::RecordInvalid)
+      end
+
+      it 'validates policy names are unique within account only' do
+        create(:assignment_policy, account: account1, name: 'Default')
+        
+        # Same name in different account should be allowed
+        expect {
+          create(:assignment_policy, account: account2, name: 'Default')
+        }.not_to raise_error
+
+        # Same name in same account should fail
+        expect {
+          create(:assignment_policy, account: account1, name: 'Default')
+        }.to raise_error(ActiveRecord::RecordInvalid)
+      end
+    end
+
+    context 'Enterprise capacity policies' do
+      before do
+        stub_const('Enterprise', Module.new)
+        enterprise_policy_class = Class.new(ApplicationRecord) do
+          include AccountCacheRevalidator
+          
+          self.table_name = 'enterprise_agent_capacity_policies'
+          
+          belongs_to :account
+          validates :name, presence: true, uniqueness: { scope: :account_id }
+        end
+        
+        stub_const('Enterprise::AgentCapacityPolicy', enterprise_policy_class)
+      end
+
+      it 'prevents cross-account agent assignment' do
+        # This test would verify that users from one account 
+        # cannot be assigned to capacity policies from another account
+        expect(true).to be true  # Placeholder - would need actual enterprise models
+      end
+    end
+  end
+
+  describe 'SQL injection prevention' do
+    let(:policy) { create(:assignment_policy, account: account1) }
+    let(:orchestrator) { AssignmentV2::AssignmentOrchestrator.new(inbox1) }
+
+    it 'uses parameterized queries in conversation fetching' do
+      # Test that malicious input in policy configuration doesn't lead to SQL injection
+      malicious_input = "'; DROP TABLE users; --"
+      
+      # Should not raise SQL errors or cause injection
+      expect {
+        # This would test priority ordering with potentially malicious data
+        conversations = orchestrator.send(:fetch_prioritized_conversations, 10)
+      }.not_to raise_error
+    end
+
+    it 'safely handles Arel SQL in longest_waiting priority' do
+      policy.update!(conversation_priority: 'longest_waiting')
+      conversations = orchestrator.send(:fetch_prioritized_conversations, 10)
+      
+      # Should use parameterized Arel queries, not string interpolation
+      expect(conversations).to be_an(ActiveRecord::Relation)
+    end
+  end
+
+  describe 'Redis key isolation' do
+    let(:policy) { create(:assignment_policy, account: account1) }
+    let(:rate_limiter1) { AssignmentV2::RateLimiter.new(policy) }
+    let(:rate_limiter2) { AssignmentV2::RateLimiter.new(policy) }
+
+    it 'uses agent-specific Redis keys to prevent data leaks' do
+      key1 = rate_limiter1.send(:rate_limit_key, user1, 123456)
+      key2 = rate_limiter1.send(:rate_limit_key, user2, 123456)
+      
+      expect(key1).to include(user1.id.to_s)
+      expect(key2).to include(user2.id.to_s)
+      expect(key1).not_to eq(key2)
+    end
+
+    it 'includes window in Redis keys for temporal isolation' do
+      window1 = 123456
+      window2 = 123457
+      
+      key1 = rate_limiter1.send(:rate_limit_key, user1, window1)
+      key2 = rate_limiter1.send(:rate_limit_key, user1, window2)
+      
+      expect(key1).not_to eq(key2)
+      expect(key1).to include(window1.to_s)
+      expect(key2).to include(window2.to_s)
+    end
+  end
+
+  describe 'Authorization checks' do
+    let(:policy) { create(:assignment_policy, account: account1) }
+    let(:selector) { AssignmentV2::RoundRobinSelector.new(inbox1, policy) }
+
+    it 'only selects agents who are inbox members' do
+      create(:account_user, account: account1, user: user1, role: 'agent')
+      create(:account_user, account: account2, user: user2, role: 'agent')
+      
+      # Only user1 should be eligible (user2 is not an inbox member)
+      eligible_agents = selector.send(:compute_eligible_agents)
+      expect(eligible_agents).not_to include(user2.id)
+    end
+
+    it 'only selects agents from the same account' do
+      create(:inbox_member, inbox: inbox1, user: user1)
+      create(:inbox_member, inbox: inbox1, user: user2)  # Cross-account member
+      create(:account_user, account: account1, user: user1, role: 'agent')
+      create(:account_user, account: account2, user: user2, role: 'agent')
+      
+      eligible_agents = selector.send(:compute_eligible_agents)
+      expect(eligible_agents).to include(user1.id)
+      expect(eligible_agents).not_to include(user2.id)
+    end
+  end
+
+  describe 'Input validation and sanitization' do
+    it 'validates assignment policy limits are reasonable' do
+      expect {
+        create(:assignment_policy, 
+          account: account1,
+          fair_distribution_limit: 101  # Over limit
+        )
+      }.to raise_error(ActiveRecord::RecordInvalid)
+
+      expect {
+        create(:assignment_policy, 
+          account: account1,
+          fair_distribution_window: 30  # Under minimum
+        )
+      }.to raise_error(ActiveRecord::RecordInvalid)
+    end
+
+    it 'validates enterprise capacity limits are reasonable' do
+      # Test would validate that conversation limits can't be set to extreme values
+      # This prevents resource exhaustion attacks
+      expect(true).to be true  # Placeholder
+    end
+
+    it 'sanitizes policy names and descriptions' do
+      policy = create(:assignment_policy, 
+        account: account1,
+        name: 'Test Policy',
+        description: 'A test description'
+      )
+      
+      expect(policy.name).to eq('Test Policy')
+      expect(policy.description).to eq('A test description')
+      expect(policy.name.length).to be <= 255
+      expect(policy.description.length).to be <= 1000
+    end
+  end
+
+  describe 'Rate limiting protection' do
+    let(:policy) { create(:assignment_policy, account: account1, fair_distribution_limit: 5) }
+    let(:rate_limiter) { AssignmentV2::RateLimiter.new(policy) }
+
+    before do
+      allow(Redis::Alfred).to receive(:get).and_return('10')  # Over limit
+      allow(Redis::Alfred).to receive(:multi)
+      allow(Rails.logger).to receive(:error)
+    end
+
+    it 'prevents assignment when agent is over rate limit' do
+      expect(rate_limiter.agent_within_limits?(user1)).to be false
+    end
+
+    it 'handles Redis failures gracefully without blocking assignments' do
+      allow(Redis::Alfred).to receive(:get).and_raise(Redis::CannotConnectError)
+      
+      # Should return 0 on Redis failure, effectively disabling rate limiting
+      expect(rate_limiter.agent_within_limits?(user1)).to be true
+    end
+  end
+end
\ No newline at end of file