From 30ee4a61fc164a62621376afd5243629c77d2a87 Mon Sep 17 00:00:00 2001
From: Tanmay Deep Sharma <tanmaydeepsharma21@gmail.com>
Date: Fri, 10 Jul 2026 10:00:07 +0530
Subject: [PATCH] feat(automations): add delayed execution for automation rules

---
 .../accounts/automation_rules_controller.rb   |  17 +-
 app/javascript/dashboard/featureFlags.js      |   1 +
 .../dashboard/helper/automationHelper.js      |   6 +
 .../dashboard/i18n/locale/en/automation.json  |  16 +-
 .../settings/automation/AddAutomationRule.vue |   1 +
 .../automation/AutomationRuleForm.vue         | 100 +++++
 .../settings/automation/AutomationRuleRow.vue |  11 +
 .../dashboard/settings/automation/Index.vue   |  14 +
 .../process_pending_execution_job.rb          |  53 +++
 .../trigger_pending_executions_job.rb         |  42 ++
 app/jobs/trigger_scheduled_items_job.rb       |   3 +
 app/listeners/automation_rule_listener.rb     |  17 +-
 app/models/account.rb                         |   1 +
 app/models/automation_rule.rb                 |  35 +-
 .../automation_rule_pending_execution.rb      | 116 +++++
 app/models/conversation.rb                    |   7 +
 .../partials/_automation_rule.json.jbuilder   |   1 +
 config/features.yml                           |   5 +
 config/installation_config.yml                |   7 +
 ...add_execution_delay_to_automation_rules.rb |   5 +
 ..._add_status_changed_at_to_conversations.rb |   5 +
 ...eate_automation_rule_pending_executions.rb |  21 +
 db/schema.rb                                  |  22 +-
 ...delayed-automations-implementation-plan.md | 332 +++++++++++++++
 docs/delayed-automations-overview.md          | 149 +++++++
 docs/delayed-automations.md                   | 335 +++++++++++++++
 ...e-based-automations-phasing-and-rollout.md | 395 ++++++++++++++++++
 .../automation_rules_controller_spec.rb       |  57 +++
 .../automation_rule_pending_executions.rb     |  10 +
 .../process_pending_execution_job_spec.rb     | 119 ++++++
 .../trigger_pending_executions_job_spec.rb    |  52 +++
 .../automation_rule_listener_spec.rb          |  33 ++
 .../automation_rule_pending_execution_spec.rb | 169 ++++++++
 spec/models/automation_rule_spec.rb           |  38 ++
 spec/models/conversation_spec.rb              |  24 ++
 35 files changed, 2204 insertions(+), 15 deletions(-)
 create mode 100644 app/jobs/automation_rules/process_pending_execution_job.rb
 create mode 100644 app/jobs/automation_rules/trigger_pending_executions_job.rb
 create mode 100644 app/models/automation_rule_pending_execution.rb
 create mode 100644 db/migrate/20260709060000_add_execution_delay_to_automation_rules.rb
 create mode 100644 db/migrate/20260709060100_add_status_changed_at_to_conversations.rb
 create mode 100644 db/migrate/20260709060200_create_automation_rule_pending_executions.rb
 create mode 100644 docs/delayed-automations-implementation-plan.md
 create mode 100644 docs/delayed-automations-overview.md
 create mode 100644 docs/delayed-automations.md
 create mode 100644 docs/time-based-automations-phasing-and-rollout.md
 create mode 100644 spec/factories/automation_rule_pending_executions.rb
 create mode 100644 spec/jobs/automation_rules/process_pending_execution_job_spec.rb
 create mode 100644 spec/jobs/automation_rules/trigger_pending_executions_job_spec.rb
 create mode 100644 spec/models/automation_rule_pending_execution_spec.rb

diff --git a/app/controllers/api/v1/accounts/automation_rules_controller.rb b/app/controllers/api/v1/accounts/automation_rules_controller.rb
index 0840d0eea..4c1b2b373 100644
--- a/app/controllers/api/v1/accounts/automation_rules_controller.rb
+++ b/app/controllers/api/v1/accounts/automation_rules_controller.rb
@@ -3,6 +3,7 @@ class Api::V1::Accounts::AutomationRulesController < Api::V1::Accounts::BaseCont
 
   before_action :check_authorization
   before_action :fetch_automation_rule, only: [:show, :update, :destroy, :clone]
+  before_action :ensure_execution_delay_allowed, only: [:create, :update]
 
   def index
     @automation_rules = Current.account.automation_rules
@@ -55,13 +56,27 @@ class Api::V1::Accounts::AutomationRulesController < Api::V1::Accounts::BaseCont
   private
 
   def automation_rules_permit
+    permitted_attributes = [:name, :description, :event_name, :active]
+    permitted_attributes << :execution_delay if delayed_automations_enabled?
+
     params.permit(
-      :name, :description, :event_name, :active,
+      *permitted_attributes,
       conditions: [:attribute_key, :filter_operator, :query_operator, :custom_attribute_type, { values: [] }],
       actions: [:action_name, { action_params: [] }]
     )
   end
 
+  def ensure_execution_delay_allowed
+    return if delayed_automations_enabled?
+    return if params[:execution_delay].blank?
+
+    render json: { error: 'Delayed automations are not enabled for this account.' }, status: :unprocessable_entity
+  end
+
+  def delayed_automations_enabled?
+    Current.account.feature_enabled?('delayed_automations')
+  end
+
   def fetch_automation_rule
     @automation_rule = Current.account.automation_rules.find_by(id: params[:id])
   end
diff --git a/app/javascript/dashboard/featureFlags.js b/app/javascript/dashboard/featureFlags.js
index b9f59eba0..a14b1e3a2 100644
--- a/app/javascript/dashboard/featureFlags.js
+++ b/app/javascript/dashboard/featureFlags.js
@@ -10,6 +10,7 @@ export const FEATURE_FLAGS = {
   CANNED_RESPONSES: 'canned_responses',
   CRM: 'crm',
   CUSTOM_ATTRIBUTES: 'custom_attributes',
+  DELAYED_AUTOMATIONS: 'delayed_automations',
   INBOX_MANAGEMENT: 'inbox_management',
   INTEGRATIONS: 'integrations',
   LABELS: 'labels',
diff --git a/app/javascript/dashboard/helper/automationHelper.js b/app/javascript/dashboard/helper/automationHelper.js
index 8aed8dcda..c66dc204c 100644
--- a/app/javascript/dashboard/helper/automationHelper.js
+++ b/app/javascript/dashboard/helper/automationHelper.js
@@ -208,6 +208,12 @@ export const generateAutomationPayload = payload => {
   return automation;
 };
 
+export const formatDelay = minutes => {
+  if (minutes % 1440 === 0) return `${minutes / 1440}d`;
+  if (minutes % 60 === 0) return `${minutes / 60}h`;
+  return `${minutes}m`;
+};
+
 export const isCustomAttribute = (attrs, key) => {
   return attrs.find(attr => attr.key === key);
 };
diff --git a/app/javascript/dashboard/i18n/locale/en/automation.json b/app/javascript/dashboard/i18n/locale/en/automation.json
index 2c4852dc8..805674eba 100644
--- a/app/javascript/dashboard/i18n/locale/en/automation.json
+++ b/app/javascript/dashboard/i18n/locale/en/automation.json
@@ -28,6 +28,18 @@
           "PLACEHOLDER": "Please select one",
           "ERROR": "Event is required"
         },
+        "EXECUTE": {
+          "LABEL": "Execute",
+          "IMMEDIATELY": "Immediately",
+          "AFTER_DELAY": "After a delay of",
+          "UNITS": {
+            "MINUTES": "Minutes",
+            "HOURS": "Hours",
+            "DAYS": "Days"
+          },
+          "ERROR": "Delay must be between 10 minutes and 30 days",
+          "HELP_TEXT": "Delayed rules run only if the conditions still hold when the delay ends, and apply to conversations with activity after the rule is created."
+        },
         "CONDITIONS": {
           "LABEL": "Conditions"
         },
@@ -49,7 +61,9 @@
         "CREATED_ON": "Created on",
         "ACTIONS": "Actions"
       },
