Compare commits

..
Author SHA1 Message Date
aakashb95 5cdd2bc14c fix(captain): mark discarded generations 2026-07-22 23:55:13 +05:30
aakashb95 8c6b70670c fix(captain): discard stale v2 responses 2026-07-22 23:21:16 +05:30
Shivam MishraandGitHub ddb0535a93 perf: reuse resolved count for reopen rate (#15122)
This improves the Captain overview by loading reporting metrics and FAQ
stats from separate endpoints. Range changes now refresh only the
metrics, while reopen-rate calculation reuses the resolved conversation
count to avoid redundant database queries.

## What changed

- Split Captain overview metrics and FAQ stats into separate APIs.
- Fetch FAQ stats independently from range-based metrics.
- Reuse resolved conversation totals when calculating reopen rate.
- Skip the reopen query when there are no resolved conversations.
2026-07-22 22:03:25 +05:30
Sivin VargheseandGitHub 42cbf7d3b9 fix: stray backslash after hard breaks before formatted list items (#15112) 2026-07-22 20:07:00 +05:30
887897ea98 fix: lock agent quota checks (#15029)
# Pull Request Template

## Description

Locks the agent quota check to the account row while creating account
users. This fixes a race where concurrent agent-create requests could
all observe the same remaining seat before any `account_users` row was
inserted.

The API continues to return the existing `402 Account limit exceeded.
Please purchase more licenses` response when the limit is reached. Bulk
create now preflights the requested email count while holding the
account lock, then creates each agent through the same locked builder
path. The Enterprise custom-role hook now no-ops when create did not
produce an agent.

Fixes:
[CW-7039](https://linear.app/chatwoot/issue/CW-7039/race-condition-in-agent-creation-bypasses-plan-agent-seat-limit)

## Type of change

- [x] Bug fix (non-breaking change which fixes an issue)

## How Has This Been Tested?

- `POSTGRES_DATABASE=chatwoot_test_c20f_agent_quota REDIS_DB=9 bundle
exec rspec spec/builders/agent_builder_spec.rb
spec/enterprise/builders/agent_builder_spec.rb
spec/controllers/api/v1/accounts/agents_controller_spec.rb
spec/enterprise/controllers/api/v1/accounts/agents_controller_spec.rb
spec/enterprise/controllers/enterprise/api/v1/accounts/agents_controller_spec.rb`
- `bundle exec rubocop app/builders/agent_builder.rb
app/controllers/api/v1/accounts/agents_controller.rb
enterprise/app/controllers/enterprise/api/v1/accounts/agents_controller.rb
spec/builders/agent_builder_spec.rb
spec/enterprise/controllers/api/v1/accounts/agents_controller_spec.rb`
- `git diff --check`
- One-off threaded Rails validation with 8 concurrent `AgentBuilder`
calls against an account with one remaining seat: `created: 1`,
`limited: 7`, final `count=2`, `limit=2`.

## Checklist:

- [x] My code follows the style guidelines of this project
- [x] I have performed a self-review of my code
- [x] I have commented on my code, particularly in hard-to-understand
areas
- [ ] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes
- [ ] Any dependent changes have been merged and published in downstream
modules

Co-authored-by: Muhsin Keloth <muhsinkeramam@gmail.com>
2026-07-22 18:43:00 +05:30
8aee518149 fix(integrations): restrict Linear/Notion/Shopify hook deletion to admins (#15126)
Non-admin agents could delete an account's Linear, Notion, or Shopify
integration through the dedicated integration endpoints, which — unlike
the generic hooks endpoint — never checked the caller's role. This
restores the intended admin-only boundary for removing an integration.

## Closes
- https://linear.app/chatwoot/issue/CW-7383
- https://linear.app/chatwoot/issue/CW-7384
- https://linear.app/chatwoot/issue/CW-7189

## How to reproduce
As a non-admin **agent**, `DELETE
/api/v1/accounts/:id/integrations/{linear,notion,shopify}` returned
`200` and removed the account-wide integration. After this change it
returns `401` and the integration is preserved; administrators can still
remove it.

## What changed
- Route integration-hook deletion through `HookPolicy` (admin-only) via
a shared `Integrations::BaseController`, matching the generic hooks
controller.

Co-authored-by: Vishnu Narayanan <iamwishnu@gmail.com>
2026-07-22 17:53:25 +05:30
Vishnu NarayananandGitHub 5733b822e3 fix: guard widget email-transcript button against repeat clicks (#15095)
## Description

The widget email-transcript button (`ChatFooter.vue`) had no client-side
guard. Its visibility depends only on whether the contact has an email,
and the click handler fired a request on every click with no in-flight
lock, no disabled state, and no post-send handling. A user could
therefore trigger a large number of duplicate transcript emails from a
single conversation just by clicking repeatedly.

This adds a re-entry guard in the handler, disables the button while a
send is in flight, and applies a short cooldown (15s) after a successful
send. Normal use is unaffected: the button sends once, shows the success
toast, then briefly disables and automatically re-enables so a genuine
later re-request still works. On failure the button stays enabled so the
user can retry immediately. The cooldown timer is cleared on unmount.

Using a timed cooldown (rather than a permanent post-send lock) also
avoids the button getting stuck disabled if a resolved conversation is
reopened and later re-resolved.

This is the client-side complement to the server-side rate limit added
in #15085.

Fixes https://linear.app/chatwoot/issue/CW-7640
2026-07-22 17:36:36 +05:30
69 changed files with 624 additions and 1912 deletions
+19 -3
View File
@@ -2,6 +2,14 @@
# It initializes with necessary attributes and provides a perform method
# to create a user and account user in a transaction.
class AgentBuilder
LIMIT_EXCEEDED_MESSAGE = 'Account limit exceeded. Please purchase more licenses'.freeze
class LimitExceededError < StandardError
def initialize
super(AgentBuilder::LIMIT_EXCEEDED_MESSAGE)
end
end
# Initializes an AgentBuilder with necessary attributes.
# @param email [String] the email of the user.
# @param name [String] the name of the user.
@@ -14,15 +22,23 @@ class AgentBuilder
# Creates a user and account user in a transaction.
# @return [User] the created user.
def perform
ActiveRecord::Base.transaction do
@user = find_or_create_user
create_account_user
account.with_lock do
raise LimitExceededError unless can_add_agent?
ActiveRecord::Base.transaction do
@user = find_or_create_user
create_account_user
end
end
@user
end
private
def can_add_agent?
account.usage_limits[:agents] > account.account_users.count
end
# Finds a user by email or creates a new one with a temporary password.
# @return [User] the found or created user.
def find_or_create_user
@@ -1,8 +1,6 @@
class Api::V1::Accounts::AgentsController < Api::V1::Accounts::BaseController
before_action :fetch_agent, except: [:create, :index, :bulk_create]
before_action :check_authorization
before_action :validate_limit, only: [:create]
before_action :validate_limit_for_bulk_create, only: [:bulk_create]
def index
@agents = agents
@@ -20,6 +18,8 @@ class Api::V1::Accounts::AgentsController < Api::V1::Accounts::BaseController
)
@agent = builder.perform
rescue AgentBuilder::LimitExceededError => e
render_payment_required(e.message)
end
def update
@@ -36,25 +36,13 @@ class Api::V1::Accounts::AgentsController < Api::V1::Accounts::BaseController
def bulk_create
emails = params[:emails]
emails.each do |email|
builder = AgentBuilder.new(
email: email,
name: email.split('@').first,
inviter: current_user,
account: Current.account
)
begin
builder.perform
rescue ActiveRecord::RecordInvalid => e
Rails.logger.info "[Agent#bulk_create] ignoring email #{email}, errors: #{e.record.errors}"
end
end
bulk_create_agents(emails)
# This endpoint is used to bulk create agents during onboarding
# onboarding_step key in present in Current account custom attributes, since this is a one time operation
Current.account.custom_attributes.delete('onboarding_step')
Current.account.save!
clear_onboarding_step
head :ok
rescue AgentBuilder::LimitExceededError => e
render_payment_required(e.message)
end
private
@@ -87,22 +75,33 @@ class Api::V1::Accounts::AgentsController < Api::V1::Accounts::BaseController
@agents ||= Current.account.users.order_by_full_name.includes(:account_users, { avatar_attachment: [:blob] })
end
def validate_limit_for_bulk_create
limit_available = params[:emails].count <= available_agent_count
def bulk_create_agents(emails)
Current.account.with_lock do
raise AgentBuilder::LimitExceededError if emails.count > available_agent_count
render_payment_required('Account limit exceeded. Please purchase more licenses') unless limit_available
emails.each { |email| create_agent_from_email(email) }
end
end
def validate_limit
render_payment_required('Account limit exceeded. Please purchase more licenses') unless can_add_agent?
def create_agent_from_email(email)
builder = AgentBuilder.new(
email: email,
name: email.split('@').first,
inviter: current_user,
account: Current.account
)
builder.perform
rescue ActiveRecord::RecordInvalid => e
Rails.logger.info "[Agent#bulk_create] ignoring email #{email}, errors: #{e.record.errors}"
end
def clear_onboarding_step
Current.account.custom_attributes.delete('onboarding_step')
Current.account.save!
end
def available_agent_count
Current.account.usage_limits[:agents] - agents.count
end
def can_add_agent?
available_agent_count.positive?
Current.account.usage_limits[:agents] - Current.account.account_users.count
end
def delete_user_record(agent)
@@ -3,7 +3,6 @@ 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
@@ -49,9 +48,6 @@ class Api::V1::Accounts::AutomationRulesController < Api::V1::Accounts::BaseCont
def clone
automation_rule = Current.account.automation_rules.find_by(id: params[:automation_rule_id])
new_rule = automation_rule.dup
# dup copies execution_delay; drop it when the feature is off so clone can't create new
# delayed rules that create/update would reject.
new_rule.execution_delay = nil unless delayed_automations_enabled?
new_rule.save!
@automation_rule = new_rule
end
@@ -59,27 +55,13 @@ 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(
*permitted_attributes,
:name, :description, :event_name, :active,
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
@@ -0,0 +1,9 @@
class Api::V1::Accounts::Integrations::BaseController < Api::V1::Accounts::BaseController
private
# Managing an integration hook (create/update/destroy) is admin-only, enforced via HookPolicy.
# Subclasses opt in per action with `before_action :check_authorization, only: [...]`.
def check_authorization
authorize(:hook)
end
end
@@ -1,4 +1,4 @@
class Api::V1::Accounts::Integrations::HooksController < Api::V1::Accounts::BaseController
class Api::V1::Accounts::Integrations::HooksController < Api::V1::Accounts::Integrations::BaseController
before_action :fetch_hook, except: [:create]
before_action :check_authorization
@@ -35,10 +35,6 @@ class Api::V1::Accounts::Integrations::HooksController < Api::V1::Accounts::Base
@hook = Current.account.hooks.find(params[:id])
end
def check_authorization
authorize(:hook)
end
def permitted_params
params.require(:hook).permit(:app_id, :inbox_id, :status, settings: {})
end
@@ -1,6 +1,7 @@
class Api::V1::Accounts::Integrations::LinearController < Api::V1::Accounts::BaseController
class Api::V1::Accounts::Integrations::LinearController < Api::V1::Accounts::Integrations::BaseController
before_action :fetch_conversation, only: [:create_issue, :link_issue, :unlink_issue, :linked_issues]
before_action :fetch_hook, only: [:destroy]
before_action :check_authorization, only: [:destroy]
def destroy
revoke_linear_token
@@ -1,5 +1,6 @@
class Api::V1::Accounts::Integrations::NotionController < Api::V1::Accounts::BaseController
class Api::V1::Accounts::Integrations::NotionController < Api::V1::Accounts::Integrations::BaseController
before_action :fetch_hook, only: [:destroy]
before_action :check_authorization, only: [:destroy]
def destroy
@hook.destroy!
@@ -1,7 +1,8 @@
class Api::V1::Accounts::Integrations::ShopifyController < Api::V1::Accounts::BaseController
class Api::V1::Accounts::Integrations::ShopifyController < Api::V1::Accounts::Integrations::BaseController
include Shopify::IntegrationHelper
before_action :setup_shopify_context, only: [:orders]
before_action :fetch_hook, except: [:auth]
before_action :check_authorization, only: [:destroy]
before_action :validate_contact, only: [:orders]
def auth
@@ -26,13 +26,20 @@ class CaptainAssistant extends ApiClient {
});
}
getStats({ assistantId, range, signal }) {
getMetrics({ assistantId, range, signal }) {
const requestConfig = {
params: { range, timezone_offset: getTimezoneOffset() },
};
if (signal) requestConfig.signal = signal;
return axios.get(`${this.url}/${assistantId}/stats`, requestConfig);
return axios.get(`${this.url}/${assistantId}/metrics`, requestConfig);
}
getFaqStats({ assistantId, signal }) {
const requestConfig = {};
if (signal) requestConfig.signal = signal;
return axios.get(`${this.url}/${assistantId}/faq_stats`, requestConfig);
}
getSummary({ assistantId, range, stats }) {
@@ -8,6 +8,8 @@ import {
EditorState,
Selection,
imageResizeView,
toggleMark,
wrapInList,
} from '@chatwoot/prosemirror-schema';
import {
suggestionsPlugin,
@@ -17,8 +19,6 @@ import imagePastePlugin from '@chatwoot/prosemirror-schema/src/plugins/image';
import embedPreviewPlugin from '@chatwoot/prosemirror-schema/src/plugins/embedPreview';
import trailingParagraphPlugin from '@chatwoot/prosemirror-schema/src/plugins/trailingParagraph';
import { embeds as markdownEmbeds } from 'dashboard/helper/markdownEmbeds';
import { toggleMark } from 'prosemirror-commands';
import { wrapInList } from 'prosemirror-schema-list';
import { toggleBlockType } from '@chatwoot/prosemirror-schema/src/menu/common';
import { checkFileSizeLimit } from 'shared/helpers/FileHelper';
import { isEscape } from 'shared/helpers/KeyboardHelpers';
-1
View File
@@ -14,7 +14,6 @@ export const FEATURE_FLAGS = {
CRM: 'crm',
CUSTOM_ATTRIBUTES: 'custom_attributes',
DATA_IMPORT: 'data_import',
DELAYED_AUTOMATIONS: 'delayed_automations',
API_AND_WEBHOOKS: 'api_and_webhooks',
INBOX_MANAGEMENT: 'inbox_management',
INTEGRATIONS: 'integrations',
@@ -208,12 +208,6 @@ 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);
};
@@ -1,4 +1,6 @@
import {
InputRule,
inputRules,
MessageMarkdownSerializer,
MessageMarkdownTransformer,
messageSchema,
@@ -9,7 +11,6 @@ import * as Sentry from '@sentry/vue';
import camelcaseKeys from 'camelcase-keys';
import { FORMATTING, MARKDOWN_PATTERNS } from 'dashboard/constants/editor';
import { INBOX_TYPES, TWILIO_CHANNEL_MEDIUM } from 'dashboard/helper/inbox';
import { InputRule, inputRules } from 'prosemirror-inputrules';
/**
* Extract text from markdown, and remove all images, code blocks, links, headers, bold, italic, lists etc.
@@ -28,35 +28,6 @@
"PLACEHOLDER": "Please select one",
"ERROR": "Event is required"
},
"EXECUTE": {
"LABEL": "Delayed execution",
"AFTER_DELAY": "Run after",
"UNITS": {
"MINUTES": "Minutes",
"HOURS": "Hours",
"DAYS": "Days"
},
"ERROR": "Delay must be between 10 minutes and 30 days",
"ENDS_IF_LABEL": "Won't run if",
"ENDS_IF": {
"STATUS": "the conversation's status changes, or its conditions no longer match.",
"CUSTOMER_REPLY": "the customer replies, or the conditions no longer match.",
"AGENT_REPLY": "an agent replies, or the conditions no longer match.",
"GENERIC": "there is a new reply on the conversation, or the conditions no longer match."
},
"HELP_TEXT": "Only applies to conversations with activity after the rule is created."
},
"TRIGGER": {
"LABEL": "Trigger",
"WHEN_LABEL": "When",
"STATUS_LABEL": "Status is",
"INBOX_LABEL": "Inbox",
"OPTIONS": {
"CUSTOMER_UNRESPONSIVE": "Customer unresponsive",
"AGENT_UNRESPONSIVE": "Teammate unresponsive",
"CONVERSATION_STATUS": "Conversation in a status"
}
},
"CONDITIONS": {
"LABEL": "Conditions"
},
@@ -78,13 +49,7 @@
"CREATED_ON": "Created on",
"ACTIONS": "Actions"
},
"404": "No automation rules found",
"SECTIONS": {
"INSTANT": "Automations",
"DELAYED": "Delayed execution"
},
"DELAY_BADGE": "Runs after {delay}",
"DELAY_DISABLED_BANNER": "Delayed execution is disabled for this account. Delayed rules won't run until it is enabled again."
"404": "No automation rules found"
},
"DELETE": {
"TITLE": "Delete Automation Rule",
@@ -26,25 +26,28 @@ const canDrilldown = computed(() => checkPermissions(['administrator']));
const selectedRange = ref('this_month');
const assistantId = computed(() => route.params.assistantId);
const stats = ref(null);
const isFetching = ref(false);
const metricStats = ref(null);
const faqStats = ref(null);
const isFetchingMetrics = ref(false);
// Increments on every fetch so a response (or retry) from a superseded
// range/assistant can't clobber the latest request's state.
let fetchToken = 0;
let abortController = null;
let metricsFetchToken = 0;
let faqStatsFetchToken = 0;
let metricsAbortController = null;
let faqStatsAbortController = null;
const fetchStats = async () => {
fetchToken += 1;
const token = fetchToken;
abortController?.abort();
abortController = new AbortController();
const { signal } = abortController;
stats.value = null;
isFetching.value = true;
const fetchMetrics = async () => {
metricsFetchToken += 1;
const token = metricsFetchToken;
metricsAbortController?.abort();
metricsAbortController = new AbortController();
const { signal } = metricsAbortController;
metricStats.value = null;
isFetchingMetrics.value = true;
const requestStats = () =>
CaptainAssistant.getStats({
const requestMetrics = () =>
CaptainAssistant.getMetrics({
assistantId: assistantId.value,
range: selectedRange.value,
signal,
@@ -52,25 +55,54 @@ const fetchStats = async () => {
let data = null;
try {
({ data } = await requestStats());
({ data } = await requestMetrics());
} catch {
// One silent retry before giving up, unless the request was aborted.
try {
if (token === fetchToken && !signal.aborted)
({ data } = await requestStats());
if (token === metricsFetchToken && !signal.aborted)
({ data } = await requestMetrics());
} catch {
data = null;
}
}
if (token !== fetchToken || signal.aborted) return;
stats.value = data;
isFetching.value = false;
if (token !== metricsFetchToken || signal.aborted) return;
metricStats.value = data;
isFetchingMetrics.value = false;
};
onUnmounted(() => abortController?.abort());
const fetchFaqStats = async () => {
faqStatsFetchToken += 1;
const token = faqStatsFetchToken;
faqStatsAbortController?.abort();
faqStatsAbortController = new AbortController();
const { signal } = faqStatsAbortController;
faqStats.value = null;
watch([selectedRange, assistantId], fetchStats, { immediate: true });
try {
const { data } = await CaptainAssistant.getFaqStats({
assistantId: assistantId.value,
signal,
});
if (token === faqStatsFetchToken && !signal.aborted) faqStats.value = data;
} catch {
if (token === faqStatsFetchToken && !signal.aborted) faqStats.value = null;
}
};
const summaryStats = computed(() => {
if (!metricStats.value || !faqStats.value) return null;
return { ...metricStats.value, knowledge: faqStats.value };
});
onUnmounted(() => {
metricsAbortController?.abort();
faqStatsAbortController?.abort();
});
watch([selectedRange, assistantId], fetchMetrics, { immediate: true });
watch(assistantId, fetchFaqStats, { immediate: true });
// `direction` says whether a rising trend is good ('up'), bad ('down'), or
// neutral, so we can colour the delta independently of its sign.
@@ -90,7 +122,7 @@ const formatDuration = hours =>
hours >= 100 ? `${Math.round(hours / 24)}d` : `${hours}h`;
const metricFor = (statKey, formatValue, direction, trendKind = 'percent') => {
const data = stats.value?.[statKey];
const data = metricStats.value?.[statKey];
if (!data) return { value: '—', trend: '', trendGood: null };
const sign = data.trend > 0 ? '+' : '';
@@ -184,9 +216,9 @@ const closeDrilldown = () => {
<div class="flex flex-col gap-6 pb-8">
<InboxBanner />
<CoverageBanner :knowledge="stats?.knowledge" />
<CoverageBanner :knowledge="faqStats ?? undefined" />
<WelcomeCard :range="selectedRange" :stats="stats" />
<WelcomeCard :range="selectedRange" :stats="summaryStats" />
<div
class="grid grid-cols-1 gap-px overflow-hidden border rounded-xl sm:grid-cols-2 lg:grid-cols-3 bg-n-weak border-n-weak"
@@ -199,13 +231,15 @@ const closeDrilldown = () => {
:trend="metric.trend"
:hint="metric.hint"
:trend-good="metric.trendGood"
:loading="isFetching"
:clickable="canDrilldown && Boolean(metric.metric) && !isFetching"
:loading="isFetchingMetrics"
:clickable="
canDrilldown && Boolean(metric.metric) && !isFetchingMetrics
"
@click="openDrilldown(metric)"
/>
</div>
<KnowledgeCard :knowledge="stats?.knowledge" />
<KnowledgeCard :knowledge="faqStats ?? undefined" />
<QuickLinks />
</div>
@@ -10,7 +10,6 @@ const START_VALUE = {
name: null,
description: null,
event_name: 'conversation_created',
execution_delay: null,
conditions: [
{
attribute_key: 'status',
@@ -6,11 +6,6 @@ import { useOperators } from 'dashboard/components-next/filter/operators';
import ConditionRow from 'dashboard/components-next/filter/ConditionRow.vue';
import AutomationActionInput from 'dashboard/components/widgets/AutomationActionInput.vue';
import NextButton from 'dashboard/components-next/button/Button.vue';
import ToggleSwitch from 'dashboard/components-next/switch/Switch.vue';
import FilterSelect from 'dashboard/components-next/filter/inputs/FilterSelect.vue';
import MultiSelect from 'dashboard/components-next/filter/inputs/MultiSelect.vue';
import DurationInput from 'dashboard/components-next/input/DurationInput.vue';
import { DURATION_UNITS } from 'dashboard/components-next/input/constants';
import Dialog from 'dashboard/components-next/dialog/Dialog.vue';
import {
generateAutomationPayload,
@@ -19,7 +14,6 @@ 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({
@@ -77,28 +71,6 @@ const INPUT_TYPE_MAP = {
date: 'date',
};
const DEFAULT_DELAY_MINUTES = 240; // 4 hours
const MIN_DELAY_MINUTES = 10;
const MAX_DELAY_MINUTES = 43200; // 30 days
// A delayed rule is expressed as one meaningful trigger instead of a raw event + conditions. Each
// trigger maps to the automation's event_name plus a preset condition: message_type for the two
// unresponsive cases (reply-chase / awaiting-agent), or a chosen status for conversation_updated.
const DELAYED_TRIGGERS = [
{ key: 'conversation_status', eventName: 'conversation_updated' },
{
key: 'customer_unresponsive',
eventName: 'message_created',
messageType: 'outgoing',
},
{
key: 'agent_unresponsive',
eventName: 'message_created',
messageType: 'incoming',
},
];
const DEFAULT_TRIGGER = DELAYED_TRIGGERS[0].key;
const DEFAULT_TRIGGER_STATUS = 'pending';
const { t } = useI18n();
const { isCloudFeatureEnabled } = useAccount();
const { operators } = useOperators();
@@ -106,7 +78,6 @@ const { operators } = useOperators();
const dialogRef = ref(null);
const conditionsRef = useTemplateRef('conditionsRef');
const errors = ref({});
const isDelayed = ref(false);
const isEditMode = computed(() => props.mode === 'edit');
@@ -213,167 +184,6 @@ const hasActionErrors = computed(() =>
Object.keys(errors.value).some(key => key.startsWith('action_'))
);
const allowsDelayedExecution = computed(() =>
isCloudFeatureEnabled(FEATURE_FLAGS.DELAYED_AUTOMATIONS)
);
// Trigger controls for the delayed flow. They own event_name + conditions while the wait is on.
// selectedTrigger / triggerStatus are plain values (FilterSelect); triggerInboxes is an array of
// { id, name } (MultiSelect) — empty means the rule applies to every inbox.
const selectedTrigger = ref(DEFAULT_TRIGGER);
const triggerStatus = ref(DEFAULT_TRIGGER_STATUS);
const triggerInboxes = ref([]);
const isStatusTrigger = computed(
() => selectedTrigger.value === 'conversation_status'
);
// FilterSelect expects { label, value }.
const statusSelectOptions = computed(() =>
(props.getConditionDropdownValues('status') || [])
.filter(option => option.id !== 'all')
.map(option => ({ value: option.id, label: option.name }))
);
const triggerSelectOptions = computed(() =>
DELAYED_TRIGGERS.map(trigger => ({
value: trigger.key,
label: t(
`AUTOMATION.ADD.FORM.TRIGGER.OPTIONS.${trigger.key.toUpperCase()}`
),
}))
);
// MultiSelect expects (and returns) { id, name } options.
const inboxOptions = computed(
() => props.getConditionDropdownValues('inbox_id') || []
);
// What ends the wait, mirroring the backend episode that arms the rule. Shown to the user so
// they can predict when the rule runs. Conversation rules key on status; message rules key on
// the reply that ends the wait (customer reply for outgoing, agent reply for incoming).
const waitEndsKey = computed(() => {
if (eventName.value !== 'message_created') return 'STATUS';
const messageType = (automation.value?.conditions || []).find(
condition => condition.attribute_key === 'message_type'
);
const raw = Array.isArray(messageType?.values)
? messageType.values[0]
: messageType?.values;
// Raw create-mode values are strings ('outgoing'); edit-mode values are option objects.
const value = raw && typeof raw === 'object' ? raw.id : raw;
if (value === 'outgoing') return 'CUSTOMER_REPLY';
if (value === 'incoming') return 'AGENT_REPLY';
return 'GENERIC';
});
// DurationInput holds the wait in minutes and clamps to [MIN, MAX]; the unit is display-only.
const delayMinutes = ref(DEFAULT_DELAY_MINUTES);
const delayUnit = ref(DURATION_UNITS.HOURS);
const executionDelayInvalid = computed(
() => isDelayed.value && !Number.isFinite(delayMinutes.value)
);
// Show the wait in the largest whole unit (240 min → 4 hours). Passed in by open() rather than
// read from `automation`, whose model prop only settles a tick later.
const syncDelayFromDelay = delay => {
isDelayed.value = Boolean(delay);
const minutes = delay || DEFAULT_DELAY_MINUTES;
if (minutes % 1440 === 0) delayUnit.value = DURATION_UNITS.DAYS;
else if (minutes % 60 === 0) delayUnit.value = DURATION_UNITS.HOURS;
else delayUnit.value = DURATION_UNITS.MINUTES;
delayMinutes.value = minutes;
};
watch([isDelayed, delayMinutes], () => {
if (!automation.value || !allowsDelayedExecution.value) return;
automation.value.execution_delay = isDelayed.value
? delayMinutes.value
: null;
});
const buildTriggerCondition = (attributeKey, values) => ({
attribute_key: attributeKey,
filter_operator: 'equal_to',
values,
query_operator: 'and',
custom_attribute_type: '',
});
// Write the selected trigger (plus optional inbox scope) onto the rule's event_name + conditions.
const applyDelayedTrigger = () => {
const trigger = DELAYED_TRIGGERS.find(
item => item.key === selectedTrigger.value
);
if (!automation.value || !trigger) return;
automation.value.event_name = trigger.eventName;
const conditions = [
trigger.messageType
? buildTriggerCondition('message_type', trigger.messageType)
: buildTriggerCondition('status', triggerStatus.value),
];
if (triggerInboxes.value.length) {
conditions.push(
buildTriggerCondition(
'inbox_id',
triggerInboxes.value.map(inbox => inbox.id)
)
);
}
automation.value.conditions = conditions;
};
// A single value is a raw string in create mode and an option object ({ id }) after edit-mode
// formatting; return its plain value either way.
const rawConditionValue = condition => {
const raw = Array.isArray(condition?.values)
? condition.values[0]
: condition?.values;
return raw && typeof raw === 'object' ? raw.id : raw;
};
// Populate the trigger controls from an existing delayed rule when editing.
const hydrateTriggerFromAutomation = () => {
const conditions = automation.value?.conditions || [];
const byKey = key => conditions.find(c => c.attribute_key === key);
const messageType = rawConditionValue(byKey('message_type'));
if (messageType === 'incoming') selectedTrigger.value = 'agent_unresponsive';
else if (messageType === 'outgoing')
selectedTrigger.value = 'customer_unresponsive';
else {
selectedTrigger.value = 'conversation_status';
triggerStatus.value =
rawConditionValue(byKey('status')) || DEFAULT_TRIGGER_STATUS;
}
const inboxValues = byKey('inbox_id')?.values || [];
const inboxIds = inboxValues.map(value =>
value && typeof value === 'object' ? value.id : value
);
triggerInboxes.value = inboxOptions.value.filter(inbox =>
inboxIds.includes(inbox.id)
);
};
// Turning the wait on (create) sets the default trigger's event + conditions.
watch(isDelayed, delayed => {
if (delayed && automation.value && !isEditMode.value) applyDelayedTrigger();
});
// Any trigger-control change re-derives event_name + conditions. After hydration this simply
// re-writes the same values, so it stays idempotent (no reference change → no loop).
watch([selectedTrigger, triggerStatus, triggerInboxes], () => {
if (isDelayed.value) applyDelayedTrigger();
});
// Opening an existing delayed rule mirrors its event/conditions into the trigger controls.
watch(
() => automation.value,
() => {
if (isDelayed.value && automation.value) hydrateTriggerFromAutomation();
}
);
watch(
() => automation.value,
() => {
@@ -406,9 +216,8 @@ const syncCustomAttributeTypes = () => {
});
};
const open = (executionDelay = null) => {
const open = () => {
resetValidation();
syncDelayFromDelay(executionDelay);
dialogRef.value?.open();
};
@@ -421,13 +230,8 @@ 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);
}
};
@@ -462,163 +266,84 @@ defineExpose({ open, close });
:error="errors.description ? $t('AUTOMATION.ADD.FORM.DESC.ERROR') : ''"
:placeholder="$t('AUTOMATION.ADD.FORM.DESC.PLACEHOLDER')"
/>
<!-- Wait Start (choose the delay first, then the trigger) -->
<div v-if="allowsDelayedExecution" class="mb-6">
<div class="flex items-center justify-between gap-4">
<label class="mb-0" :class="{ error: errors.execution_delay }">
{{ $t('AUTOMATION.ADD.FORM.EXECUTE.LABEL') }}
</label>
<ToggleSwitch v-model="isDelayed" />
</div>
<div v-if="isDelayed" class="flex flex-wrap items-center gap-2 mt-2">
<span class="text-sm text-n-slate-11">
{{ $t('AUTOMATION.ADD.FORM.EXECUTE.AFTER_DELAY') }}
</span>
<div class="flex items-center gap-2 w-64">
<DurationInput
v-model="delayMinutes"
v-model:unit="delayUnit"
:min="MIN_DELAY_MINUTES"
:max="MAX_DELAY_MINUTES"
/>
</div>
</div>
<span
v-if="isDelayed && executionDelayInvalid"
class="text-xs text-n-ruby-9"
>
{{ $t('AUTOMATION.ADD.FORM.EXECUTE.ERROR') }}
</span>
</div>
<!-- Wait End -->
<!-- Delayed trigger: a curated event + condition, in place of raw Event/Conditions -->
<div v-if="isDelayed" class="mb-6">
<label class="mb-1">
{{ $t('AUTOMATION.ADD.FORM.TRIGGER.LABEL') }}
</label>
<div
class="flex flex-col gap-3 p-4 outline outline-1 -outline-offset-1 rounded-xl outline-n-weak dark:outline-n-strong"
>
<div class="flex items-center gap-3 min-h-8">
<span class="w-20 shrink-0 text-sm text-n-slate-11">
{{ $t('AUTOMATION.ADD.FORM.TRIGGER.WHEN_LABEL') }}
</span>
<FilterSelect
v-model="selectedTrigger"
:options="triggerSelectOptions"
/>
</div>
<div v-if="isStatusTrigger" class="flex items-center gap-3 min-h-8">
<span class="w-20 shrink-0 text-sm text-n-slate-11">
{{ $t('AUTOMATION.ADD.FORM.TRIGGER.STATUS_LABEL') }}
</span>
<FilterSelect
v-model="triggerStatus"
:options="statusSelectOptions"
/>
</div>
<div class="flex items-center gap-3 min-h-8">
<span class="w-20 shrink-0 text-sm text-n-slate-11">
{{ $t('AUTOMATION.ADD.FORM.TRIGGER.INBOX_LABEL') }}
</span>
<MultiSelect v-model="triggerInboxes" :options="inboxOptions" />
</div>
</div>
<p class="text-xs text-n-slate-11 pt-2 mb-0">
<span class="text-n-slate-12 font-medium">
{{ $t('AUTOMATION.ADD.FORM.EXECUTE.ENDS_IF_LABEL') }}
</span>
{{ $t(`AUTOMATION.ADD.FORM.EXECUTE.ENDS_IF.${waitEndsKey}`) }}
</p>
<p class="text-xs text-n-slate-11 pt-1 mb-0">
{{ $t('AUTOMATION.ADD.FORM.EXECUTE.HELP_TEXT') }}
</p>
</div>
<!-- Instant flow: raw Event + Conditions -->
<template v-else>
<div class="mb-6">
<label :class="{ error: errors.event_name }">
{{ $t('AUTOMATION.ADD.FORM.EVENT.LABEL') }}
<select
v-model="automation.event_name"
class="m-0"
@change="onEventChange()"
<div class="mb-6">
<label :class="{ error: errors.event_name }">
{{ $t('AUTOMATION.ADD.FORM.EVENT.LABEL') }}
<select
v-model="automation.event_name"
class="m-0"
@change="onEventChange()"
>
<option
v-for="event in automationRuleEvents"
:key="event.key"
:value="event.key"
>
<option
v-for="event in automationRuleEvents"
:key="event.key"
:value="event.key"
>
{{ event.value }}
</option>
</select>
<span v-if="errors.event_name" class="message">
{{ $t('AUTOMATION.ADD.FORM.EVENT.ERROR') }}
</span>
</label>
<p
v-if="!isEditMode && hasAutomationMutated"
class="text-xs text-right text-n-teal-10 pt-1"
>
{{ $t('AUTOMATION.FORM.RESET_MESSAGE') }}
</p>
</div>
<!-- Conditions Start -->
<section class="mb-5">
<label>
{{ $t('AUTOMATION.ADD.FORM.CONDITIONS.LABEL') }}
</label>
<ul
class="grid gap-4 list-none p-3 mb-4 outline outline-1 rounded-xl -outline-offset-1"
:class="
hasConditionErrors
? 'outline-n-ruby-5 bg-n-ruby-2/50'
: 'outline-n-weak dark:outline-n-strong'
"
>
<template v-for="(condition, i) in automation.conditions" :key="i">
<ConditionRow
v-if="i === 0"
ref="conditionsRef"
v-model:attribute-key="automation.conditions[i].attribute_key"
v-model:filter-operator="
automation.conditions[i].filter_operator
"
v-model:values="automation.conditions[i].values"
:filter-types="filterTypes"
:show-query-operator="false"
@remove="removeFilter(i)"
/>
<ConditionRow
v-else
ref="conditionsRef"
v-model:attribute-key="automation.conditions[i].attribute_key"
v-model:filter-operator="
automation.conditions[i].filter_operator
"
v-model:query-operator="
automation.conditions[i - 1].query_operator
"
v-model:values="automation.conditions[i].values"
:filter-types="filterTypes"
show-query-operator
@remove="removeFilter(i)"
/>
</template>
<div>
<NextButton
icon="i-lucide-plus"
blue
faded
sm
:label="$t('AUTOMATION.ADD.CONDITION_BUTTON_LABEL')"
@click="appendNewCondition"
/>
</div>
</ul>
</section>
<!-- Conditions End -->
</template>
{{ event.value }}
</option>
</select>
<span v-if="errors.event_name" class="message">
{{ $t('AUTOMATION.ADD.FORM.EVENT.ERROR') }}
</span>
</label>
<p
v-if="!isEditMode && hasAutomationMutated"
class="text-xs text-right text-n-teal-10 pt-1"
>
{{ $t('AUTOMATION.FORM.RESET_MESSAGE') }}
</p>
</div>
<!-- Conditions Start -->
<section class="mb-5">
<label>
{{ $t('AUTOMATION.ADD.FORM.CONDITIONS.LABEL') }}
</label>
<ul
class="grid gap-4 list-none p-3 mb-4 outline outline-1 rounded-xl -outline-offset-1"
:class="
hasConditionErrors
? 'outline-n-ruby-5 bg-n-ruby-2/50'
: 'outline-n-weak dark:outline-n-strong'
"
>
<template v-for="(condition, i) in automation.conditions" :key="i">
<ConditionRow
v-if="i === 0"
ref="conditionsRef"
v-model:attribute-key="automation.conditions[i].attribute_key"
v-model:filter-operator="automation.conditions[i].filter_operator"
v-model:values="automation.conditions[i].values"
:filter-types="filterTypes"
:show-query-operator="false"
@remove="removeFilter(i)"
/>
<ConditionRow
v-else
ref="conditionsRef"
v-model:attribute-key="automation.conditions[i].attribute_key"
v-model:filter-operator="automation.conditions[i].filter_operator"
v-model:query-operator="
automation.conditions[i - 1].query_operator
"
v-model:values="automation.conditions[i].values"
:filter-types="filterTypes"
show-query-operator
@remove="removeFilter(i)"
/>
</template>
<div>
<NextButton
icon="i-lucide-plus"
blue
faded
sm
:label="$t('AUTOMATION.ADD.CONDITION_BUTTON_LABEL')"
@click="appendNewCondition"
/>
</div>
</ul>
</section>
<!-- Conditions End -->
<!-- Actions Start -->
<section>
<label>
@@ -1,7 +1,6 @@
<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';
@@ -44,16 +43,6 @@ 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 }}
@@ -34,33 +34,29 @@ const {
const { formatAutomation } = useEditableAutomation();
const syncAutomationFromSelected = (source = props.selectedResponse) => {
if (!source?.conditions) return;
manifestCustomAttributes();
automation.value = formatAutomation(
source,
allCustomAttributes.value,
automationTypes,
AUTOMATION_ACTION_TYPES
);
};
// Format from the rule passed to open(): the prop updates a tick later, so at open() time
// automation still holds the previously selected rule (its execution_delay hydrates the form).
const open = rule => {
syncAutomationFromSelected(rule);
formRef.value?.open(rule?.execution_delay);
};
const open = () => formRef.value?.open();
const close = () => formRef.value?.close();
const onSave = (payload, mode) => {
emit('saveAutomation', payload, mode);
};
watch(() => props.selectedResponse, syncAutomationFromSelected, {
immediate: true,
});
watch(
() => props.selectedResponse,
value => {
if (!value?.conditions) return;
manifestCustomAttributes();
automation.value = formatAutomation(
value,
allCustomAttributes.value,
automationTypes,
AUTOMATION_ACTION_TYPES
);
},
{ immediate: true }
);
defineExpose({ open, close });
</script>
@@ -35,31 +35,6 @@ const filteredRecords = computed(() => {
if (!query) return records.value;
return picoSearch(records.value, query, ['name', 'description']);
});
// Delayed (wait) rules run on a different lifecycle, so list them in their own section.
const hasDelayedRecords = computed(() =>
records.value.some(automation => automation.execution_delay)
);
const sections = computed(() => {
const instant = [];
const delayed = [];
filteredRecords.value.forEach(automation =>
(automation.execution_delay ? delayed : instant).push(automation)
);
return [
{
key: 'instant',
label: t('AUTOMATION.LIST.SECTIONS.INSTANT'),
items: instant,
},
{
key: 'delayed',
label: t('AUTOMATION.LIST.SECTIONS.DELAYED'),
items: delayed,
},
].filter(section => section.items.length);
});
const uiFlags = computed(() => getters['automations/getUIFlags'].value);
const accountId = computed(() => getters.getCurrentAccountId.value);
@@ -77,14 +52,6 @@ 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');
@@ -107,7 +74,7 @@ const hideAddPopup = () => {
const openEditPopup = response => {
selectedAutomation.value = { ...response };
editDialogRef.value?.open(response);
editDialogRef.value?.open();
};
const hideEditPopup = () => {
editDialogRef.value?.close();
@@ -161,11 +128,11 @@ const submitAutomation = async (payload, mode) => {
hideAddPopup();
hideEditPopup();
} catch (error) {
const fallbackMessage =
const errorMessage =
mode === 'edit'
? t('AUTOMATION.EDIT.API.ERROR_MESSAGE')
: t('AUTOMATION.ADD.API.ERROR_MESSAGE');
useAlert(error?.response?.data?.error || fallbackMessage);
useAlert(errorMessage);
}
};
const toggleAutomation = async ({ id, name, status }) => {
@@ -245,49 +212,25 @@ 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>
<template v-if="filteredRecords.length">
<div
v-for="section in sections"
:key="section.key"
class="mb-6 last:mb-0"
>
<h4
v-if="hasDelayedRecords"
class="mb-2 text-sm font-medium text-n-slate-11"
>
{{ section.label }}
</h4>
<BaseTable :headers="tableHeaders" :items="section.items">
<template #row="{ items }">
<AutomationRuleRow
v-for="automation in items"
:key="automation.id"
:automation="automation"
:loading="loading[automation.id]"
@clone="cloneAutomation"
@toggle="toggleAutomation"
@edit="openEditPopup"
@delete="openDeletePopup"
/>
</template>
</BaseTable>
</div>
</template>
<BaseTable
v-else
:headers="tableHeaders"
:items="[]"
:items="filteredRecords"
:no-data-message="
searchQuery ? $t('AUTOMATION.NO_RESULTS') : $t('AUTOMATION.LIST.404')
"
>
<template #row />
<template #row="{ items }">
<AutomationRuleRow
v-for="automation in items"
:key="automation.id"
:automation="automation"
:loading="loading[automation.id]"
@clone="cloneAutomation"
@toggle="toggleAutomation"
@edit="openEditPopup"
@delete="openDeletePopup"
/>
</template>
</BaseTable>
</template>
+37 -12
View File
@@ -11,6 +11,8 @@ import { IFrameHelper } from '../helpers/utils';
import { CHATWOOT_ON_START_CONVERSATION } from '../constants/sdkEvents';
import { emitter } from 'shared/helpers/mitt';
const TRANSCRIPT_COOLDOWN_MS = 15000;
export default {
components: {
ChatInputWrap,
@@ -24,6 +26,9 @@ export default {
data() {
return {
inReplyTo: null,
isSendingTranscript: false,
transcriptCooldown: false,
transcriptCooldownTimer: null,
};
},
computed: {
@@ -57,6 +62,9 @@ export default {
mounted() {
emitter.on(BUS_EVENTS.TOGGLE_REPLY_TO_MESSAGE, this.toggleReplyTo);
},
beforeUnmount() {
clearTimeout(this.transcriptCooldownTimer);
},
methods: {
...mapActions('conversation', ['sendMessage', 'sendAttachment']),
...mapActions('conversationAttributes', ['getAttributes']),
@@ -90,19 +98,35 @@ export default {
toggleReplyTo(message) {
this.inReplyTo = message;
},
startTranscriptCooldown() {
this.transcriptCooldown = true;
clearTimeout(this.transcriptCooldownTimer);
this.transcriptCooldownTimer = setTimeout(() => {
this.transcriptCooldown = false;
}, TRANSCRIPT_COOLDOWN_MS);
},
async sendTranscript() {
if (this.hasEmail) {
try {
await sendEmailTranscript();
emitter.emit(BUS_EVENTS.SHOW_ALERT, {
message: this.$t('EMAIL_TRANSCRIPT.SEND_EMAIL_SUCCESS'),
type: 'success',
});
} catch (error) {
emitter.$emit(BUS_EVENTS.SHOW_ALERT, {
message: this.$t('EMAIL_TRANSCRIPT.SEND_EMAIL_ERROR'),
});
}
if (
!this.hasEmail ||
this.isSendingTranscript ||
this.transcriptCooldown
) {
return;
}
this.isSendingTranscript = true;
try {
await sendEmailTranscript();
this.startTranscriptCooldown();
emitter.emit(BUS_EVENTS.SHOW_ALERT, {
message: this.$t('EMAIL_TRANSCRIPT.SEND_EMAIL_SUCCESS'),
type: 'success',
});
} catch (error) {
emitter.emit(BUS_EVENTS.SHOW_ALERT, {
message: this.$t('EMAIL_TRANSCRIPT.SEND_EMAIL_ERROR'),
});
} finally {
this.isSendingTranscript = false;
}
},
},
@@ -144,6 +168,7 @@ export default {
v-if="showEmailTranscriptButton"
type="clear"
class="font-normal"
:disabled="isSendingTranscript || transcriptCooldown"
@click="sendTranscript"
>
{{ $t('EMAIL_TRANSCRIPT.BUTTON_TEXT') }}
@@ -1,65 +0,0 @@
class AutomationRules::ProcessPendingExecutionJob < ApplicationJob
queue_as :medium
discard_on ActiveJob::DeserializationError
def perform(pending_execution)
return if delayed_automations_disabled?
# Account flag off pauses (not skips): leave the row pending so re-enabling resumes it.
return unless pending_execution.account.feature_enabled?('delayed_automations')
# Atomic claim: a duplicate enqueue (overlapping sweep or stale reclaim) loses here and returns.
return unless pending_execution.claim!
skip_reason = skip_reason_for(pending_execution)
return pending_execution.update!(status: :skipped, skip_reason: skip_reason) if skip_reason
execute(pending_execution)
rescue StandardError => e
# Row stays `processing`; the next sweep reclaims and retries it once the lock goes stale.
ChatwootExceptionTracker.new(e, account: pending_execution.account).capture_exception
end
private
def skip_reason_for(pending_execution)
return 'expired' if pending_execution.due_at < AutomationRulePendingExecution::DUE_WINDOW.ago
structural_skip_reason(pending_execution) || behavioral_skip_reason(pending_execution)
end
def structural_skip_reason(pending_execution)
rule = pending_execution.automation_rule
return 'rule_inactive' if rule.nil? || !rule.active?
return 'conversation_gone' if pending_execution.conversation.nil?
nil
end
def behavioral_skip_reason(pending_execution)
return 'episode_moved' unless pending_execution.episode_current?
return AutomationRulePendingExecution::CONDITIONS_CHANGED_SKIP 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
@@ -1,11 +0,0 @@
class AutomationRules::ResumePausedExecutionsJob < ApplicationJob
# Enqueued the moment the account flag flips back on, ahead of the next sweep, so overdue rows
# are rescheduled before that sweep's per-row jobs could mark them expired.
queue_as :medium
discard_on ActiveJob::DeserializationError
def perform(account)
AutomationRulePendingExecution.reschedule_paused(account)
end
end
@@ -1,32 +0,0 @@
class AutomationRules::TriggerPendingExecutionsJob < ApplicationJob
queue_as :scheduled_jobs
DEFAULT_SWEEP_LIMIT = 1000
def perform
return if delayed_automations_disabled?
started_at = Time.current
purged = AutomationRulePendingExecution.purge_terminal!
rows = AutomationRulePendingExecution.sweepable.for_enabled_accounts.order(:due_at).limit(sweep_limit).to_a
rows.each { |row| AutomationRules::ProcessPendingExecutionJob.perform_later(row) }
log_summary(enqueued: rows.size, capped: rows.size >= sweep_limit, purged: purged, 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(enqueued:, capped:, purged:, started_at:)
summary = { event: 'completed', enqueued: enqueued, capped: capped, purged: purged, duration_ms: ((Time.current - started_at) * 1000).round }
Rails.logger.info("[AutomationRules::TriggerPendingExecutionsJob] #{summary.to_json}")
end
end
-3
View File
@@ -19,9 +19,6 @@ class TriggerScheduledItemsJob < ApplicationJob
# Job to sync whatsapp templates
Channels::Whatsapp::TemplatesSyncSchedulerJob.perform_later
# Job to trigger pending executions
AutomationRules::TriggerPendingExecutionsJob.perform_later
end
end
+2 -15
View File
@@ -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
execute_rule(rule, account, message.conversation, message: message) if conditions_match.present?
::AutomationRules::ActionService.new(rule, account, message.conversation).perform if conditions_match.present?
end
end
@@ -52,20 +52,7 @@ class AutomationRuleListener < BaseListener
rules.each do |rule|
conditions_match = ::AutomationRules::ConditionsFilterService.new(rule, conversation, { changed_attributes: changed_attributes }).perform
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
AutomationRules::ActionService.new(rule, account, conversation).perform if conditions_match.present?
end
end
-6
View File
@@ -65,7 +65,6 @@ 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
@@ -112,7 +111,6 @@ class Account < ApplicationRecord
before_validation :validate_limit_keys
after_create_commit :notify_creation
after_update_commit :clear_unread_conversation_counts_cache, if: :saved_change_to_feature_conversation_unread_counts?
after_update_commit :resume_delayed_automations, if: -> { saved_change_to_feature_delayed_automations? && feature_delayed_automations? }
after_destroy :remove_account_sequences
def agents
@@ -191,10 +189,6 @@ class Account < ApplicationRecord
::Conversations::UnreadCounts::Store.clear_account!(id)
end
def resume_delayed_automations
AutomationRules::ResumePausedExecutionsJob.perform_later(self)
end
trigger.after(:insert).for_each(:row) do
"execute format('create sequence IF NOT EXISTS conv_dpid_seq_%s', NEW.id);"
end
+10 -50
View File
@@ -2,17 +2,16 @@
#
# 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
# execution_delay :integer
# 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
# name :string not null
# created_at :datetime not null
# updated_at :datetime not null
# account_id :bigint not null
#
# Indexes
#
@@ -22,13 +21,7 @@ class AutomationRule < ApplicationRecord
include Rails.application.routes.url_helpers
include Reauthorizable
EXECUTION_DELAY_RANGE = (10..43_200) # minutes: 10 min to 30 days
# Conversation-level delayed rules key their episode on status; only status and attributes
# that never change after the delay (inbox) are safe to also filter on.
DELAYED_CONVERSATION_ATTRIBUTES = %w[status inbox_id].freeze
belongs_to :account
has_many :pending_executions, class_name: 'AutomationRulePendingExecution', dependent: :delete_all
has_many_attached :files
validate :json_conditions_format
@@ -36,13 +29,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
validate :execution_delay_supported_event
after_update_commit :reauthorized!, if: -> { saved_change_to_conditions? }
# Discard rows armed under the old definition; they re-arm on the next matching event.
after_update :discard_stale_pending_executions, if: :execution_config_changed?
scope :active, -> { where(active: true) }
@@ -107,34 +95,6 @@ class AutomationRule < ApplicationRecord
end
end
# The fire-time re-check cannot reconstruct changed_attributes, so delayed rules
# cannot use attribute_changed conditions.
def execution_delay_supported_conditions
return if execution_delay.blank? || conditions.blank?
return if conditions.none? { |obj| obj['filter_operator'] == 'attribute_changed' }
errors.add(:execution_delay, 'cannot be used with attribute_changed conditions.')
end
# Conversation-level episodes key on status_changed_at alone. Mutable attributes would collapse
# distinct periods into one episode, so only status and immutable filters (inbox) are allowed.
def execution_delay_supported_event
return if execution_delay.blank? || conditions.blank? || event_name == 'message_created'
return if conditions.all? { |obj| DELAYED_CONVERSATION_ATTRIBUTES.include?(obj['attribute_key']) }
errors.add(:execution_delay, 'only supports status and inbox conditions for conversation-level events.')
end
def execution_config_changed?
saved_change_to_execution_delay? || saved_change_to_event_name? ||
saved_change_to_conditions? || saved_change_to_actions?
end
def discard_stale_pending_executions
# armed = pending + stale processing, which the sweep would otherwise reclaim.
pending_executions.armed.delete_all
end
def validate_single_condition(condition)
query_operator = condition['query_operator']
@@ -1,180 +0,0 @@
# == 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
# A processing row whose lock is older than this is treated as abandoned and reclaimed.
STALE_PROCESSING_TIMEOUT = 15.minutes
# Terminal rows are purged after this to keep the table bounded.
RETENTION_WINDOW = 30.days
# 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
belongs_to :account
belongs_to :message, optional: true
enum status: { pending: 0, processing: 1, executed: 2, skipped: 3 }
# Rows a sweep should hand to a worker: due pending rows, plus processing rows whose lock went stale.
scope :sweepable, lambda {
pending.where(due_at: ..Time.current).or(processing.where(updated_at: ...STALE_PROCESSING_TIMEOUT.ago))
}
# Non-terminal rows still bound to fire (a stale processing row is reclaimed by the sweep).
scope :armed, -> { where(status: [statuses[:pending], statuses[:processing]]) }
# Excludes rows whose account paused delayed automations, so one disabled account's backlog
# can't fill the sweep limit and starve enabled accounts (paused rows resume on re-enable).
scope :for_enabled_accounts, -> { joins(:account).merge(Account.feature_delayed_automations) }
def self.schedule(rule:, conversation:, message: nil)
key = arm_episode_key_for(conversation, message)
anchor = arm_anchor_for(conversation, message)
create!(
automation_rule: rule, conversation: conversation, account_id: conversation.account_id,
message_id: message&.id, episode_key: key, due_at: rule.execution_delay.minutes.since(anchor)
)
rescue ActiveRecord::RecordNotUnique
rearm_or_advance_episode(rule, conversation, key, message, anchor)
end
# The episode is already armed. Status episodes keep their first clock (a status change would
# give a new key), so only message episodes advance or re-arm here.
def self.rearm_or_advance_episode(rule, conversation, key, message, anchor)
return unless message
row = find_by!(automation_rule_id: rule.id, conversation_id: conversation.id, episode_key: key)
# Jobs can arrive out of order; only a strictly newer message advances or re-arms, so a late
# older message can't pull due_at backwards and fire before the delay elapses.
return unless message.id > row.message_id
due_at = rule.execution_delay.minutes.since(anchor)
if row.condition_skipped?
# A message episode key can recur (no new incoming reply) while conditions swing back into
# match, so a later qualifying message re-arms the condition-only skip instead of dropping.
row.update!(status: :pending, skip_reason: nil, due_at: due_at, message_id: message.id)
elsif !row.terminal?
# Track the newest qualifying message. Reply-chase advances due_at with each agent reply;
# awaiting-agent keeps its first clock (its anchor is the stable waiting_since, so due_at is
# unchanged). Re-anchoring a row still processing (its worker died mid-run) back to pending
# also keeps a stale reclaim from firing the old clock instead of the latest one.
row.update!(status: :pending, due_at: due_at, message_id: message.id)
end
end
# The wait is measured from when the qualifying event happened, not when this (possibly
# backlogged or retried) listener runs, so a late dispatch still fires on schedule. Mirrors
# the timestamps the episode keys track.
def self.arm_anchor_for(conversation, message)
if message.nil?
conversation.status_changed_at.presence || conversation.created_at
elsif message.incoming?
conversation.waiting_since.presence || message.created_at
else
message.created_at
end
end
# waiting_since is written just after MESSAGE_CREATED dispatches, so it can still be nil when
# an awaiting-agent episode arms. It becomes the starting message's created_at, so use that
# here; the strict fire-time key (episode_key_for) then matches once waiting_since is settled.
def self.arm_episode_key_for(conversation, message)
return episode_key_for(conversation, message) unless message&.incoming? && conversation.waiting_since.blank?
"awaiting_agent:#{microsecond_stamp(message.created_at)}"
end
# Microsecond integer, not a float: epoch seconds carry ~16 significant digits, past float64's
# precision, so an in-memory timestamp (arm time) and its DB-reloaded value (fire time) would
# round to different floats. strftime is exact on both. Sub-second distinguishes rapid episodes.
def self.microsecond_stamp(time)
time&.strftime('%s%6N') || '0'
end
# Episode keys identify one qualifying stretch of conversation state; when the recomputed
# key no longer matches, the episode ended and the pending action is cancelled at fire time.
def self.episode_key_for(conversation, message)
if message.nil?
# Sub-second precision so a resolve→reopen inside one second still ends the episode.
# Integer microseconds (not a float) so an in-memory arm and a DB-reloaded fire agree.
"status:#{microsecond_stamp(conversation.status_changed_at.presence || conversation.created_at)}"
elsif message.incoming?
# waiting_since is cleared on agent/bot reply, so a reply invalidates this episode. Strict
# here: at fire time a nil waiting_since means the agent replied (episode ended).
"awaiting_agent:#{microsecond_stamp(conversation.waiting_since)}"
else
# A new customer message changes the max incoming id, invalidating this episode.
"reply_chase:#{conversation.messages.incoming.maximum(:id) || 0}"
end
end
def self.purge_terminal!
where(status: [statuses[:executed], statuses[:skipped]], updated_at: ...RETENTION_WINDOW.ago)
.in_batches(of: 1000).delete_all
end
# Rows that came due while an account had delayed automations paused would expire the moment
# the sweep reaches them on resume. Reset their clock so pause/resume replays them (still
# subject to the fire-time episode/condition re-checks) instead of silently dropping them.
def self.reschedule_paused(account)
overdue = pending.where(account_id: account.id, due_at: ...DUE_WINDOW.ago)
overdue.find_each { |row| row.update!(due_at: Time.current) }
end
# Atomic claim: only one worker can move a row into processing, so a row re-enqueued by an
# overlapping sweep (or after a stale reclaim) cannot double-execute. Refreshing updated_at
# renews the lock, keeping the row out of the stale window while this worker holds it.
def claim!
with_lock do
next false unless claimable?
update!(status: :processing, updated_at: Time.current)
true
end
end
def episode_current?
self.class.episode_key_for(conversation, message) == episode_key
end
def condition_skipped?
skipped? && skip_reason == CONDITIONS_CHANGED_SKIP
end
def terminal?
executed? || skipped?
end
private
def claimable?
# due_at guard: a reply-chase reschedule can push due_at forward after this row was enqueued;
# such a row must wait for a later sweep instead of firing early.
(pending? && due_at <= Time.current) || (processing? && updated_at < STALE_PROCESSING_TIMEOUT.ago)
end
end
-7
View File
@@ -15,7 +15,6 @@
# priority :integer
# snoozed_until :datetime
# status :integer default("open"), not null
# status_changed_at :datetime
# uuid :uuid not null
# waiting_since :datetime
# created_at :datetime not null
@@ -125,10 +124,8 @@ 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
@@ -275,10 +272,6 @@ 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
@@ -7,5 +7,4 @@ 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?
-5
View File
@@ -268,8 +268,3 @@
display_name: WhatsApp Embedded Signup Flow
enabled: false
column: feature_flags_ext_1
- name: delayed_automations
display_name: Delayed Automations
enabled: false
chatwoot_internal: true
column: feature_flags_ext_1
-7
View File
@@ -567,10 +567,3 @@
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
+2 -1
View File
@@ -66,7 +66,8 @@ Rails.application.routes.draw do
resources :assistants do
member do
post :playground
get :stats
get :metrics
get :faq_stats
get :summary
get :drilldown
end
@@ -1,5 +0,0 @@
class AddExecutionDelayToAutomationRules < ActiveRecord::Migration[7.0]
def change
add_column :automation_rules, :execution_delay, :integer
end
end
@@ -1,5 +0,0 @@
class AddStatusChangedAtToConversations < ActiveRecord::Migration[7.0]
def change
add_column :conversations, :status_changed_at, :datetime
end
end
@@ -1,21 +0,0 @@
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
-20
View File
@@ -277,24 +277,6 @@ ActiveRecord::Schema[7.1].define(version: 2026_07_13_184351) 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
@@ -305,7 +287,6 @@ ActiveRecord::Schema[7.1].define(version: 2026_07_13_184351) 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
@@ -807,7 +788,6 @@ ActiveRecord::Schema[7.1].define(version: 2026_07_13_184351) 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"
@@ -37,6 +37,23 @@ class Captain::AssistantStatsBuilder
build_metrics(current, previous)
end
# Approved/pending FAQ counts and the document total in a single round trip.
def faq_stats
approved, pending, documents = Captain::AssistantResponse.by_assistant(assistant.id).reorder(nil).pick(
Arel.sql("COUNT(*) FILTER (WHERE status = #{Captain::AssistantResponse.statuses['approved']})"),
Arel.sql("COUNT(*) FILTER (WHERE status = #{Captain::AssistantResponse.statuses['pending']})"),
Arel.sql("(SELECT COUNT(*) FROM captain_documents WHERE assistant_id = #{assistant.id.to_i})")
)
total = approved + pending
{
approved: approved,
pending: pending,
documents: documents,
coverage: total.zero? ? 0 : (approved.to_f / total * 100).round
}
end
private
attr_reader :window
@@ -56,8 +73,7 @@ class Captain::AssistantStatsBuilder
handoff_rate: pack(current[:handoff], previous[:handoff], :point),
hours_saved: pack(current[:hours_saved], previous[:hours_saved], :percent),
reopen_rate: pack(current[:reopen], previous[:reopen], :point),
conversation_depth: pack(current[:depth], previous[:depth], :absolute),
knowledge: knowledge
conversation_depth: pack(current[:depth], previous[:depth], :absolute)
}
end
@@ -73,7 +89,7 @@ class Captain::AssistantStatsBuilder
auto_resolution: rate(resolution[:resolved], handled),
handoff: rate(resolution[:handoff], handled),
hours_saved: (public_count * SECONDS_SAVED_PER_REPLY / 3600.0).round,
reopen: reopen_rate(range),
reopen: reopen_rate(range, resolution[:resolved]),
depth: depth_conversations.zero? ? 0 : (public_count.to_f / depth_conversations).round(1)
}
end
@@ -158,7 +174,9 @@ class Captain::AssistantStatsBuilder
# derived from the assistant's handled conversations (not current inbox membership) so a later
# inbox reassignment doesn't drop historical resolves, and covers both the evaluated (inference)
# and time-based (bot) resolve paths so the denominator matches auto_resolution_rate.
def reopen_rate(range)
def reopen_rate(range, resolved_count)
return 0 if resolved_count.zero?
resolved_scope = account.reporting_events
.where(name: RESOLVED_EVENT_NAMES, created_at: range,
conversation_id: handled_scope(range).select(:conversation_id))
@@ -178,24 +196,7 @@ class Captain::AssistantStatsBuilder
'ON resolves.conversation_id = reporting_events.conversation_id ' \
'AND reporting_events.event_end_time >= resolves.event_end_time')
.distinct.count('reporting_events.conversation_id')
rate(reopened, resolved_scope.distinct.count(:conversation_id))
end
# Approved/pending FAQ counts and the document total in a single round trip.
def knowledge
approved, pending, documents = Captain::AssistantResponse.by_assistant(assistant.id).reorder(nil).pick(
Arel.sql("COUNT(*) FILTER (WHERE status = #{Captain::AssistantResponse.statuses['approved']})"),
Arel.sql("COUNT(*) FILTER (WHERE status = #{Captain::AssistantResponse.statuses['pending']})"),
Arel.sql("(SELECT COUNT(*) FROM captain_documents WHERE assistant_id = #{assistant.id.to_i})")
)
total = approved + pending
{
approved: approved,
pending: pending,
documents: documents,
coverage: total.zero? ? 0 : (approved.to_f / total * 100).round
}
rate(reopened, resolved_count)
end
def rate(numerator, denominator)
@@ -1,7 +1,7 @@
class Api::V1::Accounts::Captain::AssistantsController < Api::V1::Accounts::BaseController
before_action -> { check_authorization(Captain::Assistant) }
before_action :set_assistant, only: [:show, :update, :destroy, :playground, :stats, :summary, :drilldown]
before_action :set_assistant, only: [:show, :update, :destroy, :playground, :metrics, :faq_stats, :summary, :drilldown]
def index
@assistants = account_assistants.ordered
@@ -42,10 +42,14 @@ class Api::V1::Accounts::Captain::AssistantsController < Api::V1::Accounts::Base
@tools = assistant.available_agent_tools
end
def stats
def metrics
render json: Captain::AssistantStatsBuilder.new(@assistant, params[:range], params[:timezone_offset]).metrics
end
def faq_stats
render json: Captain::AssistantStatsBuilder.new(@assistant).faq_stats
end
def summary
window = Captain::AssistantStatsWindow.new(params[:range], params[:timezone_offset])
result = cached_or_generated_summary(window, summary_stats)
@@ -1,6 +1,8 @@
module Enterprise::Api::V1::Accounts::AgentsController
def create
super
return if @agent.blank?
associate_agent_with_custom_role
end
@@ -7,16 +7,19 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
retry_on ActiveStorage::FileNotFoundError, attempts: 3, wait: 2.seconds
retry_on Faraday::BadRequestError, attempts: 3, wait: 2.seconds
def perform(conversation, assistant)
def perform(conversation, assistant, trigger_message_id = nil)
@conversation = conversation
@inbox = conversation.inbox
@assistant = assistant
@trigger_message_id = trigger_message_id if captain_v2_enabled?
return unless conversation_pending?
Current.executed_by = @assistant
if captain_v2_enabled?
return unless trigger_message_current?
generate_response_with_v2
else
generate_and_process_response
@@ -45,11 +48,18 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
end
def generate_response_with_v2
runner_service = Captain::Assistant::AgentRunnerService.new(assistant: @assistant, conversation: @conversation)
runner_service = Captain::Assistant::AgentRunnerService.new(
assistant: @assistant,
conversation: @conversation,
trigger_message_id: @trigger_message_id
)
message_history = Captain::Conversation::MessageHistoryBuilderService.new(conversation: @conversation).perform
@response = runner_service.generate_response(message_history: message_history)
@run_result = runner_service.last_run_result
return if runner_service.response_discarded?
return unless trigger_message_current?
process_response
end
@@ -77,16 +87,24 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
process_v1_handoff
elsif conversation_pending?
message = nil
ActiveRecord::Base.transaction do
message = create_messages
Rails.logger.info("[CAPTAIN][ResponseBuilderJob] Incrementing response usage for #{account.id}")
account.increment_response_usage
end
capture_assistant_session(result_message: message, credits_consumed: 1.0)
process_standard_response
end
end
def process_standard_response
message = nil
ActiveRecord::Base.transaction do
next unless trigger_message_current?
message = create_messages
Rails.logger.info("[CAPTAIN][ResponseBuilderJob] Incrementing response usage for #{account.id}")
account.increment_response_usage
end
return unless message
capture_assistant_session(result_message: message, credits_consumed: 1.0)
end
def v1_handoff_requested?
legacy_v1_handoff_token? || classifier_v1_handoff_requested?
end
@@ -117,6 +135,8 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
end
def process_v2_handoff
return unless trigger_message_current?
# HandoffTool already ran bot_handoff! + OOO inside the agent loop. Preserve
# waiting_since so this message doesn't clear the timestamp it left in place.
I18n.with_locale(@assistant.account.locale) do
@@ -152,7 +172,7 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
@response ||= {}
@response['action_source'] ||= 'error'
@response['action_reason'] ||= error_action_reason(error)
process_v1_handoff if conversation_pending?
process_v1_handoff if conversation_pending? && (!captain_v2_enabled? || trigger_message_current?)
true
end
@@ -181,4 +201,14 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
status = Conversation.uncached { Conversation.where(id: @conversation.id).pick(:status) }
status == 'pending' || status == Conversation.statuses[:pending]
end
def trigger_message_current?
return true if @trigger_message_id.blank?
latest_incoming_message_id = Conversation.uncached do
@conversation.messages.incoming.maximum(:id)
end
latest_incoming_message_id == @trigger_message_id
end
end
@@ -7,7 +7,11 @@ class Captain::AssistantPolicy < ApplicationPolicy
true
end
def stats?
def metrics?
true
end
def faq_stats?
true
end
@@ -9,11 +9,12 @@ class Captain::Assistant::AgentRunnerService
attr_reader :last_run_result
def initialize(assistant:, conversation: nil, callbacks: {}, source: nil)
def initialize(assistant:, conversation: nil, callbacks: {}, source: nil, trigger_message_id: nil)
@assistant = assistant
@conversation = conversation
@callbacks = callbacks
@source = source
@trigger_message_id = trigger_message_id
@handoff_tool_called = false
end
@@ -31,6 +32,10 @@ class Captain::Assistant::AgentRunnerService
error_response(e.message)
end
def response_discarded?
@response_discarded == true
end
private
def build_context(message_history)
@@ -151,10 +156,9 @@ class Captain::Assistant::AgentRunnerService
track_handoff_usage(tool_name, handoff_tool_name, context_wrapper)
end
if ChatwootApp.otel_enabled?
runner.on_run_complete do |_agent_name, _result, context_wrapper|
write_credits_used_metadata(context_wrapper)
end
runner.on_run_complete do |_agent_name, _result, context_wrapper|
@response_discarded = trigger_message_stale?
write_run_metadata(context_wrapper) if ChatwootApp.otel_enabled?
end
runner
end
@@ -169,11 +173,22 @@ class Captain::Assistant::AgentRunnerService
@handoff_tool_called = true
end
def write_credits_used_metadata(context_wrapper)
def write_run_metadata(context_wrapper)
root_span = context_wrapper&.context&.dig(:__otel_tracing, :root_span)
return unless root_span
root_span.set_attribute(format(ATTR_LANGFUSE_METADATA, 'credit_used'), @handoff_tool_called ? 'false' : 'true')
root_span.set_attribute(format(ATTR_LANGFUSE_METADATA, 'discarded'), response_discarded?.to_s)
root_span.set_attribute(format(ATTR_LANGFUSE_METADATA, 'credit_used'), (!@handoff_tool_called && !response_discarded?).to_s)
end
def trigger_message_stale?
return false if @trigger_message_id.blank? || @conversation.blank?
latest_incoming_message_id = Conversation.uncached do
@conversation.messages.incoming.maximum(:id)
end
latest_incoming_message_id != @trigger_message_id
end
def runner
@@ -13,7 +13,8 @@ class Captain::Assistant::InstrumentationAttributeProvider
def generation_attributes(_context_wrapper, _chat, message)
{
format(ATTR_LANGFUSE_OBSERVATION_METADATA, 'generation_stage') => generation_stage(message)
format(ATTR_LANGFUSE_OBSERVATION_METADATA, 'generation_stage') => generation_stage(message),
format(ATTR_LANGFUSE_OBSERVATION_METADATA, 'discarded') => @service.send(:trigger_message_stale?).to_s
}
end
@@ -23,6 +23,7 @@ module Captain::Assistant::RunnerStateHelper
timezone: @conversation&.inbox&.timezone.presence || 'UTC'
}
state[:source] = @source if @source.present?
state[:trigger_message_id] = @trigger_message_id if @trigger_message_id.present?
build_conversation_state(state) if @conversation
state
@@ -31,6 +31,7 @@ module Enterprise::MessageTemplates::HookExecutionService
def schedule_captain_response
job_args = [conversation, conversation.inbox.captain_assistant]
job_args << message.id if conversation.account.feature_enabled?('captain_integration_v2')
if message.attachments.blank?
Captain::Conversation::ResponseBuilderJob.perform_later(*job_args)
@@ -5,6 +5,7 @@ class Captain::Tools::HandoffTool < Captain::Tools::BasePublicTool
def perform(tool_context, reason: nil)
conversation = find_conversation(tool_context.state)
return 'Conversation not found' unless conversation
return 'Handoff skipped because a newer customer message arrived' unless trigger_message_current?(tool_context.state, conversation)
# Log the handoff with reason
log_tool_usage('tool_handoff', {
@@ -23,6 +24,17 @@ class Captain::Tools::HandoffTool < Captain::Tools::BasePublicTool
private
def trigger_message_current?(state, conversation)
trigger_message_id = state&.dig(:trigger_message_id)
return true if trigger_message_id.blank?
latest_incoming_message_id = Conversation.uncached do
conversation.messages.incoming.maximum(:id)
end
latest_incoming_message_id == trigger_message_id
end
def trigger_handoff(tool_context, conversation, reason)
# post the reason as a private note
note = conversation.messages.create!(
+1 -4
View File
@@ -34,7 +34,7 @@
"@amplitude/analytics-browser": "^2.11.10",
"@breezystack/lamejs": "^1.2.7",
"@chatwoot/ninja-keys": "1.2.3",
"@chatwoot/prosemirror-schema": "1.3.22",
"@chatwoot/prosemirror-schema": "1.3.23",
"@chatwoot/utils": "^0.0.56",
"@formkit/core": "^1.7.2",
"@formkit/vue": "^1.7.2",
@@ -86,9 +86,6 @@
"mitt": "^3.0.1",
"opus-recorder": "^8.0.5",
"pinia": "^3.0.4",
"prosemirror-commands": "^1.7.1",
"prosemirror-inputrules": "^1.4.0",
"prosemirror-schema-list": "^1.5.1",
"qrcode": "^1.5.4",
"semver": "7.6.3",
"snakecase-keys": "^8.0.1",
+6 -22
View File
@@ -25,8 +25,8 @@ importers:
specifier: 1.2.3
version: 1.2.3
'@chatwoot/prosemirror-schema':
specifier: 1.3.22
version: 1.3.22
specifier: 1.3.23
version: 1.3.23
'@chatwoot/utils':
specifier: ^0.0.56
version: 0.0.56
@@ -180,15 +180,6 @@ importers:
pinia:
specifier: ^3.0.4
version: 3.0.4(typescript@5.6.2)(vue@3.5.12(typescript@5.6.2))
prosemirror-commands:
specifier: ^1.7.1
version: 1.7.1
prosemirror-inputrules:
specifier: ^1.4.0
version: 1.4.0
prosemirror-schema-list:
specifier: ^1.5.1
version: 1.5.1
qrcode:
specifier: ^1.5.4
version: 1.5.4
@@ -461,8 +452,8 @@ packages:
'@chatwoot/ninja-keys@1.2.3':
resolution: {integrity: sha512-xM8d9P5ikDMZm2WbaCTk/TW5HFauylrU3cJ75fq5je6ixKwyhl/0kZbVN/vbbZN4+AUX/OaSIn6IJbtCgIF67g==}
'@chatwoot/prosemirror-schema@1.3.22':
resolution: {integrity: sha512-0r+PT8xhQLCKCpoV9k9XVTTRECs/0Nr37wbcLsRS7yvc7WkF9FY05z2hGCRJReWmTOcmmshHtb042LVP+MyB/w==}
'@chatwoot/prosemirror-schema@1.3.23':
resolution: {integrity: sha512-jGxbWELCdlVI64BJiE1wT84ekJHYDXXKiluQIKT3aKPEjPwMR48umKF3A0yHjKoR7IIxCC9oM77TvXOA0ebLtw==}
'@chatwoot/utils@0.0.56':
resolution: {integrity: sha512-A6dmPLfTSrW4qYNY73btyi4PqpfzcXRSaucscZTQdzNqF6G/QUdgnBmHtho8HeiYby/kSHXaSxLJj+0dx3yEQQ==}
@@ -4001,9 +3992,6 @@ packages:
prosemirror-tables@1.5.0:
resolution: {integrity: sha512-VMx4zlYWm7aBlZ5xtfJHpqa3Xgu3b7srV54fXYnXgsAcIGRqKSrhiK3f89omzzgaAgAtDOV4ImXnLKhVfheVNQ==}
prosemirror-transform@1.10.0:
resolution: {integrity: sha512-9UOgFSgN6Gj2ekQH5CTDJ8Rp/fnKR2IkYfGdzzp5zQMFsS4zDllLVx/+jGcX86YlACpG7UR5fwAXiWzxqWtBTg==}
prosemirror-transform@1.12.0:
resolution: {integrity: sha512-GxboyN4AMIsoHNtz5uf2r2Ru551i5hWeCMD6E2Ib4Eogqoub0NflniaBPVQ4MrGE5yZ8JV9tUHg9qcZTTrcN4w==}
@@ -5136,7 +5124,7 @@ snapshots:
hotkeys-js: 3.8.7
lit: 2.2.6
'@chatwoot/prosemirror-schema@1.3.22':
'@chatwoot/prosemirror-schema@1.3.23':
dependencies:
markdown-it-sup: 2.0.0
prosemirror-commands: 1.7.1
@@ -9035,7 +9023,7 @@ snapshots:
dependencies:
prosemirror-model: 1.22.3
prosemirror-state: 1.4.3
prosemirror-transform: 1.10.0
prosemirror-transform: 1.12.0
prosemirror-state@1.4.3:
dependencies:
@@ -9051,10 +9039,6 @@ snapshots:
prosemirror-transform: 1.12.0
prosemirror-view: 1.34.1
prosemirror-transform@1.10.0:
dependencies:
prosemirror-model: 1.22.3
prosemirror-transform@1.12.0:
dependencies:
prosemirror-model: 1.22.3
+18
View File
@@ -23,6 +23,12 @@ RSpec.describe AgentBuilder, type: :model do
end
describe '#perform' do
it 'locks the account while checking and creating the agent' do
expect(account).to receive(:with_lock).and_call_original
agent_builder.perform
end
context 'when user does not exist' do
it 'creates a new user' do
expect { agent_builder.perform }.to change(User, :count).by(1)
@@ -67,5 +73,17 @@ RSpec.describe AgentBuilder, type: :model do
expect(user.encrypted_password).not_to be_empty
end
end
context 'when the account has reached its agent limit' do
before do
allow(account).to receive(:usage_limits).and_return({ agents: account.account_users.count })
end
it 'raises a limit exceeded error without creating a user' do
expect { agent_builder.perform }.to raise_error(described_class::LimitExceededError, described_class::LIMIT_EXCEEDED_MESSAGE)
expect(User.from_email(email)).to be_nil
end
end
end
end
@@ -451,71 +451,4 @@ RSpec.describe 'Api::V1::Accounts::AutomationRulesController', type: :request do
end
end
end
describe 'execution_delay handling' do
let(:delayed_rule_params) do
{
name: 'Delayed rule',
event_name: 'conversation_updated',
execution_delay: 240,
conditions: [{ attribute_key: 'status', filter_operator: 'equal_to', values: ['pending'], query_operator: nil }],
actions: [{ action_name: 'add_label', action_params: ['stale'] }]
}
end
context 'when the delayed_automations feature is enabled' do
before { account.enable_features!('delayed_automations') }
it 'persists and serializes execution_delay' do
post "/api/v1/accounts/#{account.id}/automation_rules",
headers: administrator.create_new_auth_token,
params: delayed_rule_params
expect(response).to have_http_status(:success)
body = JSON.parse(response.body, symbolize_names: true)
expect(body[:execution_delay]).to eq(240)
expect(account.automation_rules.last.execution_delay).to eq(240)
end
it 'copies execution_delay on clone' do
automation_rule = create(:automation_rule, account: account, execution_delay: 240)
post "/api/v1/accounts/#{account.id}/automation_rules/#{automation_rule.id}/clone",
headers: administrator.create_new_auth_token
expect(response).to have_http_status(:success)
expect(account.automation_rules.last.execution_delay).to eq(240)
end
end
context 'when the delayed_automations feature is disabled' do
it 'rejects a payload carrying execution_delay with 422' do
post "/api/v1/accounts/#{account.id}/automation_rules",
headers: administrator.create_new_auth_token,
params: delayed_rule_params
expect(response).to have_http_status(:unprocessable_entity)
expect(account.automation_rules.count).to eq(0)
end
it 'still accepts payloads without execution_delay' do
post "/api/v1/accounts/#{account.id}/automation_rules",
headers: administrator.create_new_auth_token,
params: delayed_rule_params.except(:execution_delay)
expect(response).to have_http_status(:success)
expect(account.automation_rules.last.execution_delay).to be_nil
end
it 'strips execution_delay when cloning an existing delayed rule' do
automation_rule = create(:automation_rule, account: account, execution_delay: 240)
post "/api/v1/accounts/#{account.id}/automation_rules/#{automation_rule.id}/clone",
headers: administrator.create_new_auth_token
expect(response).to have_http_status(:success)
expect(account.automation_rules.last.execution_delay).to be_nil
end
end
end
end
@@ -13,7 +13,9 @@ RSpec.describe 'Linear Integration API', type: :request do
end
describe 'DELETE /api/v1/accounts/:account_id/integrations/linear' do
it 'deletes the linear integration' do
let(:admin) { create(:user, account: account, role: :administrator) }
it 'deletes the linear integration when the user is an administrator' do
# Stub the HTTP call to Linear's revoke endpoint
allow(HTTParty).to receive(:post).with(
'https://api.linear.app/oauth/revoke',
@@ -21,11 +23,19 @@ RSpec.describe 'Linear Integration API', type: :request do
).and_return(instance_double(HTTParty::Response, success?: true))
delete "/api/v1/accounts/#{account.id}/integrations/linear",
headers: agent.create_new_auth_token,
headers: admin.create_new_auth_token,
as: :json
expect(response).to have_http_status(:ok)
expect(account.hooks.count).to eq(0)
end
it 'returns unauthorized for an agent and keeps the integration' do
delete "/api/v1/accounts/#{account.id}/integrations/linear",
headers: agent.create_new_auth_token,
as: :json
expect(response).to have_http_status(:unauthorized)
expect(account.hooks.count).to eq(1)
end
end
describe 'GET /api/v1/accounts/:account_id/integrations/linear/teams' do
@@ -159,15 +159,17 @@ RSpec.describe 'Shopify Integration API', type: :request do
end
describe 'DELETE /api/v1/accounts/:account_id/integrations/shopify' do
let(:admin) { create(:user, account: account, role: :administrator) }
before do
create(:integrations_hook, :shopify, account: account)
end
context 'when it is an authenticated user' do
context 'when it is an administrator' do
it 'deletes the shopify integration' do
expect do
delete "/api/v1/accounts/#{account.id}/integrations/shopify",
headers: agent.create_new_auth_token,
headers: admin.create_new_auth_token,
as: :json
end.to change { account.hooks.count }.by(-1)
@@ -175,6 +177,18 @@ RSpec.describe 'Shopify Integration API', type: :request do
end
end
context 'when it is an agent' do
it 'returns unauthorized and keeps the integration' do
expect do
delete "/api/v1/accounts/#{account.id}/integrations/shopify",
headers: agent.create_new_auth_token,
as: :json
end.not_to(change { account.hooks.count })
expect(response).to have_http_status(:unauthorized)
end
end
context 'when it is an unauthenticated user' do
it 'returns unauthorized' do
delete "/api/v1/accounts/#{account.id}/integrations/shopify",
@@ -27,7 +27,7 @@ RSpec.describe Captain::AssistantStatsBuilder do
expect(metrics.keys).to contain_exactly(
:conversations_handled, :auto_resolution_rate, :handoff_rate,
:hours_saved, :reopen_rate, :conversation_depth, :knowledge
:hours_saved, :reopen_rate, :conversation_depth
)
expect(metrics[:conversations_handled]).to include(:current, :previous, :trend)
end
@@ -229,7 +229,7 @@ RSpec.describe Captain::AssistantStatsBuilder do
end
end
describe '#metrics knowledge' do
describe '#faq_stats' do
before do
create_list(:captain_assistant_response, 3, assistant: assistant, account: account, status: :approved)
create(:captain_assistant_response, assistant: assistant, account: account, status: :pending)
@@ -237,7 +237,7 @@ RSpec.describe Captain::AssistantStatsBuilder do
end
it 'returns approved, pending, document counts and coverage' do
knowledge = described_class.new(assistant, '30').metrics[:knowledge]
knowledge = described_class.new(assistant).faq_stats
expect(knowledge).to eq(approved: 3, pending: 1, documents: 2, coverage: 75)
end
@@ -245,7 +245,7 @@ RSpec.describe Captain::AssistantStatsBuilder do
it 'reports zero coverage when there are no responses' do
Captain::AssistantResponse.where(assistant: assistant).delete_all
knowledge = described_class.new(assistant, '30').metrics[:knowledge]
knowledge = described_class.new(assistant).faq_stats
expect(knowledge[:coverage]).to eq(0)
end
@@ -21,6 +21,27 @@ RSpec.describe 'Agents API', type: :request do
expect(response).to have_http_status(:payment_required)
expect(response.body).to include('Account limit exceeded. Please purchase more licenses')
end
it 'prevents adding an agent if the last seat is consumed before creation' do
account.update!(limits: { agents: account.account_users.count + 1 })
competing_agent_created = false
allow(AgentBuilder).to receive(:new).and_wrap_original do |method, *args|
unless competing_agent_created
create(:user, account: account, role: :agent)
competing_agent_created = true
end
method.call(*args)
end
post "/api/v1/accounts/#{account.id}/agents", params: params, headers: admin.create_new_auth_token, as: :json
expect(response).to have_http_status(:payment_required)
expect(response.body).to include('Account limit exceeded. Please purchase more licenses')
expect(User.from_email(params[:email])).to be_nil
expect(account.account_users.count).to eq(account.usage_limits[:agents])
end
end
end
@@ -13,6 +13,7 @@ RSpec.describe Captain::Conversation::ResponseBuilderJob, type: :job do
let(:mock_action_classifier_service) { instance_double(Captain::Llm::AssistantActionClassifierService) }
let(:mock_false_promise_service) { instance_double(Captain::Llm::AssistantFalsePromiseService) }
let(:assistant_model) { Llm::Models.default_model_for('assistant') }
let(:trigger_message) { conversation.messages.find_by!(content: 'Hello') }
before do
create(:message, conversation: conversation, content: 'Hello', message_type: :incoming)
@@ -23,6 +24,7 @@ RSpec.describe Captain::Conversation::ResponseBuilderJob, type: :job do
allow(Captain::Assistant::AgentRunnerService).to receive(:new).and_return(mock_agent_runner_service)
allow(mock_agent_runner_service).to receive(:generate_response).and_return({ 'response' => 'Hey, welcome to Captain V2' })
allow(mock_agent_runner_service).to receive(:last_run_result).and_return(nil)
allow(mock_agent_runner_service).to receive(:response_discarded?).and_return(false)
allow(Captain::Llm::AssistantActionClassifierService).to receive(:new).and_return(mock_action_classifier_service)
allow(mock_action_classifier_service).to receive(:classify).and_return({ 'action' => 'continue' })
allow(Captain::Llm::AssistantFalsePromiseService).to receive(:new).and_return(mock_false_promise_service)
@@ -358,14 +360,28 @@ RSpec.describe Captain::Conversation::ResponseBuilderJob, type: :job do
it 'uses Captain::Assistant::AgentRunnerService' do
expect(Captain::Assistant::AgentRunnerService).to receive(:new).with(
assistant: assistant,
conversation: conversation
conversation: conversation,
trigger_message_id: trigger_message.id
)
expect(Captain::Llm::AssistantChatService).not_to receive(:new)
described_class.perform_now(conversation, assistant)
described_class.perform_now(conversation, assistant, trigger_message.id)
expect(conversation.messages.last.content).to eq('Hey, welcome to Captain V2')
end
it 'discards a response when a newer message arrives during generation' do
allow(mock_agent_runner_service).to receive(:generate_response) do
create(:message, conversation: conversation, content: 'New context', message_type: :incoming)
{ 'response' => 'Stale response', 'handoff_tool_called' => false }
end
allow(mock_agent_runner_service).to receive(:response_discarded?).and_return(true)
described_class.perform_now(conversation, assistant, trigger_message.id)
expect(conversation.messages.outgoing.count).to eq(0)
expect(account.reload.usage_limits[:captain][:responses][:consumed]).to eq(0)
end
it 'passes message history with resolution markers to agent runner service' do
same_second = Time.current.change(usec: 0)
conversation.messages.find_by!(content: 'Hello').update!(created_at: same_second, updated_at: same_second)
@@ -28,6 +28,37 @@ RSpec.describe Captain::Tools::HandoffTool, type: :model do
describe '#perform' do
context 'when conversation exists' do
context 'with a trigger message id' do
let(:trigger_message) do
create(:message, conversation: conversation, account: account, inbox: inbox, message_type: :incoming, created_at: same_second)
end
let(:same_second) { Time.current.change(usec: 0) }
let(:tool_context) do
Struct.new(:state).new({ conversation: { id: conversation.id }, trigger_message_id: trigger_message.id })
end
it 'hands off when the trigger message is still the latest message' do
trigger_message
expect do
result = tool.perform(tool_context, reason: 'Customer needs specialized support')
expect(result).to include('Conversation handed off')
end.to change(Message, :count).by(1)
end
it 'skips the handoff when a newer message has arrived' do
trigger_message
conversation.update!(status: :pending)
create(:message, conversation: conversation, account: account, inbox: inbox, message_type: :incoming, created_at: same_second)
expect do
result = tool.perform(tool_context, reason: 'Customer needs specialized support')
expect(result).to eq('Handoff skipped because a newer customer message arrived')
end.not_to change(Message, :count)
expect(conversation.reload.status).to eq('pending')
end
end
context 'with reason provided' do
it 'creates a private note with reason and hands off conversation' do
reason = 'Customer needs specialized support'
@@ -12,7 +12,7 @@ RSpec.describe Captain::AssistantPolicy, type: :policy do
let(:administrator_context) { { user: administrator, account: account, account_user: account.account_users.first } }
let(:agent_context) { { user: agent, account: account, account_user: account.account_users.first } }
permissions :index?, :show?, :playground? do
permissions :index?, :show?, :playground?, :metrics?, :faq_stats? do
context 'when administrator' do
it { expect(assistant_policy).to permit(administrator_context, assistant) }
end
@@ -52,6 +52,12 @@ RSpec.describe Captain::Assistant::AgentRunnerService do
expect(service.instance_variable_get(:@callbacks)).to eq(callbacks)
end
it 'accepts the trigger message id' do
service = described_class.new(assistant: assistant, conversation: conversation, trigger_message_id: 123)
expect(service.instance_variable_get(:@trigger_message_id)).to eq(123)
end
end
describe '#generate_response' do
@@ -99,6 +105,18 @@ RSpec.describe Captain::Assistant::AgentRunnerService do
service.generate_response(message_history: message_history)
end
it 'adds the trigger message id to the runner state' do
service = described_class.new(assistant: assistant, conversation: conversation, trigger_message_id: 123)
expect(mock_runner).to receive(:run).with(
'I need help with my account',
context: hash_including(state: hash_including(trigger_message_id: 123)),
max_turns: 10
)
service.generate_response(message_history: message_history)
end
context 'when the latest user message is multimodal' do
let(:multimodal_message_history) do
[
@@ -446,6 +464,7 @@ RSpec.describe Captain::Assistant::AgentRunnerService do
attributes = provider.generation_attributes(nil, nil, message)
expect(attributes['langfuse.observation.metadata.generation_stage']).to eq('final_response')
expect(attributes['langfuse.observation.metadata.discarded']).to eq('false')
end
it 'marks tool call generations separately from final responses' do
@@ -456,6 +475,18 @@ RSpec.describe Captain::Assistant::AgentRunnerService do
expect(attributes['langfuse.observation.metadata.generation_stage']).to eq('tool_call')
end
it 'marks a generation as discarded when a newer message has arrived' do
trigger_message = create(:message, conversation: conversation, message_type: :incoming)
runner_service = described_class.new(assistant: assistant, conversation: conversation, trigger_message_id: trigger_message.id)
attribute_provider = Captain::Assistant::InstrumentationAttributeProvider.new(runner_service)
message = instance_double(RubyLLM::Message, tool_calls: {})
create(:message, conversation: conversation, message_type: :incoming)
attributes = attribute_provider.generation_attributes(nil, nil, message)
expect(attributes['langfuse.observation.metadata.discarded']).to eq('true')
end
end
describe '#build_state' do
@@ -574,6 +605,7 @@ RSpec.describe Captain::Assistant::AgentRunnerService do
tool_complete_callback.call(Captain::Tools::HandoffTool.new(assistant).name, 'ok', context_wrapper)
expect(root_span).to receive(:set_attribute).with('langfuse.trace.metadata.discarded', 'false')
expect(root_span).to receive(:set_attribute).with('langfuse.trace.metadata.credit_used', 'false')
run_complete_callback.call('assistant', nil, context_wrapper)
end
@@ -588,6 +620,7 @@ RSpec.describe Captain::Assistant::AgentRunnerService do
tool_complete_callback = block
runner
end
allow(runner).to receive(:on_run_complete).and_return(runner)
service.send(:add_usage_metadata_callback, runner)
@@ -599,15 +632,24 @@ RSpec.describe Captain::Assistant::AgentRunnerService do
expect(context_wrapper.context[:captain_v2_handoff_tool_called]).to be true
end
it 'does not register OTEL run callback when OTEL is disabled' do
service = described_class.new(assistant: assistant, conversation: conversation)
it 'tracks discarded responses when OTEL is disabled' do
trigger_message = create(:message, conversation: conversation, message_type: :incoming)
service = described_class.new(assistant: assistant, conversation: conversation, trigger_message_id: trigger_message.id)
runner = instance_double(Agents::AgentRunner)
run_complete_callback = nil
allow(ChatwootApp).to receive(:otel_enabled?).and_return(false)
allow(runner).to receive(:on_tool_complete).and_return(runner)
expect(runner).not_to receive(:on_run_complete)
allow(runner).to receive(:on_run_complete) do |&block|
run_complete_callback = block
runner
end
service.send(:add_usage_metadata_callback, runner)
create(:message, conversation: conversation, message_type: :incoming)
run_complete_callback.call('assistant', nil, Struct.new(:context).new({}))
expect(service.response_discarded?).to be true
end
it 'sets credit_used=true when handoff tool is not used' do
@@ -629,9 +671,38 @@ RSpec.describe Captain::Assistant::AgentRunnerService do
service.send(:add_usage_metadata_callback, runner)
expect(root_span).to receive(:set_attribute).with('langfuse.trace.metadata.discarded', 'false')
expect(root_span).to receive(:set_attribute).with('langfuse.trace.metadata.credit_used', 'true')
run_complete_callback.call('assistant', nil, context_wrapper)
end
it 'marks the trace discarded and does not use credit when a newer message arrived' do
trigger_message = create(:message, conversation: conversation, message_type: :incoming)
service = described_class.new(assistant: assistant, conversation: conversation, trigger_message_id: trigger_message.id)
runner = instance_double(Agents::AgentRunner)
run_complete_callback = nil
span_class = Class.new do
def set_attribute(*); end
end
root_span = instance_double(span_class)
context_wrapper = Struct.new(:context).new({ __otel_tracing: { root_span: root_span } })
allow(ChatwootApp).to receive(:otel_enabled?).and_return(true)
allow(runner).to receive(:on_tool_complete).and_return(runner)
allow(runner).to receive(:on_run_complete) do |&block|
run_complete_callback = block
runner
end
service.send(:add_usage_metadata_callback, runner)
create(:message, conversation: conversation, message_type: :incoming)
expect(root_span).to receive(:set_attribute).with('langfuse.trace.metadata.discarded', 'true')
expect(root_span).to receive(:set_attribute).with('langfuse.trace.metadata.credit_used', 'false')
run_complete_callback.call('assistant', nil, context_wrapper)
expect(service.response_discarded?).to be true
end
end
describe 'constants' do
@@ -22,9 +22,13 @@ RSpec.describe MessageTemplates::HookExecutionService do
end
it 'schedules captain response job for incoming messages on pending conversations' do
expect(Captain::Conversation::ResponseBuilderJob).to receive(:perform_later).with(conversation, assistant)
allow(account).to receive(:feature_enabled?).and_call_original
allow(account).to receive(:feature_enabled?).with('captain_integration_v2').and_return(true)
allow(Captain::Conversation::ResponseBuilderJob).to receive(:perform_later)
create(:message, conversation: conversation, message_type: :incoming, account: account)
message = create(:message, conversation: conversation, message_type: :incoming, account: account)
expect(Captain::Conversation::ResponseBuilderJob).to have_received(:perform_later).with(conversation, assistant, message.id)
end
end
@@ -1,11 +0,0 @@
FactoryBot.define do
factory :automation_rule_pending_execution do
account
automation_rule { association :automation_rule, account: account }
conversation { association :conversation, account: account }
# Derive from production so the row is episode_current (matches the conversation's status).
episode_key { AutomationRulePendingExecution.episode_key_for(conversation, nil) }
due_at { 1.hour.from_now }
status { :pending }
end
end
@@ -1,155 +0,0 @@
require 'rails_helper'
RSpec.describe AutomationRules::ProcessPendingExecutionJob do
subject(:job) { described_class.new }
let(:account) { create(:account) }
let(:conversation) { create(:conversation, account: account, status: :pending) }
let(:rule) do
create(:automation_rule, account: account, event_name: 'conversation_updated', execution_delay: 60,
conditions: [{ 'values' => ['pending'], 'attribute_key' => 'status', 'query_operator' => nil,
'filter_operator' => 'equal_to' }],
actions: [{ 'action_name' => 'add_label', 'action_params' => ['stale'] }])
end
let(:pending_execution) do
AutomationRulePendingExecution.schedule(rule: rule, conversation: conversation)
# The sweep only enqueues due rows, so make it due before the job runs.
AutomationRulePendingExecution.last.tap { |row| row.update!(due_at: 1.minute.ago) }
end
before do
GlobalConfig.clear_cache
account.enable_features!('delayed_automations')
end
it 'runs the actions and marks the row executed when every guard passes' do
job.perform(pending_execution.reload)
expect(pending_execution.reload).to be_executed
expect(conversation.reload.label_list).to include('stale')
end
it 'skips with rule_inactive when the rule was disabled' do
rule.update!(active: false)
job.perform(pending_execution.reload)
expect(pending_execution.reload).to be_skipped
expect(pending_execution.skip_reason).to eq('rule_inactive')
expect(conversation.reload.label_list).to be_empty
end
it 'pauses (keeps pending) while the account flag is off, then fires when re-enabled' do
account.disable_features!('delayed_automations')
job.perform(pending_execution.reload)
expect(pending_execution.reload).to be_pending
expect(conversation.reload.label_list).to be_empty
account.enable_features!('delayed_automations')
described_class.new.perform(pending_execution.reload)
expect(pending_execution.reload).to be_executed
expect(conversation.reload.label_list).to include('stale')
end
it 'skips with episode_moved when the conversation left the armed status' do
pending_execution
conversation.update!(status: :resolved)
job.perform(pending_execution.reload)
expect(pending_execution.reload).to be_skipped
expect(pending_execution.skip_reason).to eq('episode_moved')
expect(conversation.reload.label_list).to be_empty
end
it 'skips with conditions_changed when the conversation drifts but the episode is intact' do
# A message_created (reply-chase) rule whose extra condition is the conversation status.
message_rule = create(:automation_rule, account: account, event_name: 'message_created', execution_delay: 60,
conditions: [{ 'values' => ['pending'], 'attribute_key' => 'status',
'query_operator' => nil, 'filter_operator' => 'equal_to' }],
actions: [{ 'action_name' => 'add_label', 'action_params' => ['stale'] }])
agent_reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
AutomationRulePendingExecution.schedule(rule: message_rule, conversation: conversation, message: agent_reply)
row = AutomationRulePendingExecution.last.tap { |r| r.update!(due_at: 1.minute.ago) }
# Status change fails the condition but leaves the reply_chase episode (max incoming id) intact.
conversation.update!(status: :open)
job.perform(row)
expect(row.reload).to be_skipped
expect(row.skip_reason).to eq('conditions_changed')
end
it 'skips with expired when the row is past the due window' do
pending_execution.update!(due_at: 4.days.ago)
job.perform(pending_execution.reload)
expect(pending_execution.reload).to be_skipped
expect(pending_execution.skip_reason).to eq('expired')
expect(conversation.reload.label_list).to be_empty
end
it 'leaves the row untouched without executing when the kill switch is set' do
create(:installation_config, name: 'DISABLE_DELAYED_AUTOMATIONS', serialized_value: { value: true }.with_indifferent_access)
GlobalConfig.clear_cache
job.perform(pending_execution.reload)
expect(pending_execution.reload).to be_pending
expect(conversation.reload.label_list).to be_empty
end
it 'runs the actions once when the same row is processed twice concurrently' do
allow(AutomationRules::ActionService).to receive(:new).and_call_original
duplicate = AutomationRulePendingExecution.find(pending_execution.id)
job.perform(pending_execution.reload)
described_class.new.perform(duplicate)
expect(AutomationRules::ActionService).to have_received(:new).once
expect(pending_execution.reload).to be_executed
end
it 'leaves the row processing and reports the error when an action blows up' do
action_service = instance_double(AutomationRules::ActionService)
allow(AutomationRules::ActionService).to receive(:new).and_return(action_service)
allow(action_service).to receive(:perform).and_raise(StandardError, 'boom')
allow(ChatwootExceptionTracker).to receive(:new).and_call_original
job.perform(pending_execution.reload)
expect(pending_execution.reload).to be_processing
expect(ChatwootExceptionTracker).to have_received(:new)
end
it 'sends the follow-up exactly once for the reply-chase story' do
message_rule = create(:automation_rule, account: account, event_name: 'message_created', execution_delay: 60,
conditions: [{ 'values' => ['outgoing'], 'attribute_key' => 'message_type',
'query_operator' => nil, 'filter_operator' => 'equal_to' }],
actions: [{ 'action_name' => 'send_message', 'action_params' => ['Just checking in'] }])
agent_reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
AutomationRulePendingExecution.schedule(rule: message_rule, conversation: conversation, message: agent_reply)
row = AutomationRulePendingExecution.last.tap { |r| r.update!(due_at: 1.minute.ago) }
job.perform(row.reload)
expect(row.reload).to be_executed
expect(conversation.messages.outgoing.where(content: 'Just checking in').count).to eq(1)
end
it 'cancels the follow-up when the customer replied before it was due' do
message_rule = create(:automation_rule, account: account, event_name: 'message_created', execution_delay: 60,
conditions: [{ 'values' => ['outgoing'], 'attribute_key' => 'message_type',
'query_operator' => nil, 'filter_operator' => 'equal_to' }],
actions: [{ 'action_name' => 'send_message', 'action_params' => ['Just checking in'] }])
agent_reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
AutomationRulePendingExecution.schedule(rule: message_rule, conversation: conversation, message: agent_reply)
row = AutomationRulePendingExecution.last.tap { |r| r.update!(due_at: 1.minute.ago) }
create(:message, conversation: conversation, account: account, message_type: :incoming)
job.perform(row.reload)
expect(row.reload).to be_skipped
expect(row.skip_reason).to eq('episode_moved')
expect(conversation.messages.outgoing.pluck(:content)).not_to include('Just checking in')
end
end
@@ -1,62 +0,0 @@
require 'rails_helper'
RSpec.describe AutomationRules::TriggerPendingExecutionsJob do
subject(:job) { described_class.new }
let(:account) { create(:account) }
let(:conversation) { create(:conversation, account: account) }
before do
GlobalConfig.clear_cache
account.enable_features!('delayed_automations')
end
it 'enqueues a per-row job for due pending rows but not future ones' do
due_row = create(:automation_rule_pending_execution, account: account, conversation: conversation, due_at: 1.minute.ago)
future_row = create(:automation_rule_pending_execution, account: account, due_at: 1.hour.from_now)
expect { job.perform }.to have_enqueued_job(AutomationRules::ProcessPendingExecutionJob).exactly(:once)
expect(AutomationRules::ProcessPendingExecutionJob).to have_been_enqueued.with(due_row)
expect(AutomationRules::ProcessPendingExecutionJob).not_to have_been_enqueued.with(future_row)
end
it 're-enqueues stale processing rows so they get retried' do
stale_row = travel_to(20.minutes.ago) do
create(:automation_rule_pending_execution, account: account, conversation: conversation, status: :processing, due_at: 19.minutes.from_now)
end
expect { job.perform }.to have_enqueued_job(AutomationRules::ProcessPendingExecutionJob).with(stale_row)
end
it 'caps enqueues at the configured sweep limit' do
create(:installation_config, name: 'AUTOMATION_PENDING_EXECUTIONS_SWEEP_LIMIT', serialized_value: { value: 1 }.with_indifferent_access)
create_list(:automation_rule_pending_execution, 2, account: account, due_at: 1.minute.ago)
expect { job.perform }.to have_enqueued_job(AutomationRules::ProcessPendingExecutionJob).exactly(:once)
end
it 'purges terminal rows past the retention window' do
old_row = travel_to(31.days.ago) { create(:automation_rule_pending_execution, account: account, status: :executed) }
job.perform
expect { old_row.reload }.to raise_error(ActiveRecord::RecordNotFound)
end
it 'skips rows for accounts with delayed automations disabled so they cannot starve others' do
enabled_row = create(:automation_rule_pending_execution, account: account, conversation: conversation, due_at: 1.minute.ago)
disabled_account = create(:account) # delayed_automations off by default
create(:automation_rule_pending_execution, account: disabled_account, due_at: 2.minutes.ago)
expect { job.perform }.to have_enqueued_job(AutomationRules::ProcessPendingExecutionJob).exactly(:once)
expect(AutomationRules::ProcessPendingExecutionJob).to have_been_enqueued.with(enabled_row)
end
it 'does nothing when the kill switch is set' do
create(:installation_config, name: 'DISABLE_DELAYED_AUTOMATIONS', serialized_value: { value: true }.with_indifferent_access)
GlobalConfig.clear_cache
create(:automation_rule_pending_execution, account: account, conversation: conversation, due_at: 1.minute.ago)
expect { job.perform }.not_to have_enqueued_job(AutomationRules::ProcessPendingExecutionJob)
end
end
@@ -247,37 +247,4 @@ describe AutomationRuleListener do
end
end
end
describe 'delayed rules' do
let!(:automation_rule) { create(:automation_rule, event_name: 'conversation_updated', account: account, execution_delay: 60) }
let(:event) do
Events::Base.new('conversation_updated', Time.zone.now, { conversation: conversation, changed_attributes: {} })
end
before { allow(condition_match).to receive(:present?).and_return(true) }
context 'when the delayed_automations feature is enabled' do
before { account.enable_features!('delayed_automations') }
it 'records a pending execution instead of running actions' do
expect { listener.conversation_updated(event) }.to change(AutomationRulePendingExecution, :count).by(1)
expect(AutomationRules::ActionService).not_to have_received(:new)
expect(AutomationRulePendingExecution.last.due_at).to be_within(5.seconds).of(60.minutes.from_now)
end
it 'still runs rules without a delay immediately' do
automation_rule.update!(execution_delay: nil)
expect { listener.conversation_updated(event) }.not_to change(AutomationRulePendingExecution, :count)
expect(AutomationRules::ActionService).to have_received(:new).with(automation_rule, account, conversation)
end
end
context 'when the delayed_automations feature is disabled' do
it 'neither arms a pending execution nor falls back to immediate execution' do
expect { listener.conversation_updated(event) }.not_to change(AutomationRulePendingExecution, :count)
expect(AutomationRules::ActionService).not_to have_received(:new)
end
end
end
end
+1 -17
View File
@@ -112,21 +112,6 @@ RSpec.describe Account do
end
end
describe 'resuming delayed automations' do
let(:account) { create(:account) }
it 'enqueues the resume job when delayed_automations is turned back on' do
expect { account.enable_features!('delayed_automations') }
.to have_enqueued_job(AutomationRules::ResumePausedExecutionsJob).with(account)
end
it 'does not enqueue the resume job when the flag is turned off' do
account.enable_features!('delayed_automations')
expect { account.disable_features!('delayed_automations') }
.not_to have_enqueued_job(AutomationRules::ResumePausedExecutionsJob)
end
end
describe 'feature flag columns' do
let(:account) { described_class.new(name: 'Test Account') }
@@ -137,8 +122,7 @@ RSpec.describe Account do
feature_data_import: 1 << 1,
feature_api_and_webhooks: 1 << 2,
feature_whatsapp_reconfigure: 1 << 3,
feature_whatsapp_embedded_signup_inbox_creation: 1 << 4,
feature_delayed_automations: 1 << 5
feature_whatsapp_embedded_signup_inbox_creation: 1 << 4
)
expect(described_class.flag_mapping['feature_flags_ext_1'][:feature_whatsapp_manual_transfer]).to eq(1)
expect(described_class.flag_mapping['feature_flags_ext_1'][:feature_data_import]).to eq(2)
@@ -1,305 +0,0 @@
require 'rails_helper'
RSpec.describe AutomationRulePendingExecution do
let(:account) { create(:account) }
let(:conversation) { create(:conversation, account: account) }
let(:rule) do
create(:automation_rule, account: account, event_name: 'conversation_updated', execution_delay: 60,
actions: [{ 'action_name' => 'add_label', 'action_params' => ['stale'] }])
end
describe '.episode_key_for' do
it 'derives status episodes from status_changed_at' do
expect(described_class.episode_key_for(conversation, nil)).to eq("status:#{conversation.status_changed_at.strftime('%s%6N')}")
end
it 'matches between an in-memory arm and a DB-reloaded fire (no float rounding drift)' do
conversation.status_changed_at = Time.zone.at(1_784_102_080.844761923r)
arm_key = described_class.episode_key_for(conversation, nil)
conversation.save!
expect(arm_key).to eq(described_class.episode_key_for(conversation.reload, nil))
end
it 'falls back to created_at when status_changed_at is blank' do
conversation.update!(status_changed_at: nil)
expect(described_class.episode_key_for(conversation.reload, nil)).to eq("status:#{conversation.created_at.strftime('%s%6N')}")
end
it 'derives awaiting_agent episodes from waiting_since (sub-second) for incoming messages' do
message = create(:message, conversation: conversation, account: account, message_type: :incoming)
expect(described_class.episode_key_for(conversation.reload, message)).to eq("awaiting_agent:#{conversation.waiting_since.strftime('%s%6N')}")
end
it 'distinguishes two waiting periods that fall within the same second' do
message = create(:message, conversation: conversation, account: account, message_type: :incoming)
first_key = described_class.episode_key_for(conversation.reload, message)
# Agent replies then customer re-waits within the same second: keys must differ.
conversation.update!(waiting_since: conversation.waiting_since + 0.4)
expect(described_class.episode_key_for(conversation.reload, message)).not_to eq(first_key)
end
it 'arms an awaiting_agent episode from the message created_at when waiting_since is not yet written' do
message = create(:message, conversation: conversation, account: account, message_type: :incoming)
# Simulate the race where the listener arms before update_waiting_since commits.
conversation.update!(waiting_since: nil)
armed_key = described_class.arm_episode_key_for(conversation.reload, message)
# Once waiting_since settles to the message's created_at, the strict fire-time key matches.
conversation.update!(waiting_since: message.created_at)
expect(armed_key).to eq("awaiting_agent:#{message.created_at.strftime('%s%6N')}")
expect(armed_key).to eq(described_class.episode_key_for(conversation.reload, message))
end
it 'derives reply_chase episodes from the max incoming message id for outgoing messages' do
incoming = create(:message, conversation: conversation, account: account, message_type: :incoming)
outgoing = create(:message, conversation: conversation, account: account, message_type: :outgoing)
expect(described_class.episode_key_for(conversation.reload, outgoing)).to eq("reply_chase:#{incoming.id}")
end
it 'uses 0 for reply_chase when there is no incoming message' do
outgoing = create(:message, conversation: conversation, account: account, message_type: :outgoing)
expect(described_class.episode_key_for(conversation.reload, outgoing)).to eq('reply_chase:0')
end
end
describe '.schedule' do
it 'creates a pending row due after the rule delay' do
described_class.schedule(rule: rule, conversation: conversation)
row = described_class.last
expect(row).to have_attributes(account_id: account.id, conversation_id: conversation.id, status: 'pending')
expect(row.due_at).to be_within(5.seconds).of(60.minutes.from_now)
end
it 'anchors due_at to the event time, not when a backlogged listener runs' do
conversation.update!(status_changed_at: 30.minutes.ago)
described_class.schedule(rule: rule, conversation: conversation)
# A 60-minute rule on a status that changed 30 minutes ago is already 30 minutes into its wait.
expect(described_class.last.due_at).to be_within(5.seconds).of(30.minutes.from_now)
end
it 'does not reset the clock for a repeated status episode' do
described_class.schedule(rule: rule, conversation: conversation)
original_due_at = described_class.last.due_at
travel_to(30.minutes.from_now) { described_class.schedule(rule: rule, conversation: conversation) }
expect(described_class.count).to eq(1)
expect(described_class.last.due_at).to be_within(1.second).of(original_due_at)
end
it 'moves the clock and anchor for a repeated reply_chase episode' do
first_reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
described_class.schedule(rule: rule, conversation: conversation, message: first_reply)
travel_to(30.minutes.from_now) do
second_reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
described_class.schedule(rule: rule, conversation: conversation, message: second_reply)
# due_at is re-anchored to the new reply's created_at, not the original schedule time.
expect(described_class.count).to eq(1)
expect(described_class.last.message_id).to eq(second_reply.id)
expect(described_class.last.due_at).to be_within(5.seconds).of(60.minutes.from_now)
end
end
it 'does not let a late older reply move the reply_chase clock backwards' do
older_reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
newer_reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
# The newer reply's job runs first and arms the episode.
described_class.schedule(rule: rule, conversation: conversation, message: newer_reply)
armed_due_at = described_class.last.due_at
# The older reply's job arrives late; it must not pull the clock or message_id back.
described_class.schedule(rule: rule, conversation: conversation, message: older_reply)
expect(described_class.count).to eq(1)
expect(described_class.last.message_id).to eq(newer_reply.id)
expect(described_class.last.due_at).to eq(armed_due_at)
end
it 'does not re-arm an executed reply_chase episode' do
reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
described_class.schedule(rule: rule, conversation: conversation, message: reply)
described_class.last.update!(status: :executed)
described_class.schedule(rule: rule, conversation: conversation, message: reply)
expect(described_class.count).to eq(1)
expect(described_class.last).to be_executed
end
it 're-arms a condition-skipped episode when a newer qualifying message arrives' do
first_reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
described_class.schedule(rule: rule, conversation: conversation, message: first_reply)
described_class.last.update!(status: :skipped, skip_reason: 'conditions_changed')
second_reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
described_class.schedule(rule: rule, conversation: conversation, message: second_reply)
row = described_class.last
expect(described_class.count).to eq(1)
expect(row).to be_pending
expect(row.skip_reason).to be_nil
expect(row.message_id).to eq(second_reply.id)
end
it 're-anchors a reply_chase row stuck in processing when a newer reply arrives' do
first_reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
described_class.schedule(rule: rule, conversation: conversation, message: first_reply)
# The worker claimed the row and then died, leaving it in processing with the old clock.
described_class.last.update!(status: :processing)
travel_to(30.minutes.from_now) do
second_reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
described_class.schedule(rule: rule, conversation: conversation, message: second_reply)
row = described_class.last
expect(described_class.count).to eq(1)
expect(row).to be_pending
expect(row.message_id).to eq(second_reply.id)
expect(row.due_at).to be_within(5.seconds).of(60.minutes.from_now)
end
end
it 'does not re-arm an episode skipped for a non-condition reason' do
reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
described_class.schedule(rule: rule, conversation: conversation, message: reply)
described_class.last.update!(status: :skipped, skip_reason: 'episode_moved')
newer_reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
described_class.schedule(rule: rule, conversation: conversation, message: newer_reply)
expect(described_class.count).to eq(1)
expect(described_class.last).to be_skipped
end
it 'keeps the awaiting_agent clock but tracks the newest incoming message' do
first_message = create(:message, conversation: conversation, account: account, message_type: :incoming)
described_class.schedule(rule: rule, conversation: conversation, message: first_message)
original_due_at = described_class.last.due_at
second_message = create(:message, conversation: conversation, account: account, message_type: :incoming)
travel_to(30.minutes.from_now) { described_class.schedule(rule: rule, conversation: conversation, message: second_message) }
row = described_class.last
expect(described_class.count).to eq(1)
# The wait still counts from waiting_since, so the clock is unchanged...
expect(row.due_at).to be_within(1.second).of(original_due_at)
# ...but the row tracks the newest qualifying message, not the stale first one.
expect(row.message_id).to eq(second_message.id)
end
it 'keeps the newest message when an older incoming collision arrives last' do
first_message = create(:message, conversation: conversation, account: account, message_type: :incoming)
described_class.schedule(rule: rule, conversation: conversation, message: first_message)
older = create(:message, conversation: conversation, account: account, message_type: :incoming)
newer = create(:message, conversation: conversation, account: account, message_type: :incoming)
# The newer message re-arms first; a late older-message collision must not overwrite it.
described_class.schedule(rule: rule, conversation: conversation, message: newer)
described_class.schedule(rule: rule, conversation: conversation, message: older)
expect(described_class.count).to eq(1)
expect(described_class.last.message_id).to eq(newer.id)
end
end
describe '#episode_current?' do
it 'is true while the conversation stays in the armed status' do
described_class.schedule(rule: rule, conversation: conversation)
expect(described_class.last.episode_current?).to be(true)
end
it 'is false after a status transition' do
described_class.schedule(rule: rule, conversation: conversation)
conversation.update!(status: :resolved)
expect(described_class.last.reload.episode_current?).to be(false)
end
it 'is false for awaiting_agent episodes once the agent replies (waiting_since cleared)' do
message = create(:message, conversation: conversation, account: account, message_type: :incoming)
described_class.schedule(rule: rule, conversation: conversation, message: message)
conversation.update!(waiting_since: nil)
expect(described_class.last.episode_current?).to be(false)
end
it 'is false for reply_chase episodes once the customer replies' do
reply = create(:message, conversation: conversation, account: account, message_type: :outgoing)
described_class.schedule(rule: rule, conversation: conversation, message: reply)
create(:message, conversation: conversation, account: account, message_type: :incoming)
expect(described_class.last.episode_current?).to be(false)
end
end
describe '#claim!' do
let(:row) { create(:automation_rule_pending_execution, account: account, conversation: conversation, due_at: 1.minute.ago) }
it 'claims a pending row exactly once so a duplicate enqueue cannot double-fire' do
expect(row.claim!).to be(true)
expect(row.reload).to be_processing
expect(described_class.find(row.id).claim!).to be(false)
end
it 'does not claim a row whose due_at was pushed into the future (reply-chase reschedule)' do
row.update!(due_at: 1.hour.from_now)
expect(row.claim!).to be(false)
end
it 'does not claim terminal rows' do
row.update!(status: :executed)
expect(row.claim!).to be(false)
end
it 'reclaims a processing row only after its lock goes stale' do
row.update!(status: :processing)
expect(row.claim!).to be(false)
travel_to(20.minutes.from_now) { expect(row.claim!).to be(true) }
end
end
describe '.sweepable' do
it 'selects due pending rows and stale processing rows, but not future or fresh ones' do
due = create(:automation_rule_pending_execution, account: account, conversation: conversation, due_at: 1.minute.ago)
create(:automation_rule_pending_execution, account: account, due_at: 1.hour.from_now)
create(:automation_rule_pending_execution, account: account, status: :processing)
stale = travel_to(20.minutes.ago) do
create(:automation_rule_pending_execution, account: account, conversation: conversation, status: :processing)
end
expect(described_class.sweepable).to contain_exactly(due, stale)
end
end
describe '.purge_terminal!' do
it 'deletes terminal rows past the retention window and keeps everything else' do
old_executed = travel_to(31.days.ago) { create(:automation_rule_pending_execution, account: account, status: :executed) }
recent_skipped = create(:automation_rule_pending_execution, account: account, status: :skipped)
pending = create(:automation_rule_pending_execution, account: account, conversation: conversation)
described_class.purge_terminal!
expect(described_class.pluck(:id)).to contain_exactly(recent_skipped.id, pending.id)
expect { old_executed.reload }.to raise_error(ActiveRecord::RecordNotFound)
end
end
describe '.reschedule_paused' do
it 'resets rows overdue past the window so a resumed account replays them instead of expiring' do
expired = create(:automation_rule_pending_execution, account: account, conversation: conversation, due_at: 5.days.ago)
within_window = create(:automation_rule_pending_execution, account: account, due_at: 2.days.ago)
described_class.reschedule_paused(account)
expect(expired.reload.due_at).to be_within(5.seconds).of(Time.current)
expect(within_window.reload.due_at).to be_within(5.seconds).of(2.days.ago)
end
end
end
-114
View File
@@ -137,118 +137,4 @@ RSpec.describe AutomationRule do
end
end
end
describe 'execution_delay validations' do
let(:rule) { build(:automation_rule, account: create(:account)) }
it 'allows nil (immediate execution)' do
rule.execution_delay = nil
expect(rule).to be_valid
end
it 'allows delays between 10 minutes and 30 days' do
rule.execution_delay = 240
expect(rule).to be_valid
end
it 'rejects delays below 10 minutes' do
rule.execution_delay = 5
expect(rule).not_to be_valid
expect(rule.errors[:execution_delay]).to be_present
end
it 'rejects delays above 30 days' do
rule.execution_delay = 43_201
expect(rule).not_to be_valid
end
it 'rejects non-integer delays' do
rule.execution_delay = 10.5
expect(rule).not_to be_valid
end
it 'rejects a delay combined with an attribute_changed condition' do
rule.execution_delay = 60
rule.conditions = [{ 'attribute_key' => 'status', 'filter_operator' => 'attribute_changed',
'values' => { 'from' => ['open'], 'to' => ['pending'] }, 'query_operator' => nil }]
expect(rule).not_to be_valid
expect(rule.errors[:execution_delay]).to include('cannot be used with attribute_changed conditions.')
end
it 'rejects a delayed conversation-level rule with a mutable non-status condition' do
rule.event_name = 'conversation_updated'
rule.execution_delay = 60
rule.conditions = [{ 'attribute_key' => 'priority', 'filter_operator' => 'equal_to', 'values' => ['urgent'], 'query_operator' => nil }]
expect(rule).not_to be_valid
expect(rule.errors[:execution_delay]).to include('only supports status and inbox conditions for conversation-level events.')
end
it 'allows a delayed conversation-level rule with only status conditions' do
rule.event_name = 'conversation_updated'
rule.execution_delay = 60
rule.conditions = [{ 'attribute_key' => 'status', 'filter_operator' => 'equal_to', 'values' => ['pending'], 'query_operator' => nil }]
expect(rule).to be_valid
end
it 'allows a delayed conversation_created rule (arms on creation)' do
rule.event_name = 'conversation_created'
rule.execution_delay = 10
rule.conditions = [{ 'attribute_key' => 'status', 'filter_operator' => 'equal_to', 'values' => ['open'], 'query_operator' => nil }]
expect(rule).to be_valid
end
it 'allows a delayed conversation-level rule scoped by status and inbox (immutable)' do
rule.event_name = 'conversation_updated'
rule.execution_delay = 60
rule.conditions = [{ 'attribute_key' => 'status', 'filter_operator' => 'equal_to', 'values' => ['pending'], 'query_operator' => 'AND' },
{ 'attribute_key' => 'inbox_id', 'filter_operator' => 'equal_to', 'values' => [1], 'query_operator' => nil }]
expect(rule).to be_valid
end
it 'allows a delayed message_created rule with a non-status condition' do
rule.event_name = 'message_created'
rule.execution_delay = 60
rule.conditions = [{ 'attribute_key' => 'message_type', 'filter_operator' => 'equal_to', 'values' => ['outgoing'], 'query_operator' => nil }]
expect(rule).to be_valid
end
end
describe 'discarding stale pending executions on edit' do
let(:account) { create(:account) }
let(:conversation) { create(:conversation, account: account, status: :pending) }
let(:status_condition) { { 'attribute_key' => 'status', 'filter_operator' => 'equal_to', 'values' => ['pending'], 'query_operator' => nil } }
let(:rule) do
create(:automation_rule, account: account, event_name: 'conversation_updated', execution_delay: 60,
conditions: [status_condition], actions: [{ 'action_name' => 'add_label', 'action_params' => ['stale'] }])
end
before { AutomationRulePendingExecution.schedule(rule: rule, conversation: conversation) }
it 'discards armed rows when the actions change' do
rule.update!(actions: [{ 'action_name' => 'add_label', 'action_params' => ['urgent'] }])
expect(rule.pending_executions.pending).to be_empty
end
it 'discards armed rows when the delay changes' do
rule.update!(execution_delay: 120)
expect(rule.pending_executions.pending).to be_empty
end
it 'discards a stale processing row that the sweep would otherwise reclaim' do
rule.pending_executions.first.update!(status: :processing)
rule.update!(actions: [{ 'action_name' => 'add_label', 'action_params' => ['urgent'] }])
expect(rule.pending_executions.armed).to be_empty
end
it 'frees the episode slot so the new definition re-arms for the same episode' do
rule.update!(actions: [{ 'action_name' => 'add_label', 'action_params' => ['urgent'] }])
AutomationRulePendingExecution.schedule(rule: rule, conversation: conversation)
expect(rule.pending_executions.pending.count).to eq(1)
end
it 'leaves armed rows untouched on a name-only edit' do
rule.update!(name: 'Renamed rule')
expect(rule.pending_executions.pending.count).to eq(1)
end
end
end
-24
View File
@@ -1230,28 +1230,4 @@ RSpec.describe Conversation do
end
end
end
describe '#status_changed_at' do
let(:conversation) { create(:conversation) }
it 'is set on create' do
expect(conversation.status_changed_at).to be_present
end
it 'is updated on every status transition' do
original = conversation.status_changed_at
travel_to(1.hour.from_now) { conversation.update!(status: :resolved) }
expect(conversation.reload.status_changed_at).to be > original
end
it 'is untouched by non-status saves' do
original = conversation.status_changed_at
travel_to(1.hour.from_now) { conversation.update!(priority: :high) }
expect(conversation.reload.status_changed_at).to be_within(1.second).of(original)
end
end
end