fix(automations): dont fire rescheduled rows early; resume paused rows past expiry

- claimable? now requires the row to still be due, so a reply-chase reschedule that
  pushes due_at forward after the sweep enqueued the row no longer fires early.
- Re-enabling delayed_automations on an account reschedules its overdue pending rows
  (past DUE_WINDOW) via a job enqueued ahead of the next sweep, so a pause longer than
  the expiry window resumes those rows instead of expiring them.

app/jobs/automation_rules/resume_paused_executions_job.rb

@@ -0,0 +1,11 @@
+class AutomationRules::ResumePausedExecutionsJob < ApplicationJob
+  # Enqueued the moment the account flag flips back on, ahead of the next sweep, so overdue rows
+  # are rescheduled before that sweep's per-row jobs could mark them expired.
+  queue_as :medium
+
+  discard_on ActiveJob::DeserializationError
+
+  def perform(account)
+    AutomationRulePendingExecution.reschedule_paused(account)
+  end
+end

app/models/account.rb

@@ -112,6 +112,7 @@ class Account < ApplicationRecord
   before_validation :validate_limit_keys
   after_create_commit :notify_creation
   after_update_commit :clear_unread_conversation_counts_cache, if: :saved_change_to_feature_conversation_unread_counts?
+  after_update_commit :resume_delayed_automations, if: -> { saved_change_to_feature_delayed_automations? && feature_delayed_automations? }
   after_destroy :remove_account_sequences
 
   def agents
@@ -186,6 +187,10 @@ class Account < ApplicationRecord
     ::Conversations::UnreadCounts::Store.clear_account!(id)
   end
 
+  def resume_delayed_automations
+    AutomationRules::ResumePausedExecutionsJob.perform_later(self)
+  end
+
   trigger.after(:insert).for_each(:row) do
     "execute format('create sequence IF NOT EXISTS conv_dpid_seq_%s', NEW.id);"
   end

app/models/automation_rule_pending_execution.rb

@@ -102,6 +102,14 @@ class AutomationRulePendingExecution < ApplicationRecord
       .in_batches(of: 1000).delete_all
   end
 
+  # Rows that came due while an account had delayed automations paused would expire the moment
+  # the sweep reaches them on resume. Reset their clock so pause/resume replays them (still
+  # subject to the fire-time episode/condition re-checks) instead of silently dropping them.
+  def self.reschedule_paused(account)
+    overdue = pending.where(account_id: account.id, due_at: ...DUE_WINDOW.ago)
+    overdue.find_each { |row| row.update!(due_at: Time.current) }
+  end
+
   # Atomic claim: only one worker can move a row into processing, so a row re-enqueued by an
   # overlapping sweep (or after a stale reclaim) cannot double-execute. Refreshing updated_at
   # renews the lock, keeping the row out of the stale window while this worker holds it.
@@ -121,6 +129,8 @@ class AutomationRulePendingExecution < ApplicationRecord
   private
 
   def claimable?
-    pending? || (processing? && updated_at < STALE_PROCESSING_TIMEOUT.ago)
+    # due_at guard: a reply-chase reschedule can push due_at forward after this row was enqueued;
+    # such a row must wait for a later sweep instead of firing early.
+    (pending? && due_at <= Time.current) || (processing? && updated_at < STALE_PROCESSING_TIMEOUT.ago)
   end
 end