feat: Inject "Start" node for snippet before running the whole snippet workflow.

# Conflicts:
#	api/controllers/console/snippets/payloads.py
#	api/controllers/console/snippets/snippet_workflow.py
#	api/services/snippet_service.py

api/controllers/console/__init__.py

@@ -116,6 +116,12 @@ from .explore import (
     trial,
 )
 
+# Import evaluation controllers
+from .evaluation import evaluation
+
+# Import snippet controllers
+from .snippets import snippet_workflow
+
 # Import tag controllers
 from .tag import tags
 
@@ -129,6 +135,7 @@ from .workspace import (
     model_providers,
     models,
     plugin,
+    snippets,
     tool_providers,
     trigger_providers,
     workspace,
@@ -166,6 +173,7 @@ __all__ = [
     "datasource_content_preview",
     "email_register",
     "endpoint",
+    "evaluation",
     "extension",
     "external",
     "feature",
@@ -199,6 +207,8 @@ __all__ = [
     "saved_message",
     "setup",
     "site",
+    "snippet_workflow",
+    "snippets",
     "spec",
     "statistic",
     "tags",

api/controllers/console/evaluation/__init__.py

@@ -0,0 +1 @@
+# Evaluation controller module

api/controllers/console/evaluation/evaluation.py

@@ -0,0 +1,320 @@
+import logging
+from collections.abc import Callable
+from functools import wraps
+from typing import ParamSpec, TypeVar, Union
+from urllib.parse import quote
+
+from flask import Response, request
+from flask_restx import Resource, fields
+from pydantic import BaseModel
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import NotFound
+
+from controllers.common.schema import register_schema_models
+from controllers.console import console_ns
+from controllers.console.wraps import (
+    account_initialization_required,
+    edit_permission_required,
+    setup_required,
+)
+from core.file import helpers as file_helpers
+from extensions.ext_database import db
+from libs.helper import TimestampField
+from libs.login import current_account_with_tenant, login_required
+from models import App
+from models.model import UploadFile
+from models.snippet import CustomizedSnippet
+from services.evaluation_service import EvaluationService
+
+logger = logging.getLogger(__name__)
+
+P = ParamSpec("P")
+R = TypeVar("R")
+
+# Valid evaluation target types
+EVALUATE_TARGET_TYPES = {"app", "snippets"}
+
+
+class VersionQuery(BaseModel):
+    """Query parameters for version endpoint."""
+
+    version: str
+
+
+register_schema_models(
+    console_ns,
+    VersionQuery,
+)
+
+
+# Response field definitions
+file_info_fields = {
+    "id": fields.String,
+    "name": fields.String,
+}
+
+evaluation_log_fields = {
+    "created_at": TimestampField,
+    "created_by": fields.String,
+    "test_file": fields.Nested(
+        console_ns.model(
+            "EvaluationTestFile",
+            file_info_fields,
+        )
+    ),
+    "result_file": fields.Nested(
+        console_ns.model(
+            "EvaluationResultFile",
+            file_info_fields,
+        ),
+        allow_null=True,
+    ),
+    "version": fields.String,
+}
+
+evaluation_log_list_model = console_ns.model(
+    "EvaluationLogList",
+    {
+        "data": fields.List(fields.Nested(console_ns.model("EvaluationLog", evaluation_log_fields))),
+    },
+)
+
+customized_matrix_fields = {
+    "evaluation_workflow_id": fields.String,
+    "input_fields": fields.Raw,
+    "output_fields": fields.Raw,
+}
+
+condition_fields = {
+    "name": fields.List(fields.String),
+    "comparison_operator": fields.String,
+    "value": fields.String,
+}
+
+judgement_conditions_fields = {
+    "logical_operator": fields.String,
+    "conditions": fields.List(fields.Nested(console_ns.model("EvaluationCondition", condition_fields))),
+}
+
+evaluation_detail_fields = {
+    "evaluation_model": fields.String,
+    "evaluation_model_provider": fields.String,
+    "customized_matrix": fields.Nested(
+        console_ns.model("EvaluationCustomizedMatrix", customized_matrix_fields),
+        allow_null=True,
+    ),
+    "judgement_conditions": fields.Nested(
+        console_ns.model("EvaluationJudgementConditions", judgement_conditions_fields),
+        allow_null=True,
+    ),
+}
+
+evaluation_detail_model = console_ns.model("EvaluationDetail", evaluation_detail_fields)
+
+
+def get_evaluation_target(view_func: Callable[P, R]):
+    """
+    Decorator to resolve polymorphic evaluation target (app or snippet).
+
+    Validates the target_type parameter and fetches the corresponding
+    model (App or CustomizedSnippet) with tenant isolation.
+    """
+
+    @wraps(view_func)
+    def decorated_view(*args: P.args, **kwargs: P.kwargs):
+        target_type = kwargs.get("evaluate_target_type")
+        target_id = kwargs.get("evaluate_target_id")
+
+        if target_type not in EVALUATE_TARGET_TYPES:
+            raise NotFound(f"Invalid evaluation target type: {target_type}")
+
+        _, current_tenant_id = current_account_with_tenant()
+
+        target_id = str(target_id)
+
+        # Remove path parameters
+        del kwargs["evaluate_target_type"]
+        del kwargs["evaluate_target_id"]
+
+        target: Union[App, CustomizedSnippet] | None = None
+
+        if target_type == "app":
+            target = (
+                db.session.query(App).where(App.id == target_id, App.tenant_id == current_tenant_id).first()
+            )
+        elif target_type == "snippets":
+            target = (
+                db.session.query(CustomizedSnippet)
+                .where(CustomizedSnippet.id == target_id, CustomizedSnippet.tenant_id == current_tenant_id)
+                .first()
+            )
+
+        if not target:
+            raise NotFound(f"{str(target_type)} not found")
+
+        kwargs["target"] = target
+        kwargs["target_type"] = target_type
+
+        return view_func(*args, **kwargs)
+
+    return decorated_view
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/dataset-template/download")
+class EvaluationDatasetTemplateDownloadApi(Resource):
+    @console_ns.doc("download_evaluation_dataset_template")
+    @console_ns.response(200, "Template file streamed as XLSX attachment")
+    @console_ns.response(400, "Invalid target type or excluded app mode")
+    @console_ns.response(404, "Target not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    @edit_permission_required
+    def post(self, target: Union[App, CustomizedSnippet], target_type: str):
+        """
+        Download evaluation dataset template.
+
+        Generates an XLSX template based on the target's input parameters
+        and streams it directly as a file attachment.
+        """
+        try:
+            xlsx_content, filename = EvaluationService.generate_dataset_template(
+                target=target,
+                target_type=target_type,
+            )
+        except ValueError as e:
+            return {"message": str(e)}, 400
+
+        encoded_filename = quote(filename)
+        response = Response(
+            xlsx_content,
+            mimetype="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+        )
+        response.headers["Content-Disposition"] = f"attachment; filename*=UTF-8''{encoded_filename}"
+        response.headers["Content-Length"] = str(len(xlsx_content))
+        return response
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation")
+class EvaluationDetailApi(Resource):
+    @console_ns.doc("get_evaluation_detail")
+    @console_ns.response(200, "Evaluation details retrieved successfully", evaluation_detail_model)
+    @console_ns.response(404, "Target not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    def get(self, target: Union[App, CustomizedSnippet], target_type: str):
+        """
+        Get evaluation details for the target.
+
+        Returns evaluation configuration including model settings,
+        customized matrix, and judgement conditions.
+        """
+        # TODO: Implement actual evaluation detail retrieval
+        # This is a placeholder implementation
+        return {
+            "evaluation_model": None,
+            "evaluation_model_provider": None,
+            "customized_matrix": None,
+            "judgement_conditions": None,
+        }
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation/logs")
+class EvaluationLogsApi(Resource):
+    @console_ns.doc("get_evaluation_logs")
+    @console_ns.response(200, "Evaluation logs retrieved successfully", evaluation_log_list_model)
+    @console_ns.response(404, "Target not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    def get(self, target: Union[App, CustomizedSnippet], target_type: str):
+        """
+        Get offline evaluation logs for the target.
+
+        Returns a list of evaluation runs with test files,
+        result files, and version information.
+        """
+        # TODO: Implement actual evaluation logs retrieval
+        # This is a placeholder implementation
+        return {
+            "data": [],
+        }
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation/files/<uuid:file_id>")
+class EvaluationFileDownloadApi(Resource):
+    @console_ns.doc("download_evaluation_file")
+    @console_ns.response(200, "File download URL generated successfully")
+    @console_ns.response(404, "Target or file not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    def get(self, target: Union[App, CustomizedSnippet], target_type: str, file_id: str):
+        """
+        Download evaluation test file or result file.
+
+        Looks up the specified file, verifies it belongs to the same tenant,
+        and returns file info and download URL.
+        """
+        file_id = str(file_id)
+        _, current_tenant_id = current_account_with_tenant()
+
+        with Session(db.engine, expire_on_commit=False) as session:
+            stmt = select(UploadFile).where(
+                UploadFile.id == file_id,
+                UploadFile.tenant_id == current_tenant_id,
+            )
+            upload_file = session.execute(stmt).scalar_one_or_none()
+
+        if not upload_file:
+            raise NotFound("File not found")
+
+        download_url = file_helpers.get_signed_file_url(upload_file_id=upload_file.id, as_attachment=True)
+
+        return {
+            "id": upload_file.id,
+            "name": upload_file.name,
+            "size": upload_file.size,
+            "extension": upload_file.extension,
+            "mime_type": upload_file.mime_type,
+            "created_at": int(upload_file.created_at.timestamp()) if upload_file.created_at else None,
+            "download_url": download_url,
+        }
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation/version")
+class EvaluationVersionApi(Resource):
+    @console_ns.doc("get_evaluation_version_detail")
+    @console_ns.expect(console_ns.models.get(VersionQuery.__name__))
+    @console_ns.response(200, "Version details retrieved successfully")
+    @console_ns.response(404, "Target or version not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    def get(self, target: Union[App, CustomizedSnippet], target_type: str):
+        """
+        Get evaluation target version details.
+
+        Returns the workflow graph for the specified version.
+        """
+        version = request.args.get("version")
+
+        if not version:
+            return {"message": "version parameter is required"}, 400
+
+        # TODO: Implement actual version detail retrieval
+        # For now, return the current graph if available
+        graph = {}
+        if target_type == "snippets" and isinstance(target, CustomizedSnippet):
+            graph = target.graph_dict
+
+        return {
+            "graph": graph,
+        }

api/controllers/console/snippets/payloads.py

@@ -0,0 +1,102 @@
+from typing import Any, Literal
+
+from pydantic import BaseModel, Field
+
+
+class SnippetListQuery(BaseModel):
+    """Query parameters for listing snippets."""
+
+    page: int = Field(default=1, ge=1, le=99999)
+    limit: int = Field(default=20, ge=1, le=100)
+    keyword: str | None = None
+
+
+class IconInfo(BaseModel):
+    """Icon information model."""
+
+    icon: str | None = None
+    icon_type: Literal["emoji", "image"] | None = None
+    icon_background: str | None = None
+    icon_url: str | None = None
+
+
+class InputFieldDefinition(BaseModel):
+    """Input field definition for snippet parameters."""
+
+    default: str | None = None
+    hint: bool | None = None
+    label: str | None = None
+    max_length: int | None = None
+    options: list[str] | None = None
+    placeholder: str | None = None
+    required: bool | None = None
+    type: str | None = None  # e.g., "text-input"
+
+
+class CreateSnippetPayload(BaseModel):
+    """Payload for creating a new snippet."""
+
+    name: str = Field(..., min_length=1, max_length=255)
+    description: str | None = Field(default=None, max_length=2000)
+    type: Literal["node", "group"] = "node"
+    icon_info: IconInfo | None = None
+    graph: dict[str, Any] | None = None
+    input_fields: list[InputFieldDefinition] | None = Field(default_factory=list)
+
+
+class UpdateSnippetPayload(BaseModel):
+    """Payload for updating a snippet."""
+
+    name: str | None = Field(default=None, min_length=1, max_length=255)
+    description: str | None = Field(default=None, max_length=2000)
+    icon_info: IconInfo | None = None
+
+
+class SnippetDraftSyncPayload(BaseModel):
+    """Payload for syncing snippet draft workflow."""
+
+    graph: dict[str, Any]
+    hash: str | None = None
+    environment_variables: list[dict[str, Any]] | None = None
+    conversation_variables: list[dict[str, Any]] | None = None
+    input_variables: list[dict[str, Any]] | None = None
+
+
+class WorkflowRunQuery(BaseModel):
+    """Query parameters for workflow runs."""
+
+    last_id: str | None = None
+    limit: int = Field(default=20, ge=1, le=100)
+
+
+class SnippetDraftRunPayload(BaseModel):
+    """Payload for running snippet draft workflow."""
+
+    inputs: dict[str, Any]
+    files: list[dict[str, Any]] | None = None
+
+
+class SnippetDraftNodeRunPayload(BaseModel):
+    """Payload for running a single node in snippet draft workflow."""
+
+    inputs: dict[str, Any]
+    query: str = ""
+    files: list[dict[str, Any]] | None = None
+
+
+class SnippetIterationNodeRunPayload(BaseModel):
+    """Payload for running an iteration node in snippet draft workflow."""
+
+    inputs: dict[str, Any] | None = None
+
+
+class SnippetLoopNodeRunPayload(BaseModel):
+    """Payload for running a loop node in snippet draft workflow."""
+
+    inputs: dict[str, Any] | None = None
+
+
+class PublishWorkflowPayload(BaseModel):
+    """Payload for publishing snippet workflow."""
+
+    knowledge_base_setting: dict[str, Any] | None = None

api/controllers/console/snippets/snippet_workflow.py

@@ -0,0 +1,540 @@
+import logging
+from collections.abc import Callable
+from functools import wraps
+from typing import ParamSpec, TypeVar
+
+from flask import request
+from flask_restx import Resource, marshal_with
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import InternalServerError, NotFound
+
+from controllers.common.schema import register_schema_models
+from controllers.console import console_ns
+from controllers.console.app.error import DraftWorkflowNotExist, DraftWorkflowNotSync
+from controllers.console.app.workflow import workflow_model
+from controllers.console.app.workflow_run import (
+    workflow_run_detail_model,
+    workflow_run_node_execution_list_model,
+    workflow_run_node_execution_model,
+    workflow_run_pagination_model,
+)
+from controllers.console.snippets.payloads import (
+    PublishWorkflowPayload,
+    SnippetDraftNodeRunPayload,
+    SnippetDraftRunPayload,
+    SnippetDraftSyncPayload,
+    SnippetIterationNodeRunPayload,
+    SnippetLoopNodeRunPayload,
+    WorkflowRunQuery,
+)
+from controllers.console.wraps import (
+    account_initialization_required,
+    edit_permission_required,
+    setup_required,
+)
+from core.app.apps.base_app_queue_manager import AppQueueManager
+from core.app.entities.app_invoke_entities import InvokeFrom
+from core.workflow.graph_engine.manager import GraphEngineManager
+from extensions.ext_database import db
+from factories import variable_factory
+from libs import helper
+from libs.helper import TimestampField
+from libs.login import current_account_with_tenant, login_required
+from models.snippet import CustomizedSnippet
+from services.errors.app import WorkflowHashNotEqualError
+from services.snippet_generate_service import SnippetGenerateService
+from services.snippet_service import SnippetService
+
+logger = logging.getLogger(__name__)
+
+P = ParamSpec("P")
+R = TypeVar("R")
+
+# Register Pydantic models with Swagger
+register_schema_models(
+    console_ns,
+    SnippetDraftSyncPayload,
+    SnippetDraftNodeRunPayload,
+    SnippetDraftRunPayload,
+    SnippetIterationNodeRunPayload,
+    SnippetLoopNodeRunPayload,
+    WorkflowRunQuery,
+    PublishWorkflowPayload,
+)
+
+
+class SnippetNotFoundError(Exception):
+    """Snippet not found error."""
+
+    pass
+
+
+def get_snippet(view_func: Callable[P, R]):
+    """Decorator to fetch and validate snippet access."""
+
+    @wraps(view_func)
+    def decorated_view(*args: P.args, **kwargs: P.kwargs):
+        if not kwargs.get("snippet_id"):
+            raise ValueError("missing snippet_id in path parameters")
+
+        _, current_tenant_id = current_account_with_tenant()
+
+        snippet_id = str(kwargs.get("snippet_id"))
+        del kwargs["snippet_id"]
+
+        snippet = SnippetService.get_snippet_by_id(
+            snippet_id=snippet_id,
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        kwargs["snippet"] = snippet
+
+        return view_func(*args, **kwargs)
+
+    return decorated_view
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft")
+class SnippetDraftWorkflowApi(Resource):
+    @console_ns.doc("get_snippet_draft_workflow")
+    @console_ns.response(200, "Draft workflow retrieved successfully", workflow_model)
+    @console_ns.response(404, "Snippet or draft workflow not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    @marshal_with(workflow_model)
+    def get(self, snippet: CustomizedSnippet):
+        """Get draft workflow for snippet."""
+        snippet_service = SnippetService()
+        workflow = snippet_service.get_draft_workflow(snippet=snippet)
+
+        if not workflow:
+            raise DraftWorkflowNotExist()
+
+        return workflow
+
+    @console_ns.doc("sync_snippet_draft_workflow")
+    @console_ns.expect(console_ns.models.get(SnippetDraftSyncPayload.__name__))
+    @console_ns.response(200, "Draft workflow synced successfully")
+    @console_ns.response(400, "Hash mismatch")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet):
+        """Sync draft workflow for snippet."""
+        current_user, _ = current_account_with_tenant()
+
+        payload = SnippetDraftSyncPayload.model_validate(console_ns.payload or {})
+
+        try:
+            environment_variables_list = payload.environment_variables or []
+            environment_variables = [
+                variable_factory.build_environment_variable_from_mapping(obj) for obj in environment_variables_list
+            ]
+            conversation_variables_list = payload.conversation_variables or []
+            conversation_variables = [
+                variable_factory.build_conversation_variable_from_mapping(obj) for obj in conversation_variables_list
+            ]
+            snippet_service = SnippetService()
+            workflow = snippet_service.sync_draft_workflow(
+                snippet=snippet,
+                graph=payload.graph,
+                unique_hash=payload.hash,
+                account=current_user,
+                environment_variables=environment_variables,
+                conversation_variables=conversation_variables,
+                input_variables=payload.input_variables,
+            )
+        except WorkflowHashNotEqualError:
+            raise DraftWorkflowNotSync()
+
+        return {
+            "result": "success",
+            "hash": workflow.unique_hash,
+            "updated_at": TimestampField().format(workflow.updated_at or workflow.created_at),
+        }
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/config")
+class SnippetDraftConfigApi(Resource):
+    @console_ns.doc("get_snippet_draft_config")
+    @console_ns.response(200, "Draft config retrieved successfully")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def get(self, snippet: CustomizedSnippet):
+        """Get snippet draft workflow configuration limits."""
+        return {
+            "parallel_depth_limit": 3,
+        }
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/publish")
+class SnippetPublishedWorkflowApi(Resource):
+    @console_ns.doc("get_snippet_published_workflow")
+    @console_ns.response(200, "Published workflow retrieved successfully", workflow_model)
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    @marshal_with(workflow_model)
+    def get(self, snippet: CustomizedSnippet):
+        """Get published workflow for snippet."""
+        if not snippet.is_published:
+            return None
+
+        snippet_service = SnippetService()
+        workflow = snippet_service.get_published_workflow(snippet=snippet)
+
+        return workflow
+
+    @console_ns.doc("publish_snippet_workflow")
+    @console_ns.expect(console_ns.models.get(PublishWorkflowPayload.__name__))
+    @console_ns.response(200, "Workflow published successfully")
+    @console_ns.response(400, "No draft workflow found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet):
+        """Publish snippet workflow."""
+        current_user, _ = current_account_with_tenant()
+        snippet_service = SnippetService()
+
+        with Session(db.engine) as session:
+            snippet = session.merge(snippet)
+            try:
+                workflow = snippet_service.publish_workflow(
+                    session=session,
+                    snippet=snippet,
+                    account=current_user,
+                )
+                workflow_created_at = TimestampField().format(workflow.created_at)
+                session.commit()
+            except ValueError as e:
+                return {"message": str(e)}, 400
+
+        return {
+            "result": "success",
+            "created_at": workflow_created_at,
+        }
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/default-workflow-block-configs")
+class SnippetDefaultBlockConfigsApi(Resource):
+    @console_ns.doc("get_snippet_default_block_configs")
+    @console_ns.response(200, "Default block configs retrieved successfully")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def get(self, snippet: CustomizedSnippet):
+        """Get default block configurations for snippet workflow."""
+        snippet_service = SnippetService()
+        return snippet_service.get_default_block_configs()
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflow-runs")
+class SnippetWorkflowRunsApi(Resource):
+    @console_ns.doc("list_snippet_workflow_runs")
+    @console_ns.response(200, "Workflow runs retrieved successfully", workflow_run_pagination_model)
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @marshal_with(workflow_run_pagination_model)
+    def get(self, snippet: CustomizedSnippet):
+        """List workflow runs for snippet."""
+        query = WorkflowRunQuery.model_validate(
+            {
+                "last_id": request.args.get("last_id"),
+                "limit": request.args.get("limit", type=int, default=20),
+            }
+        )
+        args = {
+            "last_id": query.last_id,
+            "limit": query.limit,
+        }
+
+        snippet_service = SnippetService()
+        result = snippet_service.get_snippet_workflow_runs(snippet=snippet, args=args)
+
+        return result
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflow-runs/<uuid:run_id>")
+class SnippetWorkflowRunDetailApi(Resource):
+    @console_ns.doc("get_snippet_workflow_run_detail")
+    @console_ns.response(200, "Workflow run detail retrieved successfully", workflow_run_detail_model)
+    @console_ns.response(404, "Workflow run not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @marshal_with(workflow_run_detail_model)
+    def get(self, snippet: CustomizedSnippet, run_id):
+        """Get workflow run detail for snippet."""
+        run_id = str(run_id)
+
+        snippet_service = SnippetService()
+        workflow_run = snippet_service.get_snippet_workflow_run(snippet=snippet, run_id=run_id)
+
+        if not workflow_run:
+            raise NotFound("Workflow run not found")
+
+        return workflow_run
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflow-runs/<uuid:run_id>/node-executions")
+class SnippetWorkflowRunNodeExecutionsApi(Resource):
+    @console_ns.doc("list_snippet_workflow_run_node_executions")
+    @console_ns.response(200, "Node executions retrieved successfully", workflow_run_node_execution_list_model)
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @marshal_with(workflow_run_node_execution_list_model)
+    def get(self, snippet: CustomizedSnippet, run_id):
+        """List node executions for a workflow run."""
+        run_id = str(run_id)
+
+        snippet_service = SnippetService()
+        node_executions = snippet_service.get_snippet_workflow_run_node_executions(
+            snippet=snippet,
+            run_id=run_id,
+        )
+
+        return {"data": node_executions}
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/nodes/<string:node_id>/run")
+class SnippetDraftNodeRunApi(Resource):
+    @console_ns.doc("run_snippet_draft_node")
+    @console_ns.doc(description="Run a single node in snippet draft workflow (single-step debugging)")
+    @console_ns.doc(params={"snippet_id": "Snippet ID", "node_id": "Node ID"})
+    @console_ns.expect(console_ns.models.get(SnippetDraftNodeRunPayload.__name__))
+    @console_ns.response(200, "Node run completed successfully", workflow_run_node_execution_model)
+    @console_ns.response(404, "Snippet or draft workflow not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @marshal_with(workflow_run_node_execution_model)
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet, node_id: str):
+        """
+        Run a single node in snippet draft workflow.
+
+        Executes a specific node with provided inputs for single-step debugging.
+        Returns the node execution result including status, outputs, and timing.
+        """
+        current_user, _ = current_account_with_tenant()
+        payload = SnippetDraftNodeRunPayload.model_validate(console_ns.payload or {})
+
+        user_inputs = payload.inputs
+
+        # Get draft workflow for file parsing
+        snippet_service = SnippetService()
+        draft_workflow = snippet_service.get_draft_workflow(snippet=snippet)
+        if not draft_workflow:
+            raise NotFound("Draft workflow not found")
+
+        files = SnippetGenerateService.parse_files(draft_workflow, payload.files)
+
+        workflow_node_execution = SnippetGenerateService.run_draft_node(
+            snippet=snippet,
+            node_id=node_id,
+            user_inputs=user_inputs,
+            account=current_user,
+            query=payload.query,
+            files=files,
+        )
+
+        return workflow_node_execution
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/nodes/<string:node_id>/last-run")
+class SnippetDraftNodeLastRunApi(Resource):
+    @console_ns.doc("get_snippet_draft_node_last_run")
+    @console_ns.doc(description="Get last run result for a node in snippet draft workflow")
+    @console_ns.doc(params={"snippet_id": "Snippet ID", "node_id": "Node ID"})
+    @console_ns.response(200, "Node last run retrieved successfully", workflow_run_node_execution_model)
+    @console_ns.response(404, "Snippet, draft workflow, or node last run not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @marshal_with(workflow_run_node_execution_model)
+    def get(self, snippet: CustomizedSnippet, node_id: str):
+        """
+        Get the last run result for a specific node in snippet draft workflow.
+
+        Returns the most recent execution record for the given node,
+        including status, inputs, outputs, and timing information.
+        """
+        snippet_service = SnippetService()
+        draft_workflow = snippet_service.get_draft_workflow(snippet=snippet)
+        if not draft_workflow:
+            raise NotFound("Draft workflow not found")
+
+        node_exec = snippet_service.get_snippet_node_last_run(
+            snippet=snippet,
+            workflow=draft_workflow,
+            node_id=node_id,
+        )
+        if node_exec is None:
+            raise NotFound("Node last run not found")
+
+        return node_exec
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/iteration/nodes/<string:node_id>/run")
+class SnippetDraftRunIterationNodeApi(Resource):
+    @console_ns.doc("run_snippet_draft_iteration_node")
+    @console_ns.doc(description="Run draft workflow iteration node for snippet")
+    @console_ns.doc(params={"snippet_id": "Snippet ID", "node_id": "Node ID"})
+    @console_ns.expect(console_ns.models.get(SnippetIterationNodeRunPayload.__name__))
+    @console_ns.response(200, "Iteration node run started successfully (SSE stream)")
+    @console_ns.response(404, "Snippet or draft workflow not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet, node_id: str):
+        """
+        Run a draft workflow iteration node for snippet.
+
+        Iteration nodes execute their internal sub-graph multiple times over an input list.
+        Returns an SSE event stream with iteration progress and results.
+        """
+        current_user, _ = current_account_with_tenant()
+        args = SnippetIterationNodeRunPayload.model_validate(console_ns.payload or {}).model_dump(exclude_none=True)
+
+        try:
+            response = SnippetGenerateService.generate_single_iteration(
+                snippet=snippet, user=current_user, node_id=node_id, args=args, streaming=True
+            )
+
+            return helper.compact_generate_response(response)
+        except ValueError as e:
+            raise e
+        except Exception:
+            logger.exception("internal server error.")
+            raise InternalServerError()
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/loop/nodes/<string:node_id>/run")
+class SnippetDraftRunLoopNodeApi(Resource):
+    @console_ns.doc("run_snippet_draft_loop_node")
+    @console_ns.doc(description="Run draft workflow loop node for snippet")
+    @console_ns.doc(params={"snippet_id": "Snippet ID", "node_id": "Node ID"})
+    @console_ns.expect(console_ns.models.get(SnippetLoopNodeRunPayload.__name__))
+    @console_ns.response(200, "Loop node run started successfully (SSE stream)")
+    @console_ns.response(404, "Snippet or draft workflow not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet, node_id: str):
+        """
+        Run a draft workflow loop node for snippet.
+
+        Loop nodes execute their internal sub-graph repeatedly until a condition is met.
+        Returns an SSE event stream with loop progress and results.
+        """
+        current_user, _ = current_account_with_tenant()
+        args = SnippetLoopNodeRunPayload.model_validate(console_ns.payload or {})
+
+        try:
+            response = SnippetGenerateService.generate_single_loop(
+                snippet=snippet, user=current_user, node_id=node_id, args=args, streaming=True
+            )
+
+            return helper.compact_generate_response(response)
+        except ValueError as e:
+            raise e
+        except Exception:
+            logger.exception("internal server error.")
+            raise InternalServerError()
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/run")
+class SnippetDraftWorkflowRunApi(Resource):
+    @console_ns.doc("run_snippet_draft_workflow")
+    @console_ns.expect(console_ns.models.get(SnippetDraftRunPayload.__name__))
+    @console_ns.response(200, "Draft workflow run started successfully (SSE stream)")
+    @console_ns.response(404, "Snippet or draft workflow not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet):
+        """
+        Run draft workflow for snippet.
+
+        Executes the snippet's draft workflow with the provided inputs
+        and returns an SSE event stream with execution progress and results.
+        """
+        current_user, _ = current_account_with_tenant()
+
+        payload = SnippetDraftRunPayload.model_validate(console_ns.payload or {})
+        args = payload.model_dump(exclude_none=True)
+
+        try:
+            response = SnippetGenerateService.generate(
+                snippet=snippet,
+                user=current_user,
+                args=args,
+                invoke_from=InvokeFrom.DEBUGGER,
+                streaming=True,
+            )
+
+            return helper.compact_generate_response(response)
+        except ValueError as e:
+            raise e
+        except Exception:
+            logger.exception("internal server error.")
+            raise InternalServerError()
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflow-runs/tasks/<string:task_id>/stop")
+class SnippetWorkflowTaskStopApi(Resource):
+    @console_ns.doc("stop_snippet_workflow_task")
+    @console_ns.response(200, "Task stopped successfully")
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet, task_id: str):
+        """
+        Stop a running snippet workflow task.
+
+        Uses both the legacy stop flag mechanism and the graph engine
+        command channel for backward compatibility.
+        """
+        # Stop using both mechanisms for backward compatibility
+        # Legacy stop flag mechanism (without user check)
+        AppQueueManager.set_stop_flag_no_user_check(task_id)
+
+        # New graph engine command channel mechanism
+        GraphEngineManager.send_stop_command(task_id)
+
+        return {"result": "success"}

api/controllers/console/workspace/snippets.py

@@ -0,0 +1,202 @@
+import logging
+
+from flask import request
+from flask_restx import Resource, marshal, marshal_with
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import NotFound
+
+from controllers.common.schema import register_schema_models
+from controllers.console import console_ns
+from controllers.console.snippets.payloads import (
+    CreateSnippetPayload,
+    SnippetListQuery,
+    UpdateSnippetPayload,
+)
+from controllers.console.wraps import (
+    account_initialization_required,
+    edit_permission_required,
+    setup_required,
+)
+from extensions.ext_database import db
+from fields.snippet_fields import snippet_fields, snippet_list_fields, snippet_pagination_fields
+from libs.login import current_account_with_tenant, login_required
+from models.snippet import SnippetType
+from services.snippet_service import SnippetService
+
+logger = logging.getLogger(__name__)
+
+# Register Pydantic models with Swagger
+register_schema_models(
+    console_ns,
+    SnippetListQuery,
+    CreateSnippetPayload,
+    UpdateSnippetPayload,
+)
+
+# Create namespace models for marshaling
+snippet_model = console_ns.model("Snippet", snippet_fields)
+snippet_list_model = console_ns.model("SnippetList", snippet_list_fields)
+snippet_pagination_model = console_ns.model("SnippetPagination", snippet_pagination_fields)
+
+
+@console_ns.route("/workspaces/current/customized-snippets")
+class CustomizedSnippetsApi(Resource):
+    @console_ns.doc("list_customized_snippets")
+    @console_ns.expect(console_ns.models.get(SnippetListQuery.__name__))
+    @console_ns.response(200, "Snippets retrieved successfully", snippet_pagination_model)
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self):
+        """List customized snippets with pagination and search."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        query_params = request.args.to_dict()
+        query = SnippetListQuery.model_validate(query_params)
+
+        snippets, total, has_more = SnippetService.get_snippets(
+            tenant_id=current_tenant_id,
+            page=query.page,
+            limit=query.limit,
+            keyword=query.keyword,
+        )
+
+        return {
+            "data": marshal(snippets, snippet_list_fields),
+            "page": query.page,
+            "limit": query.limit,
+            "total": total,
+            "has_more": has_more,
+        }, 200
+
+    @console_ns.doc("create_customized_snippet")
+    @console_ns.expect(console_ns.models.get(CreateSnippetPayload.__name__))
+    @console_ns.response(201, "Snippet created successfully", snippet_model)
+    @console_ns.response(400, "Invalid request or name already exists")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def post(self):
+        """Create a new customized snippet."""
+        current_user, current_tenant_id = current_account_with_tenant()
+
+        payload = CreateSnippetPayload.model_validate(console_ns.payload or {})
+
+        try:
+            snippet_type = SnippetType(payload.type)
+        except ValueError:
+            snippet_type = SnippetType.NODE
+
+        try:
+            snippet = SnippetService.create_snippet(
+                tenant_id=current_tenant_id,
+                name=payload.name,
+                description=payload.description,
+                snippet_type=snippet_type,
+                icon_info=payload.icon_info.model_dump() if payload.icon_info else None,
+                graph=payload.graph,
+                input_fields=[f.model_dump() for f in payload.input_fields] if payload.input_fields else None,
+                account=current_user,
+            )
+        except ValueError as e:
+            return {"message": str(e)}, 400
+
+        return marshal(snippet, snippet_fields), 201
+
+
+@console_ns.route("/workspaces/current/customized-snippets/<uuid:snippet_id>")
+class CustomizedSnippetDetailApi(Resource):
+    @console_ns.doc("get_customized_snippet")
+    @console_ns.response(200, "Snippet retrieved successfully", snippet_model)
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, snippet_id: str):
+        """Get customized snippet details."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        snippet = SnippetService.get_snippet_by_id(
+            snippet_id=str(snippet_id),
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        return marshal(snippet, snippet_fields), 200
+
+    @console_ns.doc("update_customized_snippet")
+    @console_ns.expect(console_ns.models.get(UpdateSnippetPayload.__name__))
+    @console_ns.response(200, "Snippet updated successfully", snippet_model)
+    @console_ns.response(400, "Invalid request or name already exists")
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def patch(self, snippet_id: str):
+        """Update customized snippet."""
+        current_user, current_tenant_id = current_account_with_tenant()
+
+        snippet = SnippetService.get_snippet_by_id(
+            snippet_id=str(snippet_id),
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        payload = UpdateSnippetPayload.model_validate(console_ns.payload or {})
+        update_data = payload.model_dump(exclude_unset=True)
+
+        if "icon_info" in update_data and update_data["icon_info"] is not None:
+            update_data["icon_info"] = payload.icon_info.model_dump() if payload.icon_info else None
+
+        if not update_data:
+            return {"message": "No valid fields to update"}, 400
+
+        try:
+            with Session(db.engine, expire_on_commit=False) as session:
+                snippet = session.merge(snippet)
+                snippet = SnippetService.update_snippet(
+                    session=session,
+                    snippet=snippet,
+                    account_id=current_user.id,
+                    data=update_data,
+                )
+                session.commit()
+        except ValueError as e:
+            return {"message": str(e)}, 400
+
+        return marshal(snippet, snippet_fields), 200
+
+    @console_ns.doc("delete_customized_snippet")
+    @console_ns.response(204, "Snippet deleted successfully")
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def delete(self, snippet_id: str):
+        """Delete customized snippet."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        snippet = SnippetService.get_snippet_by_id(
+            snippet_id=str(snippet_id),
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        with Session(db.engine) as session:
+            snippet = session.merge(snippet)
+            SnippetService.delete_snippet(
+                session=session,
+                snippet=snippet,
+            )
+            session.commit()
+
+        return "", 204

api/fields/snippet_fields.py

@@ -0,0 +1,45 @@
+from flask_restx import fields
+
+from fields.member_fields import simple_account_fields
+from libs.helper import TimestampField
+
+# Snippet list item fields (lightweight for list display)
+snippet_list_fields = {
+    "id": fields.String,
+    "name": fields.String,
+    "description": fields.String,
+    "type": fields.String,
+    "version": fields.Integer,
+    "use_count": fields.Integer,
+    "is_published": fields.Boolean,
+    "icon_info": fields.Raw,
+    "created_at": TimestampField,
+    "updated_at": TimestampField,
+}
+
+# Full snippet fields (includes creator info and graph data)
+snippet_fields = {
+    "id": fields.String,
+    "name": fields.String,
+    "description": fields.String,
+    "type": fields.String,
+    "version": fields.Integer,
+    "use_count": fields.Integer,
+    "is_published": fields.Boolean,
+    "icon_info": fields.Raw,
+    "graph": fields.Raw(attribute="graph_dict"),
+    "input_fields": fields.Raw(attribute="input_fields_list"),
+    "created_by": fields.Nested(simple_account_fields, attribute="created_by_account", allow_null=True),
+    "created_at": TimestampField,
+    "updated_by": fields.Nested(simple_account_fields, attribute="updated_by_account", allow_null=True),
+    "updated_at": TimestampField,
+}
+
+# Pagination response fields
+snippet_pagination_fields = {
+    "data": fields.List(fields.Nested(snippet_list_fields)),
+    "page": fields.Integer,
+    "limit": fields.Integer,
+    "total": fields.Integer,
+    "has_more": fields.Boolean,
+}

api/migrations/versions/2026_01_29_1200-1c05e80d2380_add_customized_snippets_table.py

@@ -0,0 +1,83 @@
+"""add_customized_snippets_table
+
+Revision ID: 1c05e80d2380
+Revises: 788d3099ae3a
+Create Date: 2026-01-29 12:00:00.000000
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+import models as models
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
+
+# revision identifiers, used by Alembic.
+revision = "1c05e80d2380"
+down_revision = "788d3099ae3a"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    conn = op.get_bind()
+
+    if _is_pg(conn):
+        op.create_table(
+            "customized_snippets",
+            sa.Column("id", models.types.StringUUID(), server_default=sa.text("uuidv7()"), nullable=False),
+            sa.Column("tenant_id", models.types.StringUUID(), nullable=False),
+            sa.Column("name", sa.String(length=255), nullable=False),
+            sa.Column("description", sa.Text(), nullable=True),
+            sa.Column("type", sa.String(length=50), server_default=sa.text("'node'"), nullable=False),
+            sa.Column("workflow_id", models.types.StringUUID(), nullable=True),
+            sa.Column("is_published", sa.Boolean(), server_default=sa.text("false"), nullable=False),
+            sa.Column("version", sa.Integer(), server_default=sa.text("1"), nullable=False),
+            sa.Column("use_count", sa.Integer(), server_default=sa.text("0"), nullable=False),
+            sa.Column("icon_info", postgresql.JSONB(astext_type=sa.Text()), nullable=True),
+            sa.Column("graph", sa.Text(), nullable=True),
+            sa.Column("input_fields", sa.Text(), nullable=True),
+            sa.Column("created_by", models.types.StringUUID(), nullable=True),
+            sa.Column("created_at", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
+            sa.Column("updated_by", models.types.StringUUID(), nullable=True),
+            sa.Column("updated_at", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
+            sa.PrimaryKeyConstraint("id", name="customized_snippet_pkey"),
+            sa.UniqueConstraint("tenant_id", "name", name="customized_snippet_tenant_name_key"),
+        )
+    else:
+        op.create_table(
+            "customized_snippets",
+            sa.Column("id", models.types.StringUUID(), nullable=False),
+            sa.Column("tenant_id", models.types.StringUUID(), nullable=False),
+            sa.Column("name", sa.String(length=255), nullable=False),
+            sa.Column("description", models.types.LongText(), nullable=True),
+            sa.Column("type", sa.String(length=50), server_default=sa.text("'node'"), nullable=False),
+            sa.Column("workflow_id", models.types.StringUUID(), nullable=True),
+            sa.Column("is_published", sa.Boolean(), server_default=sa.text("false"), nullable=False),
+            sa.Column("version", sa.Integer(), server_default=sa.text("1"), nullable=False),
+            sa.Column("use_count", sa.Integer(), server_default=sa.text("0"), nullable=False),
+            sa.Column("icon_info", models.types.AdjustedJSON(astext_type=sa.Text()), nullable=True),
+            sa.Column("graph", models.types.LongText(), nullable=True),
+            sa.Column("input_fields", models.types.LongText(), nullable=True),
+            sa.Column("created_by", models.types.StringUUID(), nullable=True),
+            sa.Column("created_at", sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+            sa.Column("updated_by", models.types.StringUUID(), nullable=True),
+            sa.Column("updated_at", sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+            sa.PrimaryKeyConstraint("id", name="customized_snippet_pkey"),
+            sa.UniqueConstraint("tenant_id", "name", name="customized_snippet_tenant_name_key"),
+        )
+
+    with op.batch_alter_table("customized_snippets", schema=None) as batch_op:
+        batch_op.create_index("customized_snippet_tenant_idx", ["tenant_id"], unique=False)
+
+
+def downgrade():
+    with op.batch_alter_table("customized_snippets", schema=None) as batch_op:
+        batch_op.drop_index("customized_snippet_tenant_idx")
+
+    op.drop_table("customized_snippets")

api/models/__init__.py

@@ -81,6 +81,7 @@ from .provider import (
     TenantDefaultModel,
     TenantPreferredModelProvider,
 )
+from .snippet import CustomizedSnippet, SnippetType
 from .source import DataSourceApiKeyAuthBinding, DataSourceOauthBinding
 from .task import CeleryTask, CeleryTaskSet
 from .tools import (
@@ -140,6 +141,7 @@ __all__ = [
     "Conversation",
     "ConversationVariable",
     "CreatorUserRole",
+    "CustomizedSnippet",
     "DataSourceApiKeyAuthBinding",
     "DataSourceOauthBinding",
     "Dataset",
@@ -184,6 +186,7 @@ __all__ = [
     "RecommendedApp",
     "SavedMessage",
     "Site",
+    "SnippetType",
     "Tag",
     "TagBinding",
     "Tenant",

api/models/snippet.py

@@ -0,0 +1,101 @@
+import json
+from datetime import datetime
+from enum import StrEnum
+from typing import Any
+
+import sqlalchemy as sa
+from sqlalchemy import DateTime, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from libs.uuid_utils import uuidv7
+
+from .account import Account
+from .base import Base
+from .engine import db
+from .types import AdjustedJSON, LongText, StringUUID
+
+
+class SnippetType(StrEnum):
+    """Snippet Type Enum"""
+
+    NODE = "node"
+    GROUP = "group"
+
+
+class CustomizedSnippet(Base):
+    """
+    Customized Snippet Model
+
+    Stores reusable workflow components (nodes or node groups) that can be
+    shared across applications within a workspace.
+    """
+
+    __tablename__ = "customized_snippets"
+    __table_args__ = (
+        sa.PrimaryKeyConstraint("id", name="customized_snippet_pkey"),
+        sa.Index("customized_snippet_tenant_idx", "tenant_id"),
+        sa.UniqueConstraint("tenant_id", "name", name="customized_snippet_tenant_name_key"),
+    )
+
+    id: Mapped[str] = mapped_column(StringUUID, default=lambda: str(uuidv7()))
+    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    name: Mapped[str] = mapped_column(String(255), nullable=False)
+    description: Mapped[str | None] = mapped_column(LongText, nullable=True)
+    type: Mapped[str] = mapped_column(String(50), nullable=False, server_default=sa.text("'node'"))
+
+    # Workflow reference for published version
+    workflow_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
+
+    # State flags
+    is_published: Mapped[bool] = mapped_column(sa.Boolean, nullable=False, server_default=sa.text("false"))
+    version: Mapped[int] = mapped_column(sa.Integer, nullable=False, server_default=sa.text("1"))
+    use_count: Mapped[int] = mapped_column(sa.Integer, nullable=False, server_default=sa.text("0"))
+
+    # Visual customization
+    icon_info: Mapped[dict | None] = mapped_column(AdjustedJSON, nullable=True)
+
+    # Snippet configuration (stored as JSON text)
+    input_fields: Mapped[str | None] = mapped_column(LongText, nullable=True)
+
+    # Audit fields
+    created_by: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, server_default=func.current_timestamp())
+    updated_by: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime, nullable=False, server_default=func.current_timestamp(), onupdate=func.current_timestamp()
+    )
+
+    @property
+    def graph_dict(self) -> dict[str, Any]:
+        """Get graph from associated workflow."""
+        if self.workflow_id:
+            from .workflow import Workflow
+
+            workflow = db.session.get(Workflow, self.workflow_id)
+            if workflow:
+                return json.loads(workflow.graph) if workflow.graph else {}
+        return {}
+
+    @property
+    def input_fields_list(self) -> list[dict[str, Any]]:
+        """Parse input_fields JSON to list."""
+        return json.loads(self.input_fields) if self.input_fields else []
+
+    @property
+    def created_by_account(self) -> Account | None:
+        """Get the account that created this snippet."""
+        if self.created_by:
+            return db.session.get(Account, self.created_by)
+        return None
+
+    @property
+    def updated_by_account(self) -> Account | None:
+        """Get the account that last updated this snippet."""
+        if self.updated_by:
+            return db.session.get(Account, self.updated_by)
+        return None
+
+    @property
+    def version_str(self) -> str:
+        """Get version as string for API response."""
+        return str(self.version)

api/models/workflow.py

@@ -67,6 +67,7 @@ class WorkflowType(StrEnum):
     WORKFLOW = "workflow"
     CHAT = "chat"
     RAG_PIPELINE = "rag-pipeline"
+    SNIPPET = "snippet"
 
     @classmethod
     def value_of(cls, value: str) -> "WorkflowType":

api/services/evaluation_service.py

@@ -0,0 +1,178 @@
+import io
+import logging
+from typing import Union
+
+from openpyxl import Workbook
+from openpyxl.styles import Alignment, Border, Font, PatternFill, Side
+from openpyxl.utils import get_column_letter
+
+from models.model import App, AppMode
+from models.snippet import CustomizedSnippet
+from services.snippet_service import SnippetService
+from services.workflow_service import WorkflowService
+
+logger = logging.getLogger(__name__)
+
+
+class EvaluationService:
+    """
+    Service for evaluation-related operations.
+
+    Provides functionality to generate evaluation dataset templates
+    based on App or Snippet input parameters.
+    """
+
+    # Excluded app modes that don't support evaluation templates
+    EXCLUDED_APP_MODES = {AppMode.RAG_PIPELINE}
+
+    @classmethod
+    def generate_dataset_template(
+        cls,
+        target: Union[App, CustomizedSnippet],
+        target_type: str,
+    ) -> tuple[bytes, str]:
+        """
+        Generate evaluation dataset template as XLSX bytes.
+
+        Creates an XLSX file with headers based on the evaluation target's input parameters.
+        The first column is index, followed by input parameter columns.
+
+        :param target: App or CustomizedSnippet instance
+        :param target_type: Target type string ("app" or "snippet")
+        :return: Tuple of (xlsx_content_bytes, filename)
+        :raises ValueError: If target type is not supported or app mode is excluded
+        """
+        # Validate target type
+        if target_type == "app":
+            if not isinstance(target, App):
+                raise ValueError("Invalid target: expected App instance")
+            if AppMode.value_of(target.mode) in cls.EXCLUDED_APP_MODES:
+                raise ValueError(f"App mode '{target.mode}' does not support evaluation templates")
+            input_fields = cls._get_app_input_fields(target)
+        elif target_type == "snippet":
+            if not isinstance(target, CustomizedSnippet):
+                raise ValueError("Invalid target: expected CustomizedSnippet instance")
+            input_fields = cls._get_snippet_input_fields(target)
+        else:
+            raise ValueError(f"Unsupported target type: {target_type}")
+
+        # Generate XLSX template
+        xlsx_content = cls._generate_xlsx_template(input_fields, target.name)
+
+        # Build filename
+        truncated_name = target.name[:10] + "..." if len(target.name) > 10 else target.name
+        filename = f"{truncated_name}-evaluation-dataset.xlsx"
+
+        return xlsx_content, filename
+
+    @classmethod
+    def _get_app_input_fields(cls, app: App) -> list[dict]:
+        """
+        Get input fields from App's workflow.
+
+        :param app: App instance
+        :return: List of input field definitions
+        """
+        workflow_service = WorkflowService()
+        workflow = workflow_service.get_published_workflow(app_model=app)
+        if not workflow:
+            workflow = workflow_service.get_draft_workflow(app_model=app)
+
+        if not workflow:
+            return []
+
+        # Get user input form from workflow
+        user_input_form = workflow.user_input_form()
+        return user_input_form
+
+    @classmethod
+    def _get_snippet_input_fields(cls, snippet: CustomizedSnippet) -> list[dict]:
+        """
+        Get input fields from Snippet.
+
+        Tries to get from snippet's own input_fields first,
+        then falls back to workflow's user_input_form.
+
+        :param snippet: CustomizedSnippet instance
+        :return: List of input field definitions
+        """
+        # Try snippet's own input_fields first
+        input_fields = snippet.input_fields_list
+        if input_fields:
+            return input_fields
+
+        # Fallback to workflow's user_input_form
+        snippet_service = SnippetService()
+        workflow = snippet_service.get_published_workflow(snippet=snippet)
+        if not workflow:
+            workflow = snippet_service.get_draft_workflow(snippet=snippet)
+
+        if workflow:
+            return workflow.user_input_form()
+
+        return []
+
+    @classmethod
+    def _generate_xlsx_template(cls, input_fields: list[dict], target_name: str) -> bytes:
+        """
+        Generate XLSX template file content.
+
+        Creates a workbook with:
+        - First row as header row with "index" and input field names
+        - Styled header with background color and borders
+        - Empty data rows ready for user input
+
+        :param input_fields: List of input field definitions
+        :param target_name: Name of the target (for sheet name)
+        :return: XLSX file content as bytes
+        """
+        wb = Workbook()
+        ws = wb.active
+
+        sheet_name = "Evaluation Dataset"
+        ws.title = sheet_name
+
+        header_font = Font(bold=True, color="FFFFFF")
+        header_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid")
+        header_alignment = Alignment(horizontal="center", vertical="center")
+        thin_border = Border(
+            left=Side(style="thin"),
+            right=Side(style="thin"),
+            top=Side(style="thin"),
+            bottom=Side(style="thin"),
+        )
+
+        # Build header row
+        headers = ["index"]
+
+        for field in input_fields:
+            field_label = field.get("label") or field.get("variable")
+            headers.append(field_label)
+
+        # Write header row
+        for col_idx, header in enumerate(headers, start=1):
+            cell = ws.cell(row=1, column=col_idx, value=header)
+            cell.font = header_font
+            cell.fill = header_fill
+            cell.alignment = header_alignment
+            cell.border = thin_border
+
+        # Set column widths
+        ws.column_dimensions["A"].width = 10  # index column
+        for col_idx in range(2, len(headers) + 1):
+            ws.column_dimensions[get_column_letter(col_idx)].width = 20
+
+        # Add one empty row with row number for user reference
+        for col_idx in range(1, len(headers) + 1):
+            cell = ws.cell(row=2, column=col_idx, value="")
+            cell.border = thin_border
+            if col_idx == 1:
+                cell.value = 1
+                cell.alignment = Alignment(horizontal="center")
+
+        # Save to bytes
+        output = io.BytesIO()
+        wb.save(output)
+        output.seek(0)
+
+        return output.getvalue()

api/services/snippet_generate_service.py

@@ -0,0 +1,374 @@
+"""
+Service for generating snippet workflow executions.
+
+Uses an adapter pattern to bridge CustomizedSnippet with the App-based
+WorkflowAppGenerator. The adapter (_SnippetAsApp) provides the minimal App-like
+interface needed by the generator, avoiding modifications to core workflow
+infrastructure.
+
+Key invariants:
+- Snippets always run as WORKFLOW mode (not CHAT or ADVANCED_CHAT).
+- The adapter maps snippet.id to app_id in workflow execution records.
+- Snippet debugging has no rate limiting (max_active_requests = 0).
+
+Supported execution modes:
+- Full workflow run (generate): Runs the entire draft workflow as SSE stream.
+- Single node run (run_draft_node): Synchronous single-step debugging for regular nodes.
+- Single iteration run (generate_single_iteration): SSE stream for iteration container nodes.
+- Single loop run (generate_single_loop): SSE stream for loop container nodes.
+"""
+
+import json
+import logging
+from collections.abc import Generator, Mapping, Sequence
+from typing import Any, Union
+
+from sqlalchemy.orm import make_transient
+
+from core.app.app_config.features.file_upload.manager import FileUploadConfigManager
+from core.app.apps.workflow.app_generator import WorkflowAppGenerator
+from core.app.entities.app_invoke_entities import InvokeFrom
+from core.file.models import File
+from factories import file_factory
+from models import Account
+from models.model import AppMode, EndUser
+from models.snippet import CustomizedSnippet
+from models.workflow import Workflow, WorkflowNodeExecutionModel
+from services.snippet_service import SnippetService
+from services.workflow_service import WorkflowService
+
+logger = logging.getLogger(__name__)
+
+
+class _SnippetAsApp:
+    """
+    Minimal adapter that wraps a CustomizedSnippet to satisfy the App-like
+    interface required by WorkflowAppGenerator, WorkflowAppConfigManager,
+    and WorkflowService.run_draft_workflow_node.
+
+    Used properties:
+    - id: maps to snippet.id (stored as app_id in workflows table)
+    - tenant_id: maps to snippet.tenant_id
+    - mode: hardcoded to AppMode.WORKFLOW since snippets always run as workflows
+    - max_active_requests: defaults to 0 (no limit) for snippet debugging
+    - app_model_config_id: None (snippets don't have app model configs)
+    """
+
+    id: str
+    tenant_id: str
+    mode: str
+    max_active_requests: int
+    app_model_config_id: str | None
+
+    def __init__(self, snippet: CustomizedSnippet) -> None:
+        self.id = snippet.id
+        self.tenant_id = snippet.tenant_id
+        self.mode = AppMode.WORKFLOW.value
+        self.max_active_requests = 0
+        self.app_model_config_id = None
+
+
+class SnippetGenerateService:
+    """
+    Service for running snippet workflow executions.
+
+    Adapts CustomizedSnippet to work with the existing App-based
+    WorkflowAppGenerator infrastructure, avoiding duplication of the
+    complex workflow execution pipeline.
+    """
+
+    # Specific ID for the injected virtual Start node so it can be recognised
+    _VIRTUAL_START_NODE_ID = "__snippet_virtual_start__"
+
+    @classmethod
+    def generate(
+        cls,
+        snippet: CustomizedSnippet,
+        user: Union[Account, EndUser],
+        args: Mapping[str, Any],
+        invoke_from: InvokeFrom,
+        streaming: bool = True,
+    ) -> Union[Mapping[str, Any], Generator[Mapping[str, Any] | str, None, None]]:
+        """
+        Run a snippet's draft workflow.
+
+        Retrieves the draft workflow, adapts the snippet to an App-like proxy,
+        then delegates execution to WorkflowAppGenerator.
+
+        If the workflow graph has no Start node, a virtual Start node is injected
+        in-memory so that:
+        1. Graph validation passes (root node must have execution_type=ROOT).
+        2. User inputs are processed into the variable pool by the StartNode logic.
+
+        :param snippet: CustomizedSnippet instance
+        :param user: Account or EndUser initiating the run
+        :param args: Workflow inputs (must include "inputs" key)
+        :param invoke_from: Source of invocation (typically DEBUGGER)
+        :param streaming: Whether to stream the response
+        :return: Blocking response mapping or SSE streaming generator
+        :raises ValueError: If the snippet has no draft workflow
+        """
+        snippet_service = SnippetService()
+        workflow = snippet_service.get_draft_workflow(snippet=snippet)
+        if not workflow:
+            raise ValueError("Workflow not initialized")
+
+        # Inject a virtual Start node when the graph doesn't have one.
+        workflow = cls._ensure_start_node(workflow, snippet)
+
+        # Adapt snippet to App-like interface for WorkflowAppGenerator
+        app_proxy = _SnippetAsApp(snippet)
+
+        return WorkflowAppGenerator.convert_to_event_stream(
+            WorkflowAppGenerator().generate(
+                app_model=app_proxy,  # type: ignore[arg-type]
+                workflow=workflow,
+                user=user,
+                args=args,
+                invoke_from=invoke_from,
+                streaming=streaming,
+                call_depth=0,
+            )
+        )
+
+    @classmethod
+    def _ensure_start_node(cls, workflow: Workflow, snippet: CustomizedSnippet) -> Workflow:
+        """
+        Return *workflow* with a Start node.
+
+        If the graph already contains a Start node, the original workflow is
+        returned unchanged.  Otherwise a virtual Start node is injected and the
+        workflow object is detached from the SQLAlchemy session so the in-memory
+        change is never flushed to the database.
+        """
+        graph_dict = workflow.graph_dict
+        nodes: list[dict[str, Any]] = graph_dict.get("nodes", [])
+
+        has_start = any(node.get("data", {}).get("type") == "start" for node in nodes)
+        if has_start:
+            return workflow
+
+        modified_graph = cls._inject_virtual_start_node(
+            graph_dict=graph_dict,
+            input_fields=snippet.input_fields_list,
+        )
+
+        # Detach from session to prevent accidental DB persistence of the
+        # modified graph.  All attributes remain accessible for read.
+        make_transient(workflow)
+        workflow.graph = json.dumps(modified_graph)
+        return workflow
+
+    @classmethod
+    def _inject_virtual_start_node(
+        cls,
+        graph_dict: Mapping[str, Any],
+        input_fields: list[dict[str, Any]],
+    ) -> dict[str, Any]:
+        """
+        Build a new graph dict with a virtual Start node prepended.
+
+        The virtual Start node is wired to every existing node that has no
+        incoming edges (i.e. the current root candidates).  This guarantees:
+
+        :param graph_dict: Original graph configuration.
+        :param input_fields: Snippet input field definitions from
+            ``CustomizedSnippet.input_fields_list``.
+        :return: New graph dict containing the virtual Start node and edges.
+        """
+        nodes: list[dict[str, Any]] = list(graph_dict.get("nodes", []))
+        edges: list[dict[str, Any]] = list(graph_dict.get("edges", []))
+
+        # Identify nodes with no incoming edges.
+        nodes_with_incoming: set[str] = set()
+        for edge in edges:
+            target = edge.get("target")
+            if isinstance(target, str):
+                nodes_with_incoming.add(target)
+        root_candidate_ids = [n["id"] for n in nodes if n["id"] not in nodes_with_incoming]
+
+        # Build Start node ``variables`` from snippet input fields.
+        start_variables: list[dict[str, Any]] = []
+        for field in input_fields:
+            var: dict[str, Any] = {
+                "variable": field.get("variable", ""),
+                "label": field.get("label", field.get("variable", "")),
+                "type": field.get("type", "text-input"),
+                "required": field.get("required", False),
+                "options": field.get("options", []),
+            }
+            if field.get("max_length") is not None:
+                var["max_length"] = field["max_length"]
+            start_variables.append(var)
+
+        virtual_start_node: dict[str, Any] = {
+            "id": cls._VIRTUAL_START_NODE_ID,
+            "data": {
+                "type": "start",
+                "title": "Start",
+                "variables": start_variables,
+            },
+        }
+
+        # Create edges from virtual Start to each root candidate.
+        new_edges: list[dict[str, Any]] = [
+            {
+                "source": cls._VIRTUAL_START_NODE_ID,
+                "sourceHandle": "source",
+                "target": root_id,
+                "targetHandle": "target",
+            }
+            for root_id in root_candidate_ids
+        ]
+
+        return {
+            **graph_dict,
+            "nodes": [virtual_start_node, *nodes],
+            "edges": [*edges, *new_edges],
+        }
+
+    @classmethod
+    def run_draft_node(
+        cls,
+        snippet: CustomizedSnippet,
+        node_id: str,
+        user_inputs: Mapping[str, Any],
+        account: Account,
+        query: str = "",
+        files: Sequence[File] | None = None,
+    ) -> WorkflowNodeExecutionModel:
+        """
+        Run a single node in a snippet's draft workflow (single-step debugging).
+
+        Retrieves the draft workflow, adapts the snippet to an App-like proxy,
+        parses file inputs, then delegates to WorkflowService.run_draft_workflow_node.
+
+        :param snippet: CustomizedSnippet instance
+        :param node_id: ID of the node to run
+        :param user_inputs: User input values for the node
+        :param account: Account initiating the run
+        :param query: Optional query string
+        :param files: Optional parsed file objects
+        :return: WorkflowNodeExecutionModel with execution results
+        :raises ValueError: If the snippet has no draft workflow
+        """
+        snippet_service = SnippetService()
+        draft_workflow = snippet_service.get_draft_workflow(snippet=snippet)
+        if not draft_workflow:
+            raise ValueError("Workflow not initialized")
+
+        app_proxy = _SnippetAsApp(snippet)
+
+        workflow_service = WorkflowService()
+        return workflow_service.run_draft_workflow_node(
+            app_model=app_proxy,  # type: ignore[arg-type]
+            draft_workflow=draft_workflow,
+            node_id=node_id,
+            user_inputs=user_inputs,
+            account=account,
+            query=query,
+            files=files,
+        )
+
+    @classmethod
+    def generate_single_iteration(
+        cls,
+        snippet: CustomizedSnippet,
+        user: Union[Account, EndUser],
+        node_id: str,
+        args: Mapping[str, Any],
+        streaming: bool = True,
+    ) -> Union[Mapping[str, Any], Generator[Mapping[str, Any] | str, None, None]]:
+        """
+        Run a single iteration node in a snippet's draft workflow.
+
+        Iteration nodes are container nodes that execute their sub-graph multiple
+        times, producing many events. Therefore, this uses the full WorkflowAppGenerator
+        pipeline with SSE streaming (unlike regular single-step node run).
+
+        :param snippet: CustomizedSnippet instance
+        :param user: Account or EndUser initiating the run
+        :param node_id: ID of the iteration node to run
+        :param args: Dict containing 'inputs' key with iteration input data
+        :param streaming: Whether to stream the response (should be True)
+        :return: SSE streaming generator
+        :raises ValueError: If the snippet has no draft workflow
+        """
+        snippet_service = SnippetService()
+        workflow = snippet_service.get_draft_workflow(snippet=snippet)
+        if not workflow:
+            raise ValueError("Workflow not initialized")
+
+        app_proxy = _SnippetAsApp(snippet)
+
+        return WorkflowAppGenerator.convert_to_event_stream(
+            WorkflowAppGenerator().single_iteration_generate(
+                app_model=app_proxy,  # type: ignore[arg-type]
+                workflow=workflow,
+                node_id=node_id,
+                user=user,
+                args=args,
+                streaming=streaming,
+            )
+        )
+
+    @classmethod
+    def generate_single_loop(
+        cls,
+        snippet: CustomizedSnippet,
+        user: Union[Account, EndUser],
+        node_id: str,
+        args: Any,
+        streaming: bool = True,
+    ) -> Union[Mapping[str, Any], Generator[Mapping[str, Any] | str, None, None]]:
+        """
+        Run a single loop node in a snippet's draft workflow.
+
+        Loop nodes are container nodes that execute their sub-graph repeatedly,
+        producing many events. Therefore, this uses the full WorkflowAppGenerator
+        pipeline with SSE streaming (unlike regular single-step node run).
+
+        :param snippet: CustomizedSnippet instance
+        :param user: Account or EndUser initiating the run
+        :param node_id: ID of the loop node to run
+        :param args: Pydantic model with 'inputs' attribute containing loop input data
+        :param streaming: Whether to stream the response (should be True)
+        :return: SSE streaming generator
+        :raises ValueError: If the snippet has no draft workflow
+        """
+        snippet_service = SnippetService()
+        workflow = snippet_service.get_draft_workflow(snippet=snippet)
+        if not workflow:
+            raise ValueError("Workflow not initialized")
+
+        app_proxy = _SnippetAsApp(snippet)
+
+        return WorkflowAppGenerator.convert_to_event_stream(
+            WorkflowAppGenerator().single_loop_generate(
+                app_model=app_proxy,  # type: ignore[arg-type]
+                workflow=workflow,
+                node_id=node_id,
+                user=user,
+                args=args,  # type: ignore[arg-type]
+                streaming=streaming,
+            )
+        )
+
+    @staticmethod
+    def parse_files(workflow: Workflow, files: list[dict] | None = None) -> Sequence[File]:
+        """
+        Parse file mappings into File objects based on workflow configuration.
+
+        :param workflow: Workflow instance for file upload config
+        :param files: Raw file mapping dicts
+        :return: Parsed File objects
+        """
+        files = files or []
+        file_extra_config = FileUploadConfigManager.convert(workflow.features_dict, is_vision=False)
+        if file_extra_config is None:
+            return []
+        return file_factory.build_from_mappings(
+            mappings=files,
+            tenant_id=workflow.tenant_id,
+            config=file_extra_config,
+        )

api/services/snippet_service.py

@@ -0,0 +1,566 @@
+import json
+import logging
+from collections.abc import Mapping, Sequence
+from datetime import UTC, datetime
+from typing import Any
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session, sessionmaker
+
+from core.variables.variables import VariableBase
+from core.workflow.enums import NodeType
+from core.workflow.nodes.node_mapping import LATEST_VERSION, NODE_TYPE_CLASSES_MAPPING
+from extensions.ext_database import db
+from libs.infinite_scroll_pagination import InfiniteScrollPagination
+from models import Account
+from models.enums import WorkflowRunTriggeredFrom
+from models.snippet import CustomizedSnippet, SnippetType
+from models.workflow import (
+    Workflow,
+    WorkflowNodeExecutionModel,
+    WorkflowRun,
+    WorkflowType,
+)
+from repositories.factory import DifyAPIRepositoryFactory
+from services.errors.app import WorkflowHashNotEqualError
+
+logger = logging.getLogger(__name__)
+
+
+class SnippetService:
+    """Service for managing customized snippets."""
+
+    def __init__(self, session_maker: sessionmaker | None = None):
+        """Initialize SnippetService with repository dependencies."""
+        if session_maker is None:
+            session_maker = sessionmaker(bind=db.engine, expire_on_commit=False)
+        self._node_execution_service_repo = DifyAPIRepositoryFactory.create_api_workflow_node_execution_repository(
+            session_maker
+        )
+        self._workflow_run_repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker)
+
+    # --- CRUD Operations ---
+
+    @staticmethod
+    def get_snippets(
+        *,
+        tenant_id: str,
+        page: int = 1,
+        limit: int = 20,
+        keyword: str | None = None,
+    ) -> tuple[Sequence[CustomizedSnippet], int, bool]:
+        """
+        Get paginated list of snippets with optional search.
+
+        :param tenant_id: Tenant ID
+        :param page: Page number (1-indexed)
+        :param limit: Number of items per page
+        :param keyword: Optional search keyword for name/description
+        :return: Tuple of (snippets list, total count, has_more flag)
+        """
+        stmt = (
+            select(CustomizedSnippet)
+            .where(CustomizedSnippet.tenant_id == tenant_id)
+            .order_by(CustomizedSnippet.created_at.desc())
+        )
+
+        if keyword:
+            stmt = stmt.where(
+                CustomizedSnippet.name.ilike(f"%{keyword}%") | CustomizedSnippet.description.ilike(f"%{keyword}%")
+            )
+
+        # Get total count
+        count_stmt = select(func.count()).select_from(stmt.subquery())
+        total = db.session.scalar(count_stmt) or 0
+
+        # Apply pagination
+        stmt = stmt.limit(limit + 1).offset((page - 1) * limit)
+        snippets = list(db.session.scalars(stmt).all())
+
+        has_more = len(snippets) > limit
+        if has_more:
+            snippets = snippets[:-1]
+
+        return snippets, total, has_more
+
+    @staticmethod
+    def get_snippet_by_id(
+        *,
+        snippet_id: str,
+        tenant_id: str,
+    ) -> CustomizedSnippet | None:
+        """
+        Get snippet by ID with tenant isolation.
+
+        :param snippet_id: Snippet ID
+        :param tenant_id: Tenant ID
+        :return: CustomizedSnippet or None
+        """
+        return (
+            db.session.query(CustomizedSnippet)
+            .where(
+                CustomizedSnippet.id == snippet_id,
+                CustomizedSnippet.tenant_id == tenant_id,
+            )
+            .first()
+        )
+
+    @staticmethod
+    def create_snippet(
+        *,
+        tenant_id: str,
+        name: str,
+        description: str | None,
+        snippet_type: SnippetType,
+        icon_info: dict | None,
+        graph: dict | None,
+        input_fields: list[dict] | None,
+        account: Account,
+    ) -> CustomizedSnippet:
+        """
+        Create a new snippet.
+
+        :param tenant_id: Tenant ID
+        :param name: Snippet name (must be unique per tenant)
+        :param description: Snippet description
+        :param snippet_type: Type of snippet (node or group)
+        :param icon_info: Icon information
+        :param graph: Workflow graph structure
+        :param input_fields: Input field definitions
+        :param account: Creator account
+        :return: Created CustomizedSnippet
+        :raises ValueError: If name already exists
+        """
+        # Check if name already exists for this tenant
+        existing = (
+            db.session.query(CustomizedSnippet)
+            .where(
+                CustomizedSnippet.tenant_id == tenant_id,
+                CustomizedSnippet.name == name,
+            )
+            .first()
+        )
+        if existing:
+            raise ValueError(f"Snippet with name '{name}' already exists")
+
+        snippet = CustomizedSnippet(
+            tenant_id=tenant_id,
+            name=name,
+            description=description or "",
+            type=snippet_type.value,
+            icon_info=icon_info,
+            graph=json.dumps(graph) if graph else None,
+            input_fields=json.dumps(input_fields) if input_fields else None,
+            created_by=account.id,
+        )
+
+        db.session.add(snippet)
+        db.session.commit()
+
+        return snippet
+
+    @staticmethod
+    def update_snippet(
+        *,
+        session: Session,
+        snippet: CustomizedSnippet,
+        account_id: str,
+        data: dict,
+    ) -> CustomizedSnippet:
+        """
+        Update snippet attributes.
+
+        :param session: Database session
+        :param snippet: Snippet to update
+        :param account_id: ID of account making the update
+        :param data: Dictionary of fields to update
+        :return: Updated CustomizedSnippet
+        """
+        if "name" in data:
+            # Check if new name already exists for this tenant
+            existing = (
+                session.query(CustomizedSnippet)
+                .where(
+                    CustomizedSnippet.tenant_id == snippet.tenant_id,
+                    CustomizedSnippet.name == data["name"],
+                    CustomizedSnippet.id != snippet.id,
+                )
+                .first()
+            )
+            if existing:
+                raise ValueError(f"Snippet with name '{data['name']}' already exists")
+            snippet.name = data["name"]
+
+        if "description" in data:
+            snippet.description = data["description"]
+
+        if "icon_info" in data:
+            snippet.icon_info = data["icon_info"]
+
+        snippet.updated_by = account_id
+        snippet.updated_at = datetime.now(UTC).replace(tzinfo=None)
+
+        session.add(snippet)
+        return snippet
+
+    @staticmethod
+    def delete_snippet(
+        *,
+        session: Session,
+        snippet: CustomizedSnippet,
+    ) -> bool:
+        """
+        Delete a snippet.
+
+        :param session: Database session
+        :param snippet: Snippet to delete
+        :return: True if deleted successfully
+        """
+        session.delete(snippet)
+        return True
+
+    # --- Workflow Operations ---
+
+    def get_draft_workflow(self, snippet: CustomizedSnippet) -> Workflow | None:
+        """
+        Get draft workflow for snippet.
+
+        :param snippet: CustomizedSnippet instance
+        :return: Draft Workflow or None
+        """
+        workflow = (
+            db.session.query(Workflow)
+            .where(
+                Workflow.tenant_id == snippet.tenant_id,
+                Workflow.app_id == snippet.id,
+                Workflow.type == WorkflowType.SNIPPET.value,
+                Workflow.version == "draft",
+            )
+            .first()
+        )
+        return workflow
+
+    def get_published_workflow(self, snippet: CustomizedSnippet) -> Workflow | None:
+        """
+        Get published workflow for snippet.
+
+        :param snippet: CustomizedSnippet instance
+        :return: Published Workflow or None
+        """
+        if not snippet.workflow_id:
+            return None
+
+        workflow = (
+            db.session.query(Workflow)
+            .where(
+                Workflow.tenant_id == snippet.tenant_id,
+                Workflow.app_id == snippet.id,
+                Workflow.type == WorkflowType.SNIPPET.value,
+                Workflow.id == snippet.workflow_id,
+            )
+            .first()
+        )
+        return workflow
+
+    def sync_draft_workflow(
+        self,
+        *,
+        snippet: CustomizedSnippet,
+        graph: dict,
+        unique_hash: str | None,
+        account: Account,
+        environment_variables: Sequence[VariableBase],
+        conversation_variables: Sequence[VariableBase],
+        input_variables: list[dict] | None = None,
+    ) -> Workflow:
+        """
+        Sync draft workflow for snippet.
+
+        :param snippet: CustomizedSnippet instance
+        :param graph: Workflow graph configuration
+        :param unique_hash: Hash for conflict detection
+        :param account: Account making the change
+        :param environment_variables: Environment variables
+        :param conversation_variables: Conversation variables
+        :param input_variables: Input variables for snippet
+        :return: Synced Workflow
+        :raises WorkflowHashNotEqualError: If hash mismatch
+        """
+        workflow = self.get_draft_workflow(snippet=snippet)
+
+        if workflow and workflow.unique_hash != unique_hash:
+            raise WorkflowHashNotEqualError()
+
+        # Create draft workflow if not found
+        if not workflow:
+            workflow = Workflow(
+                tenant_id=snippet.tenant_id,
+                app_id=snippet.id,
+                features="{}",
+                type=WorkflowType.SNIPPET.value,
+                version="draft",
+                graph=json.dumps(graph),
+                created_by=account.id,
+                environment_variables=environment_variables,
+                conversation_variables=conversation_variables,
+            )
+            db.session.add(workflow)
+            db.session.flush()
+        else:
+            # Update existing draft workflow
+            workflow.graph = json.dumps(graph)
+            workflow.updated_by = account.id
+            workflow.updated_at = datetime.now(UTC).replace(tzinfo=None)
+            workflow.environment_variables = environment_variables
+            workflow.conversation_variables = conversation_variables
+
+        # Update snippet's input_fields if provided
+        if input_variables is not None:
+            snippet.input_fields = json.dumps(input_variables)
+            snippet.updated_by = account.id
+            snippet.updated_at = datetime.now(UTC).replace(tzinfo=None)
+
+        db.session.commit()
+        return workflow
+
+    def publish_workflow(
+        self,
+        *,
+        session: Session,
+        snippet: CustomizedSnippet,
+        account: Account,
+    ) -> Workflow:
+        """
+        Publish the draft workflow as a new version.
+
+        :param session: Database session
+        :param snippet: CustomizedSnippet instance
+        :param account: Account making the change
+        :return: Published Workflow
+        :raises ValueError: If no draft workflow exists
+        """
+        draft_workflow_stmt = select(Workflow).where(
+            Workflow.tenant_id == snippet.tenant_id,
+            Workflow.app_id == snippet.id,
+            Workflow.type == WorkflowType.SNIPPET.value,
+            Workflow.version == "draft",
+        )
+        draft_workflow = session.scalar(draft_workflow_stmt)
+        if not draft_workflow:
+            raise ValueError("No valid workflow found.")
+
+        # Create new published workflow
+        workflow = Workflow.new(
+            tenant_id=snippet.tenant_id,
+            app_id=snippet.id,
+            type=draft_workflow.type,
+            version=str(datetime.now(UTC).replace(tzinfo=None)),
+            graph=draft_workflow.graph,
+            features=draft_workflow.features,
+            created_by=account.id,
+            environment_variables=draft_workflow.environment_variables,
+            conversation_variables=draft_workflow.conversation_variables,
+            marked_name="",
+            marked_comment="",
+        )
+        session.add(workflow)
+
+        # Update snippet version
+        snippet.version += 1
+        snippet.is_published = True
+        snippet.workflow_id = workflow.id
+        snippet.updated_by = account.id
+        session.add(snippet)
+
+        return workflow
+
+    def get_all_published_workflows(
+        self,
+        *,
+        session: Session,
+        snippet: CustomizedSnippet,
+        page: int,
+        limit: int,
+    ) -> tuple[Sequence[Workflow], bool]:
+        """
+        Get all published workflow versions for snippet.
+
+        :param session: Database session
+        :param snippet: CustomizedSnippet instance
+        :param page: Page number
+        :param limit: Items per page
+        :return: Tuple of (workflows list, has_more flag)
+        """
+        if not snippet.workflow_id:
+            return [], False
+
+        stmt = (
+            select(Workflow)
+            .where(
+                Workflow.app_id == snippet.id,
+                Workflow.type == WorkflowType.SNIPPET.value,
+                Workflow.version != "draft",
+            )
+            .order_by(Workflow.version.desc())
+            .limit(limit + 1)
+            .offset((page - 1) * limit)
+        )
+
+        workflows = list(session.scalars(stmt).all())
+        has_more = len(workflows) > limit
+        if has_more:
+            workflows = workflows[:-1]
+
+        return workflows, has_more
+
+    # --- Default Block Configs ---
+
+    def get_default_block_configs(self) -> list[dict]:
+        """
+        Get default block configurations for all node types.
+
+        :return: List of default configurations
+        """
+        default_block_configs: list[dict[str, Any]] = []
+        for node_class_mapping in NODE_TYPE_CLASSES_MAPPING.values():
+            node_class = node_class_mapping[LATEST_VERSION]
+            default_config = node_class.get_default_config()
+            if default_config:
+                default_block_configs.append(dict(default_config))
+
+        return default_block_configs
+
+    def get_default_block_config(self, node_type: str, filters: dict | None = None) -> Mapping[str, object] | None:
+        """
+        Get default config for specific node type.
+
+        :param node_type: Node type string
+        :param filters: Optional filters
+        :return: Default configuration or None
+        """
+        node_type_enum = NodeType(node_type)
+
+        if node_type_enum not in NODE_TYPE_CLASSES_MAPPING:
+            return None
+
+        node_class = NODE_TYPE_CLASSES_MAPPING[node_type_enum][LATEST_VERSION]
+        default_config = node_class.get_default_config(filters=filters)
+        if not default_config:
+            return None
+
+        return default_config
+
+    # --- Workflow Run Operations ---
+
+    def get_snippet_workflow_runs(
+        self,
+        *,
+        snippet: CustomizedSnippet,
+        args: dict,
+    ) -> InfiniteScrollPagination:
+        """
+        Get paginated workflow runs for snippet.
+
+        :param snippet: CustomizedSnippet instance
+        :param args: Request arguments (last_id, limit)
+        :return: InfiniteScrollPagination result
+        """
+        limit = int(args.get("limit", 20))
+        last_id = args.get("last_id")
+
+        triggered_from_values = [
+            WorkflowRunTriggeredFrom.DEBUGGING,
+        ]
+
+        return self._workflow_run_repo.get_paginated_workflow_runs(
+            tenant_id=snippet.tenant_id,
+            app_id=snippet.id,
+            triggered_from=triggered_from_values,
+            limit=limit,
+            last_id=last_id,
+        )
+
+    def get_snippet_workflow_run(
+        self,
+        *,
+        snippet: CustomizedSnippet,
+        run_id: str,
+    ) -> WorkflowRun | None:
+        """
+        Get workflow run details.
+
+        :param snippet: CustomizedSnippet instance
+        :param run_id: Workflow run ID
+        :return: WorkflowRun or None
+        """
+        return self._workflow_run_repo.get_workflow_run_by_id(
+            tenant_id=snippet.tenant_id,
+            app_id=snippet.id,
+            run_id=run_id,
+        )
+
+    def get_snippet_workflow_run_node_executions(
+        self,
+        *,
+        snippet: CustomizedSnippet,
+        run_id: str,
+    ) -> Sequence[WorkflowNodeExecutionModel]:
+        """
+        Get workflow run node execution list.
+
+        :param snippet: CustomizedSnippet instance
+        :param run_id: Workflow run ID
+        :return: List of WorkflowNodeExecutionModel
+        """
+        workflow_run = self.get_snippet_workflow_run(snippet=snippet, run_id=run_id)
+        if not workflow_run:
+            return []
+
+        node_executions = self._node_execution_service_repo.get_executions_by_workflow_run(
+            tenant_id=snippet.tenant_id,
+            app_id=snippet.id,
+            workflow_run_id=workflow_run.id,
+        )
+
+        return node_executions
+
+    # --- Node Execution Operations ---
+
+    def get_snippet_node_last_run(
+        self,
+        *,
+        snippet: CustomizedSnippet,
+        workflow: Workflow,
+        node_id: str,
+    ) -> WorkflowNodeExecutionModel | None:
+        """
+        Get the most recent execution for a specific node in a snippet workflow.
+
+        :param snippet: CustomizedSnippet instance
+        :param workflow: Workflow instance
+        :param node_id: Node identifier
+        :return: WorkflowNodeExecutionModel or None
+        """
+        return self._node_execution_service_repo.get_node_last_execution(
+            tenant_id=snippet.tenant_id,
+            app_id=snippet.id,
+            workflow_id=workflow.id,
+            node_id=node_id,
+        )
+
+    # --- Use Count ---
+
+    @staticmethod
+    def increment_use_count(
+        *,
+        session: Session,
+        snippet: CustomizedSnippet,
+    ) -> None:
+        """
+        Increment the use_count when snippet is used.
+
+        :param session: Database session
+        :param snippet: CustomizedSnippet instance
+        """
+        snippet.use_count += 1
+        session.add(snippet)