diff --git a/app/jobs/automation_rules/process_pending_execution_job.rb b/app/jobs/automation_rules/process_pending_execution_job.rb index 1fab8e0c5..e01199fdf 100644 --- a/app/jobs/automation_rules/process_pending_execution_job.rb +++ b/app/jobs/automation_rules/process_pending_execution_job.rb @@ -37,7 +37,7 @@ class AutomationRules::ProcessPendingExecutionJob < ApplicationJob def behavioral_skip_reason(pending_execution) return 'episode_moved' unless pending_execution.episode_current? - return 'conditions_changed' unless conditions_still_match?(pending_execution) + return AutomationRulePendingExecution::CONDITIONS_CHANGED_SKIP unless conditions_still_match?(pending_execution) nil end diff --git a/app/models/automation_rule_pending_execution.rb b/app/models/automation_rule_pending_execution.rb index 455312f64..e811b7a28 100644 --- a/app/models/automation_rule_pending_execution.rb +++ b/app/models/automation_rule_pending_execution.rb @@ -29,6 +29,9 @@ class AutomationRulePendingExecution < ApplicationRecord STALE_PROCESSING_TIMEOUT = 15.minutes # Terminal rows are purged after this to keep the table bounded. RETENTION_WINDOW = 30.days + # Skip reason for a row cancelled only because conditions no longer matched at fire time; unlike + # other terminal reasons, a later qualifying message can re-arm it (see .schedule). + CONDITIONS_CHANGED_SKIP = 'conditions_changed'.freeze belongs_to :automation_rule belongs_to :conversation @@ -57,16 +60,28 @@ class AutomationRulePendingExecution < ApplicationRecord message_id: message&.id, episode_key: key, due_at: rule.execution_delay.minutes.since(anchor) ) rescue ActiveRecord::RecordNotUnique - # Episode already armed. Reply-chase tracks the latest agent reply, so its clock moves; - # status / awaiting-agent episodes keep the first clock. A terminal row is never re-armed. - return unless message && !message.incoming? + rearm_or_advance_episode(rule, conversation, key, message, anchor) + end + + # The episode is already armed. Status episodes keep their first clock (a status change would + # give a new key), so only message episodes advance or re-arm here. + def self.rearm_or_advance_episode(rule, conversation, key, message, anchor) + return unless message row = find_by!(automation_rule_id: rule.id, conversation_id: conversation.id, episode_key: key) - # Jobs can arrive out of order; only a newer reply moves the clock, so a late older reply - # can't pull due_at backwards and fire before the delay elapses since the latest reply. - return unless row.pending? && message.id > row.message_id + # Jobs can arrive out of order; only a strictly newer message advances or re-arms, so a late + # older message can't pull due_at backwards and fire before the delay elapses. + return unless message.id > row.message_id - row.update!(due_at: rule.execution_delay.minutes.since(anchor), message_id: message.id) + due_at = rule.execution_delay.minutes.since(anchor) + if row.condition_skipped? + # A message episode key can recur (no new incoming reply) while conditions swing back into + # match, so a later qualifying message re-arms the condition-only skip instead of dropping. + row.update!(status: :pending, skip_reason: nil, due_at: due_at, message_id: message.id) + elsif row.pending? && !message.incoming? + # Reply-chase tracks the latest agent reply; awaiting-agent keeps its first clock. + row.update!(due_at: due_at, message_id: message.id) + end end # The wait is measured from when the qualifying event happened, not when this (possibly @@ -144,6 +159,10 @@ class AutomationRulePendingExecution < ApplicationRecord self.class.episode_key_for(conversation, message) == episode_key end + def condition_skipped? + skipped? && skip_reason == CONDITIONS_CHANGED_SKIP + end + private def claimable?