-
+
-
+
{{ data.label || 'Shopify' }}
diff --git a/app/javascript/dashboard/i18n/locale/en/integrations.json b/app/javascript/dashboard/i18n/locale/en/integrations.json
index f51cedb7f..b92cf95ba 100644
--- a/app/javascript/dashboard/i18n/locale/en/integrations.json
+++ b/app/javascript/dashboard/i18n/locale/en/integrations.json
@@ -735,25 +735,18 @@
},
"WORKFLOWS": {
"TITLE": "Workflows",
- "DESCRIPTION": "Automate actions triggered by conversation events with a visual workflow builder.",
+ "DESCRIPTION": "Build interactive workflows that Captain can execute during conversations.",
"ADD_BUTTON": "New Workflow",
"DEFAULT_NAME": "New Workflow",
"EMPTY_MESSAGE": "No workflows yet. Create one to automate conversation actions.",
- "TRIGGERS": {
- "CONVERSATION_CREATED": "Conversation Created",
- "MESSAGE_CREATED": "Message Created",
- "CONVERSATION_RESOLVED": "Conversation Resolved"
- },
"PALETTE": {
- "TRIGGERS": "Triggers",
+ "INTERACTIVE": "Interactive",
"LOGIC": "Logic",
"ACTIONS": "Actions",
"SHOPIFY": "Shopify"
},
"NODES": {
- "TRIGGER_CONVERSATION_CREATED": "Conversation Created",
- "TRIGGER_MESSAGE_CREATED": "Message Created",
- "TRIGGER_CONVERSATION_RESOLVED": "Conversation Resolved",
+ "COLLECT_INPUT": "Collect Input",
"CONDITION": "Condition",
"SEND_MESSAGE": "Send Message",
"ADD_LABEL": "Add Label",
@@ -776,6 +769,8 @@
"ATTRIBUTE": "Attribute",
"OPERATOR": "Operator",
"VALUE": "Value",
+ "INPUT_KEY": "Input Key",
+ "PROMPT": "Prompt",
"APPLY": "Apply",
"DELETE": "Delete Node",
"CONDITION_TRUE": "True",
diff --git a/app/javascript/dashboard/routes/dashboard/captain/assistants/workflows/Editor.vue b/app/javascript/dashboard/routes/dashboard/captain/assistants/workflows/Editor.vue
index 07a1ea87e..9c0a0024d 100644
--- a/app/javascript/dashboard/routes/dashboard/captain/assistants/workflows/Editor.vue
+++ b/app/javascript/dashboard/routes/dashboard/captain/assistants/workflows/Editor.vue
@@ -9,10 +9,10 @@ import { VueFlow, useVueFlow } from '@vue-flow/core';
import WorkflowToolbar from 'dashboard/components-next/captain/workflows/WorkflowToolbar.vue';
import NodePalette from 'dashboard/components-next/captain/workflows/NodePalette.vue';
import NodeConfigPanel from 'dashboard/components-next/captain/workflows/NodeConfigPanel.vue';
-import TriggerNode from 'dashboard/components-next/captain/workflows/nodes/TriggerNode.vue';
import ActionNode from 'dashboard/components-next/captain/workflows/nodes/ActionNode.vue';
import ConditionNode from 'dashboard/components-next/captain/workflows/nodes/ConditionNode.vue';
import ShopifyNode from 'dashboard/components-next/captain/workflows/nodes/ShopifyNode.vue';
+import CollectInputNode from 'dashboard/components-next/captain/workflows/nodes/CollectInputNode.vue';
const { t } = useI18n();
const route = useRoute();
@@ -32,9 +32,7 @@ const nodes = ref([]);
const edges = ref([]);
const nodeTypes = {
- trigger_conversation_created: TriggerNode,
- trigger_message_created: TriggerNode,
- trigger_conversation_resolved: TriggerNode,
+ collect_input: CollectInputNode,
condition: ConditionNode,
send_message: ActionNode,
add_label: ActionNode,
@@ -115,14 +113,13 @@ const deleteNode = nodeId => {
}
};
-const saveWorkflow = async ({ name, description, trigger_event, enabled }) => {
+const saveWorkflow = async ({ name, description, enabled }) => {
try {
await store.dispatch('captainWorkflows/update', {
id: workflowId.value,
assistantId: assistantId.value,
name,
description,
- trigger_event,
enabled,
nodes: nodes.value.map(n => ({
id: n.id,
@@ -192,3 +189,40 @@ onMounted(async () => {
+
+
diff --git a/app/javascript/dashboard/routes/dashboard/captain/assistants/workflows/Index.vue b/app/javascript/dashboard/routes/dashboard/captain/assistants/workflows/Index.vue
index 026747bb0..3edc35f57 100644
--- a/app/javascript/dashboard/routes/dashboard/captain/assistants/workflows/Index.vue
+++ b/app/javascript/dashboard/routes/dashboard/captain/assistants/workflows/Index.vue
@@ -20,20 +20,11 @@ const uiFlags = useMapGetter('captainWorkflows/getUIFlags');
const isFetching = computed(() => uiFlags.value.fetchingList);
const workflows = useMapGetter('captainWorkflows/getRecords');
-const TRIGGER_LABELS = {
- conversation_created:
- 'CAPTAIN.ASSISTANTS.WORKFLOWS.TRIGGERS.CONVERSATION_CREATED',
- message_created: 'CAPTAIN.ASSISTANTS.WORKFLOWS.TRIGGERS.MESSAGE_CREATED',
- conversation_resolved:
- 'CAPTAIN.ASSISTANTS.WORKFLOWS.TRIGGERS.CONVERSATION_RESOLVED',
-};
-
const createWorkflow = async () => {
try {
const workflow = await store.dispatch('captainWorkflows/create', {
assistantId: assistantId.value,
name: t('CAPTAIN.ASSISTANTS.WORKFLOWS.DEFAULT_NAME'),
- trigger_event: 'conversation_created',
nodes: [],
edges: [],
});
@@ -144,9 +135,6 @@ onMounted(() => {
>
{{ workflow.description }}
-
- {{ $t(TRIGGER_LABELS[workflow.trigger_event]) }}
-
{ where(enabled: true) }
- scope :for_event, ->(event) { where(trigger_event: event) }
end
diff --git a/enterprise/app/models/captain/workflow_execution.rb b/enterprise/app/models/captain/workflow_execution.rb
index 9ecde31bd..91424cfc7 100644
--- a/enterprise/app/models/captain/workflow_execution.rb
+++ b/enterprise/app/models/captain/workflow_execution.rb
@@ -6,5 +6,5 @@ class Captain::WorkflowExecution < ApplicationRecord
belongs_to :conversation, optional: true
belongs_to :contact, optional: true
- enum :status, { pending: 0, running: 1, completed: 2, failed: 3 }
+ enum :status, { pending: 0, running: 1, completed: 2, failed: 3, waiting_for_input: 4 }
end
diff --git a/enterprise/app/services/captain/llm/assistant_chat_service.rb b/enterprise/app/services/captain/llm/assistant_chat_service.rb
index 5a2976e39..3b06388a7 100644
--- a/enterprise/app/services/captain/llm/assistant_chat_service.rb
+++ b/enterprise/app/services/captain/llm/assistant_chat_service.rb
@@ -1,10 +1,11 @@
class Captain::Llm::AssistantChatService < Llm::BaseAiService
include Captain::ChatHelper
- def initialize(assistant: nil, conversation_id: nil)
+ def initialize(assistant: nil, conversation: nil, conversation_id: nil)
super()
@assistant = assistant
+ @conversation = conversation
@conversation_id = conversation_id
@messages = [system_message]
@@ -28,16 +29,28 @@ class Captain::Llm::AssistantChatService < Llm::BaseAiService
private
def build_tools
- [Captain::Tools::SearchDocumentationService.new(@assistant, user: nil)]
+ tools = [Captain::Tools::SearchDocumentationService.new(@assistant, user: nil)]
+ if @conversation && @assistant.workflows.enabled.exists?
+ tools << Captain::Tools::ExecuteWorkflowService.new(@assistant, conversation: @conversation)
+ end
+ tools
end
def system_message
{
role: 'system',
- content: Captain::Llm::SystemPromptsService.assistant_response_generator(@assistant.name, @assistant.config['product_name'], @assistant.config)
+ content: Captain::Llm::SystemPromptsService.assistant_response_generator(
+ @assistant.name, @assistant.config['product_name'], @assistant.config,
+ workflows: @assistant.workflows.enabled,
+ active_execution: @conversation ? find_active_execution : nil
+ )
}
end
+ def find_active_execution
+ Captain::WorkflowExecution.where(conversation: @conversation, status: :waiting_for_input).last
+ end
+
def persist_message(message, message_type = 'assistant')
# No need to implement
end
diff --git a/enterprise/app/services/captain/llm/system_prompts_service.rb b/enterprise/app/services/captain/llm/system_prompts_service.rb
index ba8834c1f..59b6762bb 100644
--- a/enterprise/app/services/captain/llm/system_prompts_service.rb
+++ b/enterprise/app/services/captain/llm/system_prompts_service.rb
@@ -152,7 +152,7 @@ class Captain::Llm::SystemPromptsService
# rubocop:enable Metrics/MethodLength
# rubocop:disable Metrics/MethodLength
- def assistant_response_generator(assistant_name, product_name, config = {})
+ def assistant_response_generator(assistant_name, product_name, config = {}, workflows: [], active_execution: nil)
assistant_citation_guidelines = if config['feature_citation']
<<~CITATION_TEXT
- Always include citations for any information provided, referencing the specific source (document only - skip if it was derived from a conversation).
@@ -203,6 +203,7 @@ class Captain::Llm::SystemPromptsService
```
- If the answer is not provided in context sections, Respond to the customer and ask whether they want to talk to another support agent . If they ask to Chat with another agent, return `conversation_handoff' as the response in JSON response
#{'- You MUST provide numbered citations at the appropriate places in the text.' if config['feature_citation']}
+ #{workflow_prompt_section(workflows, active_execution)}
SYSTEM_PROMPT_MESSAGE
end
@@ -288,6 +289,39 @@ class Captain::Llm::SystemPromptsService
PROMPT
end
# rubocop:enable Metrics/MethodLength
+
+ private
+
+ def workflow_prompt_section(workflows, active_execution)
+ return '' if workflows.blank? && active_execution.blank?
+
+ sections = []
+ sections << available_workflows_prompt(workflows) if workflows.present?
+ sections << active_execution_prompt(active_execution) if active_execution.present?
+ sections.join("\n")
+ end
+
+ def available_workflows_prompt(workflows)
+ workflow_list = workflows.map { |w| "- ID: #{w.id}, Name: \"#{w.name}\", Description: \"#{w.description}\"" }.join("\n")
+ <<~WORKFLOWS
+ [Available Workflows]
+ You can execute workflows to help the customer. Use the execute_workflow tool to start or continue a workflow.
+ #{workflow_list}
+ WORKFLOWS
+ end
+
+ def active_execution_prompt(active_execution)
+ paused_node = active_execution.workflow.nodes.find { |n| n['id'] == active_execution.current_node_id }
+ input_key = paused_node&.dig('data', 'input_key') || 'user_input'
+ prompt = paused_node&.dig('data', 'prompt') || 'Please provide input'
+
+ <<~EXECUTION
+ [Active Workflow Execution]
+ Execution ID: #{active_execution.id} is waiting for input.
+ Input key: "#{input_key}", Prompt: "#{prompt}"
+ Call execute_workflow(execution_id: #{active_execution.id}, user_input: "") to continue.
+ EXECUTION
+ end
end
end
# rubocop:enable Metrics/ClassLength
diff --git a/enterprise/app/services/captain/tools/base_tool.rb b/enterprise/app/services/captain/tools/base_tool.rb
index 1ec4aaffc..12a6fe916 100644
--- a/enterprise/app/services/captain/tools/base_tool.rb
+++ b/enterprise/app/services/captain/tools/base_tool.rb
@@ -3,9 +3,10 @@ class Captain::Tools::BaseTool < RubyLLM::Tool
attr_accessor :assistant
- def initialize(assistant, user: nil)
+ def initialize(assistant, user: nil, conversation: nil)
@assistant = assistant
@user = user
+ @conversation = conversation
super()
end
diff --git a/enterprise/app/services/captain/tools/execute_workflow_service.rb b/enterprise/app/services/captain/tools/execute_workflow_service.rb
new file mode 100644
index 000000000..789b4be1b
--- /dev/null
+++ b/enterprise/app/services/captain/tools/execute_workflow_service.rb
@@ -0,0 +1,65 @@
+class Captain::Tools::ExecuteWorkflowService < Captain::Tools::BaseTool
+ def self.name
+ 'execute_workflow'
+ end
+
+ description 'Execute or continue a workflow. Use workflow_id to start a new workflow, execution_id to continue a paused one.'
+
+ param :workflow_id, desc: 'ID of workflow to start (for new executions)', required: false
+ param :execution_id, desc: 'ID of paused execution to continue', required: false
+ param :user_input, desc: 'User response for the current step', required: false
+
+ def execute(workflow_id: nil, execution_id: nil, user_input: nil)
+ if execution_id.present?
+ continue_execution(execution_id, user_input)
+ elsif workflow_id.present?
+ start_execution(workflow_id)
+ else
+ 'Provide either workflow_id to start or execution_id to continue a workflow.'
+ end
+ end
+
+ private
+
+ def start_execution(workflow_id)
+ workflow = assistant.workflows.enabled.find_by(id: workflow_id)
+ return "Workflow #{workflow_id} not found or not enabled." unless workflow
+
+ context = build_context
+ result = Captain::Workflows::ExecutionService.new(workflow, context).perform
+ format_result(result)
+ end
+
+ def continue_execution(execution_id, user_input)
+ execution = Captain::WorkflowExecution.find_by(id: execution_id, status: :waiting_for_input)
+ return "Execution #{execution_id} not found or not waiting for input." unless execution
+
+ result = Captain::Workflows::ExecutionService.resume(execution, user_input)
+ format_result(result)
+ end
+
+ def build_context
+ context = { account: assistant.account }
+ if @conversation
+ context[:conversation] = @conversation
+ context[:contact] = @conversation.contact
+ end
+ context
+ end
+
+ def format_result(result)
+ return result.to_json unless result.is_a?(Hash)
+
+ case result[:status]
+ when 'input_required'
+ "Workflow paused. Need user input: #{result[:prompt]} (input_key: #{result[:input_key]}). " \
+ "To continue, call execute_workflow(execution_id: #{result[:execution_id]}, user_input: \"\")."
+ when 'completed'
+ "Workflow completed successfully. Actions taken: #{result[:actions].to_json}"
+ when 'failed'
+ "Workflow failed: #{result[:error]}"
+ else
+ result.to_json
+ end
+ end
+end
diff --git a/enterprise/app/services/captain/workflows/execution_service.rb b/enterprise/app/services/captain/workflows/execution_service.rb
index ffcf4f8f5..ebe832ccd 100644
--- a/enterprise/app/services/captain/workflows/execution_service.rb
+++ b/enterprise/app/services/captain/workflows/execution_service.rb
@@ -4,21 +4,79 @@ class Captain::Workflows::ExecutionService
def initialize(workflow, context)
@workflow = workflow
@context = context.with_indifferent_access
+ @paused = false
end
def perform
@execution = create_execution
execution.update!(status: :running, started_at: Time.current)
-
traverse_graph
- execution.update!(status: :completed, completed_at: Time.current)
+ finalize_execution
rescue StandardError => e
- execution&.update!(status: :failed, completed_at: Time.current, error_message: e.message)
- Rails.logger.error("[CaptainWorkflow] Workflow #{workflow.id} failed: #{e.message}")
+ handle_failure(e)
+ end
+
+ def self.resume(execution, user_input)
+ context = build_resume_context(execution, user_input)
+ service = new(execution.workflow, context)
+ service.resume_from(execution)
+ end
+
+ def resume_from(existing_execution)
+ @execution = existing_execution
+ @paused = false
+ execution.update!(status: :running)
+
+ resume_from_paused_node
+ finalize_execution
+ rescue StandardError => e
+ handle_failure(e, 'resume')
+ end
+
+ def self.build_resume_context(execution, user_input)
+ context = execution.context_store.with_indifferent_access
+ context[:account] = execution.workflow.account
+ context[:conversation] = execution.conversation if execution.conversation
+ context[:contact] = execution.contact if execution.contact
+
+ paused_node = execution.workflow.nodes.find { |n| n['id'] == execution.current_node_id }
+ input_key = paused_node&.dig('data', 'input_key') || 'user_input'
+ context[input_key] = user_input
+ context
end
private
+ def resume_from_paused_node
+ nodes = workflow.nodes
+ edges = workflow.edges
+ paused_node = nodes.find { |n| n['id'] == execution.current_node_id }
+ return unless paused_node
+
+ execute_node(paused_node, nodes, edges, Set.new)
+ end
+
+ def finalize_execution
+ if @paused
+ save_paused_state
+ { status: 'input_required', prompt: @pause_result[:prompt], input_key: @pause_result[:input_key], execution_id: execution.id }
+ else
+ execution.update!(status: :completed, completed_at: Time.current)
+ { status: 'completed', actions: execution.execution_log }
+ end
+ end
+
+ def save_paused_state
+ execution.update!(status: :waiting_for_input, current_node_id: @pause_node_id, context_store: serializable_context)
+ end
+
+ def handle_failure(error, label = nil)
+ execution&.update!(status: :failed, completed_at: Time.current, error_message: error.message)
+ tag = label ? " #{label}" : ''
+ Rails.logger.error("[CaptainWorkflow] Workflow #{workflow.id}#{tag} failed: #{error.message}")
+ { status: 'failed', error: error.message }
+ end
+
def create_execution
Captain::WorkflowExecution.create!(
workflow: workflow,
@@ -35,31 +93,65 @@ class Captain::Workflows::ExecutionService
edges = workflow.edges
return if nodes.blank?
- trigger_node = nodes.find { |n| n['type']&.start_with?('trigger_') }
- return unless trigger_node
+ entry_nodes = find_entry_nodes(nodes, edges)
+ return if entry_nodes.empty?
visited = Set.new
- execute_node(trigger_node, nodes, edges, visited)
+ entry_nodes.each do |entry_node|
+ break if @paused
+
+ execute_node(entry_node, nodes, edges, visited)
+ end
+ end
+
+ def find_entry_nodes(nodes, edges)
+ target_ids = edges.to_set { |e| e['target'] }
+ nodes.reject { |n| target_ids.include?(n['id']) }
end
def execute_node(node, nodes, edges, visited)
+ return if @paused
+
node_id = node['id']
return if visited.include?(node_id)
visited.add(node_id)
+ result = run_node_executor(node)
+ return unless result
+
+ return pause_at(node_id, result) if input_required?(result)
+
+ follow_edges(node_id, nodes, edges, result, visited)
+ end
+
+ def run_node_executor(node)
executor_class = Captain::Workflows::NodeRegistry.resolve(node['type'])
unless executor_class
- log_step(node_id, node['type'], 'skipped', { reason: 'unknown node type' })
- return
+ log_step(node['id'], node['type'], 'skipped', { reason: 'unknown node type' })
+ return nil
end
- executor = executor_class.new(node, context)
- result = executor.execute
- log_step(node_id, node['type'], 'completed', result)
+ result = executor_class.new(node, context).execute
+ log_step(node['id'], node['type'], 'completed', result)
+ result
+ end
+ def input_required?(result)
+ result.is_a?(Hash) && result[:status] == 'input_required'
+ end
+
+ def pause_at(node_id, result)
+ @paused = true
+ @pause_node_id = node_id
+ @pause_result = result
+ end
+
+ def follow_edges(node_id, nodes, edges, result, visited)
next_edges = find_next_edges(node_id, edges, result)
next_edges.each do |edge|
+ break if @paused
+
target_node = nodes.find { |n| n['id'] == edge['target'] }
execute_node(target_node, nodes, edges, visited) if target_node
end
@@ -85,4 +177,12 @@ class Captain::Workflows::ExecutionService
}
execution.save!
end
+
+ def serializable_context
+ context.each_with_object({}) do |(key, value), hash|
+ next if value.is_a?(ActiveRecord::Base)
+
+ hash[key] = value
+ end
+ end
end
diff --git a/enterprise/app/services/captain/workflows/node_registry.rb b/enterprise/app/services/captain/workflows/node_registry.rb
index 3a86cf19d..855e0a57f 100644
--- a/enterprise/app/services/captain/workflows/node_registry.rb
+++ b/enterprise/app/services/captain/workflows/node_registry.rb
@@ -1,8 +1,6 @@
class Captain::Workflows::NodeRegistry
REGISTRY = {
- 'trigger_conversation_created' => Captain::Workflows::Nodes::TriggerNode,
- 'trigger_message_created' => Captain::Workflows::Nodes::TriggerNode,
- 'trigger_conversation_resolved' => Captain::Workflows::Nodes::TriggerNode,
+ 'collect_input' => Captain::Workflows::Nodes::CollectInputNode,
'condition' => Captain::Workflows::Nodes::ConditionNode,
'send_message' => Captain::Workflows::Nodes::SendMessageNode,
'add_label' => Captain::Workflows::Nodes::AddLabelNode,
diff --git a/enterprise/app/services/captain/workflows/nodes/collect_input_node.rb b/enterprise/app/services/captain/workflows/nodes/collect_input_node.rb
new file mode 100644
index 000000000..8ae46b39d
--- /dev/null
+++ b/enterprise/app/services/captain/workflows/nodes/collect_input_node.rb
@@ -0,0 +1,11 @@
+class Captain::Workflows::Nodes::CollectInputNode < Captain::Workflows::Nodes::BaseNode
+ def execute
+ input_key = node_data['input_key'] || 'user_input'
+
+ if context[input_key].present?
+ { collected: context[input_key] }
+ else
+ { status: 'input_required', prompt: node_data['prompt'] || 'Please provide input', input_key: input_key }
+ end
+ end
+end