refactor(automations): job-side claim for delayed executions, remove rubocop disables

app/jobs/automation_rules/process_pending_execution_job.rb

@@ -1,24 +1,30 @@
 class AutomationRules::ProcessPendingExecutionJob < ApplicationJob
   queue_as :medium
 
-  discard_on ActiveRecord::RecordNotFound
+  discard_on ActiveJob::DeserializationError
 
   def perform(pending_execution)
-    # The sweep also checks this, but rows already enqueued when the switch flips must not
-    # keep firing; they return to pending and replay or expire via the due window.
-    return pending_execution.update!(status: :pending) if delayed_automations_disabled?
+    return if delayed_automations_disabled?
+    # Atomic claim: a duplicate enqueue (overlapping sweep or stale reclaim) loses here and returns.
+    return unless pending_execution.claim!
+
+    return pending_execution.update!(status: :skipped, skip_reason: 'expired') if expired?(pending_execution)
 
     skip_reason = skip_reason_for(pending_execution)
     return pending_execution.update!(status: :skipped, skip_reason: skip_reason) if skip_reason
 
     execute(pending_execution)
   rescue StandardError => e
-    # Row stays `processing`; the sweep's stale reclaim retries it later.
+    # Row stays `processing`; the next sweep reclaims and retries it once the lock goes stale.
     ChatwootExceptionTracker.new(e, account: pending_execution.account).capture_exception
   end
 
   private
 
+  def expired?(pending_execution)
+    pending_execution.due_at < AutomationRulePendingExecution::DUE_WINDOW.ago
+  end
+
   def skip_reason_for(pending_execution)
     rule = pending_execution.automation_rule
     return 'rule_inactive' if rule.nil? || !rule.active?

app/jobs/automation_rules/trigger_pending_executions_job.rb

@@ -7,19 +7,12 @@ class AutomationRules::TriggerPendingExecutionsJob < ApplicationJob
     return if delayed_automations_disabled?
 
     started_at = Time.current
-    reclaimed = AutomationRulePendingExecution.reclaim_stale!
-    expired = AutomationRulePendingExecution.expire_overdue!
-    due_count = AutomationRulePendingExecution.due.count
+    purged = AutomationRulePendingExecution.purge_terminal!
 
-    enqueued = 0
-    AutomationRulePendingExecution.due.limit(sweep_limit).find_each(batch_size: 100) do |pending_execution|
-      next unless pending_execution.mark_processing!
+    rows = AutomationRulePendingExecution.sweepable.order(:due_at).limit(sweep_limit).to_a
+    rows.each { |row| AutomationRules::ProcessPendingExecutionJob.perform_later(row) }
 
-      AutomationRules::ProcessPendingExecutionJob.perform_later(pending_execution)
-      enqueued += 1
-    end
-
-    log_summary(due: due_count, enqueued: enqueued, expired: expired, reclaimed: reclaimed, started_at: started_at)
+    log_summary(enqueued: rows.size, capped: rows.size >= sweep_limit, purged: purged, started_at: started_at)
   end
 
   private
@@ -32,11 +25,8 @@ class AutomationRules::TriggerPendingExecutionsJob < ApplicationJob
     (InstallationConfig.find_by(name: 'AUTOMATION_PENDING_EXECUTIONS_SWEEP_LIMIT')&.value || DEFAULT_SWEEP_LIMIT).to_i
   end
 
-  def log_summary(due:, enqueued:, expired:, reclaimed:, started_at:)
-    summary = {
-      event: 'completed', due: due, enqueued: enqueued, capped: due > enqueued,
-      expired: expired, reclaimed: reclaimed, duration_ms: ((Time.current - started_at) * 1000).round
-    }
+  def log_summary(enqueued:, capped:, purged:, started_at:)
+    summary = { event: 'completed', enqueued: enqueued, capped: capped, purged: purged, duration_ms: ((Time.current - started_at) * 1000).round }
     Rails.logger.info("[AutomationRules::TriggerPendingExecutionsJob] #{summary.to_json}")
   end
 end

app/models/automation_rule_pending_execution.rb

@@ -25,8 +25,10 @@
 class AutomationRulePendingExecution < ApplicationRecord
   # Rows older than this never fire (bounds backlog replay after downtime).
   DUE_WINDOW = 3.days
