test(automations): claim due_at guard, paused-row resume, factory episode key

- process-job specs make rows due before running (the sweep only enqueues due rows).
- claim! guard: a row pushed into the future isn't claimed.
- reschedule_paused resets overdue rows; re-enabling the account flag enqueues the
  resume job.
- factory derives a current episode key so rows are episode_current.

spec/factories/automation_rule_pending_executions.rb

@@ -3,7 +3,8 @@ FactoryBot.define do
     account
     automation_rule { association :automation_rule, account: account }
     conversation { association :conversation, account: account }
-    episode_key { "status:#{Time.current.to_i}" }
+    # Derive from production so the row is episode_current (matches the conversation's status).
+    episode_key { AutomationRulePendingExecution.episode_key_for(conversation, nil) }
     due_at { 1.hour.from_now }
     status { :pending }
   end

spec/jobs/automation_rules/process_pending_execution_job_spec.rb

@@ -13,7 +13,8 @@ RSpec.describe AutomationRules::ProcessPendingExecutionJob do
   end
   let(:pending_execution) do
     AutomationRulePendingExecution.schedule(rule: rule, conversation: conversation)
-    AutomationRulePendingExecution.last
+    # The sweep only enqueues due rows, so make it due before the job runs.
+    AutomationRulePendingExecution.last.tap { |row| row.update!(due_at: 1.minute.ago) }
   end
 
   before do
@@ -69,7 +70,7 @@ RSpec.describe AutomationRules::ProcessPendingExecutionJob do
                                             actions: [{ 'action_name' => 'add_label', 'action_params' => ['stale'] }])
     agent_reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
     AutomationRulePendingExecution.schedule(rule: message_rule, conversation: conversation, message: agent_reply)
-    row = AutomationRulePendingExecution.last
+    row = AutomationRulePendingExecution.last.tap { |r| r.update!(due_at: 1.minute.ago) }
 
     # Status change fails the condition but leaves the reply_chase episode (max incoming id) intact.
     conversation.update!(status: :open)
@@ -127,7 +128,7 @@ RSpec.describe AutomationRules::ProcessPendingExecutionJob do
                                             actions: [{ 'action_name' => 'send_message', 'action_params' => ['Just checking in'] }])
     agent_reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
     AutomationRulePendingExecution.schedule(rule: message_rule, conversation: conversation, message: agent_reply)
-    row = AutomationRulePendingExecution.last
+    row = AutomationRulePendingExecution.last.tap { |r| r.update!(due_at: 1.minute.ago) }
 
     job.perform(row.reload)
 
@@ -142,7 +143,7 @@ RSpec.describe AutomationRules::ProcessPendingExecutionJob do
                                             actions: [{ 'action_name' => 'send_message', 'action_params' => ['Just checking in'] }])
     agent_reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
     AutomationRulePendingExecution.schedule(rule: message_rule, conversation: conversation, message: agent_reply)
-    row = AutomationRulePendingExecution.last
+    row = AutomationRulePendingExecution.last.tap { |r| r.update!(due_at: 1.minute.ago) }
 
     create(:message, conversation: conversation, account: account, message_type: :incoming)
     job.perform(row.reload)

spec/models/account_spec.rb

@@ -103,6 +103,21 @@ RSpec.describe Account do
     end
   end
 
+  describe 'resuming delayed automations' do
+    let(:account) { create(:account) }
+
+    it 'enqueues the resume job when delayed_automations is turned back on' do
+      expect { account.enable_features!('delayed_automations') }
+        .to have_enqueued_job(AutomationRules::ResumePausedExecutionsJob).with(account)
+    end
+
+    it 'does not enqueue the resume job when the flag is turned off' do
+      account.enable_features!('delayed_automations')
+      expect { account.disable_features!('delayed_automations') }
+        .not_to have_enqueued_job(AutomationRules::ResumePausedExecutionsJob)
+    end
+  end
+
   describe 'feature flag columns' do
     let(:account) { described_class.new(name: 'Test Account') }
 

spec/models/automation_rule_pending_execution_spec.rb

@@ -150,7 +150,7 @@ RSpec.describe AutomationRulePendingExecution do
   end
 
   describe '#claim!' do
-    let(:row) { create(:automation_rule_pending_execution, account: account, conversation: conversation) }
+    let(:row) { create(:automation_rule_pending_execution, account: account, conversation: conversation, due_at: 1.minute.ago) }
 
     it 'claims a pending row exactly once so a duplicate enqueue cannot double-fire' do
       expect(row.claim!).to be(true)
@@ -158,6 +158,11 @@ RSpec.describe AutomationRulePendingExecution do
       expect(described_class.find(row.id).claim!).to be(false)
     end
 
+    it 'does not claim a row whose due_at was pushed into the future (reply-chase reschedule)' do
+      row.update!(due_at: 1.hour.from_now)
+      expect(row.claim!).to be(false)
+    end
+
     it 'does not claim terminal rows' do
       row.update!(status: :executed)
       expect(row.claim!).to be(false)
@@ -196,4 +201,16 @@ RSpec.describe AutomationRulePendingExecution do
       expect { old_executed.reload }.to raise_error(ActiveRecord::RecordNotFound)
     end
   end
+
+  describe '.reschedule_paused' do
+    it 'resets rows overdue past the window so a resumed account replays them instead of expiring' do
+      expired = create(:automation_rule_pending_execution, account: account, conversation: conversation, due_at: 5.days.ago)
+      within_window = create(:automation_rule_pending_execution, account: account, due_at: 2.days.ago)
+
+      described_class.reschedule_paused(account)
+
+      expect(expired.reload.due_at).to be_within(5.seconds).of(Time.current)
+      expect(within_window.reload.due_at).to be_within(5.seconds).of(2.days.ago)
+    end
+  end
 end