feat: add workflow schedule trigger support

api/core/workflow/nodes/enums.py

@@ -25,6 +25,7 @@ class NodeType(StrEnum):
     DOCUMENT_EXTRACTOR = "document-extractor"
     LIST_OPERATOR = "list-operator"
     AGENT = "agent"
+    TRIGGER_SCHEDULE = "trigger-schedule"
 
 
 class ErrorStrategy(StrEnum):

api/core/workflow/nodes/node_mapping.py

@@ -19,6 +19,7 @@
 from core.workflow.nodes.start import StartNode
 from core.workflow.nodes.template_transform import TemplateTransformNode
 from core.workflow.nodes.tool import ToolNode
+from core.workflow.nodes.trigger_schedule import TriggerScheduleNode
 from core.workflow.nodes.variable_aggregator import VariableAggregatorNode
 from core.workflow.nodes.variable_assigner.v1 import VariableAssignerNode as VariableAssignerNodeV1
 from core.workflow.nodes.variable_assigner.v2 import VariableAssignerNode as VariableAssignerNodeV2
@@ -132,4 +133,8 @@
         "2": AgentNode,
         "1": AgentNode,
     },
+    NodeType.TRIGGER_SCHEDULE: {
+        LATEST_VERSION: TriggerScheduleNode,
+        "1": TriggerScheduleNode,
+    },
 }

api/core/workflow/nodes/trigger_schedule/__init__.py

@@ -0,0 +1,3 @@
+from core.workflow.nodes.trigger_schedule.trigger_schedule_node import TriggerScheduleNode
+
+__all__ = ["TriggerScheduleNode"]

api/core/workflow/nodes/trigger_schedule/entities.py

@@ -0,0 +1,20 @@
+from typing import Optional
+
+from pydantic import Field
+
+from core.workflow.nodes.base import BaseNodeData
+
+
+class TriggerScheduleNodeData(BaseNodeData):
+    """
+    Trigger Schedule Node Data
+    """
+
+    mode: str = Field(default="visual", description="Schedule mode: visual or cron")
+    frequency: Optional[str] = Field(
+        default=None, description="Frequency for visual mode: hourly, daily, weekly, monthly"
+    )
+    cron_expression: Optional[str] = Field(default=None, description="Cron expression for cron mode")
+    visual_config: Optional[dict] = Field(default=None, description="Visual configuration details")
+    timezone: str = Field(default="UTC", description="Timezone for schedule execution")
+    enabled: bool = Field(default=True, description="Whether the schedule is enabled")

api/core/workflow/nodes/trigger_schedule/trigger_schedule_node.py

@@ -0,0 +1,70 @@
+from collections.abc import Mapping
+from datetime import UTC, datetime
+from typing import Any, Optional
+
+from core.workflow.constants import SYSTEM_VARIABLE_NODE_ID
+from core.workflow.entities.node_entities import NodeRunResult
+from core.workflow.entities.workflow_node_execution import WorkflowNodeExecutionStatus
+from core.workflow.nodes.base import BaseNode
+from core.workflow.nodes.base.entities import BaseNodeData, RetryConfig
+from core.workflow.nodes.enums import ErrorStrategy, NodeType
+from core.workflow.nodes.trigger_schedule.entities import TriggerScheduleNodeData
+
+
+class TriggerScheduleNode(BaseNode):
+    _node_type = NodeType.TRIGGER_SCHEDULE
+
+    _node_data: TriggerScheduleNodeData
+
+    def init_node_data(self, data: Mapping[str, Any]) -> None:
+        self._node_data = TriggerScheduleNodeData(**data)
+
+    def _get_error_strategy(self) -> Optional[ErrorStrategy]:
+        return self._node_data.error_strategy
+
+    def _get_retry_config(self) -> RetryConfig:
+        return self._node_data.retry_config
+
+    def _get_title(self) -> str:
+        return self._node_data.title
+
+    def _get_description(self) -> Optional[str]:
+        return self._node_data.desc
+
+    def _get_default_value_dict(self) -> dict[str, Any]:
+        return self._node_data.default_value_dict
+
+    def get_base_node_data(self) -> BaseNodeData:
+        return self._node_data
+
+    @classmethod
+    def version(cls) -> str:
+        return "1"
+
+    def _run(self) -> NodeRunResult:
+        node_inputs = dict(self.graph_runtime_state.variable_pool.user_inputs)
+        system_inputs = self.graph_runtime_state.variable_pool.system_variables.to_dict()
+
+        # Set system variables as node outputs
+        for var in system_inputs:
+            node_inputs[SYSTEM_VARIABLE_NODE_ID + "." + var] = system_inputs[var]
+
+        # Add schedule-specific outputs
+        triggered_at = datetime.now(UTC)
+        node_inputs["triggered_at"] = triggered_at.isoformat()
+        node_inputs["timezone"] = self._node_data.timezone
+        node_inputs["mode"] = self._node_data.mode
+        node_inputs["enabled"] = self._node_data.enabled
+
+        # Add configuration details based on mode
+        if self._node_data.mode == "cron" and self._node_data.cron_expression:
+            node_inputs["cron_expression"] = self._node_data.cron_expression
+        elif self._node_data.mode == "visual":
+            node_inputs["frequency"] = self._node_data.frequency or "unknown"
+            if self._node_data.visual_config:
+                node_inputs["visual_config"] = self._node_data.visual_config
+
+        return NodeRunResult(
+            status=WorkflowNodeExecutionStatus.SUCCEEDED,
+            outputs=node_inputs,
+        )

