Add workflows inside captain

app/javascript/dashboard/components-next/captain/workflows/NodeConfigPanel.vue

@@ -64,6 +64,18 @@ const FIELD_CONFIGS = {
       label: 'CAPTAIN.ASSISTANTS.WORKFLOWS.CONFIG.PRIORITY',
     },
   ],
+  collect_input: [
+    {
+      key: 'input_key',
+      component: 'input',
+      label: 'CAPTAIN.ASSISTANTS.WORKFLOWS.CONFIG.INPUT_KEY',
+    },
+    {
+      key: 'prompt',
+      component: 'textarea',
+      label: 'CAPTAIN.ASSISTANTS.WORKFLOWS.CONFIG.PROMPT',
+    },
+  ],
   condition: [
     {
       key: 'attribute',

app/javascript/dashboard/components-next/captain/workflows/NodePalette.vue

@@ -5,23 +5,13 @@ const { t } = useI18n();
 
 const NODE_CATEGORIES = [
   {
-    label: 'CAPTAIN.ASSISTANTS.WORKFLOWS.PALETTE.TRIGGERS',
-    color: 'n-teal',
-    icon: 'i-lucide-zap',
+    label: 'CAPTAIN.ASSISTANTS.WORKFLOWS.PALETTE.INTERACTIVE',
+    color: 'n-violet',
+    icon: 'i-lucide-message-circle-question',
     nodes: [
       {
-        type: 'trigger_conversation_created',
-        label:
-          'CAPTAIN.ASSISTANTS.WORKFLOWS.NODES.TRIGGER_CONVERSATION_CREATED',
-      },
-      {
-        type: 'trigger_message_created',
-        label: 'CAPTAIN.ASSISTANTS.WORKFLOWS.NODES.TRIGGER_MESSAGE_CREATED',
-      },
-      {
-        type: 'trigger_conversation_resolved',
-        label:
-          'CAPTAIN.ASSISTANTS.WORKFLOWS.NODES.TRIGGER_CONVERSATION_RESOLVED',
+        type: 'collect_input',
+        label: 'CAPTAIN.ASSISTANTS.WORKFLOWS.NODES.COLLECT_INPUT',
       },
     ],
   },

app/javascript/dashboard/components-next/captain/workflows/WorkflowToolbar.vue

@@ -4,7 +4,6 @@ import { useI18n } from 'vue-i18n';
 import Button from 'dashboard/components-next/button/Button.vue';
 import Input from 'dashboard/components-next/input/Input.vue';
 import Switch from 'dashboard/components-next/switch/Switch.vue';
-import SelectMenu from 'dashboard/components-next/selectmenu/SelectMenu.vue';
 
 const props = defineProps({
   workflow: { type: Object, default: () => ({}) },
@@ -15,7 +14,6 @@ const { t } = useI18n();
 
 const name = ref('');
 const description = ref('');
-const triggerEvent = ref('conversation_created');
 const enabled = ref(false);
 
 watch(
@@ -24,38 +22,16 @@ watch(
     if (val && val.id) {
       name.value = val.name || '';
       description.value = val.description || '';
-      triggerEvent.value = val.trigger_event || 'conversation_created';
       enabled.value = val.enabled || false;
     }
   },
   { immediate: true }
 );
 
-const TRIGGER_OPTIONS = [
-  {
-    value: 'conversation_created',
-    label: t('CAPTAIN.ASSISTANTS.WORKFLOWS.TRIGGERS.CONVERSATION_CREATED'),
-  },
-  {
-    value: 'message_created',
-    label: t('CAPTAIN.ASSISTANTS.WORKFLOWS.TRIGGERS.MESSAGE_CREATED'),
-  },
-  {
-    value: 'conversation_resolved',
-    label: t('CAPTAIN.ASSISTANTS.WORKFLOWS.TRIGGERS.CONVERSATION_RESOLVED'),
-  },
-];
-
-const triggerLabel = () => {
-  const option = TRIGGER_OPTIONS.find(o => o.value === triggerEvent.value);
-  return option ? option.label : triggerEvent.value;
-};
-
 const save = () => {
   emit('save', {
     name: name.value,
     description: description.value,
-    trigger_event: triggerEvent.value,
     enabled: enabled.value,
   });
 };
@@ -72,11 +48,6 @@ const save = () => {
       class="max-w-xs"
       :placeholder="t('CAPTAIN.ASSISTANTS.WORKFLOWS.TOOLBAR.NAME_PLACEHOLDER')"
     />
-    <SelectMenu
-      v-model="triggerEvent"
-      :options="TRIGGER_OPTIONS"
-      :label="triggerLabel()"
-    />
     <div class="flex items-center gap-2.5 ml-auto">
       <span class="text-xs text-n-slate-11">
         {{ t('CAPTAIN.ASSISTANTS.WORKFLOWS.TOOLBAR.ENABLED') }}

app/javascript/dashboard/components-next/captain/workflows/nodes/ActionNode.vue

@@ -8,22 +8,16 @@ defineProps({
 
 <template>
   <div
-    class="flex w-52 rounded-xl bg-n-solid-2 outline outline-1 -outline-offset-1 outline-n-container shadow-sm overflow-hidden"
+    class="flex rounded-lg bg-n-solid-2 outline outline-1 -outline-offset-1 outline-n-container shadow-sm"
   >
     <Handle type="target" :position="Position.Top" class="!bg-n-blue-9" />
-    <div class="w-1 shrink-0 bg-n-blue-9" />
-    <div class="flex flex-col gap-1 px-3 py-2.5 min-w-0">
-      <div class="flex items-center gap-2">
-        <span class="i-lucide-play text-n-blue-11 size-3.5 shrink-0" />
-        <span class="text-xs font-medium text-n-slate-12 truncate">
-          {{ data.label || 'Action' }}
-        </span>
-      </div>
+    <div class="w-0.5 shrink-0 bg-n-blue-9" />
+    <div class="flex items-center gap-1 px-1.5 py-1 min-w-0">
+      <span class="i-lucide-play text-n-blue-11 size-2.5 shrink-0" />
       <span
-        v-if="data.description"
-        class="text-[11px] text-n-slate-10 leading-tight truncate"
+        class="text-[11px] leading-none font-medium text-n-slate-12 truncate"
       >
-        {{ data.description }}
+        {{ data.label || 'Action' }}
       </span>
     </div>
     <Handle type="source" :position="Position.Bottom" class="!bg-n-blue-9" />

app/javascript/dashboard/components-next/captain/workflows/nodes/CollectInputNode.vue

@@ -0,0 +1,27 @@
+<script setup>
+import { Handle, Position } from '@vue-flow/core';
+
+defineProps({
+  data: { type: Object, default: () => ({}) },
+});
+</script>
+
+<template>
+  <div
+    class="flex rounded-lg bg-n-solid-2 outline outline-1 -outline-offset-1 outline-n-container shadow-sm"
+  >
+    <Handle type="target" :position="Position.Top" class="!bg-n-violet-9" />
+    <div class="w-0.5 shrink-0 bg-n-violet-9" />
+    <div class="flex items-center gap-1 px-1.5 py-1 min-w-0">
+      <span
+        class="i-lucide-message-circle-question text-n-violet-11 size-2.5 shrink-0"
+      />
+      <span
+        class="text-[11px] leading-none font-medium text-n-slate-12 truncate"
+      >
+        {{ data.label || 'Collect Input' }}
+      </span>
+    </div>
+    <Handle type="source" :position="Position.Bottom" class="!bg-n-violet-9" />
+  </div>
+</template>

app/javascript/dashboard/components-next/captain/workflows/nodes/ConditionNode.vue

@@ -11,13 +11,13 @@ const { t } = useI18n();
 
 <template>
   <div
-    class="flex w-52 rounded-xl bg-n-solid-2 outline outline-1 -outline-offset-1 outline-n-container shadow-sm overflow-hidden"
+    class="flex w-32 rounded-xl bg-n-solid-2 outline outline-1 -outline-offset-1 outline-n-container shadow-sm"
   >
     <Handle type="target" :position="Position.Top" class="!bg-n-amber-9" />
     <div class="w-1 shrink-0 bg-n-amber-9" />
-    <div class="flex flex-col gap-1 px-3 py-2.5 min-w-0">
+    <div class="flex flex-col gap-0.5 px-2 py-1.5 min-w-0">
       <div class="flex items-center gap-2">
-        <span class="i-lucide-git-branch text-n-amber-11 size-3.5 shrink-0" />
+        <span class="i-lucide-git-branch text-n-amber-11 size-3 shrink-0" />
         <span class="text-xs font-medium text-n-slate-12 truncate">
           {{ data.label || 'Condition' }}
         </span>

app/javascript/dashboard/components-next/captain/workflows/nodes/ShopifyNode.vue

@@ -8,13 +8,13 @@ defineProps({
 
 <template>
   <div
-    class="flex w-52 rounded-xl bg-n-solid-2 outline outline-1 -outline-offset-1 outline-n-container shadow-sm overflow-hidden"
+    class="flex w-32 rounded-xl bg-n-solid-2 outline outline-1 -outline-offset-1 outline-n-container shadow-sm"
   >
     <Handle type="target" :position="Position.Top" class="!bg-n-teal-9" />
     <div class="w-1 shrink-0 bg-n-teal-9" />
-    <div class="flex flex-col gap-1 px-3 py-2.5 min-w-0">
+    <div class="flex flex-col gap-0.5 px-2 py-1.5 min-w-0">
       <div class="flex items-center gap-2">
-        <span class="i-lucide-shopping-bag text-n-teal-11 size-3.5 shrink-0" />
+        <span class="i-lucide-shopping-bag text-n-teal-11 size-3 shrink-0" />
         <span class="text-xs font-medium text-n-slate-12 truncate">
           {{ data.label || 'Shopify' }}
         </span>

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",

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 () => {
     </div>
   </div>
 </template>
+
+<style>
+.vue-flow__node {
+  padding: 0;
+  border-radius: 0;
+  border: none;
+  background: none;
+  box-shadow: none;
+  font-size: inherit;
+  min-width: 0;
+}
+
+.vue-flow__handle {
+  width: 7px;
+  height: 7px;
+  border-radius: 50%;
+  border: 1px solid white;
+  min-width: 0;
+  min-height: 0;
+}
+
+.vue-flow__edge-path {
+  stroke: var(--n-alpha-7);
+  stroke-width: 1.5;
+}
+
+.vue-flow__edge.selected .vue-flow__edge-path,
+.vue-flow__edge:hover .vue-flow__edge-path {
+  stroke: var(--n-slate-9);
+  stroke-width: 2;
+}
+
+.vue-flow__connection-line {
+  stroke: var(--n-slate-9);
+  stroke-width: 1.5;
+}
+</style>

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 }}
               </span>
-              <span class="text-xs text-n-slate-10">
-                {{ $t(TRIGGER_LABELS[workflow.trigger_event]) }}
-              </span>
             </div>
             <div class="flex items-center gap-3 flex-shrink-0">
               <Switch

app/javascript/entrypoints/dashboard.js

@@ -32,6 +32,7 @@ import { vResizeObserver } from '@vueuse/components';
 import { directive as onClickaway } from 'vue3-click-away';
 
 import 'floating-vue/dist/style.css';
+import '@vue-flow/core/dist/style.css';
 
 const i18n = createI18n({
   legacy: false, // https://github.com/intlify/vue-i18n/issues/1902

db/migrate/20260227100003_add_pause_resume_to_captain_workflow_executions.rb

@@ -0,0 +1,6 @@
+class AddPauseResumeToCaptainWorkflowExecutions < ActiveRecord::Migration[7.1]
+  def change
+    add_column :captain_workflow_executions, :current_node_id, :string
+    add_column :captain_workflow_executions, :context_store, :jsonb, default: {}
+  end
+end

db/schema.rb

@@ -10,7 +10,7 @@
 #
 # It's strongly recommended that you check this file into your version control system.
 
-ActiveRecord::Schema[7.1].define(version: 2026_02_27_100002) do
+ActiveRecord::Schema[7.1].define(version: 2026_02_27_100003) do
   # These extensions should be enabled to support this database
   enable_extension "pg_stat_statements"
   enable_extension "pg_trgm"
@@ -423,6 +423,8 @@ ActiveRecord::Schema[7.1].define(version: 2026_02_27_100002) do
     t.jsonb "execution_log", default: []
     t.datetime "created_at", null: false
     t.datetime "updated_at", null: false
+    t.string "current_node_id"
+    t.jsonb "context_store", default: {}
     t.index ["account_id"], name: "index_captain_workflow_executions_on_account_id"
     t.index ["contact_id"], name: "index_captain_workflow_executions_on_contact_id"
     t.index ["conversation_id"], name: "index_captain_workflow_executions_on_conversation_id"

enterprise/app/jobs/captain/conversation/response_builder_job.rb

@@ -28,7 +28,11 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
   delegate :account, :inbox, to: :@conversation
 
   def generate_and_process_response
-    @response = Captain::Llm::AssistantChatService.new(assistant: @assistant, conversation_id: @conversation.display_id).generate_response(
+    @response = Captain::Llm::AssistantChatService.new(
+      assistant: @assistant,
+      conversation: @conversation,
+      conversation_id: @conversation.display_id
+    ).generate_response(
       message_history: collect_previous_messages
     )
     process_response

enterprise/app/jobs/captain/workflows/execution_job.rb

@@ -1,7 +0,0 @@
-class Captain::Workflows::ExecutionJob < ApplicationJob
-  queue_as :default
-
-  def perform(workflow, context)
-    Captain::Workflows::ExecutionService.new(workflow, context).perform
-  end
-end

enterprise/app/listeners/captain_listener.rb

@@ -1,18 +1,6 @@
 class CaptainListener < BaseListener
   include ::Events::Types
 
-  def conversation_created(event)
-    conversation = extract_conversation_and_account(event)[0]
-    trigger_workflows(conversation, 'conversation_created')
-  end
-
-  def message_created(event)
-    message = extract_message_and_account(event)[0]
-    return unless message.conversation
-
-    trigger_workflows(message.conversation, 'message_created', message: message)
-  end
-
   def conversation_resolved(event)
     conversation = extract_conversation_and_account(event)[0]
     assistant = conversation.inbox.captain_assistant
@@ -21,28 +9,5 @@ class CaptainListener < BaseListener
 
     Captain::Llm::ContactNotesService.new(assistant, conversation).generate_and_update_notes if assistant.config['feature_memory'].present?
     Captain::Llm::ConversationFaqService.new(assistant, conversation).generate_and_deduplicate if assistant.config['feature_faq'].present?
-
-    trigger_workflows(conversation, 'conversation_resolved')
-  end
-
-  private
-
-  def trigger_workflows(conversation, event, extra_context = {})
-    return unless conversation.inbox.captain_active?
-
-    assistant = conversation.inbox.captain_assistant
-    return unless assistant
-
-    workflows = assistant.workflows.enabled.for_event(event)
-    workflows.each do |workflow|
-      context = {
-        account: conversation.account,
-        conversation: conversation,
-        contact: conversation.contact,
-        event: event
-      }.merge(extra_context)
-
-      Captain::Workflows::ExecutionJob.perform_later(workflow, context)
-    end
   end
 end

enterprise/app/models/captain/workflow.rb

@@ -6,11 +6,8 @@ class Captain::Workflow < ApplicationRecord
   has_many :executions, class_name: 'Captain::WorkflowExecution', dependent: :destroy_async
 
   validates :name, presence: true
-  validates :trigger_event, presence: true
   validates :assistant_id, presence: true
   validates :account_id, presence: true
-  validates :trigger_event, inclusion: { in: %w[conversation_created message_created conversation_resolved] }
 
   scope :enabled, -> { where(enabled: true) }
-  scope :for_event, ->(event) { where(trigger_event: event) }
 end

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

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

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: "<user's response>") to continue.
+      EXECUTION
+    end
   end
 end
 # rubocop:enable Metrics/ClassLength

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
 

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: \"<user's response>\")."
+    when 'completed'
+      "Workflow completed successfully. Actions taken: #{result[:actions].to_json}"
+    when 'failed'
+      "Workflow failed: #{result[:error]}"
+    else
+      result.to_json
+    end
+  end
+end

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

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,

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