Merge branch 'feature/cw-7513' into feature/cw-7513-specs

- 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/controllers/api/v1/accounts/automation_rules_controller_spec.rb

@@ -451,4 +451,71 @@ RSpec.describe 'Api::V1::Accounts::AutomationRulesController', type: :request do
       end
     end
   end
+
+  describe 'execution_delay handling' do
+    let(:delayed_rule_params) do
+      {
+        name: 'Delayed rule',
+        event_name: 'conversation_updated',
+        execution_delay: 240,
+        conditions: [{ attribute_key: 'status', filter_operator: 'equal_to', values: ['pending'], query_operator: nil }],
+        actions: [{ action_name: 'add_label', action_params: ['stale'] }]
+      }
+    end
+
+    context 'when the delayed_automations feature is enabled' do
+      before { account.enable_features!('delayed_automations') }
+
+      it 'persists and serializes execution_delay' do
+        post "/api/v1/accounts/#{account.id}/automation_rules",
+             headers: administrator.create_new_auth_token,
+             params: delayed_rule_params
+
+        expect(response).to have_http_status(:success)
+        body = JSON.parse(response.body, symbolize_names: true)
+        expect(body[:execution_delay]).to eq(240)
+        expect(account.automation_rules.last.execution_delay).to eq(240)
+      end
+
+      it 'copies execution_delay on clone' do
+        automation_rule = create(:automation_rule, account: account, execution_delay: 240)
+
+        post "/api/v1/accounts/#{account.id}/automation_rules/#{automation_rule.id}/clone",
+             headers: administrator.create_new_auth_token
+
+        expect(response).to have_http_status(:success)
+        expect(account.automation_rules.last.execution_delay).to eq(240)
+      end
+    end
+
+    context 'when the delayed_automations feature is disabled' do
+      it 'rejects a payload carrying execution_delay with 422' do
+        post "/api/v1/accounts/#{account.id}/automation_rules",
+             headers: administrator.create_new_auth_token,
+             params: delayed_rule_params
+
+        expect(response).to have_http_status(:unprocessable_entity)
+        expect(account.automation_rules.count).to eq(0)
+      end
+
+      it 'still accepts payloads without execution_delay' do
+        post "/api/v1/accounts/#{account.id}/automation_rules",
+             headers: administrator.create_new_auth_token,
+             params: delayed_rule_params.except(:execution_delay)
+
+        expect(response).to have_http_status(:success)
+        expect(account.automation_rules.last.execution_delay).to be_nil
+      end
+
+      it 'strips execution_delay when cloning an existing delayed rule' do
+        automation_rule = create(:automation_rule, account: account, execution_delay: 240)
+
+        post "/api/v1/accounts/#{account.id}/automation_rules/#{automation_rule.id}/clone",
+             headers: administrator.create_new_auth_token
+
+        expect(response).to have_http_status(:success)
+        expect(account.automation_rules.last.execution_delay).to be_nil
+      end
+    end
+  end
 end

spec/factories/automation_rule_pending_executions.rb

@@ -0,0 +1,11 @@
+FactoryBot.define do
+  factory :automation_rule_pending_execution do
+    account
+    automation_rule { association :automation_rule, account: account }
+    conversation { association :conversation, account: account }
+    # 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
+end

spec/jobs/automation_rules/process_pending_execution_job_spec.rb