api/docker/entrypoint.sh

@@ -34,10 +34,10 @@ if [[ "${MODE}" == "worker" ]]; then
   if [[ -z "${CELERY_QUEUES}" ]]; then
     if [[ "${EDITION}" == "CLOUD" ]]; then
       # Cloud edition: separate queues for dataset and trigger tasks
-      DEFAULT_QUEUES="dataset,mail,ops_trace,app_deletion,plugin,workflow_storage,workflow_professional,workflow_team,workflow_sandbox"
+      DEFAULT_QUEUES="dataset,mail,ops_trace,app_deletion,plugin,workflow_storage,workflow_professional,workflow_team,workflow_sandbox,schedule"
     else
       # Community edition (SELF_HOSTED): dataset and workflow have separate queues
-      DEFAULT_QUEUES="dataset,mail,ops_trace,app_deletion,plugin,workflow_storage,workflow"
+      DEFAULT_QUEUES="dataset,mail,ops_trace,app_deletion,plugin,workflow_storage,workflow,schedule"
     fi
   else
     DEFAULT_QUEUES="${CELERY_QUEUES}"
@@ -68,6 +68,11 @@ if [[ "${MODE}" == "worker" ]]; then
 
 elif [[ "${MODE}" == "beat" ]]; then
   exec celery -A app.celery beat --loglevel ${LOG_LEVEL:-INFO}
+elif [[ "${MODE}" == "workflow-scheduler" ]]; then
+  echo "Starting Workflow Scheduler..."
+  exec celery -A app.celery beat \
+    --scheduler schedule.schedule_dispatch:WorkflowScheduler \
+    --loglevel ${LOG_LEVEL:-INFO}
 else
   if [[ "${DEBUG}" == "true" ]]; then
     exec flask run --host=${DIFY_BIND_ADDRESS:-0.0.0.0} --port=${DIFY_PORT:-5001} --debug

api/extensions/ext_celery.py

@@ -98,6 +98,7 @@ def __call__(self, *args: object, **kwargs: object) -> object:
 
     imports = [
         "tasks.async_workflow_tasks",  # trigger workers
+        "tasks.workflow_schedule_tasks",  # schedule trigger tasks
     ]
     day = dify_config.CELERY_BEAT_SCHEDULER_TIME
 

api/migrations/versions/2025_08_24_1313-1e06b2654c6c_add_workflow_schedule_plan.py