-      "404": "No automation rules found"
+      "404": "No automation rules found",
+      "DELAY_BADGE": "Runs after {delay}",
+      "DELAY_DISABLED_BANNER": "Delayed execution is disabled for this account. Rules with a delay will not run until it is enabled again."
     },
     "DELETE": {
       "TITLE": "Delete Automation Rule",
diff --git a/app/javascript/dashboard/routes/dashboard/settings/automation/AddAutomationRule.vue b/app/javascript/dashboard/routes/dashboard/settings/automation/AddAutomationRule.vue
index 472dfe08e..5731b2d7f 100644
--- a/app/javascript/dashboard/routes/dashboard/settings/automation/AddAutomationRule.vue
+++ b/app/javascript/dashboard/routes/dashboard/settings/automation/AddAutomationRule.vue
@@ -10,6 +10,7 @@ const START_VALUE = {
   name: null,
   description: null,
   event_name: 'conversation_created',
+  execution_delay: null,
   conditions: [
     {
       attribute_key: 'status',
diff --git a/app/javascript/dashboard/routes/dashboard/settings/automation/AutomationRuleForm.vue b/app/javascript/dashboard/routes/dashboard/settings/automation/AutomationRuleForm.vue
index d78c40749..3a6cc8a8b 100644
--- a/app/javascript/dashboard/routes/dashboard/settings/automation/AutomationRuleForm.vue
+++ b/app/javascript/dashboard/routes/dashboard/settings/automation/AutomationRuleForm.vue
@@ -14,6 +14,7 @@ import {
   showActionInput,
 } from 'dashboard/helper/automationHelper';
 import { validateAutomation } from 'dashboard/helper/validations';
+import { FEATURE_FLAGS } from 'dashboard/featureFlags';
 import { AUTOMATION_RULE_EVENTS, AUTOMATION_ACTION_TYPES } from './constants';
 
 const props = defineProps({
@@ -71,6 +72,14 @@ const INPUT_TYPE_MAP = {
   date: 'date',
 };
 
+const DELAY_UNITS = [
+  { key: 'MINUTES', factor: 1 },
+  { key: 'HOURS', factor: 60 },
+  { key: 'DAYS', factor: 1440 },
+];
+const MIN_DELAY_MINUTES = 10;
+const MAX_DELAY_MINUTES = 43200; // 30 days
+
 const { t } = useI18n();
 const { isCloudFeatureEnabled } = useAccount();
 const { operators } = useOperators();
@@ -184,6 +193,50 @@ const hasActionErrors = computed(() =>
   Object.keys(errors.value).some(key => key.startsWith('action_'))
 );
 
+const allowsDelayedExecution = computed(() =>
+  isCloudFeatureEnabled(FEATURE_FLAGS.DELAYED_AUTOMATIONS)
+);
+
+const executeMode = ref('immediate');
+const delayValue = ref(4);
+const delayUnit = ref('HOURS');
+
+const delayInMinutes = computed(() => {
+  const factor =
+    DELAY_UNITS.find(unit => unit.key === delayUnit.value)?.factor || 1;
+  return Math.round(Number(delayValue.value) * factor);
+});
+
+const executionDelayInvalid = computed(
+  () =>
+    executeMode.value === 'delayed' &&
+    (!Number.isFinite(delayInMinutes.value) ||
+      delayInMinutes.value < MIN_DELAY_MINUTES ||
+      delayInMinutes.value > MAX_DELAY_MINUTES)
+);
+
+// Hydrate the delay controls from the automation, using the largest clean unit.
+const syncDelayFromAutomation = () => {
+  const delay = automation.value?.execution_delay;
+  executeMode.value = delay ? 'delayed' : 'immediate';
+  if (!delay) {
+    delayValue.value = 4;
+    delayUnit.value = 'HOURS';
+    return;
+  }
+  const unit =
+    [...DELAY_UNITS].reverse().find(u => delay % u.factor === 0) ||
+    DELAY_UNITS[0];
+  delayUnit.value = unit.key;
+  delayValue.value = delay / unit.factor;
+};
+
+watch([executeMode, delayInMinutes], () => {
+  if (!automation.value || !allowsDelayedExecution.value) return;
+  automation.value.execution_delay =
+    executeMode.value === 'delayed' ? delayInMinutes.value : null;
+});
+
 watch(
   () => automation.value,
   () => {
@@ -218,6 +271,7 @@ const syncCustomAttributeTypes = () => {
 
 const open = () => {
   resetValidation();
+  syncDelayFromAutomation();
   dialogRef.value?.open();
 };
 
@@ -230,8 +284,13 @@ const emitSaveAutomation = () => {
   syncCustomAttributeTypes();
   const conditionsValid = isConditionsValid();
   errors.value = validateAutomation(automation.value);
+  if (allowsDelayedExecution.value && executionDelayInvalid.value) {
+    errors.value.execution_delay = true;
+  }
   if (Object.keys(errors.value).length === 0 && conditionsValid) {
     const payload = generateAutomationPayload(automation.value);
+    // The API rejects the param when the feature is off; existing values are kept server-side.
+    if (!allowsDelayedExecution.value) delete payload.execution_delay;
     emit('save', payload, props.mode);
   }
 };
@@ -293,6 +352,47 @@ defineExpose({ open, close });
           {{ $t('AUTOMATION.FORM.RESET_MESSAGE') }}
         </p>
       </div>
+      <div v-if="allowsDelayedExecution" class="mb-6">
+        <label :class="{ error: errors.execution_delay }">
+          {{ $t('AUTOMATION.ADD.FORM.EXECUTE.LABEL') }}
+        </label>
+        <div class="flex flex-wrap items-center gap-4 mt-1">
+          <label class="flex items-center gap-1.5 text-sm cursor-pointer">
+            <input v-model="executeMode" type="radio" value="immediate" />
+            {{ $t('AUTOMATION.ADD.FORM.EXECUTE.IMMEDIATELY') }}
+          </label>
+          <label class="flex items-center gap-1.5 text-sm cursor-pointer">
+            <input v-model="executeMode" type="radio" value="delayed" />
+            {{ $t('AUTOMATION.ADD.FORM.EXECUTE.AFTER_DELAY') }}
+          </label>
+          <template v-if="executeMode === 'delayed'">
+            <input
+              v-model.number="delayValue"
+              type="number"
+              min="1"
+              class="!m-0 !w-24"
+            />
+            <select v-model="delayUnit" class="!m-0 !w-32">
+              <option
+                v-for="unit in DELAY_UNITS"
+                :key="unit.key"
+                :value="unit.key"
+              >
+                {{ $t(`AUTOMATION.ADD.FORM.EXECUTE.UNITS.${unit.key}`) }}
+              </option>
+            </select>
+          </template>
+        </div>
+        <span v-if="errors.execution_delay" class="text-xs text-n-ruby-9">
+          {{ $t('AUTOMATION.ADD.FORM.EXECUTE.ERROR') }}
+        </span>
+        <p
+          v-else-if="executeMode === 'delayed'"
+          class="text-xs text-n-slate-11 pt-1 mb-0"
+        >
+          {{ $t('AUTOMATION.ADD.FORM.EXECUTE.HELP_TEXT') }}
+        </p>
+      </div>
       <!-- Conditions Start -->
       <section class="mb-5">
         <label>
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 @@
 <script setup>
 import { computed } from 'vue';
 import { messageStamp } from 'shared/helpers/timeHelper';
+import { formatDelay } from 'dashboard/helper/automationHelper';
 import Button from 'dashboard/components-next/button/Button.vue';
 import ToggleSwitch from 'dashboard/components-next/switch/Switch.vue';
 import { BaseTableRow, BaseTableCell } from 'dashboard/components-next/table';
@@ -43,6 +44,16 @@ const automationActive = computed({
           <span class="text-body-main text-n-slate-12 truncate">
             {{ automation.name }}
           </span>
+          <span
+            v-if="automation.execution_delay"
+            class="text-xs px-1.5 py-0.5 rounded-md bg-n-alpha-2 text-n-slate-11 whitespace-nowrap flex-shrink-0"
+          >
+            {{
+              $t('AUTOMATION.LIST.DELAY_BADGE', {
+                delay: formatDelay(automation.execution_delay),
+              })
+            }}
+          </span>
           <div class="w-px h-3 rounded-lg bg-n-weak flex-shrink-0" />
           <span class="text-body-main text-n-slate-11 truncate">
             {{ automation.description }}
diff --git a/app/javascript/dashboard/routes/dashboard/settings/automation/Index.vue b/app/javascript/dashboard/routes/dashboard/settings/automation/Index.vue
index 32139ccd5..9ad6efd78 100644
--- a/app/javascript/dashboard/routes/dashboard/settings/automation/Index.vue
+++ b/app/javascript/dashboard/routes/dashboard/settings/automation/Index.vue
@@ -52,6 +52,14 @@ const isSLAEnabled = computed(() =>
   getters['accounts/isFeatureEnabledonAccount'].value(accountId.value, 'sla')
 );
 
+const showDelayDisabledBanner = computed(
+  () =>
+    !getters['accounts/isFeatureEnabledonAccount'].value(
+      accountId.value,
+      'delayed_automations'
+    ) && records.value.some(automation => automation.execution_delay)
+);
+
 onMounted(() => {
   store.dispatch('inboxes/get');
   store.dispatch('agents/get');
@@ -212,6 +220,12 @@ const tableHeaders = computed(() => {
       </BaseSettingsHeader>
     </template>
     <template #body>
+      <div
+        v-if="showDelayDisabledBanner"
+        class="px-4 py-3 mb-4 text-sm rounded-lg bg-n-amber-3 text-n-amber-12"
+      >
+        {{ $t('AUTOMATION.LIST.DELAY_DISABLED_BANNER') }}
+      </div>
       <BaseTable
         :headers="tableHeaders"
         :items="filteredRecords"
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..564c0e3dc
--- /dev/null
+++ b/app/jobs/automation_rules/process_pending_execution_job.rb
@@ -0,0 +1,53 @@
+class AutomationRules::ProcessPendingExecutionJob < ApplicationJob
+  queue_as :medium
+
+  discard_on ActiveRecord::RecordNotFound
+
+  def perform(pending_execution)
+    # The sweep also checks this, but rows already enqueued when the switch flips must not
+    # keep firing; they return to pending and replay or expire via the due window.
+    return pending_execution.update!(status: :pending) if delayed_automations_disabled?
+
+    skip_reason = skip_reason_for(pending_execution)
+    return pending_execution.update!(status: :skipped, skip_reason: skip_reason) if skip_reason
+
+    execute(pending_execution)
+  rescue StandardError => e
+    # Row stays `processing`; the sweep's stale reclaim retries it later.
+    ChatwootExceptionTracker.new(e, account: pending_execution.account).capture_exception
+  end
+
+  private
+
+  def skip_reason_for(pending_execution)
+    rule = pending_execution.automation_rule
+    return 'rule_inactive' if rule.nil? || !rule.active?
+    return 'flag_disabled' unless pending_execution.account.feature_enabled?('delayed_automations')
+    return 'conversation_gone' if pending_execution.conversation.nil?
+    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
+
+  def execute(pending_execution)
+    AutomationRules::ActionService.new(
+      pending_execution.automation_rule,
+      pending_execution.account,
+      pending_execution.conversation
+    ).perform
+    pending_execution.update!(status: :executed)
+  end
+
+  def delayed_automations_disabled?
+    GlobalConfig.get('DISABLE_DELAYED_AUTOMATIONS')['DISABLE_DELAYED_AUTOMATIONS']
+  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..37a69e3a9
--- /dev/null
+++ b/app/jobs/automation_rules/trigger_pending_executions_job.rb
@@ -0,0 +1,42 @@
+class AutomationRules::TriggerPendingExecutionsJob < ApplicationJob
+  queue_as :scheduled_jobs
+
+  DEFAULT_SWEEP_LIMIT = 1000
+
+  def perform
+    return if delayed_automations_disabled?
+
+    started_at = Time.current
+    reclaimed = AutomationRulePendingExecution.reclaim_stale!
+    expired = AutomationRulePendingExecution.expire_overdue!
+    due_count = AutomationRulePendingExecution.due.count
+
+    enqueued = 0
+    AutomationRulePendingExecution.due.limit(sweep_limit).find_each(batch_size: 100) do |pending_execution|
+      next unless pending_execution.mark_processing!
+
+      AutomationRules::ProcessPendingExecutionJob.perform_later(pending_execution)
+      enqueued += 1
+    end
+
+    log_summary(due: due_count, enqueued: enqueued, expired: expired, reclaimed: reclaimed, started_at: started_at)
+  end
+
+  private
+
+  def delayed_automations_disabled?
+    GlobalConfig.get('DISABLE_DELAYED_AUTOMATIONS')['DISABLE_DELAYED_AUTOMATIONS']
+  end
+
+  def sweep_limit
+    (InstallationConfig.find_by(name: 'AUTOMATION_PENDING_EXECUTIONS_SWEEP_LIMIT')&.value || DEFAULT_SWEEP_LIMIT).to_i
+  end
+
+  def log_summary(due:, enqueued:, expired:, reclaimed:, started_at:)
+    summary = {
+      event: 'completed', due: due, enqueued: enqueued, capped: due > enqueued,
+      expired: expired, reclaimed: reclaimed, duration_ms: ((Time.current - started_at) * 1000).round
+    }
+    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..33f46c6f6 100644
--- a/app/jobs/trigger_scheduled_items_job.rb
+++ b/app/jobs/trigger_scheduled_items_job.rb
@@ -17,6 +17,9 @@ class TriggerScheduledItemsJob < ApplicationJob
     # Job to auto-resolve conversations
     Account::ConversationsResolutionSchedulerJob.perform_later
 
+    # Job to fire due delayed automation rules
+    AutomationRules::TriggerPendingExecutionsJob.perform_later
+
     # Job to sync whatsapp templates
     Channels::Whatsapp::TemplatesSyncSchedulerJob.perform_later
   end
diff --git a/app/listeners/automation_rule_listener.rb b/app/listeners/automation_rule_listener.rb
index 0515d6952..bb23b7983 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
 
@@ -52,7 +52,20 @@ class AutomationRuleListener < BaseListener
 
     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
+
+  # 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 4295b162e..a781d48ac 100644
--- a/app/models/account.rb
+++ b/app/models/account.rb
@@ -65,6 +65,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
diff --git a/app/models/automation_rule.rb b/app/models/automation_rule.rb
index 9a437bac9..9b7d578b7 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,10 @@ class AutomationRule < ApplicationRecord
   include Rails.application.routes.url_helpers
   include Reauthorizable
 
+  EXECUTION_DELAY_RANGE = (10..43_200) # minutes: 10 min to 30 days
+
   belongs_to :account
+  has_many :pending_executions, class_name: 'AutomationRulePendingExecution', dependent: :delete_all
   has_many_attached :files
 
   validate :json_conditions_format
@@ -29,6 +33,8 @@ 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
 
   after_update_commit :reauthorized!, if: -> { saved_change_to_conditions? }
 
@@ -95,6 +101,15 @@ 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
+
   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..3f4a744a2
--- /dev/null
+++ b/app/models/automation_rule_pending_execution.rb
@@ -0,0 +1,116 @@
+# == 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_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
+  # Claimed rows abandoned by a crashed worker return to the sweep after this.
+  STALE_PROCESSING_TIMEOUT = 15.minutes
+
+  belongs_to :automation_rule
+  belongs_to :conversation
+  belongs_to :account
+  belongs_to :message, optional: true
+
+  enum status: { pending: 0, processing: 1, executed: 2, skipped: 3 }
+
+  scope :due, -> { pending.where(due_at: DUE_WINDOW.ago..Time.current) }
+
+  def self.schedule(rule:, conversation:, message: nil)
+    attributes = {
+      automation_rule_id: rule.id,
+      conversation_id: conversation.id,
+      account_id: conversation.account_id,
+      message_id: message&.id,
+      episode_key: episode_key_for(conversation, message),
+      due_at: rule.execution_delay.minutes.from_now,
+      created_at: Time.current,
+      updated_at: Time.current
+    }
+
+    # Values are server-computed and the unique episode index is the real guard, so the
+    # validation-skipping conflict-handling writes are safe here (repo bulk-write convention).
+    # rubocop:disable Rails/SkipsModelValidations
+    if message && !message.incoming?
+      # Reply-chase: the clock tracks the latest agent reply. Status is excluded from the
+      # update list so an executed/skipped episode is never re-armed (run-once per episode).
+      upsert(attributes, unique_by: :uniq_automation_pending_execution_episode,
+                         on_duplicate: Arel.sql('due_at = excluded.due_at, message_id = excluded.message_id, updated_at = excluded.updated_at'))
+    else
+      # Status / awaiting-agent episodes: first event wins, the clock is not reset.
+      insert(attributes, unique_by: :uniq_automation_pending_execution_episode)
+    end
+    # rubocop:enable Rails/SkipsModelValidations
+  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.
+      "status:#{(conversation.status_changed_at.presence || conversation.created_at).to_f}"
+    elsif message.incoming?
+      # waiting_since is cleared on agent/bot reply, so a reply invalidates this episode.
+      "awaiting_agent:#{conversation.waiting_since.to_i}"
+    else
+      # A new customer message changes the max incoming id, invalidating this episode.
+      "reply_chase:#{conversation.messages.incoming.maximum(:id) || 0}"
+    end
+  end
+
+  # Bulk state flips on exceptional sets; batched so no statement outlives the global 14s
+  # statement_timeout (repo bulk-write convention, see Agents::DestroyJob).
+  # rubocop:disable Rails/SkipsModelValidations
+  def self.expire_overdue!
+    expired_count = 0
+    pending.where(due_at: ...DUE_WINDOW.ago).in_batches(of: 1000) do |batch|
+      expired_count += batch.update_all(status: statuses[:skipped], skip_reason: 'expired', updated_at: Time.current)
+    end
+    expired_count
+  end
+
+  def self.reclaim_stale!
+    reclaimed_count = 0
+    processing.where(updated_at: ...STALE_PROCESSING_TIMEOUT.ago).in_batches(of: 1000) do |batch|
+      reclaimed_count += batch.update_all(status: statuses[:pending], updated_at: Time.current)
+    end
+    reclaimed_count
+  end
+  # rubocop:enable Rails/SkipsModelValidations
+
+  # Locked claim so a row re-selected by an overlapping sweep can't double-fire
+  # (Campaign#mark_processing! pattern).
+  def mark_processing!
+    with_lock do
+      next false unless pending?
+
+      processing!
+      true
+    end
+  end
+
+  def episode_current?
+    self.class.episode_key_for(conversation, message) == episode_key
+  end
+end
diff --git a/app/models/conversation.rb b/app/models/conversation.rb
index 1e9c38c41..2f2482676 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
@@ -124,8 +125,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
 
@@ -272,6 +275,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 fe2ef2122..80a3afdf1 100644
--- a/config/features.yml
+++ b/config/features.yml
@@ -249,3 +249,8 @@
   display_name: Advanced Assignment
   enabled: false
   premium: true
+- name: delayed_automations
+  display_name: Delayed Automations
+  enabled: false
+  chatwoot_internal: true
+  column: feature_flags_ext_1
diff --git a/config/installation_config.yml b/config/installation_config.yml
index cfe64767f..0b00482b6 100644
--- a/config/installation_config.yml
+++ b/config/installation_config.yml
@@ -567,3 +567,10 @@
   value: 'https://us.cloud.langfuse.com'
   locked: false
 ## ---- End of LLM Observability ---- ##
+
+- name: DISABLE_DELAYED_AUTOMATIONS
+  display_title: 'Disable delayed automations'
+  description: 'Emergency stop for delayed automation rules: halts the pending-execution sweep and per-row execution within one tick'
+  value: false
+  locked: false
+  type: boolean
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/schema.rb b/db/schema.rb
index 5e506a475..45e09a5b3 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_06_215758) do
+ActiveRecord::Schema[7.1].define(version: 2026_07_09_060200) do
   # These extensions should be enabled to support this database
   enable_extension "pg_stat_statements"
   enable_extension "pg_trgm"
@@ -250,6 +250,24 @@ ActiveRecord::Schema[7.1].define(version: 2026_07_06_215758) 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"
+  end
+
   create_table "automation_rules", force: :cascade do |t|
     t.bigint "account_id", null: false
     t.string "name", null: false
@@ -260,6 +278,7 @@ ActiveRecord::Schema[7.1].define(version: 2026_07_06_215758) 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
 
@@ -728,6 +747,7 @@ ActiveRecord::Schema[7.1].define(version: 2026_07_06_215758) 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/docs/delayed-automations-implementation-plan.md b/docs/delayed-automations-implementation-plan.md
new file mode 100644
index 000000000..0e0023108
--- /dev/null
+++ b/docs/delayed-automations-implementation-plan.md
@@ -0,0 +1,332 @@
+# Delayed Automations — Implementation Plan (final)
+
+The build source of truth for CW-7513, target **Jul 15**. Companions:
+`docs/delayed-automations.md` (approved design — the what/why) ·
+`docs/time-based-automations-phasing-and-rollout.md` (product phases + rollout stages —
+this plan implements its Phase 0/1).
+
+Supersedes earlier drafts: feature-flag storage was extended on develop
+(`873d16f54c`, #14947 — second bitset column `accounts.feature_flags_ext_1`, 0/63 used), so
+the flag is a plain features.yml append; no repurpose migration.
+
+---
+
+## Locked decisions
+
+1. **Delay bounds**: `execution_delay` in minutes, `nil` or `10..43_200` (10 min – 30 days).
+2. **Feature flag `delayed_automations`** — appended at the end of `config/features.yml`
+   with `column: feature_flags_ext_1` (mandatory: the legacy `feature_flags` column is
+   63/63 full; the header documents this), `enabled: false`, `chatwoot_internal: true`.
+   No DB migration for the flag itself — the ext column already exists; ConfigLoader
+   reconciles the new name (with its `column` metadata) into
+   `ACCOUNT_LEVEL_FEATURE_DEFAULTS` on `db:migrate`. The flag is the **per-account stop**,
+   enforced at three layers: params (Phase 5), arming (Phase 3), fire time (Phase 4).
+3. **Instance kill switch `DISABLE_DELAYED_AUTOMATIONS`** — new entry in
+   `config/installation_config.yml` (`value: false`, `type: boolean`, `locked: false`),
+   checked in **both** sweep and per-row jobs (Phase 4). Flipping it takes effect within one
+   tick, no deploy (`InstallationConfig` `after_commit :clear_cache`).
+4. **Episode anchors, split by what armed the rule** (this is the only genuinely new logic;
+   see design doc §5.3):
+   - *Conversation events*: `"status:{status_changed_at || created_at}"` — clock never
+     resets while the conversation stays in the state; leaving + re-entering = new episode.
+   - *`message_created`, triggering message outgoing* (Story 2 — customer went quiet):
+     `"reply_chase:{max incoming message id}"`; upsert **updates `due_at` + `message_id`**
+     so the clock tracks the latest agent reply; a customer reply changes the key →
+     fire-time skip.
+   - *`message_created`, triggering message incoming* (agent went quiet — the
+     unassign-on-no-response ask from the CW-7513 comments): `"awaiting_agent:{waiting_since}"`;
+     conflict = no-op (clock counts from the first unanswered customer message). Chatwoot
+     clears `waiting_since` on agent/bot reply, so the recomputed key changes and the row
+     skips — without this anchor, an "unassign after 30 min silence" rule would unassign
+     agents who replied promptly.
+5. **`attribute_changed` conditions + delay are mutually exclusive** (model validation):
+   the re-check can't reconstruct `changed_attributes` at fire time.
+6. **Anchor message persisted** (`message_id` column): `ConditionsFilterService` scopes
+   message conditions to the triggering message, so the fire-time re-check needs it.
+7. **Retention**: `executed`/`skipped` rows kept 30 days (they are the audit trail and the
+   incident blast-radius query), pruned by the daily scheduled job.
+8. **No fallback to immediate execution anywhere**: flag off / guard failure ⇒ skip. A
+   24h-delayed `send_message` silently becoming instant is worse than not firing.
+
+## Phase 1 — Schema & config
+
+### 1a. Migration: `add_execution_delay_to_automation_rules`
+```ruby
+add_column :automation_rules, :execution_delay, :integer  # minutes, NULL = immediate
+```
+
+### 1b. Migration: `add_status_changed_at_to_conversations`
+```ruby
+add_column :conversations, :status_changed_at, :datetime  # no default, no backfill
+```
+Plain nullable no-default add (the `sla_policy_id`/`assignee_agent_bot_id` precedent).
+Readers use `status_changed_at.presence || created_at` — stable across a delay window,
+which is all the episode key needs.
+
+### 1c. Migration: `create_automation_rule_pending_executions`
+```ruby
+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                      # anchor message for message_created rules
+  t.datetime   :due_at, null: false
+  t.string     :episode_key, null: false
+  t.integer    :status, null: false, default: 0 # pending: 0, processing: 1, executed: 2, skipped: 3
+  t.string     :skip_reason                     # rule_inactive | flag_disabled | conversation_gone |
+                                                # episode_moved | conditions_changed | expired
+  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'
+```
+`processing` is the claim state (double-fire guard); `skip_reason` answers "why didn't my
+rule fire" for support and drives incident blast-radius queries.
+
+### 1d. Config files (no migration)
+- `config/features.yml`: append `delayed_automations` at the end —
+  `column: feature_flags_ext_1`, `enabled: false`, `chatwoot_internal: true`. Never reorder;
+  the entry's position within its column is permanent once merged.
+- `config/installation_config.yml`: `DISABLE_DELAYED_AUTOMATIONS` (`value: false`,
+  `type: boolean`, `locked: false`).
+
+## Phase 2 — Models
+
+### `app/models/automation_rule_pending_execution.rb` (new, flat name per repo convention)
+- `belongs_to :automation_rule, :conversation, :account`; optional `belongs_to :message`.
+- `enum status: { pending: 0, processing: 1, executed: 2, skipped: 3 }`.
+- `scope :due, -> { pending.where(due_at: 3.days.ago..Time.current) }` — **bounded window**
+  (campaign/snooze-reopen precedent) so downtime backlog is finite.
+- `self.expire_overdue!` — pending rows with `due_at < 3.days.ago` → `skipped` /
+  `skip_reason: 'expired'`, returns count (batched updates — global 14s `statement_timeout`).
+- `self.reclaim_stale!` — `processing` rows older than 15 min → back to `pending`
+  (stale-claim recovery; mechanism per Captain's `SYNC_STALE_TIMEOUT`, which uses 2h —
+  15 min fits the 5-min sweep cadence).
+- `#mark_processing!` — compare-and-set from `pending` (the `Campaign#mark_processing!`
+  precedent); the per-row job is enqueued only after a successful claim.
+- `self.schedule(rule:, conversation:, message: nil)` — computes `episode_key` + `due_at =
+  Time.current + rule.execution_delay.minutes`, then by anchor type (locked decision #4):
+  - status episodes and incoming-anchored message episodes → `insert` with `unique_by`
+    (conflict = no-op, clock not reset);
+  - outgoing-anchored message episodes → `upsert` updating `due_at` + `message_id`.
+- `#episode_current?` — recomputes the episode key from live conversation state, compares.
+
+### `app/models/automation_rule.rb`
+- `has_many :pending_executions, class_name: 'AutomationRulePendingExecution',
+  dependent: :delete_all`.
+- Validation: `execution_delay` nil or integer in `10..43_200`.
+- Validation: no `execution_delay` together with an `attribute_changed` condition.
+- Update schema annotation.
+
+### `app/models/conversation.rb`
+- `before_save -> { self.status_changed_at = Time.current if status_changed? }` — covers
+  create (status always "changes" on create) and every transition; runs before the existing
+  `notify_status_change` / `dispatch_conversation_updated_event` callbacks so listeners see
+  the fresh value.
+- `has_many :automation_rule_pending_executions, dependent: :delete_all` (also on Account).
+- Verify `status_changed_at` needs no entry in the conversation-updated watched-attributes
+  list — we don't want an extra `conversation_updated` storm; a plain column write inside
+  the same save is fine.
+
+## Phase 3 — Write path (listener)
+
+### `app/listeners/automation_rule_listener.rb`
+Replace the two `ActionService.new(...).perform if conditions_match.present?` call sites
+(the `message_created` and `process_conversation_event` loops) with a shared private method:
+
+```ruby
+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
+```
+Flag off ⇒ no arming and **no immediate fallback** (locked decision #8). Everything upstream
+(loop guard `performed_by_automation?`, auto-reply skip, rule lookup,
+`ConditionsFilterService`) is untouched.
+
+## Phase 4 — Fire path (sweep)
+
+### `app/jobs/trigger_scheduled_items_job.rb`
+Add one line: `AutomationRules::TriggerPendingExecutionsJob.perform_later`. No change to
+`config/schedule.yml` (rides the existing `*/5` cron).
+
+### `app/jobs/automation_rules/trigger_pending_executions_job.rb` (new, queue `scheduled_jobs`)
+```ruby
+# instance kill switch — MUST use GlobalConfig.get (it applies the `type: boolean` cast;
+# get_value returns the raw cached value, and super-admin edits store the STRING 'false',
+# which is truthy — the ENABLE_*_CHANNEL_HUMAN_AGENT read pattern):
+return if GlobalConfig.get('DISABLE_DELAYED_AUTOMATIONS')['DISABLE_DELAYED_AUTOMATIONS']
+
+reclaimed = AutomationRulePendingExecution.reclaim_stale!
+expired   = AutomationRulePendingExecution.expire_overdue!
+AutomationRulePendingExecution.due.limit(sweep_limit).find_each(batch_size: 100) do |pending|
+  next unless pending.mark_processing!
+  AutomationRules::ProcessPendingExecutionJob.perform_later(pending)
+end
+# end-of-run summary, JSON so New Relic ingests fields without a parsing rule:
+# [AutomationRules::TriggerPendingExecutionsJob] {"event":"completed","due":N,"enqueued":N,
+#  "capped":bool,"expired":N,"reclaimed":N,"duration_ms":N}
+```
+`sweep_limit`: constant on the job (default 1000) with an InstallationConfig override
+(Captain `ScheduleSyncsJob` pattern — a no-deploy tuning knob). Overflow rows stay `pending`
+for the next tick; log `capped: true` + remaining count — no silent truncation.
+Retention pruning lives in `Internal::TriggerDailyScheduledItemsJob`: batched
+`where(status: %i[executed skipped]).where(updated_at: ...30.days.ago)` deletes.
+
+Queue reality (`config/sidekiq.yml` is strict-priority): `scheduled_jobs` ranks below `low`,
+so the sweep can be late — `due_at <= now` semantics already tolerate that; per-row jobs on
+`medium` preempt `default` and below — the cap is what protects the instance. `WebhookJob`
+is also on `medium`, so webhook-heavy delayed rules share that budget.
+
+### `app/jobs/automation_rules/process_pending_execution_job.rb` (new, queue `medium`,
+mirrors `Sla::ProcessAppliedSlaJob`)
+`discard_on ActiveRecord::RecordNotFound`. Re-checks the kill switch first (sweep-only
+checking would let a full sweep's already-enqueued rows keep firing after the flip): if set,
+revert the row to `pending` and return — it replays or expires via the window. Guard chain,
+each failure → `skipped!` with `skip_reason`:
+1. rule exists and `active?` → `rule_inactive`
+2. account flag still enabled → `flag_disabled` (the per-account stop)
+3. conversation exists → `conversation_gone`
+4. `pending.episode_current?` → `episode_moved`
+5. `ConditionsFilterService.new(rule, conversation, { message: pending.message }).perform` —
+   the re-check (anchor message passed so message-scoped conditions evaluate correctly;
+   `changed_attributes` intentionally absent, locked decision #5) → `conditions_changed`
+6. all pass → `AutomationRules::ActionService.new(rule, account, conversation).perform`,
+   then `executed!`
+
+Per-row rescue → `ChatwootExceptionTracker.new(e, account: account).capture_exception`; the
+row stays `processing` on unexpected errors (stale reclaim retries it), it is NOT marked
+skipped.
+
+Loop safety: `ActionService#initialize` already sets `Current.executed_by = rule` and resets
+it in `ensure`, so events emitted by delayed actions carry `performed_by: rule` and are
+ignored by the listener — identical to the immediate path.
+
+## Phase 5 — API
+
+### `app/controllers/api/v1/accounts/automation_rules_controller.rb`
+Param-level flag gating (the enterprise contacts-controller precedent — `permitted_params`
+merges `:company_id` only when `feature_enabled?('companies')`): include `:execution_delay`
+in `automation_rules_permit` only when
+`Current.account.feature_enabled?('delayed_automations')`; when the param is present with
+the flag off, render 422 with a clear error (explicit beats silent stripping). `clone` works
+unchanged (`dup` copies the column).
+
+### `app/views/api/v1/accounts/automation_rules/partials/_automation_rule.json.jbuilder`
+Expose `execution_delay`.
+
+## Phase 6 — Frontend
+
+All under `app/javascript/dashboard/`:
+
+1. **`routes/dashboard/settings/automation/AutomationRuleForm.vue`** — below the Event
+   select: "Execute" radio group (Immediately / After a delay) + number input + unit select
+   (minutes / hours / days). Convert to minutes on save; hydrate back to the largest clean
+   unit on edit. Client-side validation mirrors the 10 min–30 days bounds.
+2. **`helper/automationHelper.js`** — `generateAutomationPayload` passes `execution_delay`
+   through (null when "Immediately"). Add a `formatDelay` helper for the badge label.
+3. **`composables/useAutomation.js` / `useEditableAutomation.js`** — default
+   `execution_delay: null` on create; carry the field when editing/cloning.
+4. **Rule list badge** — "Runs after 4h" chip when `execution_delay` present.
+5. **Flag gating** — add `delayed_automations` to `featureFlags.js`; render the Execute
+   control only when `isFeatureEnabledonAccount` says so. Rules that already have a delay
+   keep their badge when the flag is off, and the settings page shows a "delayed execution
+   is disabled for this account" banner (flag-off must never silently hide configured
+   behavior).
+6. **i18n** — `i18n/locale/en/automation.json` only: radio labels, unit names, validation
+   message, badge text, disabled-flag banner.
+7. Store (`store/modules/automations.js`) and API client (`api/automation.js`) are generic
+   passthroughs — no changes expected.
+
+Tailwind-only styling; Composition API `<script setup>` (form already is).
+
+## Phase 7 — Specs (in scope per design doc §7)
+
+- `spec/models/automation_rule_spec.rb` — delay bounds; `attribute_changed` × delay rejection.
+- `spec/models/automation_rule_pending_execution_spec.rb` — episode key derivation (all
+  three anchors); insert/upsert semantics (status + incoming-anchored: clock not reset;
+  outgoing-anchored: due_at tracks latest agent reply); `episode_current?` incl.
+  agent-reply-clears-`waiting_since` cancellation; `mark_processing!` claim;
+  `reclaim_stale!`; `expire_overdue!`.
+- `spec/models/conversation_spec.rb` — `status_changed_at` set on create and every
+  transition, untouched on non-status saves.
+- `spec/listeners/automation_rule_listener_spec.rb` — delayed rule records a pending
+  execution and does NOT run actions; nil delay unchanged; `performed_by` automation events
+  create no rows; flag off → no arming AND no immediate fallback.
+- `spec/jobs/automation_rules/trigger_pending_executions_job_spec.rb` — only due+pending
+  swept; kill switch no-ops the sweep; window expiry marks `expired`; cap leaves overflow
+  pending; stale `processing` reclaimed.
+- `spec/jobs/automation_rules/process_pending_execution_job_spec.rb` — full guard chain
+  with `skip_reason` per branch; kill switch reverts row to pending; happy path executes
+  actions exactly once.
+- `spec/controllers/api/v1/accounts/automation_rules_controller_spec.rb` —
+  permits/persists/serializes `execution_delay` when flag on; 422 when submitted with flag
+  off; clone copies it.
+- Story-level integration spec (cheap, high value): Story 1 and Story 2 tables from the
+  design doc as end-to-end listener→sweep specs with `travel_to`.
+
+## Manual QA script (maps to design-doc stories + rollout Stage B)
+
+1. Story 1: rule `conversation_updated` + `status = pending` + 10 min delay + add label →
+   flip to pending, touch the conversation twice (no clock reset), wait for sweep → label.
+   Repeat but resolve before due → no label, row `skipped`/`episode_moved` (resolving
+   changes `status_changed_at`, so the episode guard fires before the conditions re-check).
+2. Story 2: rule `message_created` + `message_type = outgoing` + delay + send_message →
+   agent reply, then customer reply before due → cancelled (`episode_moved`). Without
+   customer reply → follow-up sent once; the follow-up itself doesn't re-arm the rule.
+3. Agent-idle recipe (CW-7513 comments): rule `message_created` + `message_type = incoming`
+   + 30 min delay + `assign_agent: None` → customer message, no agent reply → unassigned at
+   sweep. Repeat with agent replying before due → NOT unassigned (`waiting_since` episode
+   cancelled, `episode_moved`).
+4. Story 4: pre-existing rule untouched → still fires instantly.
+5. Story 5: disable rule with pending row → sweep marks `rule_inactive`; delete rule →
+   rows gone.
+6. Flag: disable `delayed_automations` on the account → delay control hidden + banner
+   shown; armed rows skip with `flag_disabled`; no new rows armed. Set
+   `DISABLE_DELAYED_AUTOMATIONS` → sweep no-ops within one tick and queued per-row jobs
+   revert rows to pending.
+7. Local tip: temporarily drop the cron to `*/1` or run
+   `AutomationRules::TriggerPendingExecutionsJob.perform_now` from console.
+
+## Sequencing / PR split
+
+Stacked PRs off `feature/cw-7513`:
+
+1. **PR 1 — backend**: Phases 1–5 + specs. Reviewable standalone; feature is
+   **API-complete** — this is the Jul 15 degenerate path: if PR 2 slips, the canary
+   customer's rules can be created via console/API with only PR 1 shipped.
+2. **PR 2 — frontend**: Phase 6 (delay control behind the flag, badge, banner, i18n).
+
+Work order within PR 1: schema/config → conversation hook → pending-execution model
+(episode semantics are the only genuinely new logic — do them test-first) → listener branch
+→ jobs → controller/jbuilder.
+
+## Risks / edge cases to keep in view
+
+- **Upsert vs. validation races**: `insert`/`upsert` skip AR validations — fine, all values
+  are server-computed; the unique index is the real guard.
+- **`conversation_updated` storm cost**: one indexed insert-conflict per matching event, no
+  row growth per episode (unique index). Watch the unique index's bloat after rollout.
+- **Rule edited while pending**: conditions re-check uses current conditions (by design);
+  `due_at` keeps its original value (documented v1 behavior).
+- **`send_email_to_team` / webhook actions at fire time**: work unchanged — rate limit
+  (`within_email_rate_limit?`) and `Current.executed_by` both live in ActionService; verify
+  in specs, don't reimplement.
+- **Audit overlay**: `enterprise/app/models/enterprise/audit/automation_rule.rb` audits the
+  new column automatically — verify the audited payload includes `execution_delay`.
+- **Super-admin flag UI**: `selected_feature_flags=` spans both bitset columns since
+  #14947 — the per-account toggle works with zero extra code; being `chatwoot_internal`,
+  the flag is visible only on Cloud super-admin until Stage E.
+- **Retroactivity limitation** (documented, not a bug): delayed rules arm on events; a new
+  rule does nothing for conversations already sitting in the target state with no further
+  activity. Release notes + user guide + rule-form microcopy state this; closing it is a
+  Phase 2 decision (see rollout doc).
+- **Deprecation note in `lib/events/types.rb`** (opened/resolved → status_changed): no
+  impact, we add no new events.
diff --git a/docs/delayed-automations-overview.md b/docs/delayed-automations-overview.md
new file mode 100644
index 000000000..0eb74ecc8
--- /dev/null
+++ b/docs/delayed-automations-overview.md
@@ -0,0 +1,149 @@
+# Delayed automations, in five minutes
+
+**Team explainer · CW-7513** — automation rules can now wait. Set a delay on any rule and it
+runs later, but only if what triggered it is still true.
+
+> **TL;DR** — We never schedule an *action*. We schedule a *question* — "is this still
+> true?" — and only act when the answer is yes.
+
+Companion docs: [design](delayed-automations.md) ·
+[implementation plan](delayed-automations-implementation-plan.md) ·
+[phasing & rollout](time-based-automations-phasing-and-rollout.md)
+
+---
+
+## One new option on the rule form
+
+Every automation rule gets an **Execute: immediately / after a delay** choice
+(10 minutes to 30 days). Everything else — events, conditions, actions — is the automation
+feature people already know. The three things customers asked for become plain rules:
+
+| Scenario | Rule | If things change first |
+|---|---|---|
+| **Stuck in a status too long** | `Conversation Updated` + `status = pending` + **after 4h** → add label | Resolved before the 4 hours pass? Nothing happens. |
+| **Customer went quiet** | `Message Created` + `type = outgoing` + **after 24h** → send message | "Just checking in…" — sent once, cancelled automatically if the customer replies first. |
+| **Agent went quiet** | `Message Created` + `type = incoming` + **after 30m** → unassign agent | Frees the conversation for reassignment — cancelled the moment the agent actually replies. |
+
+## The core idea: delay + re-check
+
+"Has been in state X for N hours" is the same as "matched at some moment, and still matches
+N hours later." So the rule is evaluated twice — once to start the timer, once before acting.
+
+```mermaid
+flowchart LR
+    E[Event fires<br/><small>message, status change…</small>] --> M[Rule matches<br/><small>conditions pass, as today</small>]
+    M --> T[Timer armed<br/><small>due = now + delay</small>]
+    T -. delay elapses .-> R{Still true?<br/><small>conditions re-checked</small>}
+    R -- yes --> A[Actions run]
+    R -- no --> C[Cancelled silently<br/><small>reason recorded</small>]
+    style T fill:#e9f2fe,stroke:#2781f6
+    style A fill:#e4f4ec,stroke:#248a5c
+    style C fill:#faf1dc,stroke:#a97413
+```
+
+The re-check is what makes the semantics right: a follow-up message cancels itself when the
+customer replies; a "stuck in pending" label never lands on a conversation that got resolved.
+
+## Under the hood: one table and the cron we already have
+
+No new events, no future-scheduled Sidekiq jobs sitting in Redis for days. A delayed match
+writes one row; the existing 5-minute scheduler sweeps whatever is due — the same pattern
+snooze, campaigns, auto-resolve and SLA already use.
+
+```mermaid
+flowchart LR
+    E[Conversation event] --> L[Automation listener<br/><small>rule + conditions match</small>]
+    L -- no delay --> A[Actions run<br/><small>instantly, unchanged</small>]
+    L -- delay set --> P[(pending execution row<br/><small>who · which rule · due when</small>)]
+    P --> S[5-minute sweep<br/><small>picks up due rows</small>]
+    S --> G{Guards + re-check}
+    G -- still true --> A
+    G -- not anymore --> K[Skipped<br/><small>reason saved</small>]
+    style P fill:#e9f2fe,stroke:#2781f6
+    style A fill:#e4f4ec,stroke:#248a5c
+    style K fill:#faf1dc,stroke:#a97413
+```
+
+Accounts with no delayed rules create zero rows and zero cost. Repeated events for the same
+wait collapse into the same row — the timer does not reset every time someone touches the
+conversation.
+
+## One conversation, two endings
+
+The "stuck in pending" rule from above, on a real timeline:
+
+**Stays pending → the rule fires**
+
+| 10:00 | 11:30 | 14:00 |
+|---|---|---|
+| Moved to Pending — timer armed, due 14:00 | Agent adds a note — clock unchanged | Still pending → **label added** ✓ |
+
+**Resolved in time → the rule cancels itself**
+
+| 10:00 | 13:00 | 14:00 |
+|---|---|---|
+| Moved to Pending — timer armed, due 14:00 | Resolved — the wait it measured is over | **Nothing happens** — cancelled, nothing visible |
+
+Leaving and re-entering Pending starts a fresh timer. A rule fires at most once per
+"episode" — the platform owns the anti-loop guarantee; users never build "only once" hacks
+into their conditions.
+
+> **Good to know:** delayed rules apply to conversations with activity *after* the rule is
+> created. A brand-new rule doesn't retroactively scan the existing backlog — closing that
+> gap is on the roadmap (Phase 2).
+
+## "Why didn't my rule fire?"
+
+Every timer that doesn't run records exactly why. Six reasons cover every case:
+
+| Reason | What it means |
+|---|---|
+| `conditions_changed` | The conversation no longer matched at fire time — the usual, healthy cancellation. |
+| `episode_moved` | The wait it was measuring ended — customer replied, agent replied, or the status changed. |
+| `rule_inactive` | The rule was switched off or deleted while the timer was running. |
+| `flag_disabled` | Delayed automations were turned off for the account after the timer was armed. |
+| `conversation_gone` | The conversation was deleted. |
+| `expired` | The row was overdue by more than 3 days (e.g. long downtime) — we don't replay stale actions. |
+
+Rows are kept for 30 days — support and engineering can answer "what fired, what didn't,
+and why" for any conversation. An admin-facing history view ships in Phase 2.
+
+## Three ways to stop it, three different sizes
+
+| Scope | Lever | Behaviour |
+|---|---|---|
+| One rule | The existing on/off toggle | Pending timers for a disabled rule are skipped at fire time. |
+| One account | `delayed_automations` feature flag | Gates the form control, the timer arming, and firing. Off means nothing arms and nothing fires — it never falls back to running instantly. |
+| Whole instance | `DISABLE_DELAYED_AUTOMATIONS` config | Super-admin setting, takes effect within one 5-minute tick, no deploy. Checked by the sweep **and** every in-flight job. |
+
+## Where we are
+
+```mermaid
+flowchart LR
+    A[✅ Built & tested<br/><small>backend + UI, specs green,<br/>verified end-to-end in the app</small>] --> B[Internal dogfood<br/><small>our own account</small>]
+    B --> C[Canary — Jul 15<br/><small>requesting customers</small>]
+    C --> D[Cloud GA<br/><small>waved, gated on metrics</small>]
+    D --> E[Self-hosted<br/><small>next tagged release</small>]
+    style A fill:#e4f4ec,stroke:#248a5c
+    style B fill:#e9f2fe,stroke:#2781f6
+```
+
+**Covered in v1**
+
+- Delay on any rule, 10 min – 30 days
+- All three customer scenarios, including unassign-on-silence
+- Automatic cancellation on reply / status change
+- Fires at most once per episode — no loops, by design
+- Full audit trail with skip reasons
+
+**Deliberately later**
+
+- "Hours since X" conditions for already-quiet conversations — Phase 2
+- Business-hours aware delays — Phase 2
+- Admin-facing run history UI — Phase 2
+- Reminder chains ("remind ×3, then resolve") — Phase 3
+- Calendar/cron triggers — Phase 4
+
+---
+
+Code on `feature/cw-7513` · Tracking: CW-7513, parent CW-5790
diff --git a/docs/delayed-automations.md b/docs/delayed-automations.md
new file mode 100644
index 000000000..525f2a55c
--- /dev/null
+++ b/docs/delayed-automations.md
@@ -0,0 +1,335 @@
+# Delayed Automations — Time-Based Automation Rules
+
+**Status:** Design approved, pending implementation
+**Target:** July 15 release
+**Owner:** Tanmay
+
+**Key decisions**
+- Merged into existing automation rules (same model, same UI, same API) — not a separate
+  feature or a special rule type. Delay is an optional property (`execution_delay`).
+- No future-scheduled Sidekiq jobs — DB `due_at` + the existing 5-minute cron sweep
+  (`TriggerScheduledItemsJob`), the same entry point SLA uses.
+- Fire-time **re-check** of the rule's conditions is what gives "stayed in state for N hours"
+  semantics; the episode guard (key recomputation) is what auto-cancels follow-ups when the
+  customer replies.
+
+---
+
+## 1. Problem
+
+Chatwoot automation rules are purely **event-driven**. A rule fires at the instant one of five
+events occurs (`conversation_created`, `conversation_updated`, `conversation_opened`,
+`conversation_resolved`, `message_created`) and never again. There is no way to express
+*duration*:
+
+- "If a conversation stays in status X for more than N hours → add a label"
+- "If the end user doesn't reply within N hours → send a follow-up message"
+- "If a conversation is open for more than N hours → add a label"
+
+Nothing in the system wakes up N hours later to check. That is the gap this feature fills.
+
+## 2. Goals
+
+- Automation rules can be configured with a **delay**: "execute this rule N hours/minutes after
+  the triggering event, if the conditions still hold."
+- Covers the three customer asks above using **existing events, conditions, and actions**.
+- Supported actions include at least **add label** and **send message** (both already exist).
+- Fully backward compatible: existing rules keep behaving exactly as today.
+- Safe at Chatwoot Cloud volume: load must scale with *pending delayed rules*, not raw event traffic.
+
+## 3. Non-goals (v1)
+
+- **Per-action delays / multi-step sequences** (wait 24h → message → wait 24h → label).
+  v1 is one delay per rule. Sequences are a natural follow-up.
+- **Exact-second firing.** Delays are hours-scale; firing within ~5 minutes of the due time
+  is acceptable.
+- **New duration condition attributes** (`hours_in_status`, etc.). The delay + re-check model
+  covers the v1 stories without them; richer "time since X" anchors are Phase 2 — see
+  `docs/time-based-automations-phasing-and-rollout.md`.
+- **Recurring execution** (fire every N hours while the condition holds). v1 fires at most
+  once per conversation per qualifying episode.
+
+## 4. Core model: delay + re-check
+
+The design rests on one insight:
+
+> **"Has been in state X for N hours" ≡ "matched the conditions at T₀, and still matches at T₀ + N."**
+
+So instead of building a new duration-condition engine, we:
+
+1. Evaluate the rule's conditions at event time, exactly as today.
+2. If the rule has a delay, **record a pending execution** due at `now + delay` instead of
+   acting immediately.
+3. When the due time arrives, **re-evaluate the same conditions** against the conversation's
+   current state.
+   - Still matches → run the actions.
+   - No longer matches → discard silently. This is what makes the semantics correct: a
+     follow-up message is automatically cancelled when the customer replies, a "stuck in
+     pending" label is skipped if the conversation moved on.
+
+Everything downstream — `AutomationRules::ConditionsFilterService`,
+`AutomationRules::ActionService`, all existing actions — is reused unchanged.
+
+### Not a special type of rule
+
+A delayed rule is **not** a new rule type — no STI, no new class, no new `event_name`. It stays
+a plain `AutomationRule`; the only difference is whether `execution_delay` is set:
+
+```ruby
+rule.execution_delay # => nil  → executes immediately (today's path)
+rule.execution_delay # => 240 → records a pending execution due in 4 hours
+```
+
+Why a property beats a type:
+
+- Any existing rule can become delayed (and back) by setting one field, without recreating it
+  as a different kind of object.
+- The listener branch is a single `if` at the point where actions would run; everything
+  upstream (rule lookup by account+event, condition evaluation) is identical for both paths.
+- The UI stays one form. A separate type would mean a type picker, separate list filters, and
+  separate API handling — ceremony with no behavioral payoff.
+
+The only places it *acts* special are the rule-list badge ("runs after 4h") and the fire-time
+guard chain — both driven off the field, not a type.
+
+### Why not Sidekiq `perform_in`?
+
+The obvious implementation (enqueue a future Sidekiq job per matching event) is wrong at our
+volume:
+
+- `conversation_updated` fires on nearly every touch. Each match would push a job into Redis'
+  scheduled set, where it sits for **hours to days**. Redis bloats and the scheduled-set poller
+  degrades. Work scales with raw event volume — the worst thing to couple to.
+- Cancelling/deduping scheduled Sidekiq jobs is awkward (no efficient lookup by
+  rule+conversation).
+
+Instead we use the pattern Chatwoot already uses for snooze, auto-resolve, campaigns, and SLA:
+**store a `due_at` in Postgres and sweep due rows from the existing 5-minute cron**
+(`TriggerScheduledItemsJob`). One indexed row per pending (rule, conversation) pair, a bounded
+`WHERE due_at <= now` range scan, Redis untouched.
+
+## 5. Architecture
+
+### 5.1 Data model
+
+**`automation_rules.execution_delay`** — new integer column, delay in **minutes**.
+`NULL` (default) = execute immediately, i.e. today's behavior. No backfill needed.
+
+**`automation_rule_pending_executions`** — new table (name TBD):
+
+| Column | Type | Notes |
+|---|---|---|
+| `automation_rule_id` | bigint FK | rule to execute |
+| `conversation_id` | bigint FK | target conversation |
+| `account_id` | bigint FK | for scoping/cleanup |
+| `due_at` | datetime | `event_time + rule.execution_delay` |
+| `episode_key` | string | identifies the qualifying episode (see §5.3) |
+| `status` | enum | `pending` / `processing` / `executed` / `skipped` |
+| timestamps | | |
+
+Final schema (incl. the `message_id` anchor, `skip_reason`, and the `processing` claim
+state) is in the implementation plan, Phase 1c.
+
+Indexes:
+- `(status, due_at)` — the sweep query.
+- Unique `(automation_rule_id, conversation_id, episode_key)` — dedup: repeated events for the
+  same episode collapse into one row. For status- and waiting_since-anchored episodes the
+  clock is **not reset**; outgoing-anchored (reply-chase) episodes update `due_at` to track
+  the latest agent reply (impl plan, locked decision #4).
+
+**`conversations.status_changed_at`** — new datetime column, set on every status transition
+(hooked into the existing status-change path on `Conversation`). Serves two purposes:
+- It is the `episode_key` source for status-based rules, so "entered open → left → re-entered
+  open" counts as a *new* episode with a fresh timer.
+- Lets the fire-time check confirm the status held continuously, not just coincidentally at
+  both endpoints.
+
+### 5.2 Event-time flow (write path)
+
+Inside the existing `AutomationRuleListener` path — no new events:
+
+```
+event fires
+  └─ for each active rule matching (account, event_name):
+       ConditionsFilterService.perform
+         └─ matched?
+              ├─ rule.execution_delay.nil?  → ActionService.perform   (unchanged, today's path)
+              └─ rule.execution_delay set   → upsert pending execution
+                                              due_at = now + delay
+                                              (unique index ⇒ one row per episode; clock NOT
+                                               reset — except outgoing-anchored reply-chase
+                                               episodes, which update due_at to track the
+                                               latest agent reply)
+```
+
+### 5.3 Episode semantics — "don't reset the clock"
+
+`conversation_updated` fires on every touch. For a rule like *"status = pending for 4 hours"*:
+
+- First event where conditions match → pending row created, `due_at = entered_pending_at + 4h`,
+  `episode_key = status_changed_at`.
+- Every subsequent update while still pending → same episode key → upsert no-ops. The timer
+  keeps counting from when the conversation *entered* the state.
+- Conversation leaves pending, later returns → new `status_changed_at` → new episode → fresh
+  timer. The old pending row is discarded at fire time by the re-check.
+
+For message rules the anchor depends on the triggering message's direction (impl plan,
+locked decision #4): *outgoing-anchored* (customer went quiet) keys on the latest incoming
+message id, with `due_at` tracking the latest agent reply; *incoming-anchored* (agent went
+quiet) keys on `waiting_since`, which Chatwoot clears on agent/bot reply — so the reply
+cancels the pending action via the episode guard.
+
+### 5.4 Fire-time flow (sweep)
+
+Hooked into the existing **`TriggerScheduledItemsJob`** (every 5 minutes, `config/schedule.yml`
+→ `trigger_scheduled_items_job`, cron `*/5`) — the same entry point SLA piggybacks on. Note:
+SLA does *not* run every minute; it also rides this 5-minute job.
+
+```
+TriggerScheduledItemsJob (cron, */5)
+  └─ AutomationRules::TriggerPendingExecutionsJob
+       └─ PendingExecution.pending.where(due_at: ..Time.current)
+            .find_each → ProcessPendingExecutionJob (per record):
+              1. rule still exists & active?          no → mark skipped
+              2. conversation still exists?           no → mark skipped
+              3. episode still current? (anchor unchanged)
+                                                      no → mark skipped
+              4. ConditionsFilterService still true?  no → mark skipped   ← the re-check
+              5. ActionService.perform → mark executed
+```
+
+**Same pattern as SLA, but flatter.** SLA needs a 3-level fan-out
+(`TriggerSlasForAccountsJob` → per-account `ProcessAccountAppliedSlasJob` → per-record
+`ProcessAppliedSlaJob`) because it must *recompute* deadlines from conversation timestamps on
+every pass — it doesn't know in advance when a breach will happen. We do know: `due_at` is
+precomputed at event time. So we skip the per-account fan-out entirely:
+
+- One global indexed range query on `(status, due_at)`. The table only contains rows that are
+  actually waiting — accounts with no delayed rules contribute zero rows, zero cost.
+- Per-record jobs for the actual execution (guards + re-check + `ActionService`), so one slow
+  or failing conversation doesn't block the batch — this part mirrors
+  `Sla::ProcessAppliedSlaJob` exactly.
+
+**Why 5-minute cadence is enough.** Delays are hours-scale ("4 hours", "24 hours"); worst case
+a rule fires 5 minutes late, which is invisible at that scale. A 1-minute cadence would buy
+precision nobody asked for at 5× the cron load. If tighter precision is ever needed, the fix is
+cadence-only — one line in `schedule.yml` — with no design change.
+
+Notes:
+- `Current.executed_by = rule` is set exactly as in the immediate path, so events emitted by the
+  delayed actions are recognized as automation-performed and don't retrigger rules (loop guard).
+- Executed/skipped rows are kept 30 days (audit trail / incident blast-radius queries — impl
+  plan locked decision #7), pruned by the daily scheduled job.
+
+### 5.5 Volume characteristics
+
+| Concern | Answer |
+|---|---|
+| Redis scheduled-set growth | Zero — no future jobs enqueued |
+| Write amplification from update storms | Collapsed by the unique episode index (1 row per rule×conversation×episode) |
+| Sweep cost | Indexed range scan on `(status, due_at)`; scales with count of *due* rows, not traffic |
+| Accounts without delayed rules | Zero cost — no pending rows are ever created |
+| Firing precision | Within the 5-min cron window; fine for hours-scale delays |
+
+### 5.6 Frontend
+
+In the Add/Edit Automation Rule dialog, below Event (per the mockup):
+
+```
+Execute   ( • ) Immediately
+          (   ) After a delay:  [ 4 ] [ hours ▾ ]
+```
+
+- Stored as minutes in `execution_delay`; UI offers minutes/hours/days units.
+- Shown for all existing events — no new event types in the dropdown.
+- Rule list row shows a badge like "runs after 4h" for delayed rules.
+- Changes: `AutomationRuleForm` + `constants.js` untouched for conditions/actions;
+  `automation.json` i18n additions; API strong params + jbuilder gain `execution_delay`.
+
+## 6. User stories & expected behavior
+
+### Story 1 — Label conversations stuck in Pending
+
+> *As an admin, I create: Event = **Conversation Updated**, Condition = **Status equals Pending**,
+> Delay = **4 hours**, Action = **Add label `stale-pending`**.*
+
+| What happens | Behavior |
+|---|---|
+| Conversation moves to Pending at 10:00 | Rule matches → pending execution created, due 14:00 |
+| Agent adds a private note at 11:30 (another `conversation_updated`) | Same episode → no new row, **clock not reset**, still due 14:00 |
+| Conversation still Pending at 14:00–14:05 sweep | Re-check passes → label `stale-pending` added |
+| — or — conversation was resolved at 13:00 | Re-check fails at 14:00 → nothing happens, row marked skipped |
+| Conversation reopens and goes Pending again next day | New `status_changed_at` → new episode → fresh 4-hour timer |
+
+### Story 2 — Follow up when the customer goes quiet
+
+> *As a support lead, I create: Event = **Message Created**, Condition = **Message Type is
+> Outgoing**, Delay = **24 hours**, Action = **Send message** "Just checking in — did that solve
+> it for you?"*
+
+| What happens | Behavior |
+|---|---|
+| Agent replies to the customer Monday 15:00 | Pending execution created, due Tuesday 15:00 |
+| Agent sends two more replies Monday 15:10 | Anchor updates → the follow-up tracks the latest agent reply |
+| Customer replies Monday 18:00 | The reply changes the episode key; at fire time the episode guard sees it → **follow-up silently cancelled** (`episode_moved`) |
+| Customer never replies | Tuesday ~15:00–15:05 → follow-up message sent to the end user, exactly once |
+| The follow-up message itself is created | `Current.executed_by = rule` → does **not** re-arm the rule (no infinite follow-up loop) |
+
+*(v1 note: whether the follow-up re-arms after each new agent reply, or fires once per
+conversation, is decided by the episode key — default is once per "waiting" episode.)*
+
+### Story 3 — Escalation label on long-open conversations
+
+> *As an admin, I create: Event = **Conversation Created**, Condition = **Status equals Open**
+> (optionally + Inbox = Support), Delay = **48 hours**, Action = **Add label `sla-risk`**.*
+
+| What happens | Behavior |
+|---|---|
+| Conversation created Wednesday 09:00, still open Friday 09:00 | Label `sla-risk` added at the Friday ~09:00 sweep |
+| Conversation resolved Thursday | Re-check fails Friday → no label |
+| Conversation resolved Thursday, **reopened** Friday 08:00 | v1: original episode ended → no label from the old timer. (`conversation_opened` + delay can cover re-opens as a separate rule.) |
+
+### Story 4 — Existing rules are untouched
+
+> *As an existing customer, my current rules have no delay.*
+
+| What happens | Behavior |
+|---|---|
+| Any existing rule fires | `execution_delay` is `NULL` → immediate path, byte-for-byte today's behavior |
+| I edit an old rule and never touch the delay field | Still immediate |
+
+### Story 5 — Rule lifecycle while a timer is pending
+
+| What happens | Behavior |
+|---|---|
+| Admin **disables** the rule while executions are pending | Fire-time guard: rule inactive → all its pending rows skipped |
+| Admin **deletes** the rule | Pending rows removed (FK/dependent destroy) |
+| Admin **edits the delay** from 4h → 8h | Applies to new episodes; already-pending rows keep their original `due_at` (simple, predictable v1 rule) |
+| Admin edits the **conditions** | Fire-time re-check uses the *current* conditions — the edited rule is what's enforced |
+| Conversation deleted before due | Guard: skipped |
+
+## 7. Scope for July 15
+
+**In:**
+1. Migration: `automation_rules.execution_delay`, `conversations.status_changed_at`,
+   pending-executions table.
+2. Listener wiring: delayed rules record a pending execution instead of executing.
+3. Sweep job hooked into `TriggerScheduledItemsJob` with the fire-time guard chain.
+4. Frontend delay field in the rule form + list badge + i18n (`en.json` only).
+5. Specs: model validations, listener branching, sweep guards/re-check, episode dedup.
+
+**Out (follow-ups):** per-action wait steps, recurring fires, business-hours-aware delays
+(SLA's `Sla::BusinessHoursService` is the ready-made building block when we want it),
+exact-time firing.
+
+## 8. Open questions
+
+1. **Delay bounds** — enforce a min (≥ 5 min, below cron granularity is meaningless) and a max
+   (e.g. 30 days) at the model level.
+2. **Follow-up re-arm policy** (Story 2): once per waiting-episode (default) vs. once per
+   conversation ever. Proposing per-episode.
+3. **Enterprise overlay** — feature ships in OSS core (automation rules are OSS; the scan cost
+   is self-limiting since accounts without delayed rules create zero rows). Confirm no
+   enterprise gating is wanted.
+4. Table/row retention: ~~delete executed/skipped rows immediately vs. keep N days~~ —
+   **resolved**: keep 30 days (audit/blast-radius), pruned by the daily scheduled job.
diff --git a/docs/time-based-automations-phasing-and-rollout.md b/docs/time-based-automations-phasing-and-rollout.md
new file mode 100644
index 000000000..f997b09ed
--- /dev/null
+++ b/docs/time-based-automations-phasing-and-rollout.md
@@ -0,0 +1,395 @@
+# Time-Based Automations — Product Phasing & Rollout Plan
+
+**Status:** Draft for review (adversarially reviewed; implementation plan amended to match — see §5 note)
+**Owner:** Tanmay
+**Companions:** `docs/delayed-automations.md` (v1 design, approved) ·
+`docs/delayed-automations-implementation-plan.md` (v1 build plan — amended by this doc's review)
+**Tracking:** CW-7513 (v1, due Jul 15) · CW-5790 (parent tracker, "Needs Spec" — this doc is
+the spec for everything past v1)
+
+---
+
+## 1. The end goal, defined
+
+"Time-based automations" is not one feature — it is a ladder that every mature support
+platform climbs in the same order (Zendesk, Freshdesk, Intercom, HubSpot, Help Scout, Front
+all converge on it):
+
+1. Delayed action with fire-time re-check ← **v1 (CW-7513)**
+2. Run-once / anti-loop semantics owned by the platform, not the user
+3. Rich "time since X" condition vocabulary (created / entered status / last customer reply /
+   last agent reply / assigned)
+4. Business-hours-aware time accounting
+5. SLA-integrated escalation (at-risk + breach stages with actions)
+6. Multi-step sequences (follow-up → wait → follow-up → resolve, interruptible on reply)
+7. Scheduled/calendar triggers and full workflow builder
+
+**End goal for Chatwoot (12-month horizon): rungs 1–6.** Rung 7 (visual builder, cron
+triggers) is a separate product investment and stays out of this plan except as a
+compatibility constraint: nothing we ship may assume "one rule = one delayed action" so
+deeply that sequences can't be layered on.
+
+Three facts from the research shape everything below:
+
+- **The v1 engine is already the end-state engine.** Per-conversation pending executions with
+  fire-time re-check is the architecture Front and Intercom use; Zendesk/Freshdesk's hourly
+  full-table sweeps are the legacy model whose user-visible warts (whole-hour granularity,
+  ±1 h slop, "nullifying condition" hacks) we get to skip. Every later phase is additive
+  vocabulary and UI on the same table + sweep — no rewrite is on the ladder.
+- **One capability the sweep model has that ours doesn't: eventless, retroactive
+  evaluation.** An event-armed engine does nothing for conversations already sitting in the
+  target state when a rule is created. This is a real, user-visible limitation ("why didn't
+  it label my existing backlog?") — v1 documents it plainly (§3, Phase 1) and Phase 2 carries
+  a named scope decision for closing it.
+- **Community demand is mapped, with honest issue states.** Phases cite Chatwoot GitHub
+  issues as *evidence of demand*; several were bulk-closed as stale (marked below), so they
+  document the ask, not a ticket we get to close.
+
+## 2. Product phases at a glance
+
+| Phase | Ships | Evidenced demand | Target |
+|---|---|---|---|
+| **0. Engine (dark)** | Schema, pending-execution engine, sweep, guardrails, feature flag — no UI | — | with v1 PRs |
+| **1. Delayed rules** | `execution_delay` + fire-time re-check; both reply directions via episode anchors; delay UI; badge | CW-7513 (all 3 examples + unassign-on-idle from comments); #1270 *(partial: single reminder)* | **Jul 15** |
+| **2. Time vocabulary + business hours + run history** | "Time since last customer/agent reply, time in status, time since created" conditions; run inside/outside business hours; per-rule execution history UI | #6861 (open), #10455 (open), #12694 (open), #4542 *(closed-stale; documents the WhatsApp-24h ask)*; #6889 *(recipe becomes possible)* | ~6 weeks after Cloud GA |
+| **3. Follow-up chains + SLA escalation** | Repeat-with-terminator ("remind ×N then resolve"); SLA at-risk/breach as automation events; snooze-duration action | #1270 (open, full ask), #6889 (open), #10118 *(closed-stale)* | Q4 2026 |
+| **4. Scheduled triggers / builder** | Calendar/cron rule triggers, scheduled sends, visual sequences | #9484 (open) + scheduled-send cluster (#5554 closed-stale, #9771, #11486, #12146, #12823); CW-5790 original ask | 2027, own spec |
+
+Each phase is independently shippable, independently valuable, and strictly additive on the
+Phase 0 engine.
+
+## 3. Phase detail
+
+### Phase 0 — Engine, shipped dark
+
+Everything in `docs/delayed-automations-implementation-plan.md` (migrations, pending-execution
+model with claim states, listener branch, sweep jobs with caps/windows/kill switch, API
+gating, feature flag) merged with no UI exposure. `execution_delay` stays `NULL` for every
+rule, so the listener branch is dead code for 100% of traffic until the flag turns the UI on —
+this is what makes the merge risk-free.
+
+Exit criteria (measurable, and achievable while dark):
+- Engine + guard specs green.
+- Sweep running in production logging `"due":0,"enqueued":0` every 5 minutes.
+- All three CW-7513 stories exercised **on staging** with compressed delays (10 min).
+- §6 alerts created and verified end-to-end with a test trigger.
+
+(Live dogfood on Chatwoot's own account requires the flag on — that is Stage B, Phase 1.)
+
+### Phase 1 — Delayed rules (CW-7513, Jul 15)
+
+The v1 surface: "Execute: immediately / after a delay" on the existing rule form, delay badge
+in the rule list.
+
+**What "time since last reply" means in v1 — both directions, precisely:**
+- *Customer went quiet* (CW-7513 example 2, design-doc Story 2): rule anchored on an
+  **outgoing** message. Episode tracks the latest agent reply; a customer reply cancels the
+  pending follow-up at fire time.
+- *Agent went quiet* (the unassign ask in the Linear comments; the #6861 phrasing): rule
+  anchored on an **incoming** message. Episode is keyed on `waiting_since`, which Chatwoot
+  clears when an agent or bot replies — so an agent reply cancels the pending action at fire
+  time. Without this anchor the recipe would unassign agents who *did* reply promptly; the
+  implementation plan's episode-key section now specifies both anchors (this was the single
+  most important correction from review).
+
+With that, the unassign-on-no-response recipe from the Linear comments is buildable in the
+UI on day one (the agent dropdown already offers "None", serializing to the `'nil'`
+sentinel — `useAutomationValues.js:88`): *Message Created + Message Type is Incoming + delay
+30 min + Assign agent: None*. Ship it as a documented recipe. The thread's follow-up ask
+(agent online/offline as a condition) is not in v1 — logged as a Phase 3 open item.
+
+**Known limitation, stated everywhere users will hit it** (release notes, user-guide,
+rule-form microcopy): delayed rules apply to conversations that have activity *after* the
+rule is created. A brand-new "stuck in Pending 4h" rule does not retroactively arm for
+conversations already sitting in Pending. Closing this is a named Phase 2 decision.
+
+**Run-once semantics** (ladder rung 2) are built in from day one via episode keys — the
+lesson from Zendesk/Freshdesk, whose users must hand-build nullifying conditions. Never
+expose a knob that lets a rule re-fire hourly.
+
+**Observability in v1 is internal-only**: support/eng can answer "why didn't it fire" from
+the pending-executions table (`skip_reason` column). Admin-facing run history ships in
+Phase 2 — until then this is a known support load, priced in.
+
+Exit criteria:
+- Before Jul 15: the requesting customer confirms the two reply-direction readings above
+  match their expectation (don't discover a mismatch after release).
+- Canary accounts using delayed rules for 1 week: `expired` = 0 sustained, fire-lag p95
+  < 10 min, zero new Sentry groups under the two job classes.
+- Story 2's cancellation observed in production (a real customer reply cancelling a pending
+  follow-up), not just in specs.
+- Week-1 executed:skipped mix recorded as the baseline band for Stage D gating.
+
+### Phase 2 — Time vocabulary, business hours, run history
+
+What users ask for the moment v1 lands (top evidenced asks):
+
+1. **"Time since X" conditions** — new condition attributes evaluated by the same engine:
+   `hours_since_last_customer_message`, `hours_since_last_agent_message`,
+   `hours_in_current_status`, `hours_since_created`. Mechanically these are alternate
+   **episode anchors + due-at sources** for a pending execution — the rule form gains a
+   "when time elapses" trigger style that composes with existing conditions.
+   **Named scope decision — arming for quiet/pre-existing conversations:** event-armed
+   pending rows don't cover conversations with no future events. Options: (a) one-time
+   arming scan at rule save (bounded, predictable — recommended), (b) periodic eligibility
+   scan (reintroduces exactly the per-account sweep v1 avoided; needs its own cost model).
+   Decide at Phase 2 spec with Cloud row-count data from Phase 1.
+2. **Business hours** — two separable features, in this order:
+   a. *Gate*: "only run this rule inside/outside inbox business hours" (a fire-time guard
+      against `working_hours`) — the #10455/#12694 ask.
+   b. *Accounting*: "4 business hours" delays that pause the clock outside schedules —
+      reuse `Sla::BusinessHoursService`. That service lives in `enterprise/`; either
+      accounting becomes a premium capability (matching SLA) or the calculator moves to OSS
+      core. Decide at Phase 2 spec; the schema needs nothing either way.
+3. **Re-arm policy control** — expose the v1 episode semantics as an explicit choice
+   ("once per conversation" / "every time it re-qualifies"), defaulting to current behavior.
+4. **Per-rule execution history UI** — fired/skipped(+reason) list on the rule, from the
+   existing table + `skip_reason` (30-day retention shown as a visible constraint). This is
+   the Zendesk/Intercom-parity observability the feature's support load demands.
+
+Exit criteria (self-contained, not tied to issue states):
+- A rule "send a template when >23h since the last inbound customer message on a WhatsApp
+  inbox" works end-to-end (the WhatsApp 24h-session use case).
+- A rule gated to business hours provably does not fire outside them; delay *accounting*
+  verified against SLA's calculator on shared fixtures.
+- Each re-arm policy behaves per spec in the Story-2 integration test.
+- An admin can answer "why didn't my rule fire on this conversation" from the rule's history
+  page without contacting support.
+
+### Phase 3 — Follow-up chains + SLA escalation
+
+1. **Repeat-with-terminator** (#1270's full ask): "send reminder every N hours, at most K
+   times, then resolve." Schema: `max_occurrences` + occurrence counter on the pending
+   execution — the episode model already prevents overlap. Sequences-*lite*: one repeated
+   action + one terminal action, deliberately short of a builder.
+2. **SLA integration** (#6889): emit `sla_missed` / `sla_at_risk` as automation events
+   (enterprise overlay adds them to the event dropdown, the same way `sla_policy_id`
+   conditions are added today), so escalation rules compose from existing actions. SLA
+   already computes the timing; automations subscribe. (Note: #6889's literal "waited more
+   than X hours → escalate" becomes satisfiable with Phase 2 vocabulary — publish that
+   recipe with Phase 2; Phase 3 makes it SLA-native.)
+3. **Snooze composition** (#10118's ask): `snooze_conversation` action gains a duration
+   param; snooze-expiry already rides the same 5-min sweep.
+4. Evaluate the agent online/offline condition (Linear-thread follow-up ask) with
+   assignment-policy input.
+
+Exit criteria: Intercom's canonical journey (reply-chase → reminder → auto-resolve)
+buildable in two rules; SLA-escalation recipe documented; sweep p95 flat vs Phase 2 baseline.
+
+### Phase 4 — Scheduled triggers / builder (out of this plan's scope)
+
+Calendar/cron triggers ("every Monday 9am"), scheduled one-off sends, visual multi-step
+builder. Needs its own spec (CW-5790 becomes that spec once Phases 1–3 close the
+duration-based demand). The only obligation now: keep `event_name` + pending-executions
+generic enough that a `schedule` pseudo-event can create pending executions without a
+conversation event — the current schema (rule, conversation, due_at, episode) already
+permits this.
+
+## 4. Rollout plan (Phase 1 in full; later phases reuse the template)
+
+### 4.1 Gating levers
+
+| Lever | Choice for this feature |
+|---|---|
+| Per-account feature flag | `delayed_automations`, appended at the end of `config/features.yml` with **`column: feature_flags_ext_1`**. The bit-budget problem this plan originally flagged was solved on develop (`873d16f54c`, #14947): a second bitset column `accounts.feature_flags_ext_1` exists (0/63 used) and the features.yml header now mandates new flags use it — the legacy `feature_flags` column is 63/63 full. No repurpose migration; ConfigLoader reconciles the new name (with its `column` metadata) into `ACCOUNT_LEVEL_FEATURE_DEFAULTS` on migrate, and `selected_feature_flags=` / super-admin toggles span both columns since #14947. `enabled: false`, `chatwoot_internal: true` at introduction. Flag name and its position within the ext column are frozen forever once merged. |
+| What the flag gates | **Three layers, one coherent semantics — the flag is the per-account stop.** (1) *Configuration*: the delay control in the rule form (`featureFlags.js` + `isFeatureEnabledonAccount`), and server-side, `execution_delay` is accepted only when the flag is on — param-level gating per the enterprise contacts-controller precedent (`permitted_params` includes `:company_id` only when `feature_enabled?('companies')`), returning 422 when a delay is submitted with the flag off (explicit beats silent stripping). (2) *Arming*: the listener does not create pending executions for flag-off accounts — and does **not** fall back to immediate execution (a 24h-delayed message silently becoming instant is worse than skipping). (3) *Fire time*: the guard chain checks the flag and marks rows `skipped` / `skip_reason: flag_disabled`. Rules keep their delay badge when the flag is off; the settings page shows a "delayed execution is disabled for this account" banner. |
+| Instance kill switch | `DISABLE_DELAYED_AUTOMATIONS`, **declared in `config/installation_config.yml` with `type: boolean`** (so "false" means false — the bare `DISABLE_GRAVATAR` presence-check pattern would treat any non-blank string as disabled), checked in **both** the sweep job *and* the per-row job (sweep-only would let up to a full sweep's already-enqueued per-row jobs fire after the flip). Rows the per-row job refuses stay `pending` and replay or expire via the due-window. Takes effect within one tick, no deploy (`InstallationConfig` `after_commit :clear_cache`). |
+| Rule-level off | Existing `active` toggle — already in the fire-time guard chain. |
+| Premium? | **Ship non-premium** (see Decisions, §7). The `premium: true` + `ReconcilePlanFeaturesService` lever stays available with zero schema cost if the business later wants plan-gating. |
+
+One honesty note: the wave-enablement rake in Stage D is **new work**, not existing
+machinery. Its safety kit (dry-run default, `APPLY=true`, `LIMIT=n`, confirmation) is
+specified here, modeled on `assignment_v2:migrate`'s interactive confirm + `ACCOUNT_ID`
+scoping and `reporting_events_rollup.rake`'s dry-run prompt — no committed rake has the full
+kit today.
+
+### 4.2 Stages
+
+**Stage A — Dark merge (deadline: Fri Jul 10 EOD; Sat Jul 11 is slack, not the plan).**
+PR 1 (engine + flag, per the amended implementation plan) and PR 2 (UI behind flag) land on
+`develop`. Named reviewers with a same-day review SLA are a §7 decision to lock **today**.
+Confirm the weekend deploy policy: if Cloud doesn't deploy weekends, Friday's deploy is the
+last train before dogfood. §5 marks which guardrails are Stage-A-blocking vs fast-follow so
+a compressed schedule sheds the right load. Everyone is on the immediate path; the sweep
+runs and logs `"due":0,"enqueued":0`.
+
+**Stage B — Internal dogfood (Fri Jul 10 evening – Tue Jul 14).**
+Enable the flag on Chatwoot's own support account + staging via super-admin
+(`chatwoot_internal` keeps it invisible to self-hosted super-admins). The window spans a
+weekend — compensate: staging rules at 10-minute delays to compress many cycles Sat–Sun,
+plus one realistic 24h rule on the production support account; name who watches the first
+sweep ticks Saturday morning. Go/no-go **Tue Jul 14** requires at least one business day
+(Mon) of real traffic and: fire-lag p95 < 10 min, `expired` = 0, zero unexpected Sentry
+groups, every skip row's `skip_reason` explainable, **and Story 2's cancellation path
+observed against real traffic** (customer replied → follow-up cancelled).
+
+**Stage C — Canary (Wed Jul 15 — the release commitment).**
+Enable for a hand-picked cohort: the CW-7513 requesting customer, the accounts attached to
+CW-5790's three customer conversations, ~10–20 design partners. **White-glove the first
+one**: on Jul 15, set up the customer's three rules together with them, so the one-week
+canary clock starts on day one instead of waiting for self-serve discovery. Mechanism:
+super-admin toggle or `account.enable_features!('delayed_automations')`. CW-7513 is
+"released" at this stage. Delayed `send_message` rules stay within dogfood + canary
+accounts until Stage D (highest-blast-radius action; messages can't be unsent).
+
+**Degenerate paths (decided now, not during the incident):**
+- PR 2 (UI) slips → Jul 15 is still met: the feature is API-complete after PR 1; enable the
+  customer's flag and create their three rules via console/API.
+- PR 1 slips past Mon Jul 13 → customer comms + Stage C moves to Jul 16–17. The commitment
+  is the capability in the customer's hands, not a specific artifact.
+
+**Stage D — Cloud GA (earliest Jul 22; gated, not dated).**
+Entry gate: ≥5 canary accounts each with ≥1 delayed rule that completed ≥1 full execution
+cycle, and Phase 1 exit metrics holding. Waves ordered by *real* signal (an inert-flag wave
+proves nothing — the flag only unlocks configuration):
+1. **Wave 1**: accounts with ≥1 active automation rule, batched ascending by conversation
+   volume — the population that will actually create delayed rules, arriving gradually.
+2. **Wave 2**: high-volume automation accounts, after Wave 1 metrics hold for 24–48h.
+3. Zero-rule accounts get the flag via the new-account default flip below — bulk-enabling
+   them separately is ceremony.
+Pause criterion between batches: any `expired > 0` sustained or new Sentry group within 24h
+of a batch. New rake task carries the specified safety kit (dry-run default, `APPLY=true`,
+`LIMIT=n`, interactive confirmation). Simultaneously flip the new-account default:
+migration mutating `ACCOUNT_LEVEL_FEATURE_DEFAULTS['delayed_automations'].enabled = true` +
+`GlobalConfig.clear_cache` (assignment_v2 `20260409091202` precedent).
+
+**Stage E — Self-hosted GA (next tagged release after Stage D holds for 2 weeks).**
+In one PR: `enabled: true` in `features.yml` (new installs), drop `chatwoot_internal`
+(flag becomes visible in self-hosted super-admin), ship the Stage D config-flip + backfill
+migrations (existing installs get them on `db:migrate` — ConfigLoader alone never flips an
+existing install's defaults; the migration is mandatory, per the captain_tasks +
+assignment_v2 precedent pair). **Exit criteria, not follow-ups:**
+- Docs page: `DISABLE_DELAYED_AUTOMATIONS` (where to set, halts within ~5 min) and the
+  flag's exact three-layer semantics.
+- Upgrade note: "if you run a custom Sidekiq queue list, `scheduled_jobs` **and** `medium`
+  must be processed or delayed rules never fire" (stock `sidekiq.yml` is titled a *sample*
+  configuration).
+- Expectations paragraph: on strict-priority queues, low-concurrency installs will see
+  sweep lag; hours-scale delays are the design point.
+- The run-once-semantics user-guide article.
+
+**Stage F — Cleanup (2–3 weeks post-GA).**
+Keep the controller flag guard (self-hosted admins legitimately toggle account flags).
+Cleanup: retire the canary rake if one-off, close CW-7513, re-spec CW-5790 as Phase 2,
+record learnings + the Phase 1 baseline bands for Phase 2 gating.
+
+### 4.3 Rollback playbook
+
+| Symptom | Response | Blast radius |
+|---|---|---|
+| Sweep melting `medium` queue / runaway executions | Set `DISABLE_DELAYED_AUTOMATIONS` → sweep no-ops next tick **and** already-enqueued per-row jobs refuse to execute (switch is checked in both). Pending rows accumulate; the 3-day due-window caps replay on re-enable | Instance-wide stop; immediate path untouched |
+| One account's rules misbehaving | Disable the account's `delayed_automations` flag — per-account stop at all three layers (no new arming; armed rows skip with `flag_disabled`). Or deactivate the specific rule | Single account |
+| **Re-check bug: rule fires where it shouldn't** | (1) Kill switch. (2) Blast radius = executed rows for that rule in the window (`rule_id, conversation_id, updated_at` — this audit trail is why `skip_reason`/rows are kept 30 days). (3) Reversible actions (labels, assignments): remediation rake driven off that row list. (4) `send_message`: **irreversible** — comms template + a named support owner; this is why delayed sends stay dogfood/canary-only until Stage D | Bounded by the executed-rows list |
+| Bad migration / schema issue at Stage A | Standard revert; `execution_delay` is nullable with no readers when dark | Deploy-level |
+
+An explicit non-rollback: **never delete the flag or reorder features.yml** — bit positions
+are permanent (header contract in features.yml).
+
+## 5. Engine guardrails (folded into the implementation plan)
+
+> These originated as review amendments; the schema/scope/spec deltas are now **written into
+> `docs/delayed-automations-implementation-plan.md`** so the PRs are built from one source of
+> truth. This section keeps the rationale and the Stage-A-blocking split.
+
+**Stage-A-blocking** (the sweep is unsafe without them):
+- **Bounded due-window**: sweep selects `due_at: 3.days.ago..Time.current`
+  (campaign/snooze-reopen precedent); older rows → `skipped` / `skip_reason: expired`,
+  count logged — no silent truncation.
+- **Per-sweep cap**: constant on the job class with an InstallationConfig override
+  (Captain ScheduleSyncsJob pattern — gives a no-deploy tuning knob), default 1000;
+  overflow logs `capped: true` + remaining count; due rows stay due for the next tick.
+- **Claim transition**: `pending → processing` before enqueue (`Campaign#mark_processing!`
+  precedent) so a re-selected row can't double-fire; stale `processing` rows older than
+  15 min return to eligibility (mechanism per Captain's stale-claim recovery — its
+  `SYNC_STALE_TIMEOUT` is 2h; we choose 15 min to match the 5-min cadence).
+- **Kill switch checked in both jobs** (§4.1).
+- **Queue reality** (`config/sidekiq.yml` is strict-priority, no weights): per-row jobs on
+  `medium` (3rd) preempt `default` and below — the cap is what protects the instance;
+  `scheduled_jobs` (8th) sits *below* `low`, so the sweep itself can be late — `due_at <=
+  now` semantics already tolerate that. Note `WebhookJob` is **also** `queue_as :medium`,
+  so webhook-heavy delayed rules add to the same queue; covered by the cap.
+- Per-row error isolation: `discard_on ActiveRecord::RecordNotFound`; rescue →
+  `ChatwootExceptionTracker.new(e, account:).capture_exception`, continue.
+
+**Fast-follow tolerable** (days, not weeks; needed before Stage C):
+- `skip_reason` terminal values on every skip path (`rule_inactive | conversation_gone |
+  flag_disabled | episode_moved | conditions_changed | expired`).
+- Structured end-of-run summary log + dashboards (§6).
+
+**Unchanged from the implementation plan** (called out because review probed them):
+- Migration shapes: plain nullable no-default `add_column` on `conversations`; concurrent
+  indexes; retention pruning batched under the global 14s `statement_timeout`.
+- Email actions keep `within_email_rate_limit?` via ActionService reuse — verify in specs,
+  don't reimplement.
+
+## 6. Observability, metrics, alerts
+
+**Structured summary log per sweep**, emitted as JSON so New Relic ingests fields without a
+parsing rule:
+`[AutomationRules::TriggerPendingExecutionsJob] {"event":"completed","due":N,"enqueued":N,
+"capped":false,"expired":N,"reclaimed":N,"duration_ms":N}` — plus per-row terminal
+`skip_reason` stored on the row (the support-facing answer to "why didn't my rule fire"
+until Phase 2's history UI).
+
+**Alerts are pre-Stage-A tasks with owners, or they don't exist.** Each needs its NRQL/Sentry
+rule created, a notification channel, and a named acknowledger for Jul 10–22 (§7 decision):
+
+| Alert | Mechanism | Owner / channel |
+|---|---|---|
+| Sweep summary absent > 15 min | NR loss-of-signal on `event:'completed'` from the job class | *fill at §7 sign-off* |
+| `expired > 0` sustained (2+ ticks) | NRQL on the JSON field |〃 |
+| Pending-row count growing across 6 ticks | NRQL on `due` | 〃 |
+| New Sentry group under the two job classes | Sentry issue alert filtered by job-class tag | 〃 |
+
+Stage A exit includes a live end-to-end test of each alert (test trigger → channel).
+
+**Dashboards** (New Relic, existing log forwarding + newrelic-sidekiq-metrics):
+fire lag (`executed_at - due_at` p50/p95; target p95 < 10 min) · outcome mix by
+`skip_reason` (high `conditions_changed` is *healthy* — cancelled follow-ups working; any
+`expired` means starvation) · pending count + oldest-pending age · `medium` and
+`scheduled_jobs` queue latency.
+
+**Numeric gates** (replacing fuzzy language): Stage B go/no-go and Stage D pause criteria in
+§4.2; the week-1 canary executed:skipped mix is *recorded as the baseline band* — later
+deviations from that band, not an arbitrary number, are the anomaly signal.
+
+**Success metrics (product):** accounts with ≥1 delayed rule (adoption), delayed executions
+per day, executed:skipped ratio vs baseline, and the CW-7513 customer's three scenarios
+confirmed by them.
+
+## 7. Decisions needed (owners; lock before Stage A — i.e., today/tomorrow)
+
+1. **Flag name sign-off** (eng, Tanmay): `delayed_automations` on `feature_flags_ext_1` —
+   ~~resolved~~: the ext column shipped on develop (#14947), so this is a plain features.yml
+   append; the name/position freezes on merge. Non-blocking beyond PR review.
+2. **Premium or not** (product, Pranav/Sony): recommendation **non-premium** — base
+   `automations` is free, engine cost is self-limiting (no delayed rules → zero rows), and
+   Phases 2–3 carry natural premium hooks (business-hours accounting alongside SLA, SLA
+   escalation). Plan-gating later is one `ReconcilePlanFeaturesService` change.
+3. **Named reviewers + review SLA for PR 1/2** (eng leads): Friday EOD merge deadline makes
+   this load-bearing.
+4. **Alert ownership Jul 10–22** (eng/on-call): who receives and acks §6's four alerts.
+5. **Weekend deploy policy** (infra): confirms whether Stage B starts Friday evening or
+   Monday.
+6. **Delay bounds** (product): 10 min–30 days per the implementation plan — confirm.
+7. **Canary cohort** (product/support): CW-7513 customer + CW-5790 attached accounts +
+   design partners; confirm the white-glove session on Jul 15.
+8. **Phase 2 business-hours placement** (product, at Phase 2 spec): OSS gate + premium
+   accounting, or all-OSS. No schema impact either way.
+
+## 8. Timeline
+
+| Date | Milestone |
+|---|---|
+| Thu Jul 9 | This plan signed off; §7 decisions 1–5 locked; PR 1 (engine+flag, dark) in review |
+| Fri Jul 10 | PR 2 (UI behind flag); **both merged EOD** (named-reviewer SLA); deploy; Stage B flag on for internal accounts |
+| Sat–Sun Jul 11–12 | Slack for merge slip; staging compressed-delay cycles running; Saturday-morning sweep watch (named) |
+| Mon Jul 13 | First full business day of dogfood signal |
+| Tue Jul 14 | Go/no-go on §4.2 Stage B gates |
+| **Wed Jul 15** | **Stage C canary — CW-7513 delivered** (white-glove setup with the requesting customer; degenerate path: console-created rules if PR 2 slipped) |
+| Jul 22+ | Stage D Cloud GA — earliest date, entry-gated on canary evidence; waves per §4.2 + new-account default flip |
+| Next tagged release | Stage E self-hosted GA (checklist in §4.2 are exit criteria) |
+| +2–3 weeks | Stage F cleanup; Phase 2 spec kickoff (re-scope CW-5790; carries the arming-scan decision + business-hours placement) |
+| ~Sep | Phase 2 ship (time vocabulary + business hours + run history) |
+| Q4 2026 | Phase 3 (chains + SLA escalation) |
+| 2027 | Phase 4 spec (scheduled triggers / builder) |
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..d9026be29 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,61 @@ 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
+    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..8298991ae
--- /dev/null
+++ b/spec/factories/automation_rule_pending_executions.rb
@@ -0,0 +1,10 @@
+FactoryBot.define do
+  factory :automation_rule_pending_execution do
+    account
+    automation_rule { association :automation_rule, account: account }
+    conversation { association :conversation, account: account }
+    episode_key { "status:#{Time.current.to_i}" }
+    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..01fedfc05
--- /dev/null
+++ b/spec/jobs/automation_rules/process_pending_execution_job_spec.rb
@@ -0,0 +1,119 @@
+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)
+    AutomationRulePendingExecution.last.tap { |row| row.update!(status: :processing) }
+  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 'skips with flag_disabled when the account flag was turned off' do
+    account.disable_features!('delayed_automations')
+    job.perform(pending_execution.reload)
+
+    expect(pending_execution.reload).to be_skipped
+    expect(pending_execution.skip_reason).to eq('flag_disabled')
+  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 re-check no longer matches the edited rule' do
+    pending_execution
+    # Rule edited while pending: the re-check enforces the current conditions (by design).
+    rule.update!(conditions: [{ 'values' => ['open'], 'attribute_key' => 'status', 'query_operator' => nil,
+                                'filter_operator' => 'equal_to' }])
+    job.perform(pending_execution.reload)
+
+    expect(pending_execution.reload).to be_skipped
+    expect(pending_execution.skip_reason).to eq('conditions_changed')
+  end
+
+  it 'reverts the row to pending 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 'leaves the row processing and reports the error when an action blows up' do
+    allow(AutomationRules::ActionService).to receive(:new).and_raise(StandardError, 'boom')
+    allow(ChatwootExceptionTracker).to receive(:new).and_call_original
+
+    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!(status: :processing) }
+
+    job.perform(row.reload)
+
+    expect(row.reload).to be_executed
+    expect(conversation.messages.outgoing.pluck(:content)).to include('Just checking in')
+  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!(status: :processing) }
+
+    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..a45c39561
--- /dev/null
+++ b/spec/jobs/automation_rules/trigger_pending_executions_job_spec.rb
@@ -0,0 +1,52 @@
+require 'rails_helper'
+
+RSpec.describe AutomationRules::TriggerPendingExecutionsJob do
+  subject(:job) { described_class.new }
+
+  let(:account) { create(:account) }
+  let(:conversation) { create(:conversation, account: account) }
+
+  before { GlobalConfig.clear_cache }
+
+  it 'enqueues per-row jobs for due pending rows and claims them' 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).with(due_row)
+    expect(due_row.reload).to be_processing
+    expect(future_row.reload).to be_pending
+  end
+
+  it 'expires rows past the due window instead of enqueuing them' do
+    overdue_row = create(:automation_rule_pending_execution, account: account, conversation: conversation, due_at: 4.days.ago)
+
+    expect { job.perform }.not_to have_enqueued_job(AutomationRules::ProcessPendingExecutionJob)
+    expect(overdue_row.reload).to be_skipped
+    expect(overdue_row.skip_reason).to eq('expired')
+  end
+
+  it 'reclaims stale processing rows so the next sweep retries them' do
+    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 and leaves overflow pending' do
+    create(:installation_config, name: 'AUTOMATION_PENDING_EXECUTIONS_SWEEP_LIMIT', serialized_value: { value: 1 }.with_indifferent_access)
+    create_list(:automation_rule_pending_execution, 2, account: account, due_at: 1.minute.ago)
+
+    expect { job.perform }.to have_enqueued_job(AutomationRules::ProcessPendingExecutionJob).exactly(:once)
+    expect(AutomationRulePendingExecution.pending.count).to eq(1)
+  end
+
+  it 'does nothing when the kill switch is set' do
+    create(:installation_config, name: 'DISABLE_DELAYED_AUTOMATIONS', serialized_value: { value: true }.with_indifferent_access)
+    GlobalConfig.clear_cache
+    due_row = create(:automation_rule_pending_execution, account: account, conversation: conversation, due_at: 1.minute.ago)
+
+    expect { job.perform }.not_to have_enqueued_job(AutomationRules::ProcessPendingExecutionJob)
+    expect(due_row.reload).to be_pending
+  end
+end
diff --git a/spec/listeners/automation_rule_listener_spec.rb b/spec/listeners/automation_rule_listener_spec.rb
index 08085da7a..4b723d9fa 100644
--- a/spec/listeners/automation_rule_listener_spec.rb
+++ b/spec/listeners/automation_rule_listener_spec.rb
@@ -247,4 +247,37 @@ describe AutomationRuleListener do
       end
     end
   end
+
+  describe 'delayed rules' do
+    let!(:automation_rule) { create(:automation_rule, event_name: 'conversation_updated', account: account, execution_delay: 60) }
+    let(:event) do
+      Events::Base.new('conversation_updated', Time.zone.now, { conversation: conversation, changed_attributes: {} })
+    end
+
+    before { allow(condition_match).to receive(:present?).and_return(true) }
+
+    context 'when the delayed_automations feature is enabled' do
+      before { account.enable_features!('delayed_automations') }
+
+      it 'records a pending execution instead of running actions' do
+        expect { listener.conversation_updated(event) }.to change(AutomationRulePendingExecution, :count).by(1)
+        expect(AutomationRules::ActionService).not_to have_received(:new)
+        expect(AutomationRulePendingExecution.last.due_at).to be_within(5.seconds).of(60.minutes.from_now)
+      end
+
+      it 'still runs rules without a delay immediately' do
+        automation_rule.update!(execution_delay: nil)
+
+        expect { listener.conversation_updated(event) }.not_to change(AutomationRulePendingExecution, :count)
+        expect(AutomationRules::ActionService).to have_received(:new).with(automation_rule, account, conversation)
+      end
+    end
+
+    context 'when the delayed_automations feature is disabled' do
+      it 'neither arms a pending execution nor falls back to immediate execution' do
+        expect { listener.conversation_updated(event) }.not_to change(AutomationRulePendingExecution, :count)
+        expect(AutomationRules::ActionService).not_to have_received(:new)
+      end
+    end
+  end
 end
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..96f102859
--- /dev/null
+++ b/spec/models/automation_rule_pending_execution_spec.rb
@@ -0,0 +1,169 @@
+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.to_f}")
+    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.to_f}")
+    end
+
+    it 'derives awaiting_agent episodes from waiting_since 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.to_i}")
+    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 '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)
+
+      second_reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
+      travel_to(30.minutes.from_now) do
+        described_class.schedule(rule: rule, conversation: conversation, message: second_reply)
+
+        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 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 'does not reset the clock for a repeated awaiting_agent episode' 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) }
+
+      expect(described_class.count).to eq(1)
+      expect(described_class.last.due_at).to be_within(1.second).of(original_due_at)
+    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 '#mark_processing!' do
+    let(:row) { create(:automation_rule_pending_execution, account: account, conversation: conversation) }
+
+    it 'claims a pending row exactly once' do
+      expect(row.mark_processing!).to be(true)
+      expect(row.reload).to be_processing
+      expect(row.mark_processing!).to be(false)
+    end
+
+    it 'does not claim executed rows' do
+      row.update!(status: :executed)
+      expect(row.mark_processing!).to be(false)
+    end
+  end
+
+  describe '.due / .expire_overdue! / .reclaim_stale!' do
+    it 'selects only pending rows inside the due window' do
+      due = create(:automation_rule_pending_execution, account: account, conversation: conversation, due_at: 1.minute.ago)
+      create(:automation_rule_pending_execution, account: account, due_at: 1.hour.from_now)
+      create(:automation_rule_pending_execution, account: account, due_at: 1.minute.ago, status: :executed)
+
+      expect(described_class.due).to eq([due])
+    end
+
+    it 'expires pending rows older than the due window' do
+      overdue = create(:automation_rule_pending_execution, account: account, conversation: conversation, due_at: 4.days.ago)
+      fresh = create(:automation_rule_pending_execution, account: account, due_at: 1.minute.ago)
+
+      expect(described_class.expire_overdue!).to eq(1)
+      expect(overdue.reload).to be_skipped
+      expect(overdue.skip_reason).to eq('expired')
+      expect(fresh.reload).to be_pending
+    end
+
+    it 'reclaims processing rows stuck longer than the stale timeout' do
+      stale = travel_to(20.minutes.ago) do
+        create(:automation_rule_pending_execution, account: account, conversation: conversation, status: :processing)
+      end
+      recent = create(:automation_rule_pending_execution, account: account, status: :processing)
+
+      expect(described_class.reclaim_stale!).to eq(1)
+      expect(stale.reload).to be_pending
+      expect(recent.reload).to be_processing
+    end
+  end
+end
diff --git a/spec/models/automation_rule_spec.rb b/spec/models/automation_rule_spec.rb
index 7ae73496c..86d611a0f 100644
--- a/spec/models/automation_rule_spec.rb
+++ b/spec/models/automation_rule_spec.rb
@@ -137,4 +137,42 @@ 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
+  end
 end
diff --git a/spec/models/conversation_spec.rb b/spec/models/conversation_spec.rb
index 43bbab56f..d42c8200f 100644
--- a/spec/models/conversation_spec.rb
+++ b/spec/models/conversation_spec.rb
@@ -1228,4 +1228,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