-  # Claimed rows abandoned by a crashed worker return to the sweep after this.
+  # A processing row whose lock is older than this is treated as abandoned and reclaimed.
   STALE_PROCESSING_TIMEOUT = 15.minutes
+  # Terminal rows are purged after this to keep the table bounded.
+  RETENTION_WINDOW = 30.days
 
   belongs_to :automation_rule
   belongs_to :conversation
@@ -35,33 +37,24 @@ class AutomationRulePendingExecution < ApplicationRecord
 
   enum status: { pending: 0, processing: 1, executed: 2, skipped: 3 }
 
-  scope :due, -> { pending.where(due_at: DUE_WINDOW.ago..Time.current) }
+  # Rows a sweep should hand to a worker: due pending rows, plus processing rows whose lock went stale.
+  scope :sweepable, lambda {
+    pending.where(due_at: ..Time.current).or(processing.where(updated_at: ...STALE_PROCESSING_TIMEOUT.ago))
+  }
 
   def self.schedule(rule:, conversation:, message: nil)
-    attributes = {
-      automation_rule_id: rule.id,
-      conversation_id: conversation.id,
-      account_id: conversation.account_id,
-      message_id: message&.id,
-      episode_key: episode_key_for(conversation, message),
-      due_at: rule.execution_delay.minutes.from_now,
-      created_at: Time.current,
-      updated_at: Time.current
-    }
+    key = episode_key_for(conversation, message)
+    create!(
+      automation_rule: rule, conversation: conversation, account_id: conversation.account_id,
+      message_id: message&.id, episode_key: key, due_at: rule.execution_delay.minutes.from_now
+    )
+  rescue ActiveRecord::RecordNotUnique
+    # Episode already armed. Reply-chase tracks the latest agent reply, so its clock moves;
+    # status / awaiting-agent episodes keep the first clock. A terminal row is never re-armed.
+    return unless message && !message.incoming?
 
-    # Values are server-computed and the unique episode index is the real guard, so the
-    # validation-skipping conflict-handling writes are safe here (repo bulk-write convention).
-    # rubocop:disable Rails/SkipsModelValidations
-    if message && !message.incoming?
-      # Reply-chase: the clock tracks the latest agent reply. Status is excluded from the
-      # update list so an executed/skipped episode is never re-armed (run-once per episode).
-      upsert(attributes, unique_by: :uniq_automation_pending_execution_episode,
-                         on_duplicate: Arel.sql('due_at = excluded.due_at, message_id = excluded.message_id, updated_at = excluded.updated_at'))
-    else
-      # Status / awaiting-agent episodes: first event wins, the clock is not reset.
-      insert(attributes, unique_by: :uniq_automation_pending_execution_episode)
-    end
-    # rubocop:enable Rails/SkipsModelValidations
+    row = find_by!(automation_rule_id: rule.id, conversation_id: conversation.id, episode_key: key)
+    row.update!(due_at: rule.execution_delay.minutes.from_now, message_id: message.id) if row.pending?
   end
 
   # Episode keys identify one qualifying stretch of conversation state; when the recomputed
@@ -79,33 +72,19 @@ class AutomationRulePendingExecution < ApplicationRecord
     end
   end
 
-  # Bulk state flips on exceptional sets; batched so no statement outlives the global 14s
-  # statement_timeout (repo bulk-write convention, see Agents::DestroyJob).
-  # rubocop:disable Rails/SkipsModelValidations
-  def self.expire_overdue!
-    expired_count = 0
-    pending.where(due_at: ...DUE_WINDOW.ago).in_batches(of: 1000) do |batch|
-      expired_count += batch.update_all(status: statuses[:skipped], skip_reason: 'expired', updated_at: Time.current)
-    end
-    expired_count
+  def self.purge_terminal!
+    where(status: [statuses[:executed], statuses[:skipped]], updated_at: ...RETENTION_WINDOW.ago)
+      .in_batches(of: 1000).delete_all
   end
 
-  def self.reclaim_stale!
-    reclaimed_count = 0
-    processing.where(updated_at: ...STALE_PROCESSING_TIMEOUT.ago).in_batches(of: 1000) do |batch|
-      reclaimed_count += batch.update_all(status: statuses[:pending], updated_at: Time.current)
-    end
-    reclaimed_count
-  end
-  # rubocop:enable Rails/SkipsModelValidations
-
-  # Locked claim so a row re-selected by an overlapping sweep can't double-fire
-  # (Campaign#mark_processing! pattern).
-  def mark_processing!
+  # 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.
+  def claim!
     with_lock do