@@ -0,0 +1,52 @@
+"""add workflow_schedule_plan
+
+Revision ID: 1e06b2654c6c
+Revises: 4558cfabe44e
+Create Date: 2025-08-24 13:13:23.495063
+
+"""
+from alembic import op
+import models as models
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision = '1e06b2654c6c'
+down_revision = '4558cfabe44e'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('workflow_schedule_plans',
+    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+    sa.Column('app_id', models.types.StringUUID(), nullable=False),
+    sa.Column('workflow_id', models.types.StringUUID(), nullable=False),
+    sa.Column('root_node_id', sa.String(length=255), nullable=False),
+    sa.Column('cron_expression', sa.String(length=255), nullable=False),
+    sa.Column('timezone', sa.String(length=64), nullable=False),
+    sa.Column('enabled', sa.Boolean(), nullable=False),
+    sa.Column('next_run_at', sa.DateTime(timezone=True), nullable=True),
+    sa.Column('created_by', models.types.StringUUID(), nullable=False),
+    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+    sa.PrimaryKeyConstraint('id', name='workflow_schedule_plan_pkey')
+    )
+    with op.batch_alter_table('workflow_schedule_plans', schema=None) as batch_op:
+        batch_op.create_index('workflow_schedule_plan_app_workflow_idx', ['app_id', 'workflow_id'], unique=False)
+        batch_op.create_index('workflow_schedule_plan_created_at_idx', ['created_at'], unique=False)
+        batch_op.create_index('workflow_schedule_plan_tenant_enabled_next_idx', ['tenant_id', 'enabled', 'next_run_at'], unique=False)
+
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('workflow_schedule_plans', schema=None) as batch_op:
+        batch_op.drop_index('workflow_schedule_plan_tenant_enabled_next_idx')
+        batch_op.drop_index('workflow_schedule_plan_created_at_idx')
+        batch_op.drop_index('workflow_schedule_plan_app_workflow_idx')
+
+    op.drop_table('workflow_schedule_plans')
+    # ### end Alembic commands ###

api/models/workflow.py

@@ -1383,3 +1383,72 @@ def to_dict(self) -> dict:
             "triggered_at": self.triggered_at.isoformat() if self.triggered_at else None,
             "finished_at": self.finished_at.isoformat() if self.finished_at else None,
         }
+
+
+class WorkflowSchedulePlan(Base):
+    """
+    Workflow Schedule Configuration
+
+    Store schedule configurations for time-based workflow triggers.
+    Uses cron expressions with timezone support for flexible scheduling.
+
+    Attributes:
+        id: UUID primary key
+        tenant_id: Workspace ID for multi-tenancy
+        app_id: Application ID this schedule belongs to
+        workflow_id: Workflow to trigger
+        root_node_id: Starting node ID for workflow execution
+        cron_expression: Cron expression defining schedule pattern
+        timezone: Timezone for cron evaluation (e.g., 'Asia/Shanghai')
+        enabled: Whether schedule is active
+        next_run_at: Next scheduled execution time (timestamptz)
+        created_by: Creator account ID
+        created_at: Creation timestamp
+        updated_at: Last update timestamp
+    """
+
+    __tablename__ = "workflow_schedule_plans"
+    __table_args__ = (
+        sa.PrimaryKeyConstraint("id", name="workflow_schedule_plan_pkey"),
+        sa.Index("workflow_schedule_plan_tenant_enabled_next_idx", "tenant_id", "enabled", "next_run_at"),
+        sa.Index("workflow_schedule_plan_app_workflow_idx", "app_id", "workflow_id"),
+        sa.Index("workflow_schedule_plan_created_at_idx", "created_at"),
+    )
+
+    id: Mapped[str] = mapped_column(StringUUID, server_default=sa.text("uuid_generate_v4()"))
+    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    workflow_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+
+    root_node_id: Mapped[str] = mapped_column(String(255), nullable=False)
+
+    # Schedule configuration
+    cron_expression: Mapped[str] = mapped_column(String(255), nullable=False)
+    timezone: Mapped[str] = mapped_column(String(64), nullable=False)
+
+    # Schedule control
+    enabled: Mapped[bool] = mapped_column(sa.Boolean, nullable=False, default=True)
+    next_run_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
+
+    created_by: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, server_default=func.current_timestamp())
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime, nullable=False, server_default=func.current_timestamp(), onupdate=func.current_timestamp()
+    )
+
+    def to_dict(self) -> dict:
+        """Convert to dictionary representation"""
+        return {
+            "id": self.id,
+            "tenant_id": self.tenant_id,
+            "app_id": self.app_id,
+            "workflow_id": self.workflow_id,
+            "root_node_id": self.root_node_id,
+            "cron_expression": self.cron_expression,
+            "timezone": self.timezone,
+            "enabled": self.enabled,
+            "next_run_at": self.next_run_at.isoformat() if self.next_run_at else None,
+            "created_by": self.created_by,
+            "created_at": self.created_at.isoformat(),
+            "updated_at": self.updated_at.isoformat(),
+        }

api/pyproject.toml

@@ -88,6 +88,7 @@ dependencies = [
     "sseclient-py>=1.8.0",
     "httpx-sse>=0.4.0",
     "sendgrid~=6.12.3",
+    "croniter>=6.0.0",
 ]
 # Before adding new dependency, consider place it in
 # alphabet order (a-z) and suitable group.