@@ -0,0 +1,155 @@
+require 'rails_helper'
+
+RSpec.describe AutomationRules::ProcessPendingExecutionJob do
+  subject(:job) { described_class.new }
+
+  let(:account) { create(:account) }
+  let(:conversation) { create(:conversation, account: account, status: :pending) }
+  let(:rule) do
+    create(:automation_rule, account: account, event_name: 'conversation_updated', execution_delay: 60,
+                             conditions: [{ 'values' => ['pending'], 'attribute_key' => 'status', 'query_operator' => nil,
+                                            'filter_operator' => 'equal_to' }],
+                             actions: [{ 'action_name' => 'add_label', 'action_params' => ['stale'] }])
+  end
+  let(:pending_execution) do
+    AutomationRulePendingExecution.schedule(rule: rule, conversation: conversation)
+    # 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
+    GlobalConfig.clear_cache
+    account.enable_features!('delayed_automations')
+  end
+
+  it 'runs the actions and marks the row executed when every guard passes' do
+    job.perform(pending_execution.reload)
+
+    expect(pending_execution.reload).to be_executed
+    expect(conversation.reload.label_list).to include('stale')
+  end
+
+  it 'skips with rule_inactive when the rule was disabled' do
+    rule.update!(active: false)
+    job.perform(pending_execution.reload)
+
+    expect(pending_execution.reload).to be_skipped
+    expect(pending_execution.skip_reason).to eq('rule_inactive')
+    expect(conversation.reload.label_list).to be_empty
+  end
+
+  it 'pauses (keeps pending) while the account flag is off, then fires when re-enabled' do
+    account.disable_features!('delayed_automations')
+    job.perform(pending_execution.reload)
+
+    expect(pending_execution.reload).to be_pending
+    expect(conversation.reload.label_list).to be_empty
+
+    account.enable_features!('delayed_automations')
+    described_class.new.perform(pending_execution.reload)
+
+    expect(pending_execution.reload).to be_executed
+    expect(conversation.reload.label_list).to include('stale')
+  end
+
+  it 'skips with episode_moved when the conversation left the armed status' do
+    pending_execution
+    conversation.update!(status: :resolved)
+    job.perform(pending_execution.reload)
+
+    expect(pending_execution.reload).to be_skipped
+    expect(pending_execution.skip_reason).to eq('episode_moved')
+    expect(conversation.reload.label_list).to be_empty
+  end
+
+  it 'skips with conditions_changed when the conversation drifts but the episode is intact' do
+    # A message_created (reply-chase) rule whose extra condition is the conversation status.
+    message_rule = create(:automation_rule, account: account, event_name: 'message_created', execution_delay: 60,
+                                            conditions: [{ 'values' => ['pending'], 'attribute_key' => 'status',
+                                                           'query_operator' => nil, 'filter_operator' => 'equal_to' }],
+                                            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.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)
+    job.perform(row)
+
+    expect(row.reload).to be_skipped
+    expect(row.skip_reason).to eq('conditions_changed')
+  end
+
+  it 'skips with expired when the row is past the due window' do
+    pending_execution.update!(due_at: 4.days.ago)
+    job.perform(pending_execution.reload)
+
+    expect(pending_execution.reload).to be_skipped
+    expect(pending_execution.skip_reason).to eq('expired')
+    expect(conversation.reload.label_list).to be_empty
+  end
+
+  it 'leaves the row untouched without executing when the kill switch is set' do
+    create(:installation_config, name: 'DISABLE_DELAYED_AUTOMATIONS', serialized_value: { value: true }.with_indifferent_access)
+    GlobalConfig.clear_cache
+    job.perform(pending_execution.reload)
+
+    expect(pending_execution.reload).to be_pending
+    expect(conversation.reload.label_list).to be_empty
+  end
+
+  it 'runs the actions once when the same row is processed twice concurrently' do
+    allow(AutomationRules::ActionService).to receive(:new).and_call_original
+    duplicate = AutomationRulePendingExecution.find(pending_execution.id)
+
+    job.perform(pending_execution.reload)
+    described_class.new.perform(duplicate)
+
+    expect(AutomationRules::ActionService).to have_received(:new).once
+    expect(pending_execution.reload).to be_executed
+  end
+
+  it 'leaves the row processing and reports the error when an action blows up' do
+    action_service = instance_double(AutomationRules::ActionService)
+    allow(AutomationRules::ActionService).to receive(:new).and_return(action_service)
+    allow(action_service).to receive(:perform).and_raise(StandardError, 'boom')
+    allow(ChatwootExceptionTracker).to receive(:new).and_call_original
+
+    job.perform(pending_execution.reload)
+
+    expect(pending_execution.reload).to be_processing
+    expect(ChatwootExceptionTracker).to have_received(:new)
+  end
+
+  it 'sends the follow-up exactly once for the reply-chase story' do
+    message_rule = create(:automation_rule, account: account, event_name: 'message_created', execution_delay: 60,
+                                            conditions: [{ 'values' => ['outgoing'], 'attribute_key' => 'message_type',
+                                                           'query_operator' => nil, 'filter_operator' => 'equal_to' }],
+                                            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.tap { |r| r.update!(due_at: 1.minute.ago) }
+
+    job.perform(row.reload)
+
+    expect(row.reload).to be_executed
+    expect(conversation.messages.outgoing.where(content: 'Just checking in').count).to eq(1)
+  end
+
+  it 'cancels the follow-up when the customer replied before it was due' do
+    message_rule = create(:automation_rule, account: account, event_name: 'message_created', execution_delay: 60,
+                                            conditions: [{ 'values' => ['outgoing'], 'attribute_key' => 'message_type',
+                                                           'query_operator' => nil, 'filter_operator' => 'equal_to' }],
+                                            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.tap { |r| r.update!(due_at: 1.minute.ago) }
+
+    create(:message, conversation: conversation, account: account, message_type: :incoming)
+    job.perform(row.reload)
+
+    expect(row.reload).to be_skipped
+    expect(row.skip_reason).to eq('episode_moved')
+    expect(conversation.messages.outgoing.pluck(:content)).not_to include('Just checking in')
+  end
+end

spec/jobs/automation_rules/trigger_pending_executions_job_spec.rb

@@ -0,0 +1,62 @@
+require 'rails_helper'
+
+RSpec.describe AutomationRules::TriggerPendingExecutionsJob do
+  subject(:job) { described_class.new }
+
+  let(:account) { create(:account) }
+  let(:conversation) { create(:conversation, account: account) }
+
+  before do
+    GlobalConfig.clear_cache
+    account.enable_features!('delayed_automations')
+  end
+
+  it 'enqueues a per-row job for due pending rows but not future ones' do
+    due_row = create(:automation_rule_pending_execution, account: account, conversation: conversation, due_at: 1.minute.ago)
+    future_row = create(:automation_rule_pending_execution, account: account, due_at: 1.hour.from_now)
+
+    expect { job.perform }.to have_enqueued_job(AutomationRules::ProcessPendingExecutionJob).exactly(:once)
+    expect(AutomationRules::ProcessPendingExecutionJob).to have_been_enqueued.with(due_row)
+    expect(AutomationRules::ProcessPendingExecutionJob).not_to have_been_enqueued.with(future_row)
+  end
+
+  it 're-enqueues stale processing rows so they get retried' do
+    stale_row = travel_to(20.minutes.ago) do
+      create(:automation_rule_pending_execution, account: account, conversation: conversation, status: :processing, due_at: 19.minutes.from_now)
+    end
+
+    expect { job.perform }.to have_enqueued_job(AutomationRules::ProcessPendingExecutionJob).with(stale_row)
+  end
+
+  it 'caps enqueues at the configured sweep limit' do
+    create(:installation_config, name: 'AUTOMATION_PENDING_EXECUTIONS_SWEEP_LIMIT', serialized_value: { value: 1 }.with_indifferent_access)
+    create_list(:automation_rule_pending_execution, 2, account: account, due_at: 1.minute.ago)
+
+    expect { job.perform }.to have_enqueued_job(AutomationRules::ProcessPendingExecutionJob).exactly(:once)
+  end
+
+  it 'purges terminal rows past the retention window' do
+    old_row = travel_to(31.days.ago) { create(:automation_rule_pending_execution, account: account, status: :executed) }
+
+    job.perform
+
+    expect { old_row.reload }.to raise_error(ActiveRecord::RecordNotFound)
+  end
+
+  it 'skips rows for accounts with delayed automations disabled so they cannot starve others' do
+    enabled_row = create(:automation_rule_pending_execution, account: account, conversation: conversation, due_at: 1.minute.ago)
+    disabled_account = create(:account) # delayed_automations off by default
+    create(:automation_rule_pending_execution, account: disabled_account, due_at: 2.minutes.ago)
+
+    expect { job.perform }.to have_enqueued_job(AutomationRules::ProcessPendingExecutionJob).exactly(:once)
+    expect(AutomationRules::ProcessPendingExecutionJob).to have_been_enqueued.with(enabled_row)
+  end
+
+  it 'does nothing when the kill switch is set' do
+    create(:installation_config, name: 'DISABLE_DELAYED_AUTOMATIONS', serialized_value: { value: true }.with_indifferent_access)
+    GlobalConfig.clear_cache
+    create(:automation_rule_pending_execution, account: account, conversation: conversation, due_at: 1.minute.ago)
+
+    expect { job.perform }.not_to have_enqueued_job(AutomationRules::ProcessPendingExecutionJob)
+  end
+end

spec/listeners/automation_rule_listener_spec.rb

@@ -247,4 +247,37 @@ describe AutomationRuleListener do
       end
     end
   end
+
+  describe 'delayed rules' do
+    let!(:automation_rule) { create(:automation_rule, event_name: 'conversation_updated', account: account, execution_delay: 60) }
+    let(:event) do
+      Events::Base.new('conversation_updated', Time.zone.now, { conversation: conversation, changed_attributes: {} })
+    end
+
+    before { allow(condition_match).to receive(:present?).and_return(true) }
+
+    context 'when the delayed_automations feature is enabled' do
+      before { account.enable_features!('delayed_automations') }
+
+      it 'records a pending execution instead of running actions' do
+        expect { listener.conversation_updated(event) }.to change(AutomationRulePendingExecution, :count).by(1)
+        expect(AutomationRules::ActionService).not_to have_received(:new)
+        expect(AutomationRulePendingExecution.last.due_at).to be_within(5.seconds).of(60.minutes.from_now)
+      end
+
+      it 'still runs rules without a delay immediately' do
+        automation_rule.update!(execution_delay: nil)
+
+        expect { listener.conversation_updated(event) }.not_to change(AutomationRulePendingExecution, :count)
+        expect(AutomationRules::ActionService).to have_received(:new).with(automation_rule, account, conversation)
+      end
+    end
+
+    context 'when the delayed_automations feature is disabled' do
+      it 'neither arms a pending execution nor falls back to immediate execution' do
+        expect { listener.conversation_updated(event) }.not_to change(AutomationRulePendingExecution, :count)
+        expect(AutomationRules::ActionService).not_to have_received(:new)
+      end
+    end
+  end
 end

spec/models/account_spec.rb

@@ -112,6 +112,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') }
 
@@ -122,7 +137,8 @@ RSpec.describe Account do
         feature_data_import: 1 << 1,
         feature_api_and_webhooks: 1 << 2,
         feature_whatsapp_reconfigure: 1 << 3,
-        feature_whatsapp_embedded_signup_inbox_creation: 1 << 4
+        feature_whatsapp_embedded_signup_inbox_creation: 1 << 4,
+        feature_delayed_automations: 1 << 5
       )
       expect(described_class.flag_mapping['feature_flags_ext_1'][:feature_whatsapp_manual_transfer]).to eq(1)
       expect(described_class.flag_mapping['feature_flags_ext_1'][:feature_data_import]).to eq(2)

spec/models/automation_rule_pending_execution_spec.rb

@@ -0,0 +1,305 @@
+require 'rails_helper'
+
+RSpec.describe AutomationRulePendingExecution do
+  let(:account) { create(:account) }
+  let(:conversation) { create(:conversation, account: account) }
+  let(:rule) do
+    create(:automation_rule, account: account, event_name: 'conversation_updated', execution_delay: 60,
+                             actions: [{ 'action_name' => 'add_label', 'action_params' => ['stale'] }])
+  end
+
+  describe '.episode_key_for' do
+    it 'derives status episodes from status_changed_at' do
+      expect(described_class.episode_key_for(conversation, nil)).to eq("status:#{conversation.status_changed_at.strftime('%s%6N')}")
+    end
+
+    it 'matches between an in-memory arm and a DB-reloaded fire (no float rounding drift)' do
+      conversation.status_changed_at = Time.zone.at(1_784_102_080.844761923r)
+      arm_key = described_class.episode_key_for(conversation, nil)
+      conversation.save!
+      expect(arm_key).to eq(described_class.episode_key_for(conversation.reload, nil))
+    end
+
+    it 'falls back to created_at when status_changed_at is blank' do
+      conversation.update!(status_changed_at: nil)
+      expect(described_class.episode_key_for(conversation.reload, nil)).to eq("status:#{conversation.created_at.strftime('%s%6N')}")
+    end
+
+    it 'derives awaiting_agent episodes from waiting_since (sub-second) for incoming messages' do
+      message = create(:message, conversation: conversation, account: account, message_type: :incoming)
+      expect(described_class.episode_key_for(conversation.reload, message)).to eq("awaiting_agent:#{conversation.waiting_since.strftime('%s%6N')}")
+    end
+
+    it 'distinguishes two waiting periods that fall within the same second' do
+      message = create(:message, conversation: conversation, account: account, message_type: :incoming)
+      first_key = described_class.episode_key_for(conversation.reload, message)
+
+      # Agent replies then customer re-waits within the same second: keys must differ.
+      conversation.update!(waiting_since: conversation.waiting_since + 0.4)
+      expect(described_class.episode_key_for(conversation.reload, message)).not_to eq(first_key)
+    end
+
+    it 'arms an awaiting_agent episode from the message created_at when waiting_since is not yet written' do
+      message = create(:message, conversation: conversation, account: account, message_type: :incoming)
+      # Simulate the race where the listener arms before update_waiting_since commits.
+      conversation.update!(waiting_since: nil)
+      armed_key = described_class.arm_episode_key_for(conversation.reload, message)
+
+      # Once waiting_since settles to the message's created_at, the strict fire-time key matches.
+      conversation.update!(waiting_since: message.created_at)
+      expect(armed_key).to eq("awaiting_agent:#{message.created_at.strftime('%s%6N')}")
+      expect(armed_key).to eq(described_class.episode_key_for(conversation.reload, message))
+    end
+
+    it 'derives reply_chase episodes from the max incoming message id for outgoing messages' do
+      incoming = create(:message, conversation: conversation, account: account, message_type: :incoming)
+      outgoing = create(:message, conversation: conversation, account: account, message_type: :outgoing)
+      expect(described_class.episode_key_for(conversation.reload, outgoing)).to eq("reply_chase:#{incoming.id}")
+    end
+
+    it 'uses 0 for reply_chase when there is no incoming message' do
+      outgoing = create(:message, conversation: conversation, account: account, message_type: :outgoing)
+      expect(described_class.episode_key_for(conversation.reload, outgoing)).to eq('reply_chase:0')
+    end
+  end
+
+  describe '.schedule' do
+    it 'creates a pending row due after the rule delay' do
+      described_class.schedule(rule: rule, conversation: conversation)
+
+      row = described_class.last
+      expect(row).to have_attributes(account_id: account.id, conversation_id: conversation.id, status: 'pending')
+      expect(row.due_at).to be_within(5.seconds).of(60.minutes.from_now)
+    end
+
+    it 'anchors due_at to the event time, not when a backlogged listener runs' do
+      conversation.update!(status_changed_at: 30.minutes.ago)
+      described_class.schedule(rule: rule, conversation: conversation)
+
+      # A 60-minute rule on a status that changed 30 minutes ago is already 30 minutes into its wait.
+      expect(described_class.last.due_at).to be_within(5.seconds).of(30.minutes.from_now)
+    end
+
+    it 'does not reset the clock for a repeated status episode' do
+      described_class.schedule(rule: rule, conversation: conversation)
+      original_due_at = described_class.last.due_at
+
+      travel_to(30.minutes.from_now) { described_class.schedule(rule: rule, conversation: conversation) }
+
+      expect(described_class.count).to eq(1)
+      expect(described_class.last.due_at).to be_within(1.second).of(original_due_at)
+    end
+
+    it 'moves the clock and anchor for a repeated reply_chase episode' do
+      first_reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
+      described_class.schedule(rule: rule, conversation: conversation, message: first_reply)
+
+      travel_to(30.minutes.from_now) do
+        second_reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
+        described_class.schedule(rule: rule, conversation: conversation, message: second_reply)
+
+        # due_at is re-anchored to the new reply's created_at, not the original schedule time.
+        expect(described_class.count).to eq(1)
+        expect(described_class.last.message_id).to eq(second_reply.id)
+        expect(described_class.last.due_at).to be_within(5.seconds).of(60.minutes.from_now)
+      end
+    end
+
+    it 'does not let a late older reply move the reply_chase clock backwards' do
+      older_reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
+      newer_reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
+
+      # The newer reply's job runs first and arms the episode.
+      described_class.schedule(rule: rule, conversation: conversation, message: newer_reply)
+      armed_due_at = described_class.last.due_at
+
+      # The older reply's job arrives late; it must not pull the clock or message_id back.
+      described_class.schedule(rule: rule, conversation: conversation, message: older_reply)
+
+      expect(described_class.count).to eq(1)
+      expect(described_class.last.message_id).to eq(newer_reply.id)
+      expect(described_class.last.due_at).to eq(armed_due_at)
+    end
+
+    it 'does not re-arm an executed reply_chase episode' do
+      reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
+      described_class.schedule(rule: rule, conversation: conversation, message: reply)
+      described_class.last.update!(status: :executed)
+
+      described_class.schedule(rule: rule, conversation: conversation, message: reply)
+
+      expect(described_class.count).to eq(1)
+      expect(described_class.last).to be_executed
+    end
+
+    it 're-arms a condition-skipped episode when a newer qualifying message arrives' do
+      first_reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
+      described_class.schedule(rule: rule, conversation: conversation, message: first_reply)
+      described_class.last.update!(status: :skipped, skip_reason: 'conditions_changed')
+
+      second_reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
+      described_class.schedule(rule: rule, conversation: conversation, message: second_reply)
+
+      row = described_class.last
+      expect(described_class.count).to eq(1)
+      expect(row).to be_pending
+      expect(row.skip_reason).to be_nil
+      expect(row.message_id).to eq(second_reply.id)
+    end
+
+    it 're-anchors a reply_chase row stuck in processing when a newer reply arrives' do
+      first_reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
+      described_class.schedule(rule: rule, conversation: conversation, message: first_reply)
+      # The worker claimed the row and then died, leaving it in processing with the old clock.
+      described_class.last.update!(status: :processing)
+
+      travel_to(30.minutes.from_now) do
+        second_reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
+        described_class.schedule(rule: rule, conversation: conversation, message: second_reply)
+
+        row = described_class.last
+        expect(described_class.count).to eq(1)
+        expect(row).to be_pending
+        expect(row.message_id).to eq(second_reply.id)
+        expect(row.due_at).to be_within(5.seconds).of(60.minutes.from_now)
+      end
+    end
+
+    it 'does not re-arm an episode skipped for a non-condition reason' do
+      reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
+      described_class.schedule(rule: rule, conversation: conversation, message: reply)
+      described_class.last.update!(status: :skipped, skip_reason: 'episode_moved')
+
+      newer_reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
+      described_class.schedule(rule: rule, conversation: conversation, message: newer_reply)
+
+      expect(described_class.count).to eq(1)
+      expect(described_class.last).to be_skipped
+    end
+
+    it 'keeps the awaiting_agent clock but tracks the newest incoming message' do
+      first_message = create(:message, conversation: conversation, account: account, message_type: :incoming)
+      described_class.schedule(rule: rule, conversation: conversation, message: first_message)
+      original_due_at = described_class.last.due_at
+
+      second_message = create(:message, conversation: conversation, account: account, message_type: :incoming)
+      travel_to(30.minutes.from_now) { described_class.schedule(rule: rule, conversation: conversation, message: second_message) }
+
+      row = described_class.last
+      expect(described_class.count).to eq(1)
+      # The wait still counts from waiting_since, so the clock is unchanged...
+      expect(row.due_at).to be_within(1.second).of(original_due_at)
+      # ...but the row tracks the newest qualifying message, not the stale first one.
+      expect(row.message_id).to eq(second_message.id)
+    end
+
+    it 'keeps the newest message when an older incoming collision arrives last' do
+      first_message = create(:message, conversation: conversation, account: account, message_type: :incoming)
+      described_class.schedule(rule: rule, conversation: conversation, message: first_message)
+
+      older = create(:message, conversation: conversation, account: account, message_type: :incoming)
+      newer = create(:message, conversation: conversation, account: account, message_type: :incoming)
+
+      # The newer message re-arms first; a late older-message collision must not overwrite it.
+      described_class.schedule(rule: rule, conversation: conversation, message: newer)
+      described_class.schedule(rule: rule, conversation: conversation, message: older)
+
+      expect(described_class.count).to eq(1)
+      expect(described_class.last.message_id).to eq(newer.id)
+    end
+  end
+
+  describe '#episode_current?' do
+    it 'is true while the conversation stays in the armed status' do
+      described_class.schedule(rule: rule, conversation: conversation)
+      expect(described_class.last.episode_current?).to be(true)
+    end
+
+    it 'is false after a status transition' do
+      described_class.schedule(rule: rule, conversation: conversation)
+      conversation.update!(status: :resolved)
+      expect(described_class.last.reload.episode_current?).to be(false)
+    end
+
+    it 'is false for awaiting_agent episodes once the agent replies (waiting_since cleared)' do
+      message = create(:message, conversation: conversation, account: account, message_type: :incoming)
+      described_class.schedule(rule: rule, conversation: conversation, message: message)
+
+      conversation.update!(waiting_since: nil)
+      expect(described_class.last.episode_current?).to be(false)
+    end
+
+    it 'is false for reply_chase episodes once the customer replies' do
+      reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
+      described_class.schedule(rule: rule, conversation: conversation, message: reply)
+
+      create(:message, conversation: conversation, account: account, message_type: :incoming)
+      expect(described_class.last.episode_current?).to be(false)
+    end
+  end
+
+  describe '#claim!' do
+    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)
+      expect(row.reload).to be_processing
+      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)
+    end
+
+    it 'reclaims a processing row only after its lock goes stale' do
+      row.update!(status: :processing)
+      expect(row.claim!).to be(false)
+
+      travel_to(20.minutes.from_now) { expect(row.claim!).to be(true) }
+    end
+  end
+
+  describe '.sweepable' do
+    it 'selects due pending rows and stale processing rows, but not future or fresh ones' do
+      due = create(:automation_rule_pending_execution, account: account, conversation: conversation, due_at: 1.minute.ago)
+      create(:automation_rule_pending_execution, account: account, due_at: 1.hour.from_now)
+      create(:automation_rule_pending_execution, account: account, status: :processing)
+      stale = travel_to(20.minutes.ago) do
+        create(:automation_rule_pending_execution, account: account, conversation: conversation, status: :processing)
+      end
+
+      expect(described_class.sweepable).to contain_exactly(due, stale)
+    end
+  end
+
+  describe '.purge_terminal!' do
+    it 'deletes terminal rows past the retention window and keeps everything else' do
+      old_executed = travel_to(31.days.ago) { create(:automation_rule_pending_execution, account: account, status: :executed) }
+      recent_skipped = create(:automation_rule_pending_execution, account: account, status: :skipped)
+      pending = create(:automation_rule_pending_execution, account: account, conversation: conversation)
+
+      described_class.purge_terminal!
+
+      expect(described_class.pluck(:id)).to contain_exactly(recent_skipped.id, pending.id)
+      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