-      next false unless pending?
+      next false unless claimable?
 
-      processing!
+      update!(status: :processing, updated_at: Time.current)
       true
     end
   end
@@ -113,4 +92,10 @@ class AutomationRulePendingExecution < ApplicationRecord
   def episode_current?
     self.class.episode_key_for(conversation, message) == episode_key
   end
+
+  private
+
+  def claimable?
+    pending? || (processing? && updated_at < STALE_PROCESSING_TIMEOUT.ago)
+  end
 end

docs/delayed-automations-implementation-plan.md

@@ -97,20 +97,22 @@ rule fire" for support and drives incident blast-radius queries.
 ### `app/models/automation_rule_pending_execution.rb` (new, flat name per repo convention)
 - `belongs_to :automation_rule, :conversation, :account`; optional `belongs_to :message`.
 - `enum status: { pending: 0, processing: 1, executed: 2, skipped: 3 }`.
-- `scope :due, -> { pending.where(due_at: 3.days.ago..Time.current) }` — **bounded window**
-  (campaign/snooze-reopen precedent) so downtime backlog is finite.
-- `self.expire_overdue!` — pending rows with `due_at < 3.days.ago` → `skipped` /
-  `skip_reason: 'expired'`, returns count (batched updates — global 14s `statement_timeout`).
-- `self.reclaim_stale!` — `processing` rows older than 15 min → back to `pending`
-  (stale-claim recovery; mechanism per Captain's `SYNC_STALE_TIMEOUT`, which uses 2h —
-  15 min fits the 5-min sweep cadence).
-- `#mark_processing!` — compare-and-set from `pending` (the `Campaign#mark_processing!`
-  precedent); the per-row job is enqueued only after a successful claim.
+- `scope :sweepable` — due `pending` rows (`due_at <= now`) **or** `processing` rows whose
+  lock is older than `STALE_PROCESSING_TIMEOUT` (`.or` of two enum scopes). This single scope
+  replaces the old separate `due` window + `reclaim_stale!` bulk update; stale claims simply
+  become claimable again. Expiry (`due_at < 3.days.ago`) is decided per-row at fire time.
+- `self.purge_terminal!` — `executed`/`skipped` rows past `RETENTION_WINDOW` (30 days) are
+  `delete_all`ed in batches of 1000 (keeps the table bounded; `delete_all` skips no needed
+  validation and is not a `SkipsModelValidations` concern).
+- `#claim!` — atomic compare-and-set under `with_lock`: a row is claimable if `pending`, or
+  `processing` but stale. The claim happens **inside the per-row job**, not the sweep, so a
+  duplicate enqueue (overlapping sweep or reclaimed stale row) loses the claim and returns —
+  this is the double-fire guard. Refreshing `updated_at` on claim renews the lock.
 - `self.schedule(rule:, conversation:, message: nil)` — computes `episode_key` + `due_at =
-  Time.current + rule.execution_delay.minutes`, then by anchor type (locked decision #4):
-  - status episodes and incoming-anchored message episodes → `insert` with `unique_by`
-    (conflict = no-op, clock not reset);
-  - outgoing-anchored message episodes → `upsert` updating `due_at` + `message_id`.
+  Time.current + rule.execution_delay.minutes`, then `create!` guarded by the unique episode
+  index. On `RecordNotUnique`: status / incoming-anchored episodes no-op (clock not reset);
+  outgoing-anchored (reply-chase) episodes update `due_at` + `message_id` when still `pending`
+  (a terminal row is never re-armed). Validation-running writes — no `insert`/`upsert`.
 - `#episode_current?` — recomputes the episode key from live conversation state, compares.
 
 ### `app/models/automation_rule.rb`
@@ -164,15 +166,13 @@ Add one line: `AutomationRules::TriggerPendingExecutionsJob.perform_later`. No c
 # which is truthy — the ENABLE_*_CHANNEL_HUMAN_AGENT read pattern):
 return if GlobalConfig.get('DISABLE_DELAYED_AUTOMATIONS')['DISABLE_DELAYED_AUTOMATIONS']
 
-reclaimed = AutomationRulePendingExecution.reclaim_stale!
-expired   = AutomationRulePendingExecution.expire_overdue!
-AutomationRulePendingExecution.due.limit(sweep_limit).find_each(batch_size: 100) do |pending|
-  next unless pending.mark_processing!
-  AutomationRules::ProcessPendingExecutionJob.perform_later(pending)
-end
+purged = AutomationRulePendingExecution.purge_terminal!
+rows = AutomationRulePendingExecution.sweepable.order(:due_at).limit(sweep_limit).to_a
+rows.each { |row| AutomationRules::ProcessPendingExecutionJob.perform_later(row) }
+# The sweep only enqueues — the per-row job claims (double-fire guard), expires, and executes.
 # end-of-run summary, JSON so New Relic ingests fields without a parsing rule:
-# [AutomationRules::TriggerPendingExecutionsJob] {"event":"completed","due":N,"enqueued":N,
-#  "capped":bool,"expired":N,"reclaimed":N,"duration_ms":N}
+# [AutomationRules::TriggerPendingExecutionsJob] {"event":"completed","enqueued":N,
+#  "capped":bool,"purged":N,"duration_ms":N}
 ```
 `sweep_limit`: constant on the job (default 1000) with an InstallationConfig override
 (Captain `ScheduleSyncsJob` pattern — a no-deploy tuning knob). Overflow rows stay `pending`
@@ -187,10 +187,12 @@ is also on `medium`, so webhook-heavy delayed rules share that budget.
 
 ### `app/jobs/automation_rules/process_pending_execution_job.rb` (new, queue `medium`,
 mirrors `Sla::ProcessAppliedSlaJob`)
-`discard_on ActiveRecord::RecordNotFound`. Re-checks the kill switch first (sweep-only
-checking would let a full sweep's already-enqueued rows keep firing after the flip): if set,
-revert the row to `pending` and return — it replays or expires via the window. Guard chain,
-each failure → `skipped!` with `skip_reason`:
+`discard_on ActiveJob::DeserializationError` (a deleted row fails GlobalID lookup). Re-checks
+the kill switch first (sweep-only checking would let a full sweep's already-enqueued rows keep
+firing after the flip): if set, return without claiming — the row stays `pending` and replays
+or expires via the window. Then `claim!` (bail if lost), then the expiry check
+(`due_at < DUE_WINDOW.ago` → `expired`). Guard chain, each failure → `skipped!` with
+`skip_reason`:
 1. rule exists and `active?` → `rule_inactive`
 2. account flag still enabled → `flag_disabled` (the per-account stop)
 3. conversation exists → `conversation_gone`
@@ -202,8 +204,8 @@ each failure → `skipped!` with `skip_reason`:
    then `executed!`
 
 Per-row rescue → `ChatwootExceptionTracker.new(e, account: account).capture_exception`; the
-row stays `processing` on unexpected errors (stale reclaim retries it), it is NOT marked
-skipped.
+row stays `processing` on unexpected errors (the next sweep re-enqueues it once the lock goes
+stale, and `claim!` lets it run again), it is NOT marked skipped.
 
 Loop safety: `ActionService#initialize` already sets `Current.executed_by = rule` and resets
 it in `ensure`, so events emitted by delayed actions carry `performed_by: rule` and are
@@ -253,18 +255,19 @@ Tailwind-only styling; Composition API `<script setup>` (form already is).
 - `spec/models/automation_rule_pending_execution_spec.rb` — episode key derivation (all
   three anchors); insert/upsert semantics (status + incoming-anchored: clock not reset;
   outgoing-anchored: due_at tracks latest agent reply); `episode_current?` incl.
-  agent-reply-clears-`waiting_since` cancellation; `mark_processing!` claim;
-  `reclaim_stale!`; `expire_overdue!`.
+  agent-reply-clears-`waiting_since` cancellation; `claim!` (once-only + stale reclaim);
+  `sweepable` selection; `purge_terminal!`.
 - `spec/models/conversation_spec.rb` — `status_changed_at` set on create and every
   transition, untouched on non-status saves.
 - `spec/listeners/automation_rule_listener_spec.rb` — delayed rule records a pending
   execution and does NOT run actions; nil delay unchanged; `performed_by` automation events
   create no rows; flag off → no arming AND no immediate fallback.
-- `spec/jobs/automation_rules/trigger_pending_executions_job_spec.rb` — only due+pending
-  swept; kill switch no-ops the sweep; window expiry marks `expired`; cap leaves overflow
-  pending; stale `processing` reclaimed.
+- `spec/jobs/automation_rules/trigger_pending_executions_job_spec.rb` — due+pending and stale
+  `processing` enqueued (not future); kill switch no-ops the sweep; cap limits enqueues;
+  terminal rows past retention purged.
 - `spec/jobs/automation_rules/process_pending_execution_job_spec.rb` — full guard chain
-  with `skip_reason` per branch; kill switch reverts row to pending; happy path executes
+  with `skip_reason` per branch; per-row `expired`; kill switch leaves row pending; duplicate
+  enqueue executes once (claim guard); happy path executes
   actions exactly once.
 - `spec/controllers/api/v1/accounts/automation_rules_controller_spec.rb` —
   permits/persists/serializes `execution_delay` when flag on; 422 when submitted with flag

docs/time-based-automations-phasing-and-rollout.md

@@ -298,9 +298,10 @@ are permanent (header contract in features.yml).
 - **Per-sweep cap**: constant on the job class with an InstallationConfig override
   (Captain ScheduleSyncsJob pattern — gives a no-deploy tuning knob), default 1000;
   overflow logs `capped: true` + remaining count; due rows stay due for the next tick.
-- **Claim transition**: `pending → processing` before enqueue (`Campaign#mark_processing!`
-  precedent) so a re-selected row can't double-fire; stale `processing` rows older than
-  15 min return to eligibility (mechanism per Captain's stale-claim recovery — its
+- **Claim transition**: the sweep only enqueues; the per-row job does the atomic
+  `pending → processing` claim under a row lock, so a row re-enqueued by an overlapping sweep
+  (or a reclaimed stale row) loses the claim and can't double-fire. Stale `processing` rows
+  older than 15 min become claimable again (mechanism per Captain's stale-claim recovery — its
   `SYNC_STALE_TIMEOUT` is 2h; we choose 15 min to match the 5-min cadence).
 - **Kill switch checked in both jobs** (§4.1).
 - **Queue reality** (`config/sidekiq.yml` is strict-priority, no weights): per-row jobs on
@@ -308,7 +309,7 @@ are permanent (header contract in features.yml).
   `scheduled_jobs` (8th) sits *below* `low`, so the sweep itself can be late — `due_at <=
   now` semantics already tolerate that. Note `WebhookJob` is **also** `queue_as :medium`,
   so webhook-heavy delayed rules add to the same queue; covered by the cap.
-- Per-row error isolation: `discard_on ActiveRecord::RecordNotFound`; rescue →
+- Per-row error isolation: `discard_on ActiveJob::DeserializationError`; rescue →
   `ChatwootExceptionTracker.new(e, account:).capture_exception`, continue.
 
 **Fast-follow tolerable** (days, not weeks; needed before Stage C):
@@ -326,8 +327,8 @@ are permanent (header contract in features.yml).
 
 **Structured summary log per sweep**, emitted as JSON so New Relic ingests fields without a
 parsing rule:
-`[AutomationRules::TriggerPendingExecutionsJob] {"event":"completed","due":N,"enqueued":N,
-"capped":false,"expired":N,"reclaimed":N,"duration_ms":N}` — plus per-row terminal
+`[AutomationRules::TriggerPendingExecutionsJob] {"event":"completed","enqueued":N,
+"capped":false,"purged":N,"duration_ms":N}` — plus per-row terminal
 `skip_reason` stored on the row (the support-facing answer to "why didn't my rule fire"
 until Phase 2's history UI).
 

spec/jobs/automation_rules/process_pending_execution_job_spec.rb

@@ -13,7 +13,7 @@ RSpec.describe AutomationRules::ProcessPendingExecutionJob do
   end
   let(:pending_execution) do
     AutomationRulePendingExecution.schedule(rule: rule, conversation: conversation)
-    AutomationRulePendingExecution.last.tap { |row| row.update!(status: :processing) }
+    AutomationRulePendingExecution.last
   end
 
   before do
@@ -66,7 +66,16 @@ RSpec.describe AutomationRules::ProcessPendingExecutionJob do
     expect(pending_execution.skip_reason).to eq('conditions_changed')
   end
 
-  it 'reverts the row to pending when the kill switch is set' do
+  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)
@@ -75,6 +84,17 @@ RSpec.describe AutomationRules::ProcessPendingExecutionJob do
     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
     allow(AutomationRules::ActionService).to receive(:new).and_raise(StandardError, 'boom')
     allow(ChatwootExceptionTracker).to receive(:new).and_call_original
@@ -92,7 +112,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.tap { |r| r.update!(status: :processing) }
+    row = AutomationRulePendingExecution.last
 
     job.perform(row.reload)
 
@@ -107,7 +127,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.tap { |r| r.update!(status: :processing) }
+    row = AutomationRulePendingExecution.last
 
     create(:message, conversation: conversation, account: account, message_type: :incoming)
     job.perform(row.reload)

spec/jobs/automation_rules/trigger_pending_executions_job_spec.rb

@@ -8,24 +8,14 @@ RSpec.describe AutomationRules::TriggerPendingExecutionsJob do
 
   before { GlobalConfig.clear_cache }
 
-  it 'enqueues per-row jobs for due pending rows and claims them' do
+  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)
+    create(:automation_rule_pending_execution, account: account, due_at: 1.hour.from_now)
 
     expect { job.perform }.to have_enqueued_job(AutomationRules::ProcessPendingExecutionJob).exactly(:once).with(due_row)
-    expect(due_row.reload).to be_processing
-    expect(future_row.reload).to be_pending
   end
 
-  it 'expires rows past the due window instead of enqueuing them' do
-    overdue_row = create(:automation_rule_pending_execution, account: account, conversation: conversation, due_at: 4.days.ago)
-
-    expect { job.perform }.not_to have_enqueued_job(AutomationRules::ProcessPendingExecutionJob)
-    expect(overdue_row.reload).to be_skipped
-    expect(overdue_row.skip_reason).to eq('expired')
-  end
-
-  it 'reclaims stale processing rows so the next sweep retries them' do
+  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
@@ -33,20 +23,26 @@ RSpec.describe AutomationRules::TriggerPendingExecutionsJob do
     expect { job.perform }.to have_enqueued_job(AutomationRules::ProcessPendingExecutionJob).with(stale_row)
   end
 
-  it 'caps enqueues at the configured sweep limit and leaves overflow pending' do
+  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)
-    expect(AutomationRulePendingExecution.pending.count).to eq(1)
+  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 '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
-    due_row = create(:automation_rule_pending_execution, account: account, conversation: conversation, due_at: 1.minute.ago)
+    create(:automation_rule_pending_execution, account: account, conversation: conversation, due_at: 1.minute.ago)
 
     expect { job.perform }.not_to have_enqueued_job(AutomationRules::ProcessPendingExecutionJob)
