feat: implement scheduling functionality for AgentflowV2

packages/components/nodes/agentflow/Start/Start.ts

@@ -18,7 +18,7 @@ class Start_Agentflow implements INode {
     constructor() {
         this.label = 'Start'
         this.name = 'startAgentflow'
-        this.version = 1.1
+        this.version = 1.2
         this.type = 'Start'
         this.category = 'Agent Flows'
         this.description = 'Starting point of the agentflow'
@@ -40,6 +40,11 @@ class Start_Agentflow implements INode {
                         label: 'Form Input',
                         name: 'formInput',
                         description: 'Start the workflow with form inputs'
+                    },
+                    {
+                        label: 'Schedule Input',
+                        name: 'scheduleInput',
+                        description: 'Start the workflow on a recurring schedule (cron)'
                     }
                 ],
                 default: 'chatInput'
@@ -125,6 +130,144 @@ class Start_Agentflow implements INode {
                     }
                 ]
             },
+            {
+                label: 'Schedule Type',
+                name: 'scheduleType',
+                type: 'options',
+                options: [
+                    {
+                        label: 'Visual Picker',
+                        name: 'visualPicker',
+                        description: 'Use a visual picker to select schedule options'
+                    },
+                    {
+                        label: 'Cron Expression',
+                        name: 'cronExpression',
+                        description: 'Use a cron expression to define the schedule'
+                    }
+                ],
+                show: {
+                    startInputType: 'scheduleInput'
+                }
+            },
+            {
+                label: 'Cron Expression',
+                name: 'scheduleCronExpression',
+                type: 'string',
+                placeholder: '0 9 * * 1-5',
+                description:
+                    'Standard 5-field cron expression (minute hour day month weekday). Example: "0 9 * * 1-5" runs at 09:00 every weekday.',
+                show: {
+                    startInputType: 'scheduleInput',
+                    scheduleType: 'cronExpression'
+                }
+            },
+            {
+                label: 'Frequency',
+                name: 'scheduleFrequency',
+                type: 'options',
+                options: [
+                    {
+                        label: 'Hourly',
+                        name: 'hourly',
+                        description: 'Run every hour at the specified time'
+                    },
+                    {
+                        label: 'Daily',
+                        name: 'daily',
+                        description: 'Run every day at the specified time'
+                    },
+                    {
+                        label: 'Weekly',
+                        name: 'weekly',
+                        description: 'Run every week on the specified day and time'
+                    },
+                    {
+                        label: 'Monthly',
+                        name: 'monthly',
+                        description: 'Run every month on the specified date and time'
+                    }
+                ],
+                show: {
+                    startInputType: 'scheduleInput',
+                    scheduleType: 'visualPicker'
+                }
+            },
+            {
+                label: 'On Minute',
+                name: 'scheduleOnMinute',
+                type: 'number',
+                placeholder: '30',
+                description:
+                    'Minute of the hour when the schedule should run (0-59). For example, "30" means the schedule will run at the 30th minute of the hour.',
+                show: {
+                    startInputType: 'scheduleInput',
+                    scheduleType: 'visualPicker',
+                    scheduleFrequency: 'hourly'
+                }
+            },
+            {
+                label: 'On Time',
+                name: 'scheduleOnTime',
+                type: 'timePicker',
+                show: {
+                    startInputType: 'scheduleInput',
+                    scheduleType: 'visualPicker',
+                    scheduleFrequency: ['daily', 'weekly', 'monthly']
+                }
+            },
+            {
+                label: 'On Day of Week',
+                name: 'scheduleOnDayOfWeek',
+                type: 'weekDaysPicker',
+                show: {
+                    startInputType: 'scheduleInput',
+                    scheduleType: 'visualPicker',
+                    scheduleFrequency: 'weekly'
+                }
+            },
+            {
+                label: 'On Day of Month',
+                name: 'scheduleOnDayOfMonth',
+                type: 'monthDaysPicker',
+                show: {
+                    startInputType: 'scheduleInput',
+                    scheduleType: 'visualPicker',
+                    scheduleFrequency: 'monthly'
+                }
+            },
+            {
+                label: 'End Date',
+                name: 'scheduleEndDate',
+                type: 'datePicker',
+                description: 'Optional date after which the schedule will stop firing.',
+                optional: true,
+                show: {
+                    startInputType: 'scheduleInput'
+                }
+            },
+            {
+                label: 'Timezone',
+                name: 'scheduleTimezone',
+                type: 'string',
+                placeholder: 'UTC',
+                description: 'IANA timezone name, e.g. America/New_York. Defaults to UTC.',
+                optional: true,
+                show: {
+                    startInputType: 'scheduleInput'
+                }
+            },
+            {
+                label: 'Default Input',
+                name: 'scheduleDefaultInput',
+                type: 'string',
+                placeholder: 'Run the daily report',
+                description: 'Default question/input passed to the flow when it is triggered by the scheduler.',
+                rows: 4,
+                show: {
+                    startInputType: 'scheduleInput'
+                }
+            },
             {
                 label: 'Ephemeral Memory',
                 name: 'startEphemeralMemory',
@@ -213,6 +356,13 @@ class Start_Agentflow implements INode {
             outputData.form = form
         }
 
+        if (startInputType === 'scheduleInput') {
+            const defaultInput = nodeData.inputs?.scheduleDefaultInput as string
+            const effectiveInput = (typeof input === 'string' && input) || defaultInput || ''
+            inputData.question = effectiveInput
+            outputData.question = effectiveInput
+        }
+
         if (startEphemeralMemory) {
             outputData.ephemeralMemory = true
         }

packages/server/__mocks__/typeorm.ts

@@ -0,0 +1,22 @@
+/**
+ * Manual mock for 'typeorm'.
+ * All decorator factories are replaced with no-ops so TypeORM entity classes
+ * can be defined in tests without a real database connection.
+ * Used by all server-package test files via jest.mock('typeorm').
+ */
+
+const decorator = (): (() => void) => () => {}
+
+module.exports = {
+    PrimaryGeneratedColumn: decorator,
+    PrimaryColumn: decorator,
+    CreateDateColumn: decorator,
+    UpdateDateColumn: decorator,
+    Index: decorator,
+    ManyToOne: decorator,
+    OneToMany: decorator,
+    OneToOne: decorator,
+    JoinColumn: decorator,
+    Unique: decorator,
+    DataSource: jest.fn()
+}

packages/server/jest.config.js

@@ -18,6 +18,19 @@ module.exports = {
     // File extensions to recognize in module resolution
     moduleFileExtensions: ['ts', 'tsx', 'js', 'jsx', 'json', 'node'],
 
+    // uuid v10+ ships ESM-only; redirect to the CJS dist so Jest can require it.
+    // typeorm is not resolvable via pnpm symlinks in the test runner; redirect to
+    // the shared manual mock so all test files get the same decorator stubs without
+    // needing an inline jest.mock() factory.
+    moduleNameMapper: {
+        '^uuid$': '<rootDir>/node_modules/uuid/dist/index.js',
+        '^typeorm$': '<rootDir>/__mocks__/typeorm.ts'
+    },
+
+    // Include the package's own node_modules so that Jest can resolve
+    // symlinked pnpm dependencies when tests live inside src/
+    modulePaths: ['<rootDir>/node_modules'],
+
     // Display individual test results with the test suite hierarchy.
     verbose: true
 }

packages/server/package.json

@@ -126,6 +126,7 @@
         "multer-s3": "^3.0.1",
         "mysql2": "^3.11.3",
         "nanoid": "3",
+        "node-cron": "^4.2.1",
         "nodemailer": "^7.0.7",
         "openai": "6.19.0",
         "passport": "^0.7.0",

packages/server/src/Interface.Schedule.ts

@@ -0,0 +1,23 @@
+import { DataSource } from 'typeorm'
+import { IComponentNodes } from './Interface'
+import { Telemetry } from './utils/telemetry'
+import { CachePool } from './CachePool'
+import { UsageCacheManager } from './UsageCacheManager'
+
+export interface IScheduleQueueAppServer {
+    appDataSource: DataSource
+    componentNodes: IComponentNodes
+    telemetry: Telemetry
+    cachePool: CachePool
+    usageCacheManager: UsageCacheManager
+}
+
+export interface IScheduleAgentflowJobData extends IScheduleQueueAppServer {
+    scheduleRecordId: string
+    targetId: string
+    cronExpression: string
+    timezone: string
+    defaultInput?: string
+    workspaceId: string
+    scheduledAt: string // ISO string
+}

packages/server/src/commands/worker.ts

@@ -16,6 +16,7 @@ interface CustomListener extends QueueEventsListener {
 export default class Worker extends BaseCommand {
     predictionWorkerId: string
     upsertionWorkerId: string
+    scheduleWorkerId: string
 
     async run(): Promise<void> {
         logger.info('Starting Flowise Worker...')
@@ -51,6 +52,12 @@ export default class Worker extends BaseCommand {
         this.upsertionWorkerId = upsertionWorker.id
         logger.info(`Upsertion Worker ${this.upsertionWorkerId} created`)
 
+        /** Schedule */
+        const scheduleQueue = queueManager.getQueue('schedule')
+        const scheduleWorker = scheduleQueue.createWorker()
+        this.scheduleWorkerId = scheduleWorker.id
+        logger.info(`Schedule Worker ${this.scheduleWorkerId} created`)
+
         // Keep the process running
         process.stdin.resume()
     }
@@ -98,6 +105,10 @@ export default class Worker extends BaseCommand {
             const upsertWorker = queueManager.getQueue('upsert').getWorker()
             logger.info(`Shutting down Flowise Upsertion Worker ${this.upsertionWorkerId}...`)
             await upsertWorker.close()
+
+            const scheduleWorker = queueManager.getQueue('schedule').getWorker()
+            logger.info(`Shutting down Flowise Schedule Worker...`)
+            await scheduleWorker.close()
         } catch (error) {
             logger.error('There was an error shutting down Flowise Worker...', error)
             await this.failExit()

packages/server/src/controllers/chatflows/index.ts

@@ -13,6 +13,8 @@ import { WorkspaceUserErrorMessage, WorkspaceUserService } from '../../enterpris
 import { QueryRunner } from 'typeorm'
 import { GeneralErrorMessage } from '../../utils/constants'
 import { sanitizeFlowDataForPublicEndpoint } from '../../utils/sanitizeFlowData'
+import scheduleService from '../../services/schedule'
+import { ScheduleBeat } from '../../queue/ScheduleBeat'
 
 const checkIfChatflowIsValidForStreaming = async (req: Request, res: Response, next: NextFunction) => {
     try {
@@ -273,6 +275,54 @@ const checkIfChatflowHasChanged = async (req: Request, res: Response, next: Next
     }
 }
 
+const getScheduleStatus = async (req: Request, res: Response, next: NextFunction) => {
+    try {
+        if (!req.params?.id) {
+            throw new InternalFlowiseError(
+                StatusCodes.PRECONDITION_FAILED,
+                'Error: chatflowsController.getScheduleStatus - id not provided!'
+            )
+        }
+        const workspaceId = req.user?.activeWorkspaceId
+        if (!workspaceId) {
+            throw new InternalFlowiseError(StatusCodes.NOT_FOUND, 'Error: chatflowsController.getScheduleStatus - workspace not found!')
+        }
+        const status = await scheduleService.getScheduleStatus(req.params.id, workspaceId)
+        return res.json({
+            enabled: status.record?.enabled ?? false,
+            canEnable: status.canEnable,
+            reason: status.reason,
+            record: status.record
+        })
+    } catch (error) {
+        next(error)
+    }
+}
+
+const toggleScheduleEnabled = async (req: Request, res: Response, next: NextFunction) => {
+    try {
+        if (!req.params?.id) {
+            throw new InternalFlowiseError(
+                StatusCodes.PRECONDITION_FAILED,
+                'Error: chatflowsController.toggleScheduleEnabled - id not provided!'
+            )
+        }
+        const workspaceId = req.user?.activeWorkspaceId
+        if (!workspaceId) {
+            throw new InternalFlowiseError(StatusCodes.NOT_FOUND, 'Error: chatflowsController.toggleScheduleEnabled - workspace not found!')
+        }
+        const { enabled } = req.body
+        if (typeof enabled !== 'boolean') {
+            throw new InternalFlowiseError(StatusCodes.BAD_REQUEST, '"enabled" must be a boolean')
+        }
+        const record = await scheduleService.toggleScheduleEnabled(req.params.id, workspaceId, enabled)
+        await ScheduleBeat.getInstance().onScheduleChanged(record.id, enabled ? 'upsert' : 'delete')
+        return res.json(record)
+    } catch (error) {
+        next(error)
+    }
+}
+
 export default {
     checkIfChatflowIsValidForStreaming,
     checkIfChatflowIsValidForUploads,
@@ -284,5 +334,7 @@ export default {
     updateChatflow,
     getSinglePublicChatflow,
     getSinglePublicChatbotConfig,
-    checkIfChatflowHasChanged
+    checkIfChatflowHasChanged,
+    getScheduleStatus,
+    toggleScheduleEnabled
 }