spec/models/automation_rule_spec.rb

@@ -137,4 +137,118 @@ RSpec.describe AutomationRule do
       end
     end
   end
+
+  describe 'execution_delay validations' do
+    let(:rule) { build(:automation_rule, account: create(:account)) }
+
+    it 'allows nil (immediate execution)' do
+      rule.execution_delay = nil
+      expect(rule).to be_valid
+    end
+
+    it 'allows delays between 10 minutes and 30 days' do
+      rule.execution_delay = 240
+      expect(rule).to be_valid
+    end
+
+    it 'rejects delays below 10 minutes' do
+      rule.execution_delay = 5
+      expect(rule).not_to be_valid
+      expect(rule.errors[:execution_delay]).to be_present
+    end
+
+    it 'rejects delays above 30 days' do
+      rule.execution_delay = 43_201
+      expect(rule).not_to be_valid
+    end
+
+    it 'rejects non-integer delays' do
+      rule.execution_delay = 10.5
+      expect(rule).not_to be_valid
+    end
+
+    it 'rejects a delay combined with an attribute_changed condition' do
+      rule.execution_delay = 60
+      rule.conditions = [{ 'attribute_key' => 'status', 'filter_operator' => 'attribute_changed',
+                           'values' => { 'from' => ['open'], 'to' => ['pending'] }, 'query_operator' => nil }]
+      expect(rule).not_to be_valid
+      expect(rule.errors[:execution_delay]).to include('cannot be used with attribute_changed conditions.')
+    end
+
+    it 'rejects a delayed conversation-level rule with a mutable non-status condition' do
+      rule.event_name = 'conversation_updated'
+      rule.execution_delay = 60
+      rule.conditions = [{ 'attribute_key' => 'priority', 'filter_operator' => 'equal_to', 'values' => ['urgent'], 'query_operator' => nil }]
+      expect(rule).not_to be_valid
+      expect(rule.errors[:execution_delay]).to include('only supports status and inbox conditions for conversation-level events.')
+    end
+
+    it 'allows a delayed conversation-level rule with only status conditions' do
+      rule.event_name = 'conversation_updated'
+      rule.execution_delay = 60
+      rule.conditions = [{ 'attribute_key' => 'status', 'filter_operator' => 'equal_to', 'values' => ['pending'], 'query_operator' => nil }]
+      expect(rule).to be_valid
+    end
+
+    it 'allows a delayed conversation_created rule (arms on creation)' do
+      rule.event_name = 'conversation_created'
+      rule.execution_delay = 10
+      rule.conditions = [{ 'attribute_key' => 'status', 'filter_operator' => 'equal_to', 'values' => ['open'], 'query_operator' => nil }]
+      expect(rule).to be_valid
+    end
+
+    it 'allows a delayed conversation-level rule scoped by status and inbox (immutable)' do
+      rule.event_name = 'conversation_updated'
+      rule.execution_delay = 60
+      rule.conditions = [{ 'attribute_key' => 'status', 'filter_operator' => 'equal_to', 'values' => ['pending'], 'query_operator' => 'AND' },
+                         { 'attribute_key' => 'inbox_id', 'filter_operator' => 'equal_to', 'values' => [1], 'query_operator' => nil }]
+      expect(rule).to be_valid
+    end
+
+    it 'allows a delayed message_created rule with a non-status condition' do
+      rule.event_name = 'message_created'
+      rule.execution_delay = 60
+      rule.conditions = [{ 'attribute_key' => 'message_type', 'filter_operator' => 'equal_to', 'values' => ['outgoing'], 'query_operator' => nil }]
+      expect(rule).to be_valid
+    end
+  end
+
+  describe 'discarding stale pending executions on edit' do
+    let(:account) { create(:account) }
+    let(:conversation) { create(:conversation, account: account, status: :pending) }
+    let(:status_condition) { { 'attribute_key' => 'status', 'filter_operator' => 'equal_to', 'values' => ['pending'], 'query_operator' => nil } }
+    let(:rule) do
+      create(:automation_rule, account: account, event_name: 'conversation_updated', execution_delay: 60,
+                               conditions: [status_condition], actions: [{ 'action_name' => 'add_label', 'action_params' => ['stale'] }])
+    end
+
+    before { AutomationRulePendingExecution.schedule(rule: rule, conversation: conversation) }
+
+    it 'discards armed rows when the actions change' do
+      rule.update!(actions: [{ 'action_name' => 'add_label', 'action_params' => ['urgent'] }])
+      expect(rule.pending_executions.pending).to be_empty
+    end
+
+    it 'discards armed rows when the delay changes' do
+      rule.update!(execution_delay: 120)
+      expect(rule.pending_executions.pending).to be_empty
+    end
+
+    it 'discards a stale processing row that the sweep would otherwise reclaim' do
+      rule.pending_executions.first.update!(status: :processing)
+      rule.update!(actions: [{ 'action_name' => 'add_label', 'action_params' => ['urgent'] }])
+      expect(rule.pending_executions.armed).to be_empty
+    end
+
+    it 'frees the episode slot so the new definition re-arms for the same episode' do
+      rule.update!(actions: [{ 'action_name' => 'add_label', 'action_params' => ['urgent'] }])
+      AutomationRulePendingExecution.schedule(rule: rule, conversation: conversation)
+      expect(rule.pending_executions.pending.count).to eq(1)
+    end
+
+    it 'leaves armed rows untouched on a name-only edit' do
+      rule.update!(name: 'Renamed rule')
+      expect(rule.pending_executions.pending.count).to eq(1)
+    end
+  end
 end

spec/models/conversation_spec.rb

@@ -1230,4 +1230,28 @@ RSpec.describe Conversation do
       end
     end
   end
+
+  describe '#status_changed_at' do
+    let(:conversation) { create(:conversation) }
+
+    it 'is set on create' do
+      expect(conversation.status_changed_at).to be_present
+    end
+
+    it 'is updated on every status transition' do
+      original = conversation.status_changed_at
+
+      travel_to(1.hour.from_now) { conversation.update!(status: :resolved) }
+
+      expect(conversation.reload.status_changed_at).to be > original
+    end
+
+    it 'is untouched by non-status saves' do
+      original = conversation.status_changed_at
+
+      travel_to(1.hour.from_now) { conversation.update!(priority: :high) }
+
+      expect(conversation.reload.status_changed_at).to be_within(1.second).of(original)
+    end
+  end
 end