-    expect(due_row.reload).to be_pending
   end
 end

spec/models/automation_rule_pending_execution_spec.rb

@@ -121,49 +121,51 @@ RSpec.describe AutomationRulePendingExecution do
     end
   end
 
-  describe '#mark_processing!' do
+  describe '#claim!' do
     let(:row) { create(:automation_rule_pending_execution, account: account, conversation: conversation) }
 
-    it 'claims a pending row exactly once' do
-      expect(row.mark_processing!).to be(true)
+    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(row.mark_processing!).to be(false)
+      expect(described_class.find(row.id).claim!).to be(false)
     end
 
-    it 'does not claim executed rows' do
+    it 'does not claim terminal rows' do
       row.update!(status: :executed)
-      expect(row.mark_processing!).to be(false)
+      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 '.due / .expire_overdue! / .reclaim_stale!' do
-    it 'selects only pending rows inside the due window' do
+  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, due_at: 1.minute.ago, status: :executed)
-
-      expect(described_class.due).to eq([due])
-    end
-
-    it 'expires pending rows older than the due window' do
-      overdue = create(:automation_rule_pending_execution, account: account, conversation: conversation, due_at: 4.days.ago)
-      fresh = create(:automation_rule_pending_execution, account: account, due_at: 1.minute.ago)
-
-      expect(described_class.expire_overdue!).to eq(1)
-      expect(overdue.reload).to be_skipped
-      expect(overdue.skip_reason).to eq('expired')
-      expect(fresh.reload).to be_pending
-    end
-
-    it 'reclaims processing rows stuck longer than the stale timeout' do
+      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
-      recent = create(:automation_rule_pending_execution, account: account, status: :processing)
 
-      expect(described_class.reclaim_stale!).to eq(1)
-      expect(stale.reload).to be_pending
-      expect(recent.reload).to be_processing
+      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
 end