fix(automations): exclude private notes from the customer-unresponsive wait

app/javascript/dashboard/routes/dashboard/settings/automation/AutomationRuleForm.vue

@@ -313,6 +313,11 @@ const applyDelayedTrigger = () => {
       ? buildTriggerCondition('message_type', trigger.messageType)
       : buildTriggerCondition('status', triggerStatus.value),
   ];
+  // A private note is an outgoing message, so without this an internal note would read as a reply
+  // and arm the customer-unresponsive wait. Incoming messages are never private.
+  if (trigger.messageType === 'outgoing') {
+    conditions.push(buildTriggerCondition('private_note', [false]));
+  }
   if (triggerInboxes.value.length) {
     conditions.push(
       buildTriggerCondition(

app/jobs/automation_rules/process_pending_execution_job.rb

@@ -50,7 +50,10 @@ class AutomationRules::ProcessPendingExecutionJob < ApplicationJob
     ).perform.present?
   end
 
+  # Marked before the actions run: a row that dies here stays `executing`, which no sweep reclaims,
+  # so a message/email/webhook is never sent twice. Everything up to this point is still retryable.
   def execute(pending_execution)
+    pending_execution.update!(status: :executing)
     AutomationRules::ActionService.new(
       pending_execution.automation_rule,
       pending_execution.account,

app/jobs/automation_rules/trigger_pending_executions_job.rb

@@ -12,7 +12,8 @@ class AutomationRules::TriggerPendingExecutionsJob < ApplicationJob
     rows = AutomationRulePendingExecution.sweepable.for_enabled_accounts.order(:due_at).limit(sweep_limit).to_a
     rows.each { |row| AutomationRules::ProcessPendingExecutionJob.perform_later(row) }
 
-    log_summary(enqueued: rows.size, capped: rows.size >= sweep_limit, purged: purged, started_at: started_at)
+    log_summary(enqueued: rows.size, capped: rows.size >= sweep_limit, purged: purged,
+                abandoned: AutomationRulePendingExecution.abandoned.count, started_at: started_at)
   end
 
   private
@@ -25,8 +26,9 @@ class AutomationRules::TriggerPendingExecutionsJob < ApplicationJob
     (InstallationConfig.find_by(name: 'AUTOMATION_PENDING_EXECUTIONS_SWEEP_LIMIT')&.value || DEFAULT_SWEEP_LIMIT).to_i
   end
 
-  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 }
+  def log_summary(enqueued:, capped:, purged:, abandoned:, started_at:)
+    summary = { event: 'completed', enqueued: enqueued, capped: capped, purged: purged, abandoned: abandoned,
+                duration_ms: ((Time.current - started_at) * 1000).round }
     Rails.logger.info("[AutomationRules::TriggerPendingExecutionsJob] #{summary.to_json}")
   end
 end

app/listeners/automation_rule_listener.rb

@@ -46,9 +46,8 @@ class AutomationRuleListener < BaseListener
     account = conversation.account
     changed_attributes = event.data[:changed_attributes]
 
-    return unless rule_present?(event_name, account)
-
-    rules = current_account_rules(event_name, account)
+    rules = conversation_rules(event_name, account)
+    return if rules.blank?
 
     rules.each do |rule|
       conditions_match = ::AutomationRules::ConditionsFilterService.new(rule, conversation, { changed_attributes: changed_attributes }).perform
@@ -56,6 +55,17 @@ class AutomationRuleListener < BaseListener
     end
   end
 
+  # A delayed conversation rule reads as "the conversation has been in this status for N minutes",
+  # so a conversation created in that status must arm it too. Creation never dispatches
+  # CONVERSATION_UPDATED, and both paths key the episode on the same status_changed_at, so a later
+  # update arming the same episode is deduped by the unique index.
+  def conversation_rules(event_name, account)
+    rules = current_account_rules(event_name, account)
+    return rules unless event_name == 'conversation_created'
+
+    rules + current_account_rules('conversation_updated', account).where.not(execution_delay: nil)
+  end
+
   # Delayed rules record a pending execution instead of acting; the sweep re-checks and
   # runs them at due time. Flag off means no arming and no immediate fallback — a delayed
   # message silently becoming instant is worse than skipping.

app/models/automation_rule.rb

@@ -131,7 +131,8 @@ class AutomationRule < ApplicationRecord
   end
 
   def discard_stale_pending_executions
-    # armed = pending + stale processing, which the sweep would otherwise reclaim.
+    # armed = pending + processing, the rows the sweep would otherwise still run. Rows already
+    # executing are left alone: their actions are in flight and cannot be called back.
     pending_executions.armed.delete_all
   end
 

app/models/automation_rule_pending_execution.rb

@@ -38,13 +38,20 @@ class AutomationRulePendingExecution < ApplicationRecord
   belongs_to :account
   belongs_to :message, optional: true
 
-  enum status: { pending: 0, processing: 1, executed: 2, skipped: 3 }
+  # `processing` is claimed but not yet acting, so it is safe to reclaim and retry. `executing`
+  # means the actions are running: a row that dies there is never replayed, because the actions
+  # are customer-facing (messages, emails, webhooks) and repeating them is worse than dropping them.
+  enum status: { pending: 0, processing: 1, executed: 2, skipped: 3, executing: 4 }
 
   # 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))
   }
 
+  # Rows whose worker died mid-action. Nothing reclaims them; the sweep only counts them so a
+  # crash that strands customer-facing actions is visible instead of silent.
+  scope :abandoned, -> { executing.where(updated_at: ...STALE_PROCESSING_TIMEOUT.ago) }
+
   # Non-terminal rows still bound to fire (a stale processing row is reclaimed by the sweep).
   scope :armed, -> { where(status: [statuses[:pending], statuses[:processing]]) }
 
@@ -53,6 +60,11 @@ class AutomationRulePendingExecution < ApplicationRecord
   scope :for_enabled_accounts, -> { joins(:account).merge(Account.feature_delayed_automations) }
 
   def self.schedule(rule:, conversation:, message: nil)
+    # status_changed_at is only written from this feature onwards, so a conversation that predates it
+    # has no status clock. Anchoring on created_at would make every old conversation instantly
+    # overdue and fire on the next sweep; leave them for their next status change to arm.
+    return if message.nil? && conversation.status_changed_at.blank?
+
     key = arm_episode_key_for(conversation, message)
     anchor = arm_anchor_for(conversation, message)
     create!(
@@ -68,22 +80,27 @@ class AutomationRulePendingExecution < ApplicationRecord
   def self.rearm_or_advance_episode(rule, conversation, key, message, anchor)
     return unless message
 
-    row = find_by!(automation_rule_id: rule.id, conversation_id: conversation.id, episode_key: key)
-    # Jobs can arrive out of order; only a strictly newer message advances or re-arms, so a late
-    # older message can't pull due_at backwards and fire before the delay elapses.
-    return unless message.id > row.message_id
-
     due_at = rule.execution_delay.minutes.since(anchor)
-    if row.condition_skipped?
-      # A message episode key can recur (no new incoming reply) while conditions swing back into
-      # match, so a later qualifying message re-arms the condition-only skip instead of dropping.
-      row.update!(status: :pending, skip_reason: nil, due_at: due_at, message_id: message.id)
-    elsif !row.terminal?
-      # Track the newest qualifying message. Reply-chase advances due_at with each agent reply;
-      # awaiting-agent keeps its first clock (its anchor is the stable waiting_since, so due_at is
-      # unchanged). Re-anchoring a row still processing (its worker died mid-run) back to pending
-      # also keeps a stale reclaim from firing the old clock instead of the latest one.
-      row.update!(status: :pending, due_at: due_at, message_id: message.id)
+    row = find_by!(automation_rule_id: rule.id, conversation_id: conversation.id, episode_key: key)
+    # The lock (and the reload it does) makes the compare-and-write atomic. Two listeners racing on
+    # the same episode would otherwise both read the old message_id and let whichever wrote last
+    # win, so an older message could overwrite a newer one and pull due_at backwards.
+    row.with_lock do
+      # Jobs can arrive out of order; only a strictly newer message advances or re-arms, so a late
+      # older message can't pull due_at backwards and fire before the delay elapses.
+      next unless message.id > row.message_id
+
+      if row.condition_skipped?
+        # A message episode key can recur (no new incoming reply) while conditions swing back into
+        # match, so a later qualifying message re-arms the condition-only skip instead of dropping.
+        row.update!(status: :pending, skip_reason: nil, due_at: due_at, message_id: message.id)
+      elsif row.pending? || row.stale_processing?
+        # Track the newest qualifying message. Reply-chase advances due_at with each agent reply;
+        # awaiting-agent keeps its first clock (its anchor is the stable waiting_since, so due_at is
+        # unchanged). Re-anchoring a row whose worker died mid-run back to pending also keeps a
+        # stale reclaim from firing the old clock instead of the latest one.
+        row.update!(status: :pending, due_at: due_at, message_id: message.id)
+      end
     end
   end
 
@@ -92,7 +109,7 @@ class AutomationRulePendingExecution < ApplicationRecord
   # the timestamps the episode keys track.
   def self.arm_anchor_for(conversation, message)
     if message.nil?
-      conversation.status_changed_at.presence || conversation.created_at
+      conversation.status_changed_at
     elsif message.incoming?
       conversation.waiting_since.presence || message.created_at
     else
@@ -122,7 +139,7 @@ class AutomationRulePendingExecution < ApplicationRecord
     if message.nil?
       # Sub-second precision so a resolve→reopen inside one second still ends the episode.
       # Integer microseconds (not a float) so an in-memory arm and a DB-reloaded fire agree.
-      "status:#{microsecond_stamp(conversation.status_changed_at.presence || conversation.created_at)}"
+      "status:#{microsecond_stamp(conversation.status_changed_at)}"
     elsif message.incoming?
       # waiting_since is cleared on agent/bot reply, so a reply invalidates this episode. Strict
       # here: at fire time a nil waiting_since means the agent replied (episode ended).
@@ -162,6 +179,13 @@ class AutomationRulePendingExecution < ApplicationRecord
     self.class.episode_key_for(conversation, message) == episode_key
   end
 
+  # The claim renews updated_at, so a processing row past the timeout means its worker died. Only
+  # then may a re-arm take the row back: pulling a live worker's row to pending would let the sweep
+  # claim it and run the same actions alongside the worker still executing them.
+  def stale_processing?
+    processing? && updated_at < STALE_PROCESSING_TIMEOUT.ago
+  end
+
   def condition_skipped?
     skipped? && skip_reason == CONDITIONS_CHANGED_SKIP
   end
@@ -175,6 +199,6 @@ class AutomationRulePendingExecution < ApplicationRecord
   def claimable?
     # 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)
+    (pending? && due_at <= Time.current) || stale_processing?
   end
 end

spec/controllers/api/v1/accounts/automation_rules_controller_spec.rb

@@ -451,71 +451,4 @@ 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

@@ -1,11 +0,0 @@
-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

@@ -1,155 +0,0 @@
-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

@@ -1,62 +0,0 @@
-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,37 +247,4 @@ 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,21 +112,6 @@ 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') }
 
@@ -137,8 +122,7 @@ 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_delayed_automations: 1 << 5
+        feature_whatsapp_embedded_signup_inbox_creation: 1 << 4
       )
       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

@@ -1,305 +0,0 @@
-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,118 +137,4 @@ 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,28 +1230,4 @@ 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