+
+
+
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+ :placeholder="$t('AUTOMATION.ADD.FORM.DESC.PLACEHOLDER')"
+ />
+
+
+
+
-
+
+
+
+
+
+
+
diff --git a/app/javascript/dashboard/routes/dashboard/settings/automation/AutomationRuleRow.vue b/app/javascript/dashboard/routes/dashboard/settings/automation/AutomationRuleRow.vue
index 7f573e81a..b595a52f0 100644
--- a/app/javascript/dashboard/routes/dashboard/settings/automation/AutomationRuleRow.vue
+++ b/app/javascript/dashboard/routes/dashboard/settings/automation/AutomationRuleRow.vue
@@ -1,6 +1,7 @@
diff --git a/app/javascript/dashboard/routes/dashboard/settings/automation/Index.vue b/app/javascript/dashboard/routes/dashboard/settings/automation/Index.vue
index 32139ccd5..d60e6d952 100644
--- a/app/javascript/dashboard/routes/dashboard/settings/automation/Index.vue
+++ b/app/javascript/dashboard/routes/dashboard/settings/automation/Index.vue
@@ -10,7 +10,9 @@ import { useStoreGetters, useStore } from 'dashboard/composables/store';
import { picoSearch } from '@scmmishra/pico-search';
import AutomationRuleRow from './AutomationRuleRow.vue';
import Button from 'dashboard/components-next/button/Button.vue';
+import TabBar from 'dashboard/components-next/tabbar/TabBar.vue';
import { BaseTable } from 'dashboard/components-next/table';
+import { DEFAULT_DELAY_MINUTES } from './constants';
const getters = useStoreGetters();
const store = useStore();
@@ -35,9 +37,68 @@ const filteredRecords = computed(() => {
if (!query) return records.value;
return picoSearch(records.value, query, ['name', 'description']);
});
+
const uiFlags = computed(() => getters['automations/getUIFlags'].value);
const accountId = computed(() => getters.getCurrentAccountId.value);
+const isDelayedAutomationsEnabled = computed(() =>
+ getters['accounts/isFeatureEnabledonAccount'].value(
+ accountId.value,
+ 'delayed_automations'
+ )
+);
+
+const instantRecords = computed(() =>
+ filteredRecords.value.filter(automation => !automation.execution_delay)
+);
+const delayedRecords = computed(() =>
+ filteredRecords.value.filter(automation => automation.execution_delay)
+);
+
+// Accounts that can't create delayed rules, and have none left over, just see the plain list.
+const showTabs = computed(
+ () =>
+ isDelayedAutomationsEnabled.value ||
+ records.value.some(automation => automation.execution_delay)
+);
+
+const activeTab = ref('instant');
+
+const tabs = computed(() => [
+ {
+ key: 'instant',
+ label: t('AUTOMATION.LIST.TABS.INSTANT'),
+ count: instantRecords.value.length,
+ },
+ {
+ key: 'delayed',
+ label: t('AUTOMATION.LIST.TABS.DELAYED'),
+ count: delayedRecords.value.length,
+ },
+]);
+
+const activeTabIndex = computed(() =>
+ tabs.value.findIndex(tab => tab.key === activeTab.value)
+);
+
+const visibleRecords = computed(() => {
+ if (!showTabs.value) return filteredRecords.value;
+ return activeTab.value === 'delayed'
+ ? delayedRecords.value
+ : instantRecords.value;
+});
+
+const noDataMessage = computed(() => {
+ if (searchQuery.value) return t('AUTOMATION.NO_RESULTS');
+ return showTabs.value && activeTab.value === 'delayed'
+ ? t('AUTOMATION.LIST.404_DELAYED')
+ : t('AUTOMATION.LIST.404');
+});
+
+const onTabChanged = tab => {
+ activeTab.value = tab.key;
+};
+
const deleteConfirmText = computed(
() => `${t('AUTOMATION.DELETE.CONFIRM.YES')} ${selectedAutomation.value.name}`
);
@@ -52,6 +113,12 @@ const isSLAEnabled = computed(() =>
getters['accounts/isFeatureEnabledonAccount'].value(accountId.value, 'sla')
);
+const showDelayDisabledBanner = computed(
+ () =>
+ !isDelayedAutomationsEnabled.value &&
+ records.value.some(automation => automation.execution_delay)
+);
+
onMounted(() => {
store.dispatch('inboxes/get');
store.dispatch('agents/get');
@@ -66,7 +133,9 @@ onMounted(() => {
});
const openAddPopup = () => {
- addDialogRef.value?.open();
+ const startsWithWait =
+ isDelayedAutomationsEnabled.value && activeTab.value === 'delayed';
+ addDialogRef.value?.open(startsWithWait ? DEFAULT_DELAY_MINUTES : null);
};
const hideAddPopup = () => {
addDialogRef.value?.close();
@@ -74,7 +143,7 @@ const hideAddPopup = () => {
const openEditPopup = response => {
selectedAutomation.value = { ...response };
- editDialogRef.value?.open();
+ editDialogRef.value?.open(response);
};
const hideEditPopup = () => {
editDialogRef.value?.close();
@@ -128,11 +197,11 @@ const submitAutomation = async (payload, mode) => {
hideAddPopup();
hideEditPopup();
} catch (error) {
- const errorMessage =
+ const fallbackMessage =
mode === 'edit'
? t('AUTOMATION.EDIT.API.ERROR_MESSAGE')
: t('AUTOMATION.ADD.API.ERROR_MESSAGE');
- useAlert(errorMessage);
+ useAlert(error?.response?.data?.error || fallbackMessage);
}
};
const toggleAutomation = async ({ id, name, status }) => {
@@ -197,9 +266,16 @@ const tableHeaders = computed(() => {
:search-placeholder="$t('AUTOMATION.SEARCH_PLACEHOLDER')"
feature-name="automation"
>
-
+
+
+
+
- {{ $t('AUTOMATION.COUNT', { n: records.length }) }}
+ {{ $t('AUTOMATION.COUNT', { n: visibleRecords.length }) }}
@@ -212,12 +288,16 @@ const tableHeaders = computed(() => {
+
+ {{ $t('AUTOMATION.LIST.DELAY_DISABLED_BANNER') }}
+
+import { computed } from 'vue';
+import AutomationActionInput from 'dashboard/components/widgets/AutomationActionInput.vue';
+import NextButton from 'dashboard/components-next/button/Button.vue';
+import {
+ getFileName,
+ showActionInput,
+} from 'dashboard/helper/automationHelper';
+
+const props = defineProps({
+ actionTypes: {
+ type: Array,
+ required: true,
+ },
+ getActionDropdownValues: {
+ type: Function,
+ required: true,
+ },
+ files: {
+ type: Array,
+ default: () => [],
+ },
+ showFileName: {
+ type: Boolean,
+ default: false,
+ },
+ errors: {
+ type: Object,
+ default: () => ({}),
+ },
+ appendNewAction: {
+ type: Function,
+ required: true,
+ },
+ removeAction: {
+ type: Function,
+ required: true,
+ },
+ resetAction: {
+ type: Function,
+ required: true,
+ },
+});
+
+const actions = defineModel({ type: Array, required: true });
+
+const hasActionErrors = computed(() =>
+ Object.keys(props.errors).some(key => key.startsWith('action_'))
+);
+
+
+
+
+
+
+
+
diff --git a/app/javascript/dashboard/routes/dashboard/settings/automation/components/AutomationInstantTrigger.vue b/app/javascript/dashboard/routes/dashboard/settings/automation/components/AutomationInstantTrigger.vue
new file mode 100644
index 000000000..bdadacc39
--- /dev/null
+++ b/app/javascript/dashboard/routes/dashboard/settings/automation/components/AutomationInstantTrigger.vue
@@ -0,0 +1,124 @@
+
+
+
+
+
+
+
+ {{ $t('AUTOMATION.FORM.RESET_MESSAGE') }}
+
+
+
+
+
+
+
+
diff --git a/app/javascript/dashboard/routes/dashboard/settings/automation/components/AutomationRunTypeSelector.vue b/app/javascript/dashboard/routes/dashboard/settings/automation/components/AutomationRunTypeSelector.vue
new file mode 100644
index 000000000..d07094b45
--- /dev/null
+++ b/app/javascript/dashboard/routes/dashboard/settings/automation/components/AutomationRunTypeSelector.vue
@@ -0,0 +1,35 @@
+
+
+
+
+
+
+
+
+
+
diff --git a/app/javascript/dashboard/routes/dashboard/settings/automation/components/AutomationWaitCondition.vue b/app/javascript/dashboard/routes/dashboard/settings/automation/components/AutomationWaitCondition.vue
new file mode 100644
index 000000000..4689d020c
--- /dev/null
+++ b/app/javascript/dashboard/routes/dashboard/settings/automation/components/AutomationWaitCondition.vue
@@ -0,0 +1,230 @@
+
+
+
+
+
+
+
+
+
+ {{ $t('AUTOMATION.ADD.FORM.WAIT.WHEN_LABEL') }}
+
+
+
+
+
+ {{ $t('AUTOMATION.ADD.FORM.WAIT.STATUS_LABEL') }}
+
+
+
+
+
+ {{ $t('AUTOMATION.ADD.FORM.WAIT.FOR_LABEL') }}
+
+
+
+
+
+
+
+ {{ $t('AUTOMATION.ADD.FORM.WAIT.INBOX_LABEL') }}
+
+
+
+
+
+
+
+ {{ $t('AUTOMATION.ADD.FORM.WAIT.ERROR') }}
+
+
+
diff --git a/app/javascript/dashboard/routes/dashboard/settings/automation/constants.js b/app/javascript/dashboard/routes/dashboard/settings/automation/constants.js
index 3c073ec7e..4a467474e 100644
--- a/app/javascript/dashboard/routes/dashboard/settings/automation/constants.js
+++ b/app/javascript/dashboard/routes/dashboard/settings/automation/constants.js
@@ -806,3 +806,27 @@ export const AUTOMATION_ACTION_TYPES = [
inputType: 'search_select',
},
];
+
+export const DEFAULT_DELAY_MINUTES = 240; // 4 hours
+export const MIN_DELAY_MINUTES = 10;
+export const MAX_DELAY_MINUTES = 43200; // 30 days
+export const DEFAULT_TRIGGER_STATUS = 'pending';
+
+// A delayed rule is expressed as one meaningful trigger instead of a raw event + conditions. Each
+// trigger maps to the automation's event_name plus a preset condition: message_type for the two
+// unresponsive cases (reply-chase / awaiting-agent), or a chosen status for conversation_updated.
+export const DELAYED_TRIGGERS = [
+ { key: 'conversation_status', eventName: 'conversation_updated' },
+ {
+ key: 'customer_unresponsive',
+ eventName: 'message_created',
+ messageType: 'outgoing',
+ },
+ {
+ key: 'agent_unresponsive',
+ eventName: 'message_created',
+ messageType: 'incoming',
+ },
+];
+
+export const DEFAULT_TRIGGER = DELAYED_TRIGGERS[0].key;
diff --git a/app/jobs/automation_rules/process_pending_execution_job.rb b/app/jobs/automation_rules/process_pending_execution_job.rb
new file mode 100644
index 000000000..d555e0d8c
--- /dev/null
+++ b/app/jobs/automation_rules/process_pending_execution_job.rb
@@ -0,0 +1,63 @@
+class AutomationRules::ProcessPendingExecutionJob < ApplicationJob
+ queue_as :medium
+
+ discard_on ActiveJob::DeserializationError
+
+ def perform(pending_execution)
+ # Account flag off pauses (not skips): leave the row pending so re-enabling resumes it.
+ return unless pending_execution.account.feature_enabled?('delayed_automations')
+ # Atomic claim: a duplicate enqueue (overlapping sweep or stale reclaim) loses here and returns.
+ return unless pending_execution.claim!
+
+ 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 next sweep reclaims and retries it once the lock goes stale.
+ ChatwootExceptionTracker.new(e, account: pending_execution.account).capture_exception
+ end
+
+ private
+
+ def skip_reason_for(pending_execution)
+ return 'expired' if pending_execution.due_at < AutomationRulePendingExecution::DUE_WINDOW.ago
+
+ structural_skip_reason(pending_execution) || behavioral_skip_reason(pending_execution)
+ end
+
+ def structural_skip_reason(pending_execution)
+ rule = pending_execution.automation_rule
+ return 'rule_inactive' if rule.nil? || !rule.active?
+ return 'conversation_gone' if pending_execution.conversation.nil?
+
+ nil
+ end
+
+ def behavioral_skip_reason(pending_execution)
+ return 'episode_moved' unless pending_execution.episode_current?
+ return 'conditions_changed' unless conditions_still_match?(pending_execution)
+
+ nil
+ end
+
+ def conditions_still_match?(pending_execution)
+ AutomationRules::ConditionsFilterService.new(
+ pending_execution.automation_rule,
+ pending_execution.conversation,
+ { message: pending_execution.message }
+ ).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,
+ pending_execution.conversation
+ ).perform
+ pending_execution.update!(status: :executed)
+ end
+end
diff --git a/app/jobs/automation_rules/trigger_pending_executions_job.rb b/app/jobs/automation_rules/trigger_pending_executions_job.rb
new file mode 100644
index 000000000..a16ab6bde
--- /dev/null
+++ b/app/jobs/automation_rules/trigger_pending_executions_job.rb
@@ -0,0 +1,28 @@
+class AutomationRules::TriggerPendingExecutionsJob < ApplicationJob
+ queue_as :scheduled_jobs
+
+ DEFAULT_SWEEP_LIMIT = 1000
+
+ def perform
+ started_at = Time.current
+ purged = AutomationRulePendingExecution.purge_terminal!
+
+ 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,
+ abandoned: AutomationRulePendingExecution.abandoned.count, started_at: started_at)
+ end
+
+ private
+
+ def sweep_limit
+ (InstallationConfig.find_by(name: 'AUTOMATION_PENDING_EXECUTIONS_SWEEP_LIMIT')&.value || DEFAULT_SWEEP_LIMIT).to_i
+ end
+
+ 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
diff --git a/app/jobs/trigger_scheduled_items_job.rb b/app/jobs/trigger_scheduled_items_job.rb
index 0ec368c54..5b38ccc87 100644
--- a/app/jobs/trigger_scheduled_items_job.rb
+++ b/app/jobs/trigger_scheduled_items_job.rb
@@ -19,6 +19,9 @@ class TriggerScheduledItemsJob < ApplicationJob
# Job to sync whatsapp templates
Channels::Whatsapp::TemplatesSyncSchedulerJob.perform_later
+
+ # Job to trigger pending executions
+ AutomationRules::TriggerPendingExecutionsJob.perform_later
end
end
diff --git a/app/listeners/automation_rule_listener.rb b/app/listeners/automation_rule_listener.rb
index 0515d6952..e4d2da0b3 100644
--- a/app/listeners/automation_rule_listener.rb
+++ b/app/listeners/automation_rule_listener.rb
@@ -30,7 +30,7 @@ class AutomationRuleListener < BaseListener
rules.each do |rule|
conditions_match = ::AutomationRules::ConditionsFilterService.new(rule, message.conversation,
{ message: message, changed_attributes: changed_attributes }).perform
- ::AutomationRules::ActionService.new(rule, account, message.conversation).perform if conditions_match.present?
+ execute_rule(rule, account, message.conversation, message: message) if conditions_match.present?
end
end
@@ -46,13 +46,36 @@ 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
- AutomationRules::ActionService.new(rule, account, conversation).perform if conditions_match.present?
+ execute_rule(rule, account, conversation) if conditions_match.present?
+ 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.
+ def execute_rule(rule, account, conversation, message: nil)
+ if rule.execution_delay.present?
+ return unless account.feature_enabled?('delayed_automations')
+
+ AutomationRulePendingExecution.schedule(rule: rule, conversation: conversation, message: message)
+ else
+ ::AutomationRules::ActionService.new(rule, account, conversation).perform
end
end
diff --git a/app/models/account.rb b/app/models/account.rb
index 8d5d61717..20aaced28 100644
--- a/app/models/account.rb
+++ b/app/models/account.rb
@@ -68,6 +68,7 @@ class Account < ApplicationRecord
has_many :articles, dependent: :destroy_async, class_name: '::Article'
has_many :assignment_policies, dependent: :destroy_async
has_many :automation_rules, dependent: :destroy_async
+ has_many :automation_rule_pending_executions, dependent: :delete_all
has_many :macros, dependent: :destroy_async
has_many :campaigns, dependent: :destroy_async
has_many :canned_responses, dependent: :destroy_async
@@ -114,6 +115,7 @@ class Account < ApplicationRecord
before_validation :validate_limit_keys
after_create_commit :notify_creation
after_update_commit :clear_unread_conversation_counts_cache, if: :saved_change_to_feature_conversation_unread_counts?
+ after_update :resume_delayed_automations, if: -> { saved_change_to_feature_delayed_automations? && feature_delayed_automations? }
after_destroy :remove_account_sequences
def agents
@@ -196,6 +198,10 @@ class Account < ApplicationRecord
::Conversations::UnreadCounts::Store.clear_account!(id)
end
+ def resume_delayed_automations
+ AutomationRulePendingExecution.reschedule_paused(self)
+ end
+
trigger.after(:insert).for_each(:row) do
"execute format('create sequence IF NOT EXISTS conv_dpid_seq_%s', NEW.id);"
end
diff --git a/app/models/automation_rule.rb b/app/models/automation_rule.rb
index 9a437bac9..81832ff0c 100644
--- a/app/models/automation_rule.rb
+++ b/app/models/automation_rule.rb
@@ -2,16 +2,17 @@
#
# Table name: automation_rules
#
-# id :bigint not null, primary key
-# actions :jsonb not null
-# active :boolean default(TRUE), not null
-# conditions :jsonb not null
-# description :text
-# event_name :string not null
-# name :string not null
-# created_at :datetime not null
-# updated_at :datetime not null
-# account_id :bigint not null
+# id :bigint not null, primary key
+# actions :jsonb not null
+# active :boolean default(TRUE), not null
+# conditions :jsonb not null
+# description :text
+# event_name :string not null
+# execution_delay :integer
+# name :string not null
+# created_at :datetime not null
+# updated_at :datetime not null
+# account_id :bigint not null
#
# Indexes
#
@@ -21,7 +22,13 @@ class AutomationRule < ApplicationRecord
include Rails.application.routes.url_helpers
include Reauthorizable
+ EXECUTION_DELAY_RANGE = (10..43_200) # minutes: 10 min to 30 days
+ # Conversation-level delayed rules key their episode on status; only status and attributes
+ # that never change after the delay (inbox) are safe to also filter on.
+ DELAYED_CONVERSATION_ATTRIBUTES = %w[status inbox_id].freeze
+
belongs_to :account
+ has_many :pending_executions, class_name: 'AutomationRulePendingExecution', dependent: :delete_all
has_many_attached :files
validate :json_conditions_format
@@ -29,8 +36,13 @@ class AutomationRule < ApplicationRecord
validate :query_operator_presence
validate :query_operator_value
validates :account_id, presence: true
+ validates :execution_delay, numericality: { only_integer: true, in: EXECUTION_DELAY_RANGE }, allow_nil: true
+ validate :execution_delay_supported_conditions
+ validate :execution_delay_supported_event
after_update_commit :reauthorized!, if: -> { saved_change_to_conditions? }
+ # Discard rows armed under the old definition; they re-arm on the next matching event.
+ after_update :discard_stale_pending_executions, if: :execution_config_changed?
scope :active, -> { where(active: true) }
@@ -95,6 +107,37 @@ class AutomationRule < ApplicationRecord
end
end
+ # The fire-time re-check cannot reconstruct changed_attributes, so delayed rules
+ # cannot use attribute_changed conditions.
+ def execution_delay_supported_conditions
+ return if execution_delay.blank? || conditions.blank?
+ return if conditions.none? { |obj| obj['filter_operator'] == 'attribute_changed' }
+
+ errors.add(:execution_delay, 'cannot be used with attribute_changed conditions.')
+ end
+
+ # Conversation-level episodes key on status_changed_at alone. Mutable attributes would collapse
+ # distinct periods into one episode, so only status and immutable filters (inbox) are allowed.
+ def execution_delay_supported_event
+ return if execution_delay.blank? || conditions.blank? || event_name == 'message_created'
+ return if conditions.all? { |obj| DELAYED_CONVERSATION_ATTRIBUTES.include?(obj['attribute_key']) }
+
+ errors.add(:execution_delay, 'only supports status and inbox conditions for conversation-level events.')
+ end
+
+ # Deactivating counts: without it a rule turned off and back on before its due time would still
+ # run the actions the admin turned it off to stop.
+ def execution_config_changed?
+ saved_change_to_active? || saved_change_to_execution_delay? || saved_change_to_event_name? ||
+ saved_change_to_conditions? || saved_change_to_actions?
+ end
+
+ def discard_stale_pending_executions
+ # 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
+
def validate_single_condition(condition)
query_operator = condition['query_operator']
diff --git a/app/models/automation_rule_pending_execution.rb b/app/models/automation_rule_pending_execution.rb
new file mode 100644
index 000000000..264a5b5d2
--- /dev/null
+++ b/app/models/automation_rule_pending_execution.rb
@@ -0,0 +1,210 @@
+# == Schema Information
+#
+# Table name: automation_rule_pending_executions
+#
+# id :bigint not null, primary key
+# due_at :datetime not null
+# episode_key :string not null
+# skip_reason :string
+# status :integer default("pending"), not null
+# created_at :datetime not null
+# updated_at :datetime not null
+# account_id :bigint not null
+# automation_rule_id :bigint not null
+# conversation_id :bigint not null
+# message_id :bigint
+#
+# Indexes
+#
+# index_automation_pending_executions_on_status_and_updated_at (status,updated_at)
+# index_automation_rule_pending_executions_on_account_id (account_id)
+# index_automation_rule_pending_executions_on_automation_rule_id (automation_rule_id)
+# index_automation_rule_pending_executions_on_conversation_id (conversation_id)
+# index_automation_rule_pending_executions_on_status_and_due_at (status,due_at)
+# uniq_automation_pending_execution_episode (automation_rule_id,conversation_id,episode_key) UNIQUE
+#
+class AutomationRulePendingExecution < ApplicationRecord
+ # Rows older than this never fire (bounds backlog replay after downtime).
+ DUE_WINDOW = 3.days
+ # 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
+ belongs_to :account
+ belongs_to :message, optional: true
+
+ # `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 }
+
+ # Processing rows whose worker died: the claim renews updated_at, so a lock past the timeout is abandoned.
+ scope :stale_processing, -> { processing.where(updated_at: ...STALE_PROCESSING_TIMEOUT.ago) }
+
+ # Rows a sweep should hand to a worker: due pending rows, plus processing rows whose lock went stale.
+ scope :sweepable, -> { pending.where(due_at: ..Time.current).or(stale_processing) }
+
+ # 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]]) }
+
+ # Excludes rows whose account paused delayed automations, so one disabled account's backlog
+ # can't fill the sweep limit and starve enabled accounts (paused rows resume on re-enable).
+ 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!(
+ automation_rule: rule, conversation: conversation, account_id: conversation.account_id,
+ message_id: message&.id, episode_key: key, due_at: rule.execution_delay.minutes.since(anchor)
+ )
+ rescue ActiveRecord::RecordNotUnique
+ rearm_or_advance_episode(rule, conversation, key, message, anchor)
+ end
+
+ # The episode is already armed. Status episodes keep their first clock (a status change would
+ # give a new key), so only message episodes advance or re-arm here.
+ def self.rearm_or_advance_episode(rule, conversation, key, message, anchor)
+ return unless message
+
+ due_at = rule.execution_delay.minutes.since(anchor)
+ 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
+ # A row that already acted keeps its episode's single run, and a live worker keeps its row.
+ next unless row.pending? || row.skipped? || 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. A skipped row re-arms whatever
+ # the reason: its key recurs while the customer stays quiet, so leaving it terminal would
+ # suppress every later message in the episode until the row is purged.
+ row.update!(status: :pending, skip_reason: nil, due_at: due_at, message_id: message.id)
+ end
+ end
+
+ # The wait is measured from when the qualifying event happened, not when this (possibly
+ # backlogged or retried) listener runs, so a late dispatch still fires on schedule. Mirrors
+ # the timestamps the episode keys track.
+ def self.arm_anchor_for(conversation, message)
+ if message.nil?
+ conversation.status_changed_at
+ elsif message.incoming?
+ conversation.waiting_since.presence || message.created_at
+ else
+ message.created_at
+ end
+ end
+
+ # Arming keys differ from the strict fire-time keys wherever current state can already reflect the
+ # event the row waits for: MESSAGE_CREATED dispatches asynchronously, so this can run long after
+ # the message it arms.
+ def self.arm_episode_key_for(conversation, message)
+ return episode_key_for(conversation, message) if message.nil?
+
+ if message.incoming?
+ # waiting_since is written just after MESSAGE_CREATED dispatches, so it can still be nil here.
+ # It becomes the starting message's created_at, so use that; the strict fire-time key then
+ # matches once waiting_since is settled.
+ return episode_key_for(conversation, message) if conversation.waiting_since.present?
+
+ "awaiting_agent:#{microsecond_stamp(message.created_at)}"
+ else
+ # Count only the replies that predate the agent message being chased. A customer reply that
+ # landed while this job queued must end the episode at fire time, not be baked into its key.
+ "reply_chase:#{conversation.messages.incoming.where(id: ...message.id).maximum(:id) || 0}"
+ end
+ end
+
+ # Microsecond integer, not a float: epoch seconds carry ~16 significant digits, past float64's
+ # precision, so an in-memory timestamp (arm time) and its DB-reloaded value (fire time) would
+ # round to different floats. strftime is exact on both. Sub-second distinguishes rapid episodes.
+ def self.microsecond_stamp(time)
+ time&.strftime('%s%6N') || '0'
+ end
+
+ # Episode keys identify one qualifying stretch of conversation state; when the recomputed
+ # key no longer matches, the episode ended and the pending action is cancelled at fire time.
+ def self.episode_key_for(conversation, message)
+ 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)}"
+ 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).
+ "awaiting_agent:#{microsecond_stamp(conversation.waiting_since)}"
+ else
+ # A new customer message changes the max incoming id, invalidating this episode.
+ "reply_chase:#{conversation.messages.incoming.maximum(:id) || 0}"
+ end
+ end
+
+ def self.purge_terminal!
+ where(status: [statuses[:executed], statuses[:skipped]], updated_at: ...RETENTION_WINDOW.ago)
+ .in_batches(of: 1000).delete_all
+ end
+
+ # Rows that came due while an account had delayed automations paused would expire the moment
+ # the sweep reaches them on resume. Reset their clock so pause/resume replays them (still
+ # subject to the fire-time episode/condition re-checks) instead of silently dropping them.
+ # Stale processing rows go back to pending too (their worker is gone); resetting due_at alone would
+ # renew the lock and hold them out of the sweep for another timeout. A live worker keeps its row.
+ def self.reschedule_paused(account)
+ overdue = pending.or(stale_processing).where(account_id: account.id, due_at: ...DUE_WINDOW.ago)
+ overdue.find_each { |row| row.update!(status: :pending, due_at: Time.current) }
+ end
+
+ # Atomic claim: only one worker can move a row into processing, so a row re-enqueued by an
+ # overlapping sweep (or after a stale reclaim) cannot double-execute. Refreshing updated_at
+ # renews the lock, keeping the row out of the stale window while this worker holds it.
+ def claim!
+ with_lock do
+ next false unless claimable?
+
+ update!(status: :processing, updated_at: Time.current)
+ true
+ end
+ end
+
+ def episode_current?
+ 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 terminal?
+ executed? || skipped?
+ end
+
+ private
+
+ 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) || stale_processing?
+ end
+end
diff --git a/app/models/conversation.rb b/app/models/conversation.rb
index 4d26dcf47..38fb0063f 100644
--- a/app/models/conversation.rb
+++ b/app/models/conversation.rb
@@ -15,6 +15,7 @@
# priority :integer
# snoozed_until :datetime
# status :integer default("open"), not null
+# status_changed_at :datetime
# uuid :uuid not null
# waiting_since :datetime
# created_at :datetime not null
@@ -125,8 +126,10 @@ class Conversation < ApplicationRecord
has_many :notifications, as: :primary_actor, dependent: :destroy_async
has_many :attachments, through: :messages
has_many :reporting_events, dependent: :destroy_async
+ has_many :automation_rule_pending_executions, dependent: :delete_all
before_save :ensure_snooze_until_reset
+ before_save :set_status_changed_at
before_create :determine_conversation_status
before_create :ensure_waiting_since
@@ -278,6 +281,10 @@ class Conversation < ApplicationRecord
self.snoozed_until = nil unless snoozed?
end
+ def set_status_changed_at
+ self.status_changed_at = Time.current if new_record? || status_changed?
+ end
+
def ensure_waiting_since
self.waiting_since = created_at
end
diff --git a/app/views/api/v1/accounts/automation_rules/partials/_automation_rule.json.jbuilder b/app/views/api/v1/accounts/automation_rules/partials/_automation_rule.json.jbuilder
index a1047c82f..480994bcb 100644
--- a/app/views/api/v1/accounts/automation_rules/partials/_automation_rule.json.jbuilder
+++ b/app/views/api/v1/accounts/automation_rules/partials/_automation_rule.json.jbuilder
@@ -7,4 +7,5 @@ json.conditions automation_rule.conditions
json.actions automation_rule.actions
json.created_on automation_rule.created_at.to_i
json.active automation_rule.active?
+json.execution_delay automation_rule.execution_delay
json.files automation_rule.file_base_data if automation_rule.files.any?
diff --git a/config/features.yml b/config/features.yml
index 950d6e7c5..87153c290 100644
--- a/config/features.yml
+++ b/config/features.yml
@@ -268,3 +268,7 @@
display_name: WhatsApp Embedded Signup Flow
enabled: false
column: feature_flags_ext_1
+- name: delayed_automations
+ display_name: Delayed Automations
+ enabled: false
+ column: feature_flags_ext_1
diff --git a/db/migrate/20260709060000_add_execution_delay_to_automation_rules.rb b/db/migrate/20260709060000_add_execution_delay_to_automation_rules.rb
new file mode 100644
index 000000000..c37993188
--- /dev/null
+++ b/db/migrate/20260709060000_add_execution_delay_to_automation_rules.rb
@@ -0,0 +1,5 @@
+class AddExecutionDelayToAutomationRules < ActiveRecord::Migration[7.0]
+ def change
+ add_column :automation_rules, :execution_delay, :integer
+ end
+end
diff --git a/db/migrate/20260709060100_add_status_changed_at_to_conversations.rb b/db/migrate/20260709060100_add_status_changed_at_to_conversations.rb
new file mode 100644
index 000000000..7aa32cc6a
--- /dev/null
+++ b/db/migrate/20260709060100_add_status_changed_at_to_conversations.rb
@@ -0,0 +1,5 @@
+class AddStatusChangedAtToConversations < ActiveRecord::Migration[7.0]
+ def change
+ add_column :conversations, :status_changed_at, :datetime
+ end
+end
diff --git a/db/migrate/20260709060200_create_automation_rule_pending_executions.rb b/db/migrate/20260709060200_create_automation_rule_pending_executions.rb
new file mode 100644
index 000000000..9ae45b636
--- /dev/null
+++ b/db/migrate/20260709060200_create_automation_rule_pending_executions.rb
@@ -0,0 +1,21 @@
+class CreateAutomationRulePendingExecutions < ActiveRecord::Migration[7.0]
+ def change
+ create_table :automation_rule_pending_executions do |t|
+ t.references :automation_rule, null: false
+ t.references :conversation, null: false
+ t.references :account, null: false
+ t.bigint :message_id
+ t.datetime :due_at, null: false
+ t.string :episode_key, null: false
+ t.integer :status, null: false, default: 0
+ t.string :skip_reason
+
+ t.timestamps
+ end
+
+ add_index :automation_rule_pending_executions, [:status, :due_at]
+ add_index :automation_rule_pending_executions,
+ [:automation_rule_id, :conversation_id, :episode_key],
+ unique: true, name: 'uniq_automation_pending_execution_episode'
+ end
+end
diff --git a/db/migrate/20260729051500_add_status_updated_at_index_to_automation_rule_pending_executions.rb b/db/migrate/20260729051500_add_status_updated_at_index_to_automation_rule_pending_executions.rb
new file mode 100644
index 000000000..183f4e070
--- /dev/null
+++ b/db/migrate/20260729051500_add_status_updated_at_index_to_automation_rule_pending_executions.rb
@@ -0,0 +1,8 @@
+class AddStatusUpdatedAtIndexToAutomationRulePendingExecutions < ActiveRecord::Migration[7.0]
+ def change
+ # Stale reclaim, the abandoned-row alarm and the terminal purge all filter on status + updated_at;
+ # the (status, due_at) index does not serve them, and terminal rows linger for 30 days.
+ add_index :automation_rule_pending_executions, [:status, :updated_at],
+ name: 'index_automation_pending_executions_on_status_and_updated_at'
+ end
+end
diff --git a/db/schema.rb b/db/schema.rb
index 15ee142dc..75bd5480c 100644
--- a/db/schema.rb
+++ b/db/schema.rb
@@ -10,7 +10,7 @@
#
# It's strongly recommended that you check this file into your version control system.
-ActiveRecord::Schema[7.1].define(version: 2026_07_28_000001) do
+ActiveRecord::Schema[7.1].define(version: 2026_07_29_051500) do
# These extensions should be enabled to support this database
enable_extension "pg_stat_statements"
enable_extension "pg_trgm"
@@ -278,6 +278,25 @@ ActiveRecord::Schema[7.1].define(version: 2026_07_28_000001) do
t.index ["user_id", "user_type"], name: "user_index"
end
+ create_table "automation_rule_pending_executions", force: :cascade do |t|
+ t.bigint "automation_rule_id", null: false
+ t.bigint "conversation_id", null: false
+ t.bigint "account_id", null: false
+ t.bigint "message_id"
+ t.datetime "due_at", null: false
+ t.string "episode_key", null: false
+ t.integer "status", default: 0, null: false
+ t.string "skip_reason"
+ t.datetime "created_at", null: false
+ t.datetime "updated_at", null: false
+ t.index ["account_id"], name: "index_automation_rule_pending_executions_on_account_id"
+ t.index ["automation_rule_id", "conversation_id", "episode_key"], name: "uniq_automation_pending_execution_episode", unique: true
+ t.index ["automation_rule_id"], name: "index_automation_rule_pending_executions_on_automation_rule_id"
+ t.index ["conversation_id"], name: "index_automation_rule_pending_executions_on_conversation_id"
+ t.index ["status", "due_at"], name: "index_automation_rule_pending_executions_on_status_and_due_at"
+ t.index ["status", "updated_at"], name: "index_automation_pending_executions_on_status_and_updated_at"
+ end
+
create_table "automation_rules", force: :cascade do |t|
t.bigint "account_id", null: false
t.string "name", null: false
@@ -288,6 +307,7 @@ ActiveRecord::Schema[7.1].define(version: 2026_07_28_000001) do
t.datetime "created_at", null: false
t.datetime "updated_at", null: false
t.boolean "active", default: true, null: false
+ t.integer "execution_delay"
t.index ["account_id"], name: "index_automation_rules_on_account_id"
end
@@ -794,6 +814,7 @@ ActiveRecord::Schema[7.1].define(version: 2026_07_28_000001) do
t.datetime "waiting_since"
t.text "cached_label_list"
t.bigint "assignee_agent_bot_id"
+ t.datetime "status_changed_at"
t.index ["account_id", "display_id"], name: "index_conversations_on_account_id_and_display_id", unique: true
t.index ["account_id", "id"], name: "index_conversations_on_id_and_account_id"
t.index ["account_id", "inbox_id", "status", "assignee_id"], name: "conv_acid_inbid_stat_asgnid_idx"
diff --git a/spec/controllers/api/v1/accounts/automation_rules_controller_spec.rb b/spec/controllers/api/v1/accounts/automation_rules_controller_spec.rb
index 5b7c0112d..a64a4abf7 100644
--- a/spec/controllers/api/v1/accounts/automation_rules_controller_spec.rb
+++ b/spec/controllers/api/v1/accounts/automation_rules_controller_spec.rb
@@ -451,4 +451,81 @@ 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 'rejects cloning an existing delayed rule instead of turning it into an instant one' 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(:unprocessable_entity)
+ expect(account.automation_rules.count).to eq(1)
+ end
+
+ it 'still clones a rule that carries no delay' do
+ automation_rule = create(:automation_rule, account: account)
+
+ 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.count).to eq(2)
+ end
+ end
+ end
end
diff --git a/spec/factories/automation_rule_pending_executions.rb b/spec/factories/automation_rule_pending_executions.rb
new file mode 100644
index 000000000..5dccfee8b
--- /dev/null
+++ b/spec/factories/automation_rule_pending_executions.rb
@@ -0,0 +1,12 @@
+FactoryBot.define do
+ factory :automation_rule_pending_execution do
+ account
+ automation_rule { association :automation_rule, account: account }
+ conversation { association :conversation, account: account }
+ message { nil }
+ # Derive from production so the row is episode_current for the episode the message implies.
+ episode_key { AutomationRulePendingExecution.episode_key_for(conversation, message) }
+ due_at { 1.hour.from_now }
+ status { :pending }
+ end
+end
diff --git a/spec/jobs/automation_rules/process_pending_execution_job_spec.rb b/spec/jobs/automation_rules/process_pending_execution_job_spec.rb
new file mode 100644
index 000000000..44cb22520
--- /dev/null
+++ b/spec/jobs/automation_rules/process_pending_execution_job_spec.rb
@@ -0,0 +1,193 @@
+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
+ # A reply-chase rule whose action is customer-facing, so a replay is visible as a duplicate message.
+ let(:follow_up_rule) do
+ 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'] }])
+ end
+ let(:agent_reply) { create(:message, conversation: conversation, account: account, message_type: :outgoing) }
+ let(:follow_up_execution) do
+ AutomationRulePendingExecution.schedule(rule: follow_up_rule, conversation: conversation, message: agent_reply)
+ AutomationRulePendingExecution.last.tap { |row| row.update!(due_at: 1.minute.ago) }
+ end
+
+ before { account.enable_features!('delayed_automations') }
+
+ 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 '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 executing 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_executing
+ expect(ChatwootExceptionTracker).to have_received(:new)
+ end
+
+ it 'retries a row that died before the actions and still sends the follow-up exactly once' do
+ row = follow_up_execution
+ allow(AutomationRules::ConditionsFilterService).to receive(:new).and_raise(StandardError, 'boom')
+
+ job.perform(row.reload)
+
+ expect(row.reload).to be_processing
+ expect(conversation.messages.outgoing.pluck(:content)).not_to include('Just checking in')
+
+ allow(AutomationRules::ConditionsFilterService).to receive(:new).and_call_original
+ travel_to(20.minutes.from_now) do
+ expect(AutomationRulePendingExecution.sweepable).to include(row)
+ described_class.new.perform(AutomationRulePendingExecution.find(row.id))
+ end
+
+ expect(row.reload).to be_executed
+ expect(conversation.messages.outgoing.where(content: 'Just checking in').count).to eq(1)
+ end
+
+ it 'never sends the follow-up twice when the row dies after the action ran but before it was marked executed' do
+ row = follow_up_execution.reload
+ allow(row).to receive(:update!).and_call_original
+ allow(row).to receive(:update!).with(status: :executed).and_raise(ActiveRecord::StatementInvalid, 'connection lost')
+
+ job.perform(row)
+
+ # The row is left `executing`, which no sweep reclaims, so the stale retry cannot re-send.
+ expect(row.reload).to be_executing
+ expect(conversation.messages.outgoing.where(content: 'Just checking in').count).to eq(1)
+
+ travel_to(20.minutes.from_now) do
+ expect(AutomationRulePendingExecution.sweepable).not_to include(row)
+ expect(AutomationRulePendingExecution.abandoned).to include(row)
+ described_class.new.perform(AutomationRulePendingExecution.find(row.id))
+ end
+
+ expect(row.reload).to be_executing
+ expect(conversation.messages.outgoing.where(content: 'Just checking in').count).to eq(1)
+ end
+
+ it 'sends the follow-up exactly once for the reply-chase story' do
+ job.perform(follow_up_execution.reload)
+
+ expect(follow_up_execution.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 the arming job ran' do
+ agent_reply
+ create(:message, conversation: conversation, account: account, message_type: :incoming)
+ # Only now does the queued MESSAGE_CREATED job arm the row for the agent's reply.
+ row = follow_up_execution
+
+ 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
+
+ it 'cancels the follow-up when the customer replied before it was due' do
+ row = follow_up_execution
+ 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
diff --git a/spec/jobs/automation_rules/trigger_pending_executions_job_spec.rb b/spec/jobs/automation_rules/trigger_pending_executions_job_spec.rb
new file mode 100644
index 000000000..b91074a27
--- /dev/null
+++ b/spec/jobs/automation_rules/trigger_pending_executions_job_spec.rb
@@ -0,0 +1,51 @@
+require 'rails_helper'
+
+RSpec.describe AutomationRules::TriggerPendingExecutionsJob do
+ subject(:job) { described_class.new }
+
+ let(:account) { create(:account) }
+ let(:conversation) { create(:conversation, account: account) }
+
+ before { account.enable_features!('delayed_automations') }
+
+ 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
+end
diff --git a/spec/listeners/automation_rule_listener_spec.rb b/spec/listeners/automation_rule_listener_spec.rb
index 08085da7a..c083feacd 100644
--- a/spec/listeners/automation_rule_listener_spec.rb
+++ b/spec/listeners/automation_rule_listener_spec.rb
@@ -247,4 +247,74 @@ 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
+
+ # The builder's "customer unresponsive" trigger writes message_type = outgoing plus
+ # private_note = false, because a private note is an outgoing message and would otherwise
+ # start the wait without the customer ever having been replied to.
+ describe 'the curated customer-unresponsive trigger' do
+ let(:automation_rule) do
+ create(:automation_rule, account: account, event_name: 'message_created', execution_delay: 60,
+ conditions: [
+ { 'attribute_key' => 'message_type', 'filter_operator' => 'equal_to',
+ 'values' => ['outgoing'], 'query_operator' => 'and' },
+ { 'attribute_key' => 'private_note', 'filter_operator' => 'equal_to',
+ 'values' => [false], 'query_operator' => nil }
+ ],
+ actions: [{ 'action_name' => 'add_label', 'action_params' => ['stale'] }])
+ end
+
+ before do
+ allow(AutomationRules::ConditionsFilterService).to receive(:new).and_call_original
+ account.enable_features!('delayed_automations')
+ automation_rule
+ end
+
+ it 'arms the wait on a real agent reply' do
+ reply = create(:message, account: account, conversation: conversation, message_type: :outgoing)
+ event = Events::Base.new('message_created', Time.zone.now, { message: reply })
+
+ expect { listener.message_created(event) }.to change(AutomationRulePendingExecution, :count).by(1)
+ expect(AutomationRulePendingExecution.last.automation_rule).to eq(automation_rule)
+ end
+
+ it 'does not arm the wait on a private note' do
+ note = create(:message, account: account, conversation: conversation, message_type: :outgoing, private: true)
+ event = Events::Base.new('message_created', Time.zone.now, { message: note })
+
+ expect { listener.message_created(event) }.not_to change(AutomationRulePendingExecution, :count)
+ end
+ end
end
diff --git a/spec/models/account_spec.rb b/spec/models/account_spec.rb
index cbad87363..ed1760884 100644
--- a/spec/models/account_spec.rb
+++ b/spec/models/account_spec.rb
@@ -112,6 +112,34 @@ RSpec.describe Account do
end
end
+ describe 'resuming delayed automations' do
+ let(:account) { create(:account) }
+
+ it 'reschedules the overdue backlog in the same transaction that re-enables the flag' do
+ row = create(:automation_rule_pending_execution, account: account, due_at: 5.days.ago)
+
+ ActiveRecord::Base.transaction(requires_new: true) do
+ account.enable_features!('delayed_automations')
+ # Still uncommitted: no sweep can see the flag yet, and the backlog is already re-clocked,
+ # so there is no window where the account is sweepable on a stale due_at.
+ expect(row.reload.due_at).to be_within(5.seconds).of(Time.current)
+ raise ActiveRecord::Rollback
+ end
+
+ expect(account.reload.feature_delayed_automations?).to be(false)
+ expect(row.reload.due_at).to be_within(5.seconds).of(5.days.ago)
+ end
+
+ it 'leaves the backlog alone when the flag is turned off' do
+ account.enable_features!('delayed_automations')
+ row = create(:automation_rule_pending_execution, account: account, due_at: 5.days.ago)
+
+ account.disable_features!('delayed_automations')
+
+ expect(row.reload.due_at).to be_within(5.seconds).of(5.days.ago)
+ end
+ end
+
describe 'feature flag columns' do
let(:account) { described_class.new(name: 'Test Account') }
@@ -122,7 +150,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)
diff --git a/spec/models/automation_rule_pending_execution_spec.rb b/spec/models/automation_rule_pending_execution_spec.rb
new file mode 100644
index 000000000..830796edc
--- /dev/null
+++ b/spec/models/automation_rule_pending_execution_spec.rb
@@ -0,0 +1,352 @@
+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 '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 arm a status episode on a conversation that predates status_changed_at' do
+ conversation.status_changed_at = nil
+
+ expect { described_class.schedule(rule: rule, conversation: conversation) }.not_to change(described_class, :count)
+ 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 'ignores a customer reply that landed while the arming job was still queued' do
+ create(:message, conversation: conversation, account: account, message_type: :incoming)
+ agent_reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
+ # MESSAGE_CREATED dispatches asynchronously, so the customer can answer before the row is armed.
+ late_reply = create(:message, conversation: conversation, account: account, message_type: :incoming)
+
+ described_class.schedule(rule: rule, conversation: conversation, message: agent_reply)
+
+ row = described_class.last
+ expect(row.episode_key).not_to eq("reply_chase:#{late_reply.id}")
+ # The episode is already over, so the fire-time re-check cancels the row instead of chasing.
+ expect(row.episode_current?).to be(false)
+ 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 're-arms an episode skipped for a reason other than conditions, so later messages are not suppressed' do
+ reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
+ described_class.schedule(rule: rule, conversation: conversation, message: reply)
+ # The sweep was down long enough for the row to age out; a disabled rule leaves the same trail.
+ described_class.last.update!(status: :skipped, skip_reason: 'expired')
+
+ newer_reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
+ described_class.schedule(rule: rule, conversation: conversation, message: newer_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(newer_reply.id)
+ 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
+
+ it 'is true for a factory-built row anchored on a message' do
+ reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
+ row = create(:automation_rule_pending_execution, account: account, conversation: conversation, message: reply)
+
+ expect(row.episode_current?).to be(true)
+ 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
+
+ it 'hands a stale processing row back to pending instead of leaving it to expire on the next sweep' do
+ stale = travel_to(5.days.ago) do
+ create(:automation_rule_pending_execution, account: account, conversation: conversation, status: :processing)
+ end
+
+ described_class.reschedule_paused(account)
+
+ expect(stale.reload).to be_pending
+ expect(stale.due_at).to be_within(5.seconds).of(Time.current)
+ end
+
+ it 'leaves a live worker holding its own row' do
+ claimed = create(:automation_rule_pending_execution, account: account, conversation: conversation,
+ status: :processing, due_at: 5.days.ago)
+
+ described_class.reschedule_paused(account)
+
+ expect(claimed.reload).to be_processing
+ expect(claimed.due_at).to be_within(5.seconds).of(5.days.ago)
+ end
+ end
+end
diff --git a/spec/models/automation_rule_spec.rb b/spec/models/automation_rule_spec.rb
index 7ae73496c..797f392aa 100644
--- a/spec/models/automation_rule_spec.rb
+++ b/spec/models/automation_rule_spec.rb
@@ -137,4 +137,132 @@ 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 armed rows when the rule is deactivated, so reactivating cannot resurrect them' do
+ rule.update!(active: false)
+ expect(rule.pending_executions.armed).to be_empty
+
+ rule.update!(active: true)
+ expect(rule.pending_executions.armed).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 'leaves an executing row alone because its actions are already in flight' do
+ rule.pending_executions.first.update!(status: :executing)
+ rule.update!(actions: [{ 'action_name' => 'add_label', 'action_params' => ['urgent'] }])
+ expect(rule.pending_executions.executing.count).to eq(1)
+ 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
diff --git a/spec/models/conversation_spec.rb b/spec/models/conversation_spec.rb
index 15e3ebc10..29e00c204 100644
--- a/spec/models/conversation_spec.rb
+++ b/spec/models/conversation_spec.rb
@@ -1257,4 +1257,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