fix: atomically claim conversation to prevent duplicate assignment (#14495)

## Description

Fixes a bug under Assignment V2 where a single conversation could be
reassigned dozens of times in a row by the system, producing long stacks
of "Assigned to X by Automation System via <policy>" activity messages
alternating between agents. After this change each unassigned
conversation is assigned exactly once, even on busy inboxes.

## Fixes # (issue)


## Type of change

- [ ] Bug fix (non-breaking change which fixes an issue)

## How Has This Been Tested?

## How to reproduce
1. Enable `assignment_v2` on an account with at least 2 online agents in
an inbox.
2. Generate sustained resolve/snooze activity in the inbox (each one
enqueues `AutoAssignment::AssignmentJob` for the whole inbox).
3. Watch any one unassigned conversation while the jobs drain — pre-fix
it picks up multiple back-to-back "Assigned to …" activity rows
alternating between agents.


## Checklist:

- [ ] My code follows the style guidelines of this project
- [ ] I have performed a self-review of my code
- [ ] I have commented on my code, particularly in hard-to-understand
areas
- [ ] I have made corresponding changes to the documentation
- [ ] My changes generate no new warnings
- [ ] I have added tests that prove my fix is effective or that my
feature works
- [ ] New and existing unit tests pass locally with my changes
- [ ] Any dependent changes have been merged and published in downstream
modules

app/jobs/auto_assignment/assignment_job.rb

@@ -1,21 +1,53 @@
 class AutoAssignment::AssignmentJob < ApplicationJob
   queue_as :default
 
-  def perform(inbox_id:)
+  IN_FLIGHT_TTL = 5.minutes
+
+  # Coalesce per inbox: at most one AssignmentJob per inbox is in-flight
+  # (queued or running) at any time. The marker carries a token so a job only
+  # releases its own claim (a newer job may have taken it after a TTL lapse).
+  def self.enqueue_for_inbox(inbox_id)
+    key = format(::Redis::Alfred::AUTO_ASSIGNMENT_IN_FLIGHT_KEY, inbox_id: inbox_id)
+    token = SecureRandom.uuid
+    return false unless ::Redis::Alfred.set(key, token, nx: true, ex: IN_FLIGHT_TTL)
+
+    return true if perform_later(inbox_id: inbox_id, token: token)
+
+    # Enqueue was halted; release our own claim so the inbox isn't gated until the TTL.
+    ::Redis::Alfred.delete_if_equals(key, token)
+    false
+  rescue StandardError
+    # Enqueue raised after we claimed the gate; release our own claim, then re-raise.
+    ::Redis::Alfred.delete_if_equals(key, token)
+    raise
+  end
+
+  def perform(inbox_id:, token: nil)
     inbox = Inbox.find_by(id: inbox_id)
     return unless inbox
 
     service = AutoAssignment::AssignmentService.new(inbox: inbox)
-
     assigned_count = service.perform_bulk_assignment(limit: bulk_assignment_limit)
     Rails.logger.info "Assigned #{assigned_count} conversations for inbox #{inbox.id}"
   rescue StandardError => e
     Rails.logger.error "Bulk assignment failed for inbox #{inbox_id}: #{e.message}"
     raise e if Rails.env.test?
+  ensure
+    release_in_flight(inbox_id, token)
   end
 
   private
 
+  # Release the in-flight marker only if we still own it. The atomic
+  # compare-and-delete ensures a job whose TTL lapsed can't delete a newer
+  # job's claim. Tokenless (pre-deploy) jobs never claimed a key, so skip.
+  def release_in_flight(inbox_id, token)
+    return if token.nil?
+
+    key = format(::Redis::Alfred::AUTO_ASSIGNMENT_IN_FLIGHT_KEY, inbox_id: inbox_id)
+    ::Redis::Alfred.delete_if_equals(key, token)
+  end
+
   def bulk_assignment_limit
     ENV.fetch('AUTO_ASSIGNMENT_BULK_LIMIT', 100).to_i
   end

app/jobs/auto_assignment/periodic_assignment_job.rb

@@ -10,7 +10,7 @@ class AutoAssignment::PeriodicAssignmentJob < ApplicationJob
           inboxes.each do |inbox|
             next unless inbox.auto_assignment_v2_enabled?
 
-            AutoAssignment::AssignmentJob.perform_later(inbox_id: inbox.id)
+            AutoAssignment::AssignmentJob.enqueue_for_inbox(inbox.id)
           end
         end
       end

app/models/concerns/auto_assignment_handler.rb

@@ -15,8 +15,10 @@ module AutoAssignmentHandler
     return unless should_run_auto_assignment?
 
     if inbox.auto_assignment_v2_enabled?
-      # Use new assignment system
-      AutoAssignment::AssignmentJob.perform_later(inbox_id: inbox.id)
+      # Coalesces bursts of triggers per inbox. Fine if the job runs even when the
+      # surrounding save rolls back: it only scans the inbox's current unassigned
+      # conversations, so running it for an uncommitted change is harmless.
+      AutoAssignment::AssignmentJob.enqueue_for_inbox(inbox.id)
     else
       # Use legacy assignment system
       # If conversation has a team, only consider team members for assignment

app/services/auto_assignment/assignment_service.rb

@@ -72,15 +72,32 @@ class AutoAssignment::AssignmentService
   end
 
   def assign_conversation(conversation, agent)
-    Current.executed_by = inbox.assignment_policy || inbox
-    conversation.update!(assignee: agent)
-    Current.executed_by = nil
+    return false unless claim_and_assign(conversation, agent)
+
+    conversation.reload
 
     rate_limiter = build_rate_limiter(agent)
     rate_limiter.track_assignment(conversation)
 
     dispatch_assignment_event(conversation, agent)
     true
+  end
+
+  # Atomically claim the row so two bulk runs that overlap (the in-flight gate
+  # is best-effort and can lapse on TTL) can't both assign the same conversation.
+  def claim_and_assign(conversation, agent)
+    Current.executed_by = inbox.assignment_policy || inbox
+
+    Conversation.transaction do
+      locked = inbox.conversations
+                    .where(id: conversation.id, assignee_id: nil)
+                    .lock('FOR UPDATE SKIP LOCKED')
+                    .first
+      next false unless locked
+
+      locked.update!(assignee: agent)
+      true
+    end
   ensure
     Current.executed_by = nil
   end

lib/redis/alfred.rb

@@ -25,6 +25,18 @@ module Redis::Alfred
       $alfred.with { |conn| conn.del(key) }
     end
 
+    # atomic compare-and-delete (release a lock only if you still own it); WATCH/MULTI
+    # aborts the delete if the key changes between the check and the delete.
+    def delete_if_equals(key, expected_value)
+      $alfred.with do |conn|
+        conn.watch(key) do
+          next conn.unwatch unless conn.get(key) == expected_value
+
+          conn.multi { |transaction| transaction.del(key) }
+        end
+      end
+    end
+
     # increment a key by 1. throws error if key value is incompatible
     # sets key to 0 before operation if key doesn't exist
     def incr(key)

lib/redis/redis_keys.rb

@@ -73,6 +73,8 @@ module Redis::RedisKeys
   # Track conversation assignments to agents for rate limiting
   ASSIGNMENT_KEY = 'ASSIGNMENT::%<inbox_id>d::AGENT::%<agent_id>d::CONVERSATION::%<conversation_id>d'.freeze
   ASSIGNMENT_KEY_PATTERN = 'ASSIGNMENT::%<inbox_id>d::AGENT::%<agent_id>d::*'.freeze
+  # At-most-one AssignmentJob per inbox in-flight (queued or running); further enqueues are skipped
+  AUTO_ASSIGNMENT_IN_FLIGHT_KEY = 'AUTO_ASSIGNMENT_IN_FLIGHT::%<inbox_id>d'.freeze
 
   ## Account Onboarding
   ACCOUNT_ONBOARDING_ENRICHMENT = 'ONBOARDING_ENRICHMENT::%<account_id>d'.freeze

spec/jobs/auto_assignment/assignment_job_spec.rb

@@ -24,10 +24,11 @@ RSpec.describe AutoAssignment::AssignmentJob, type: :job do
           service = instance_double(AutoAssignment::AssignmentService)
           allow(AutoAssignment::AssignmentService).to receive(:new).and_return(service)
           allow(service).to receive(:perform_bulk_assignment).and_return(3)
-
-          expect(Rails.logger).to receive(:info).with("Assigned 3 conversations for inbox #{inbox.id}")
+          allow(Rails.logger).to receive(:info)
 
           described_class.new.perform(inbox_id: inbox.id)
+
+          expect(Rails.logger).to have_received(:info).with("Assigned 3 conversations for inbox #{inbox.id}")
         end
 
         it 'uses custom bulk limit from environment' do
@@ -67,16 +68,40 @@ RSpec.describe AutoAssignment::AssignmentJob, type: :job do
         service = instance_double(AutoAssignment::AssignmentService)
         allow(AutoAssignment::AssignmentService).to receive(:new).and_return(service)
         allow(service).to receive(:perform_bulk_assignment).and_raise(StandardError, 'Something went wrong')
-
-        expect(Rails.logger).to receive(:error).with("Bulk assignment failed for inbox #{inbox.id}: Something went wrong")
+        allow(Rails.logger).to receive(:error)
 
         expect do
           described_class.new.perform(inbox_id: inbox.id)
         end.to raise_error(StandardError, 'Something went wrong')
+
+        expect(Rails.logger).to have_received(:error).with("Bulk assignment failed for inbox #{inbox.id}: Something went wrong")
       end
     end
   end
 
+  describe '.enqueue_for_inbox' do
+    after { Redis::Alfred.delete(format(Redis::Alfred::AUTO_ASSIGNMENT_IN_FLIGHT_KEY, inbox_id: inbox.id)) }
+
+    it 'enqueues one run per inbox and coalesces concurrent triggers' do
+      allow(described_class).to receive(:perform_later).and_return(true)
+
+      expect(described_class.enqueue_for_inbox(inbox.id)).to be(true)
+      expect(described_class.enqueue_for_inbox(inbox.id)).to be(false)
+      expect(described_class).to have_received(:perform_later).once
+    end
+
+    it 'does not release a newer run marker when its own token is stale' do
+      key = format(Redis::Alfred::AUTO_ASSIGNMENT_IN_FLIGHT_KEY, inbox_id: inbox.id)
+      Redis::Alfred.set(key, 'newer-token', ex: 300)
+      allow(AutoAssignment::AssignmentService).to receive(:new)
+        .and_return(instance_double(AutoAssignment::AssignmentService, perform_bulk_assignment: 0))
+
+      described_class.new.perform(inbox_id: inbox.id, token: 'stale-token')
+
+      expect(Redis::Alfred.get(key)).to eq('newer-token')
+    end
+  end
+
   describe 'job configuration' do
     it 'is queued in the default queue' do
       expect(described_class.queue_name).to eq('default')

spec/jobs/auto_assignment/periodic_assignment_job_spec.rb

@@ -29,7 +29,7 @@ RSpec.describe AutoAssignment::PeriodicAssignmentJob, type: :job do
 
         it 'queues assignment job for eligible inboxes' do
           inbox_assignment_policy # ensure it exists
-          expect(AutoAssignment::AssignmentJob).to receive(:perform_later).with(inbox_id: inbox.id)
+          expect(AutoAssignment::AssignmentJob).to receive(:enqueue_for_inbox).with(inbox.id)
 
           described_class.new.perform
         end
@@ -51,8 +51,8 @@ RSpec.describe AutoAssignment::PeriodicAssignmentJob, type: :job do
 
           allow(Account).to receive(:find_in_batches).and_yield([account]).and_yield([account2])
 
-          expect(AutoAssignment::AssignmentJob).to receive(:perform_later).with(inbox_id: inbox.id)
-          expect(AutoAssignment::AssignmentJob).to receive(:perform_later).with(inbox_id: inbox2.id)
+          expect(AutoAssignment::AssignmentJob).to receive(:enqueue_for_inbox).with(inbox.id)
+          expect(AutoAssignment::AssignmentJob).to receive(:enqueue_for_inbox).with(inbox2.id)
 
           described_class.new.perform
         end
@@ -65,7 +65,7 @@ RSpec.describe AutoAssignment::PeriodicAssignmentJob, type: :job do
         end
 
         it 'does not queue assignment job' do
-          expect(AutoAssignment::AssignmentJob).not_to receive(:perform_later)
+          expect(AutoAssignment::AssignmentJob).not_to receive(:enqueue_for_inbox)
 
           described_class.new.perform
         end
@@ -78,7 +78,7 @@ RSpec.describe AutoAssignment::PeriodicAssignmentJob, type: :job do
       end
 
       it 'does not process the account' do
-        expect(AutoAssignment::AssignmentJob).not_to receive(:perform_later)
+        expect(AutoAssignment::AssignmentJob).not_to receive(:enqueue_for_inbox)
 
         described_class.new.perform
       end