feat: allow duplicate snippet names and enhance snippet import handling

# Conflicts:
#	api/tests/unit_tests/controllers/service_api/dataset/test_hit_testing.py

api/controllers/console/__init__.py

@@ -118,6 +118,7 @@ from .explore import (
     saved_message,
     trial,
 )
+from .snippets import snippet_workflow, snippet_workflow_draft_variable
 from .socketio import workflow as socketio_workflow
 
 # Import tag controllers
@@ -133,6 +134,7 @@ from .workspace import (
     model_providers,
     models,
     plugin,
+    snippets,
     tool_providers,
     trigger_providers,
     workspace,
@@ -205,6 +207,9 @@ __all__ = [
     "saved_message",
     "setup",
     "site",
+    "snippet_workflow",
+    "snippet_workflow_draft_variable",
+    "snippets",
     "socketio_workflow",
     "spec",
     "statistic",

api/controllers/console/snippets/payloads.py

@@ -0,0 +1,160 @@
+import uuid
+from typing import Any, Literal
+
+from pydantic import BaseModel, Field, field_validator
+
+
+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
+    is_published: bool | None = Field(default=None, description="Filter by published status")
+    creators: list[str] | None = Field(default=None, description="Filter by creator account IDs")
+    tag_ids: list[str] | None = Field(default=None, description="Filter by tag IDs")
+
+    @field_validator("creators", mode="before")
+    @classmethod
+    def parse_creators(cls, value: object) -> list[str] | None:
+        """Normalize creators filter from query string or list input."""
+        return cls._normalize_string_list(value)
+
+    @field_validator("tag_ids", mode="before")
+    @classmethod
+    def parse_tag_ids(cls, value: object) -> list[str] | None:
+        """Normalize and validate tag IDs from query string or list input."""
+        items = cls._normalize_string_list(value)
+        if not items:
+            return None
+        try:
+            return [str(uuid.UUID(item)) for item in items]
+        except ValueError as exc:
+            raise ValueError("Invalid UUID format in tag_ids.") from exc
+
+    @staticmethod
+    def _normalize_string_list(value: object) -> list[str] | None:
+        if value is None:
+            return None
+        if isinstance(value, str):
+            return [item.strip() for item in value.split(",") if item.strip()] or None
+        if isinstance(value, list):
+            return [str(item).strip() for item in value if str(item).strip()] or None
+        return 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
+    conversation_variables: list[dict[str, Any]] | None = Field(
+        default=None,
+        description="Ignored. Snippet workflows do not persist conversation variables.",
+    )
+    input_fields: list[dict[str, Any]] | None = None
+
+
+class SnippetWorkflowListQuery(BaseModel):
+    """Query parameters for listing snippet published workflows."""
+
+    page: int = Field(default=1, ge=1, le=99999)
+    limit: int = Field(default=10, ge=1, le=100)
+
+
+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
+
+
+class SnippetImportPayload(BaseModel):
+    """Payload for importing snippet from DSL."""
+
+    mode: str = Field(..., description="Import mode: yaml-content or yaml-url")
+    yaml_content: str | None = Field(default=None, description="YAML content (required for yaml-content mode)")
+    yaml_url: str | None = Field(default=None, description="YAML URL (required for yaml-url mode)")
+    name: str | None = Field(default=None, description="Override snippet name")
+    description: str | None = Field(default=None, description="Override snippet description")
+    snippet_id: str | None = Field(default=None, description="Snippet ID to update (optional)")
+
+
+class IncludeSecretQuery(BaseModel):
+    """Query parameter for including secret variables in export."""
+
+    include_secret: str = Field(default="false", description="Whether to include secret variables")

api/controllers/console/snippets/snippet_workflow.py

@@ -0,0 +1,638 @@
+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
+from pydantic import Field
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import BadRequest, InternalServerError, NotFound
+
+from controllers.common.schema import register_response_schema_models, register_schema_models
+from controllers.console import console_ns
+from controllers.console.app.error import DraftWorkflowNotExist, DraftWorkflowNotSync
+from controllers.console.app.workflow import (
+    RESTORE_SOURCE_WORKFLOW_MUST_BE_PUBLISHED_MESSAGE,
+    WorkflowPaginationResponse,
+    WorkflowResponse,
+)
+from controllers.console.snippets.payloads import (
+    PublishWorkflowPayload,
+    SnippetDraftNodeRunPayload,
+    SnippetDraftRunPayload,
+    SnippetDraftSyncPayload,
+    SnippetIterationNodeRunPayload,
+    SnippetLoopNodeRunPayload,
+    SnippetWorkflowListQuery,
+    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 extensions.ext_database import db
+from extensions.ext_redis import redis_client
+from fields.workflow_run_fields import (
+    WorkflowRunDetailResponse,
+    WorkflowRunNodeExecutionListResponse,
+    WorkflowRunNodeExecutionResponse,
+    WorkflowRunPaginationResponse,
+)
+from graphon.graph_engine.manager import GraphEngineManager
+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 IsDraftWorkflowError, WorkflowHashNotEqualError, WorkflowNotFoundError
+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
+class SnippetWorkflowResponse(WorkflowResponse):
+    input_fields: list[dict] = Field(default_factory=list)
+
+
+register_schema_models(
+    console_ns,
+    SnippetDraftSyncPayload,
+    SnippetDraftNodeRunPayload,
+    SnippetDraftRunPayload,
+    SnippetIterationNodeRunPayload,
+    SnippetLoopNodeRunPayload,
+    SnippetWorkflowListQuery,
+    WorkflowRunQuery,
+    PublishWorkflowPayload,
+)
+register_response_schema_models(
+    console_ns,
+    SnippetWorkflowResponse,
+    WorkflowPaginationResponse,
+    WorkflowRunPaginationResponse,
+    WorkflowRunDetailResponse,
+    WorkflowRunNodeExecutionListResponse,
+    WorkflowRunNodeExecutionResponse,
+)
+
+
+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", console_ns.models[SnippetWorkflowResponse.__name__])
+    @console_ns.response(404, "Snippet or draft workflow not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    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()
+
+        db.session.expunge(workflow)
+        workflow.conversation_variables = []
+        workflow.input_fields = snippet.input_fields_list
+        return SnippetWorkflowResponse.model_validate(workflow, from_attributes=True).model_dump(mode="json")
+
+    @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:
+            snippet_service = SnippetService()
+            workflow = snippet_service.sync_draft_workflow(
+                snippet=snippet,
+                graph=payload.graph,
+                unique_hash=payload.hash,
+                account=current_user,
+                input_fields=payload.input_fields,
+            )
+        except WorkflowHashNotEqualError:
+            raise DraftWorkflowNotSync()
+        except ValueError as e:
+            return {"message": str(e)}, 400
+
+        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", console_ns.models[SnippetWorkflowResponse.__name__])
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    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)
+
+        if not workflow:
+            return None
+
+        workflow.input_fields = snippet.input_fields_list
+        return SnippetWorkflowResponse.model_validate(workflow, from_attributes=True).model_dump(mode="json")
+
+    @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>/workflows")
+class SnippetPublishedAllWorkflowApi(Resource):
+    @console_ns.expect(console_ns.models[SnippetWorkflowListQuery.__name__])
+    @console_ns.doc("get_all_snippet_published_workflows")
+    @console_ns.doc(description="Get all published workflows for a snippet")
+    @console_ns.doc(params={"snippet_id": "Snippet ID"})
+    @console_ns.response(200, "Published workflows retrieved successfully", console_ns.models[WorkflowPaginationResponse.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def get(self, snippet: CustomizedSnippet):
+        """Get all published workflow versions for snippet."""
+        args = SnippetWorkflowListQuery.model_validate(request.args.to_dict(flat=True))
+
+        snippet_service = SnippetService()
+        with Session(db.engine) as session:
+            workflows, has_more = snippet_service.get_all_published_workflows(
+                session=session,
+                snippet=snippet,
+                page=args.page,
+                limit=args.limit,
+            )
+
+        return WorkflowPaginationResponse.model_validate(
+            {
+                "items": workflows,
+                "page": args.page,
+                "limit": args.limit,
+                "has_more": has_more,
+            },
+            from_attributes=True,
+        ).model_dump(mode="json")
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/<string:workflow_id>/restore")
+class SnippetDraftWorkflowRestoreApi(Resource):
+    @console_ns.doc("restore_snippet_workflow_to_draft")
+    @console_ns.doc(description="Restore a published snippet workflow version into the draft workflow")
+    @console_ns.doc(params={"snippet_id": "Snippet ID", "workflow_id": "Published workflow ID"})
+    @console_ns.response(200, "Workflow restored successfully")
+    @console_ns.response(400, "Source workflow must be published")
+    @console_ns.response(404, "Workflow not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet, workflow_id: str):
+        """Restore a published snippet workflow version into the draft workflow."""
+        current_user, _ = current_account_with_tenant()
+        snippet_service = SnippetService()
+
+        try:
+            workflow = snippet_service.restore_published_workflow_to_draft(
+                snippet=snippet,
+                workflow_id=workflow_id,
+                account=current_user,
+            )
+        except IsDraftWorkflowError as exc:
+            raise BadRequest(RESTORE_SOURCE_WORKFLOW_MUST_BE_PUBLISHED_MESSAGE) from exc
+        except WorkflowNotFoundError as exc:
+            raise NotFound(str(exc)) from exc
+        except ValueError as exc:
+            raise BadRequest(str(exc)) from exc
+
+        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>/workflow-runs")
+class SnippetWorkflowRunsApi(Resource):
+    @console_ns.doc("list_snippet_workflow_runs")
+    @console_ns.response(200, "Workflow runs retrieved successfully", console_ns.models[WorkflowRunPaginationResponse.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    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 WorkflowRunPaginationResponse.model_validate(result, from_attributes=True).model_dump(mode="json")
+
+
+@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", console_ns.models[WorkflowRunDetailResponse.__name__])
+    @console_ns.response(404, "Workflow run not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    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 WorkflowRunDetailResponse.model_validate(workflow_run, from_attributes=True).model_dump(mode="json")
+
+
+@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",
+        console_ns.models[WorkflowRunNodeExecutionListResponse.__name__],
+    )
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    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 WorkflowRunNodeExecutionListResponse.model_validate(
+            {"data": node_executions}, from_attributes=True
+        ).model_dump(mode="json")
+
+
+@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", console_ns.models[WorkflowRunNodeExecutionResponse.__name__]
+    )
+    @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 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 WorkflowRunNodeExecutionResponse.model_validate(
+            workflow_node_execution, from_attributes=True
+        ).model_dump(mode="json")
+
+
+@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", console_ns.models[WorkflowRunNodeExecutionResponse.__name__]
+    )
+    @console_ns.response(404, "Snippet, draft workflow, or node last run not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    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 WorkflowRunNodeExecutionResponse.model_validate(node_exec, from_attributes=True).model_dump(mode="json")
+
+
+@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(redis_client).send_stop_command(task_id)
+
+        return {"result": "success"}

api/controllers/console/snippets/snippet_workflow_draft_variable.py

@@ -0,0 +1,319 @@
+"""
+Snippet draft workflow variable APIs.
+
+Mirrors console app routes under /apps/.../workflows/draft/variables for snippet scope,
+using CustomizedSnippet.id as WorkflowDraftVariable.app_id (same invariant as snippet execution).
+
+Snippet workflows do not expose system variables (`node_id == sys`) or conversation variables
+(`node_id == conversation`): paginated list queries exclude those rows; single-variable GET/PATCH/DELETE/reset
+reject them; `GET .../system-variables` and `GET .../conversation-variables` return empty lists for API parity.
+Other routes mirror `workflow_draft_variable` app APIs under `/snippets/...`.
+"""
+
+from collections.abc import Callable
+from functools import wraps
+from typing import Any, ParamSpec, TypeVar
+
+from flask import Response, request
+from flask_restx import Resource, marshal, marshal_with
+from sqlalchemy.orm import Session
+
+from controllers.console import console_ns
+from controllers.console.app.error import DraftWorkflowNotExist
+from controllers.console.app.workflow_draft_variable import (
+    WorkflowDraftVariableListQuery,
+    WorkflowDraftVariableUpdatePayload,
+    _ensure_variable_access,
+    _file_access_controller,
+    validate_node_id,
+    workflow_draft_variable_list_model,
+    workflow_draft_variable_list_without_value_model,
+    workflow_draft_variable_model,
+)
+from controllers.console.snippets.snippet_workflow import get_snippet
+from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
+from controllers.web.error import InvalidArgumentError, NotFoundError
+from core.workflow.variable_prefixes import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
+from extensions.ext_database import db
+from factories.file_factory import build_from_mapping, build_from_mappings
+from factories.variable_factory import build_segment_with_type
+from graphon.variables.types import SegmentType
+from libs.login import current_user, login_required
+from models.snippet import CustomizedSnippet
+from models.workflow import WorkflowDraftVariable
+from services.snippet_service import SnippetService
+from services.workflow_draft_variable_service import WorkflowDraftVariableList, WorkflowDraftVariableService
+
+P = ParamSpec("P")
+R = TypeVar("R")
+
+_SNIPPET_EXCLUDED_DRAFT_VARIABLE_NODE_IDS: frozenset[str] = frozenset(
+    {SYSTEM_VARIABLE_NODE_ID, CONVERSATION_VARIABLE_NODE_ID}
+)
+
+
+def _ensure_snippet_draft_variable_row_allowed(
+    *,
+    variable: WorkflowDraftVariable,
+    variable_id: str,
+) -> None:
+    """Snippet scope only supports canvas-node draft variables; treat sys/conversation rows as not found."""
+    if variable.node_id in _SNIPPET_EXCLUDED_DRAFT_VARIABLE_NODE_IDS:
+        raise NotFoundError(description=f"variable not found, id={variable_id}")
+
+
+def _snippet_draft_var_prerequisite(f: Callable[P, R]) -> Callable[P, R]:
+    """Setup, auth, snippet resolution, and tenant edit permission (same stack as snippet workflow APIs)."""
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    @wraps(f)
+    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
+        return f(*args, **kwargs)
+
+    return wrapper
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/variables")
+class SnippetWorkflowVariableCollectionApi(Resource):
+    @console_ns.expect(console_ns.models[WorkflowDraftVariableListQuery.__name__])
+    @console_ns.doc("get_snippet_workflow_variables")
+    @console_ns.doc(description="List draft workflow variables without values (paginated, snippet scope)")
+    @console_ns.response(
+        200,
+        "Workflow variables retrieved successfully",
+        workflow_draft_variable_list_without_value_model,
+    )
+    @_snippet_draft_var_prerequisite
+    @marshal_with(workflow_draft_variable_list_without_value_model)
+    def get(self, snippet: CustomizedSnippet) -> WorkflowDraftVariableList:
+        args = WorkflowDraftVariableListQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+
+        snippet_service = SnippetService()
+        if snippet_service.get_draft_workflow(snippet=snippet) is None:
+            raise DraftWorkflowNotExist()
+
+        with Session(bind=db.engine, expire_on_commit=False) as session:
+            draft_var_srv = WorkflowDraftVariableService(session=session)
+            workflow_vars = draft_var_srv.list_variables_without_values(
+                app_id=snippet.id,
+                page=args.page,
+                limit=args.limit,
+                user_id=current_user.id,
+                exclude_node_ids=_SNIPPET_EXCLUDED_DRAFT_VARIABLE_NODE_IDS,
+            )
+
+        return workflow_vars
+
+    @console_ns.doc("delete_snippet_workflow_variables")
+    @console_ns.doc(description="Delete all draft workflow variables for the current user (snippet scope)")
+    @console_ns.response(204, "Workflow variables deleted successfully")
+    @_snippet_draft_var_prerequisite
+    def delete(self, snippet: CustomizedSnippet) -> Response:
+        draft_var_srv = WorkflowDraftVariableService(session=db.session())
+        draft_var_srv.delete_user_workflow_variables(snippet.id, user_id=current_user.id)
+        db.session.commit()
+        return Response("", 204)
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/nodes/<string:node_id>/variables")
+class SnippetNodeVariableCollectionApi(Resource):
+    @console_ns.doc("get_snippet_node_variables")
+    @console_ns.doc(description="Get variables for a specific node (snippet draft workflow)")
+    @console_ns.response(200, "Node variables retrieved successfully", workflow_draft_variable_list_model)
+    @_snippet_draft_var_prerequisite
+    @marshal_with(workflow_draft_variable_list_model)
+    def get(self, snippet: CustomizedSnippet, node_id: str) -> WorkflowDraftVariableList:
+        validate_node_id(node_id)
+        with Session(bind=db.engine, expire_on_commit=False) as session:
+            draft_var_srv = WorkflowDraftVariableService(session=session)
+            node_vars = draft_var_srv.list_node_variables(snippet.id, node_id, user_id=current_user.id)
+
+        return node_vars
+
+    @console_ns.doc("delete_snippet_node_variables")
+    @console_ns.doc(description="Delete all variables for a specific node (snippet draft workflow)")
+    @console_ns.response(204, "Node variables deleted successfully")
+    @_snippet_draft_var_prerequisite
+    def delete(self, snippet: CustomizedSnippet, node_id: str) -> Response:
+        validate_node_id(node_id)
+        srv = WorkflowDraftVariableService(db.session())
+        srv.delete_node_variables(snippet.id, node_id, user_id=current_user.id)
+        db.session.commit()
+        return Response("", 204)
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/variables/<uuid:variable_id>")
+class SnippetVariableApi(Resource):
+    @console_ns.doc("get_snippet_workflow_variable")
+    @console_ns.doc(description="Get a specific draft workflow variable (snippet scope)")
+    @console_ns.response(200, "Variable retrieved successfully", workflow_draft_variable_model)
+    @console_ns.response(404, "Variable not found")
+    @_snippet_draft_var_prerequisite
+    @marshal_with(workflow_draft_variable_model)
+    def get(self, snippet: CustomizedSnippet, variable_id: str) -> WorkflowDraftVariable:
+        draft_var_srv = WorkflowDraftVariableService(session=db.session())
+        variable = _ensure_variable_access(
+            variable=draft_var_srv.get_variable(variable_id=variable_id),
+            app_id=snippet.id,
+            variable_id=variable_id,
+        )
+        _ensure_snippet_draft_variable_row_allowed(variable=variable, variable_id=variable_id)
+        return variable
+
+    @console_ns.doc("update_snippet_workflow_variable")
+    @console_ns.doc(description="Update a draft workflow variable (snippet scope)")
+    @console_ns.expect(console_ns.models[WorkflowDraftVariableUpdatePayload.__name__])
+    @console_ns.response(200, "Variable updated successfully", workflow_draft_variable_model)
+    @console_ns.response(404, "Variable not found")
+    @_snippet_draft_var_prerequisite
+    @marshal_with(workflow_draft_variable_model)
+    def patch(self, snippet: CustomizedSnippet, variable_id: str) -> WorkflowDraftVariable:
+        draft_var_srv = WorkflowDraftVariableService(session=db.session())
+        args_model = WorkflowDraftVariableUpdatePayload.model_validate(console_ns.payload or {})
+
+        variable = _ensure_variable_access(
+            variable=draft_var_srv.get_variable(variable_id=variable_id),
+            app_id=snippet.id,
+            variable_id=variable_id,
+        )
+        _ensure_snippet_draft_variable_row_allowed(variable=variable, variable_id=variable_id)
+
+        new_name = args_model.name
+        raw_value = args_model.value
+        if new_name is None and raw_value is None:
+            return variable
+
+        new_value = None
+        if raw_value is not None:
+            if variable.value_type == SegmentType.FILE:
+                if not isinstance(raw_value, dict):
+                    raise InvalidArgumentError(description=f"expected dict for file, got {type(raw_value)}")
+                raw_value = build_from_mapping(
+                    mapping=raw_value,
+                    tenant_id=snippet.tenant_id,
+                    access_controller=_file_access_controller,
+                )
+            elif variable.value_type == SegmentType.ARRAY_FILE:
+                if not isinstance(raw_value, list):
+                    raise InvalidArgumentError(description=f"expected list for files, got {type(raw_value)}")
+                if len(raw_value) > 0 and not isinstance(raw_value[0], dict):
+                    raise InvalidArgumentError(description=f"expected dict for files[0], got {type(raw_value)}")
+                raw_value = build_from_mappings(
+                    mappings=raw_value,
+                    tenant_id=snippet.tenant_id,
+                    access_controller=_file_access_controller,
+                )
+            new_value = build_segment_with_type(variable.value_type, raw_value)
+        draft_var_srv.update_variable(variable, name=new_name, value=new_value)
+        db.session.commit()
+        return variable
+
+    @console_ns.doc("delete_snippet_workflow_variable")
+    @console_ns.doc(description="Delete a draft workflow variable (snippet scope)")
+    @console_ns.response(204, "Variable deleted successfully")
+    @console_ns.response(404, "Variable not found")
+    @_snippet_draft_var_prerequisite
+    def delete(self, snippet: CustomizedSnippet, variable_id: str) -> Response:
+        draft_var_srv = WorkflowDraftVariableService(session=db.session())
+        variable = _ensure_variable_access(
+            variable=draft_var_srv.get_variable(variable_id=variable_id),
+            app_id=snippet.id,
+            variable_id=variable_id,
+        )
+        _ensure_snippet_draft_variable_row_allowed(variable=variable, variable_id=variable_id)
+        draft_var_srv.delete_variable(variable)
+        db.session.commit()
+        return Response("", 204)
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/variables/<uuid:variable_id>/reset")
+class SnippetVariableResetApi(Resource):
+    @console_ns.doc("reset_snippet_workflow_variable")
+    @console_ns.doc(description="Reset a draft workflow variable to its default value (snippet scope)")
+    @console_ns.response(200, "Variable reset successfully", workflow_draft_variable_model)
+    @console_ns.response(204, "Variable reset (no content)")
+    @console_ns.response(404, "Variable not found")
+    @_snippet_draft_var_prerequisite
+    def put(self, snippet: CustomizedSnippet, variable_id: str) -> Response | Any:
+        draft_var_srv = WorkflowDraftVariableService(session=db.session())
+        snippet_service = SnippetService()
+        draft_workflow = snippet_service.get_draft_workflow(snippet=snippet)
+        if draft_workflow is None:
+            raise NotFoundError(
+                f"Draft workflow not found, snippet_id={snippet.id}",
+            )
+        variable = _ensure_variable_access(
+            variable=draft_var_srv.get_variable(variable_id=variable_id),
+            app_id=snippet.id,
+            variable_id=variable_id,
+        )
+        _ensure_snippet_draft_variable_row_allowed(variable=variable, variable_id=variable_id)
+
+        resetted = draft_var_srv.reset_variable(draft_workflow, variable)
+        db.session.commit()
+        if resetted is None:
+            return Response("", 204)
+        return marshal(resetted, workflow_draft_variable_model)
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/conversation-variables")
+class SnippetConversationVariableCollectionApi(Resource):
+    @console_ns.doc("get_snippet_conversation_variables")
+    @console_ns.doc(
+        description="Conversation variables are not used in snippet workflows; returns an empty list for API parity"
+    )
+    @console_ns.response(200, "Conversation variables retrieved successfully", workflow_draft_variable_list_model)
+    @_snippet_draft_var_prerequisite
+    @marshal_with(workflow_draft_variable_list_model)
+    def get(self, snippet: CustomizedSnippet) -> WorkflowDraftVariableList:
+        return WorkflowDraftVariableList(variables=[])
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/system-variables")
+class SnippetSystemVariableCollectionApi(Resource):
+    @console_ns.doc("get_snippet_system_variables")
+    @console_ns.doc(
+        description="System variables are not used in snippet workflows; returns an empty list for API parity"
+    )
+    @console_ns.response(200, "System variables retrieved successfully", workflow_draft_variable_list_model)
+    @_snippet_draft_var_prerequisite
+    @marshal_with(workflow_draft_variable_list_model)
+    def get(self, snippet: CustomizedSnippet) -> WorkflowDraftVariableList:
+        return WorkflowDraftVariableList(variables=[])
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/environment-variables")
+class SnippetEnvironmentVariableCollectionApi(Resource):
+    @console_ns.doc("get_snippet_environment_variables")
+    @console_ns.doc(description="Get environment variables from snippet draft workflow graph")
+    @console_ns.response(200, "Environment variables retrieved successfully")
+    @console_ns.response(404, "Draft workflow not found")
+    @_snippet_draft_var_prerequisite
+    def get(self, snippet: CustomizedSnippet) -> dict[str, list[dict[str, Any]]]:
+        snippet_service = SnippetService()
+        workflow = snippet_service.get_draft_workflow(snippet=snippet)
+        if workflow is None:
+            raise DraftWorkflowNotExist()
+
+        env_vars_list: list[dict[str, Any]] = []
+        for v in workflow.environment_variables:
+            env_vars_list.append(
+                {
+                    "id": v.id,
+                    "type": "env",
+                    "name": v.name,
+                    "description": v.description,
+                    "selector": v.selector,
+                    "value_type": v.value_type.exposed_type().value,
+                    "value": v.value,
+                    "edited": False,
+                    "visible": True,
+                    "editable": True,
+                }
+            )
+
+        return {"items": env_vars_list}

api/controllers/console/tag/tags.py

@@ -43,7 +43,7 @@ class TagBindingRemovePayload(BaseModel):
 
 
 class TagListQueryParam(BaseModel):
-    type: Literal["knowledge", "app", ""] = Field("", description="Tag type filter")
+    type: Literal["knowledge", "app", "snippet", ""] = Field("", description="Tag type filter")
     keyword: str | None = Field(None, description="Search keyword")
 
 
@@ -88,7 +88,10 @@ class TagListApi(Resource):
     @login_required
     @account_initialization_required
     @console_ns.doc(
-        params={"type": 'Tag type filter. Can be "knowledge" or "app".', "keyword": "Search keyword for tag name."}
+        params={
+            "type": 'Tag type filter. Can be "knowledge", "app", or "snippet".',
+            "keyword": "Search keyword for tag name.",
+        }
     )
     @console_ns.doc(responses={200: ("Success", [console_ns.models[TagResponse.__name__]])})
     def get(self):

api/controllers/console/workspace/snippets.py

@@ -0,0 +1,404 @@
+import logging
+import re
+from urllib.parse import quote
+
+from flask import Response, request
+from flask_restx import Resource, marshal
+from sqlalchemy.orm import Session
+from werkzeug.datastructures import MultiDict
+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,
+    IncludeSecretQuery,
+    SnippetImportPayload,
+    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.app_dsl_service import ImportStatus
+from services.snippet_dsl_service import SnippetDslService
+from services.snippet_service import SnippetService
+
+logger = logging.getLogger(__name__)
+_TAG_IDS_BRACKET_PATTERN = re.compile(r"^tag_ids\[(\d+)\]$")
+
+
+def _normalize_snippet_list_query_args(query_args: MultiDict[str, str]) -> dict[str, str | list[str]]:
+    normalized: dict[str, str | list[str]] = {}
+    indexed_tag_ids: list[tuple[int, str]] = []
+
+    for key in query_args:
+        match = _TAG_IDS_BRACKET_PATTERN.fullmatch(key)
+        if match:
+            indexed_tag_ids.extend((int(match.group(1)), value) for value in query_args.getlist(key))
+            continue
+
+        value = query_args.get(key)
+        if value is not None:
+            normalized[key] = value
+
+    if indexed_tag_ids:
+        normalized["tag_ids"] = [value for _, value in sorted(indexed_tag_ids)]
+
+    return normalized
+
+
+# Register Pydantic models with Swagger
+register_schema_models(
+    console_ns,
+    SnippetListQuery,
+    CreateSnippetPayload,
+    UpdateSnippetPayload,
+    SnippetImportPayload,
+    IncludeSecretQuery,
+)
+
+# 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 = SnippetListQuery.model_validate(_normalize_snippet_list_query_args(request.args))
+
+        snippets, total, has_more = SnippetService.get_snippets(
+            tenant_id=current_tenant_id,
+            page=query.page,
+            limit=query.limit,
+            keyword=query.keyword,
+            is_published=query.is_published,
+            creators=query.creators,
+            tag_ids=query.tag_ids,
+        )
+
+        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")
+    @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,
+                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")
+    @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
+
+
+@console_ns.route("/workspaces/current/customized-snippets/<uuid:snippet_id>/export")
+class CustomizedSnippetExportApi(Resource):
+    @console_ns.doc("export_customized_snippet")
+    @console_ns.doc(description="Export snippet configuration as DSL")
+    @console_ns.doc(params={"snippet_id": "Snippet ID to export"})
+    @console_ns.response(200, "Snippet exported successfully")
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def get(self, snippet_id: str):
+        """Export snippet as DSL."""
+        _, 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")
+
+        # Get include_secret parameter
+        query = IncludeSecretQuery.model_validate(request.args.to_dict())
+
+        with Session(db.engine) as session:
+            export_service = SnippetDslService(session)
+            result = export_service.export_snippet_dsl(snippet=snippet, include_secret=query.include_secret == "true")
+
+        # Set filename with .snippet extension
+        filename = f"{snippet.name}.snippet"
+        encoded_filename = quote(filename)
+        
+        response = Response(
+            result,
+            mimetype="application/x-yaml",
+        )
+        response.headers["Content-Disposition"] = f"attachment; filename*=UTF-8''{encoded_filename}"
+        response.headers["Content-Type"] = "application/x-yaml"
+        
+        return response
+
+
+@console_ns.route("/workspaces/current/customized-snippets/imports")
+class CustomizedSnippetImportApi(Resource):
+    @console_ns.doc("import_customized_snippet")
+    @console_ns.doc(description="Import snippet from DSL")
+    @console_ns.expect(console_ns.models.get(SnippetImportPayload.__name__))
+    @console_ns.response(200, "Snippet imported successfully")
+    @console_ns.response(202, "Import pending confirmation")
+    @console_ns.response(400, "Import failed")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def post(self):
+        """Import snippet from DSL."""
+        current_user, _ = current_account_with_tenant()
+        payload = SnippetImportPayload.model_validate(console_ns.payload or {})
+
+        with Session(db.engine) as session:
+            import_service = SnippetDslService(session)
+            result = import_service.import_snippet(
+                account=current_user,
+                import_mode=payload.mode,
+                yaml_content=payload.yaml_content,
+                yaml_url=payload.yaml_url,
+                snippet_id=payload.snippet_id,
+                name=payload.name,
+                description=payload.description,
+            )
+            session.commit()
+
+        # Return appropriate status code based on result
+        status = result.status
+        if status == ImportStatus.FAILED:
+            return result.model_dump(mode="json"), 400
+        elif status == ImportStatus.PENDING:
+            return result.model_dump(mode="json"), 202
+        return result.model_dump(mode="json"), 200
+
+
+@console_ns.route("/workspaces/current/customized-snippets/imports/<string:import_id>/confirm")
+class CustomizedSnippetImportConfirmApi(Resource):
+    @console_ns.doc("confirm_snippet_import")
+    @console_ns.doc(description="Confirm a pending snippet import")
+    @console_ns.doc(params={"import_id": "Import ID to confirm"})
+    @console_ns.response(200, "Import confirmed successfully")
+    @console_ns.response(400, "Import failed")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def post(self, import_id: str):
+        """Confirm a pending snippet import."""
+        current_user, _ = current_account_with_tenant()
+
+        with Session(db.engine) as session:
+            import_service = SnippetDslService(session)
+            result = import_service.confirm_import(import_id=import_id, account=current_user)
+            session.commit()
+
+        if result.status == ImportStatus.FAILED:
+            return result.model_dump(mode="json"), 400
+        return result.model_dump(mode="json"), 200
+
+
+@console_ns.route("/workspaces/current/customized-snippets/<uuid:snippet_id>/check-dependencies")
+class CustomizedSnippetCheckDependenciesApi(Resource):
+    @console_ns.doc("check_snippet_dependencies")
+    @console_ns.doc(description="Check dependencies for a snippet")
+    @console_ns.doc(params={"snippet_id": "Snippet ID"})
+    @console_ns.response(200, "Dependencies checked successfully")
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def get(self, snippet_id: str):
+        """Check dependencies for a 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:
+            import_service = SnippetDslService(session)
+            result = import_service.check_dependencies(snippet=snippet)
+
+        return result.model_dump(mode="json"), 200
+
+
+@console_ns.route("/workspaces/current/customized-snippets/<uuid:snippet_id>/use-count/increment")
+class CustomizedSnippetUseCountIncrementApi(Resource):
+    @console_ns.doc("increment_snippet_use_count")
+    @console_ns.doc(description="Increment snippet use count by 1")
+    @console_ns.doc(params={"snippet_id": "Snippet ID"})
+    @console_ns.response(200, "Use count incremented successfully")
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def post(self, snippet_id: str):
+        """Increment snippet use count when it is inserted into a workflow."""
+        _, 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.increment_use_count(session=session, snippet=snippet)
+            session.commit()
+            session.refresh(snippet)
+
+        return {"result": "success", "use_count": snippet.use_count}, 200

api/core/app/apps/workflow/app_generator.py

@@ -10,7 +10,7 @@ from typing import TYPE_CHECKING, Any, Literal, overload
 from flask import Flask, current_app
 from pydantic import ValidationError
 from sqlalchemy import select
-from sqlalchemy.orm import sessionmaker
+from sqlalchemy.orm import Session, sessionmaker
 
 import contexts
 from configs import dify_config
@@ -58,6 +58,25 @@ logger = logging.getLogger(__name__)
 
 
 class WorkflowAppGenerator(BaseAppGenerator):
+    @staticmethod
+    def _ensure_snippet_start_node_in_worker(*, session: Session, workflow: Workflow) -> Workflow:
+        """Re-apply snippet virtual Start injection after worker reloads workflow from DB."""
+        if workflow.kind_or_standard != "snippet":
+            return workflow
+
+        from models.snippet import CustomizedSnippet
+        from services.snippet_generate_service import SnippetGenerateService
+
+        snippet = session.scalar(
+            select(CustomizedSnippet).where(
+                CustomizedSnippet.id == workflow.app_id,
+                CustomizedSnippet.tenant_id == workflow.tenant_id,
+            )
+        )
+        if snippet is None:
+            return workflow
+        return SnippetGenerateService.ensure_start_node_for_worker(workflow, snippet)
+
     @staticmethod
     def _should_prepare_user_inputs(args: Mapping[str, Any]) -> bool:
         return not bool(args.get(SKIP_PREPARE_USER_INPUTS_KEY))
@@ -575,6 +594,8 @@ class WorkflowAppGenerator(BaseAppGenerator):
                 if workflow is None:
                     raise ValueError("Workflow not found")
 
+                workflow = self._ensure_snippet_start_node_in_worker(session=session, workflow=workflow)
+
                 # Determine system_user_id based on invocation source
                 is_external_api_call = application_generate_entity.invoke_from in {
                     InvokeFrom.WEB_APP,

api/core/app/apps/workflow/app_runner.py

@@ -10,6 +10,7 @@ from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerat
 from core.app.workflow.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
 from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
 from core.workflow.node_factory import get_default_root_node_id
+from core.workflow.snippet_start import get_compatible_start_aliases
 from core.workflow.system_variables import build_bootstrap_variables, build_system_variables
 from core.workflow.variable_pool_initializer import add_node_inputs_to_pool, add_variables_to_pool
 from core.workflow.workflow_entry import WorkflowEntry
@@ -115,7 +116,15 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
                 ),
             )
             root_node_id = self._root_node_id or get_default_root_node_id(self._workflow.graph_dict)
-            add_node_inputs_to_pool(variable_pool, node_id=root_node_id, inputs=inputs)
+            add_node_inputs_to_pool(
+                variable_pool,
+                node_id=root_node_id,
+                inputs=inputs,
+                aliases=get_compatible_start_aliases(
+                    workflow_kind=getattr(self._workflow, "kind_or_standard", None),
+                    root_node_id=root_node_id,
+                ),
+            )
 
             graph_runtime_state = GraphRuntimeState(variable_pool=variable_pool, start_at=time.perf_counter())
             graph = self._init_graph(

api/core/workflow/snippet_start.py

@@ -0,0 +1,21 @@
+"""Shared snippet virtual Start-node identifiers and compatibility helpers.
+
+Snippet workflows do not persist a real canvas Start node, so the backend
+injects one at runtime. Existing workflow references commonly use the public
+selector shape ``#start.<var>#``; keep that contract stable by treating the
+runtime-only snippet Start node as compatible with the legacy ``start`` id.
+"""
+
+from __future__ import annotations
+
+
+LEGACY_START_NODE_ID = "start"
+SNIPPET_VIRTUAL_START_NODE_ID = "__snippet_virtual_start__"
+
+
+def get_compatible_start_aliases(*, workflow_kind: str | None, root_node_id: str | None) -> tuple[str, ...]:
+    """Return additional selector ids that should mirror snippet Start inputs."""
+    if workflow_kind == "snippet" and root_node_id == SNIPPET_VIRTUAL_START_NODE_ID:
+        return (LEGACY_START_NODE_ID,)
+
+    return ()

api/core/workflow/variable_pool_initializer.py

@@ -10,6 +10,19 @@ def add_variables_to_pool(variable_pool: VariablePool, variables: Sequence[Varia
         variable_pool.add(variable.selector, variable)
 
 
-def add_node_inputs_to_pool(variable_pool: VariablePool, *, node_id: str, inputs: Mapping[str, Any]) -> None:
-    for key, value in inputs.items():
-        variable_pool.add((node_id, key), value)
+def add_node_inputs_to_pool(
+    variable_pool: VariablePool,
+    *,
+    node_id: str,
+    inputs: Mapping[str, Any],
+    aliases: Sequence[str] = (),
+) -> None:
+    """Store node inputs under the primary node id and any compatible aliases."""
+    node_ids: list[str] = [node_id]
+    for alias in aliases:
+        if alias not in node_ids:
+            node_ids.append(alias)
+
+    for current_node_id in node_ids:
+        for key, value in inputs.items():
+            variable_pool.add((current_node_id, key), value)

api/fields/snippet_fields.py

@@ -0,0 +1,51 @@
+from flask_restx import fields
+
+from fields.member_fields import simple_account_fields
+from libs.helper import TimestampField
+
+tag_fields = {"id": fields.String, "name": fields.String, "type": fields.String}
+
+# 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,
+    "tags": fields.List(fields.Nested(tag_fields)),
+    "created_by": fields.String,
+    "created_at": TimestampField,
+    "updated_by": fields.String,
+    "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"),
+    "tags": fields.List(fields.Nested(tag_fields)),
+    "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/models/__init__.py

@@ -98,6 +98,7 @@ from .provider import (
     TenantDefaultModel,
     TenantPreferredModelProvider,
 )
+from .snippet import CustomizedSnippet, SnippetType
 from .source import DataSourceApiKeyAuthBinding, DataSourceOauthBinding
 from .task import CeleryTask, CeleryTaskSet
 from .tools import (
@@ -123,12 +124,14 @@ from .workflow import (
     WorkflowAppLog,
     WorkflowAppLogCreatedFrom,
     WorkflowArchiveLog,
+    WorkflowKind,
     WorkflowNodeExecutionModel,
     WorkflowNodeExecutionOffload,
     WorkflowNodeExecutionTriggeredFrom,
     WorkflowPause,
     WorkflowRun,
     WorkflowType,
+    resolve_workflow_kind,
 )
 
 __all__ = [
@@ -166,6 +169,7 @@ __all__ = [
     "Conversation",
     "ConversationVariable",
     "CreatorUserRole",
+    "CustomizedSnippet",
     "DataSourceApiKeyAuthBinding",
     "DataSourceOauthBinding",
     "Dataset",
@@ -210,6 +214,7 @@ __all__ = [
     "RecommendedApp",
     "SavedMessage",
     "Site",
+    "SnippetType",
     "Tag",
     "TagBinding",
     "Tenant",
@@ -237,6 +242,7 @@ __all__ = [
     "WorkflowAppLog",
     "WorkflowAppLogCreatedFrom",
     "WorkflowArchiveLog",
+    "WorkflowKind",
     "WorkflowComment",
     "WorkflowCommentMention",
     "WorkflowCommentReply",
@@ -250,4 +256,5 @@ __all__ = [
     "WorkflowToolProvider",
     "WorkflowTriggerStatus",
     "WorkflowType",
+    "resolve_workflow_kind",
 ]

api/models/enums.py

@@ -222,6 +222,7 @@ class TagType(StrEnum):
 
     KNOWLEDGE = "knowledge"
     APP = "app"
+    SNIPPET = "snippet"
 
 
 class DatasetMetadataType(StrEnum):

api/models/model.py

@@ -2474,7 +2474,7 @@ class Tag(TypeBase):
         sa.Index("tag_name_idx", "name"),
     )
 
-    TAG_TYPE_LIST = ["knowledge", "app"]
+    TAG_TYPE_LIST = ["knowledge", "app", "snippet"]
 
     id: Mapped[str] = mapped_column(
         StringUUID, insert_default=lambda: str(uuid4()), default_factory=lambda: str(uuid4()), init=False

api/models/snippet.py

@@ -0,0 +1,117 @@
+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 .model import Tag, TagBinding
+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"),
+    )
+
+    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 tags(self):
+        """Get snippet tags."""
+        tags = db.session.scalars(
+            sa.select(Tag)
+            .join(TagBinding, Tag.id == TagBinding.tag_id)
+            .where(
+                TagBinding.target_id == self.id,
+                TagBinding.tenant_id == self.tenant_id,
+                Tag.tenant_id == self.tenant_id,
+                Tag.type == "snippet",
+            )
+        ).all()
+
+        return tags or []
+
+    @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

@@ -112,6 +112,7 @@ class WorkflowType(StrEnum):
     WORKFLOW = "workflow"
     CHAT = "chat"
     RAG_PIPELINE = "rag-pipeline"
+    SNIPPET = "snippet"
 
     @classmethod
     def value_of(cls, value: str) -> "WorkflowType":
@@ -140,6 +141,26 @@ class WorkflowType(StrEnum):
         return cls.WORKFLOW if app_mode == AppMode.WORKFLOW else cls.CHAT
 
 
+class WorkflowKind(StrEnum):
+    STANDARD = "standard"
+    SNIPPET = "snippet"
+
+    @classmethod
+    def value_of(cls, value: str) -> "WorkflowKind":
+        for kind in cls:
+            if kind.value == value:
+                return kind
+        raise ValueError(f"invalid workflow kind value {value}")
+
+
+def resolve_workflow_kind(kind: str | WorkflowKind | None) -> WorkflowKind:
+    if kind is None:
+        return WorkflowKind.STANDARD
+    if isinstance(kind, WorkflowKind):
+        return kind
+    return WorkflowKind.value_of(kind)
+
+
 class _InvalidGraphDefinitionError(Exception):
     pass
 
@@ -187,6 +208,12 @@ class Workflow(Base):  # bug
     tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
     app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
     type: Mapped[WorkflowType] = mapped_column(EnumText(WorkflowType, length=255), nullable=False)
+    kind: Mapped[WorkflowKind | None] = mapped_column(
+        EnumText(WorkflowKind, length=255),
+        nullable=True,
+        default=WorkflowKind.STANDARD,
+        server_default=sa.text("'standard'"),
+    )
     version: Mapped[str] = mapped_column(String(255), nullable=False)
     marked_name: Mapped[str] = mapped_column(String(255), default="", server_default="")
     marked_comment: Mapped[str] = mapped_column(String(255), default="", server_default="")
@@ -228,12 +255,14 @@ class Workflow(Base):  # bug
         rag_pipeline_variables: list[dict],
         marked_name: str = "",
         marked_comment: str = "",
+        kind: str | None = WorkflowKind.STANDARD.value,
     ) -> "Workflow":
         workflow = Workflow()
         workflow.id = str(uuid4())
         workflow.tenant_id = tenant_id
         workflow.app_id = app_id
         workflow.type = WorkflowType(type)
+        workflow.kind = resolve_workflow_kind(kind)
         workflow.version = version
         workflow.graph = graph
         workflow.features = features
@@ -255,6 +284,14 @@ class Workflow(Base):  # bug
     def updated_by_account(self):
         return db.session.get(Account, self.updated_by) if self.updated_by else None
 
+    @property
+    def kind_or_standard(self) -> str:
+        return self.resolved_kind.value
+
+    @property
+    def resolved_kind(self) -> WorkflowKind:
+        return resolve_workflow_kind(self.kind)
+
     @property
     def graph_dict(self) -> Mapping[str, Any]:
         # TODO(QuantumGhost): Consider caching `graph_dict` to avoid repeated JSON decoding.

api/services/snippet_dsl_service.py

@@ -0,0 +1,584 @@
+import json
+import logging
+import uuid
+from collections.abc import Mapping
+from datetime import UTC, datetime
+from enum import StrEnum
+from urllib.parse import urlparse
+
+import yaml  # type: ignore
+from packaging import version
+from pydantic import BaseModel, Field
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from core.helper import ssrf_proxy
+from core.plugin.entities.plugin import PluginDependency
+from extensions.ext_redis import redis_client
+from graphon.enums import BuiltinNodeTypes
+from graphon.model_runtime.utils.encoders import jsonable_encoder
+from models import Account
+from models.snippet import CustomizedSnippet, SnippetType
+from models.workflow import Workflow
+from services.plugin.dependencies_analysis import DependenciesAnalysisService
+from services.snippet_service import SNIPPET_FORBIDDEN_NODE_TYPES, SnippetService
+
+logger = logging.getLogger(__name__)
+
+IMPORT_INFO_REDIS_KEY_PREFIX = "snippet_import_info:"
+CHECK_DEPENDENCIES_REDIS_KEY_PREFIX = "snippet_check_dependencies:"
+IMPORT_INFO_REDIS_EXPIRY = 10 * 60  # 10 minutes
+DSL_MAX_SIZE = 10 * 1024 * 1024  # 10MB
+CURRENT_DSL_VERSION = "0.1.0"
+
+
+class ImportMode(StrEnum):
+    YAML_CONTENT = "yaml-content"
+    YAML_URL = "yaml-url"
+
+
+class ImportStatus(StrEnum):
+    COMPLETED = "completed"
+    COMPLETED_WITH_WARNINGS = "completed-with-warnings"
+    PENDING = "pending"
+    FAILED = "failed"
+
+
+class SnippetImportInfo(BaseModel):
+    id: str
+    status: ImportStatus
+    snippet_id: str | None = None
+    current_dsl_version: str = CURRENT_DSL_VERSION
+    imported_dsl_version: str = ""
+    error: str = ""
+
+
+class CheckDependenciesResult(BaseModel):
+    leaked_dependencies: list[PluginDependency] = Field(default_factory=list)
+
+
+def _check_version_compatibility(imported_version: str) -> ImportStatus:
+    """Determine import status based on version comparison"""
+    try:
+        current_ver = version.parse(CURRENT_DSL_VERSION)
+        imported_ver = version.parse(imported_version)
+    except version.InvalidVersion:
+        return ImportStatus.FAILED
+
+    # If imported version is newer than current, always return PENDING
+    if imported_ver > current_ver:
+        return ImportStatus.PENDING
+
+    # If imported version is older than current's major, return PENDING
+    if imported_ver.major < current_ver.major:
+        return ImportStatus.PENDING
+
+    # If imported version is older than current's minor, return COMPLETED_WITH_WARNINGS
+    if imported_ver.minor < current_ver.minor:
+        return ImportStatus.COMPLETED_WITH_WARNINGS
+
+    # If imported version equals or is older than current's micro, return COMPLETED
+    return ImportStatus.COMPLETED
+
+
+class SnippetPendingData(BaseModel):
+    import_mode: str
+    yaml_content: str
+    name: str | None = None
+    description: str | None = None
+    snippet_id: str | None
+
+
+class CheckDependenciesPendingData(BaseModel):
+    dependencies: list[PluginDependency]
+    snippet_id: str | None
+
+
+class SnippetDslService:
+    def __init__(self, session: Session):
+        self._session = session
+
+    def import_snippet(
+        self,
+        *,
+        account: Account,
+        import_mode: str,
+        yaml_content: str | None = None,
+        yaml_url: str | None = None,
+        snippet_id: str | None = None,
+        name: str | None = None,
+        description: str | None = None,
+    ) -> SnippetImportInfo:
+        """Import a snippet from YAML content or URL."""
+        import_id = str(uuid.uuid4())
+
+        # Validate import mode
+        try:
+            mode = ImportMode(import_mode)
+        except ValueError:
+            raise ValueError(f"Invalid import_mode: {import_mode}")
+
+        # Get YAML content
+        content: str = ""
+        if mode == ImportMode.YAML_URL:
+            if not yaml_url:
+                return SnippetImportInfo(
+                    id=import_id,
+                    status=ImportStatus.FAILED,
+                    error="yaml_url is required when import_mode is yaml-url",
+                )
+            try:
+                parsed_url = urlparse(yaml_url)
+                if parsed_url.scheme not in ["http", "https"]:
+                    return SnippetImportInfo(
+                        id=import_id,
+                        status=ImportStatus.FAILED,
+                        error="Invalid URL scheme, only http and https are allowed",
+                    )
+                response = ssrf_proxy.get(yaml_url, timeout=(10, 30))
+                if response.status_code != 200:
+                    return SnippetImportInfo(
+                        id=import_id,
+                        status=ImportStatus.FAILED,
+                        error=f"Failed to fetch YAML from URL: {response.status_code}",
+                    )
+                content = response.text
+                if len(content) > DSL_MAX_SIZE:
+                    return SnippetImportInfo(
+                        id=import_id,
+                        status=ImportStatus.FAILED,
+                        error=f"YAML content size exceeds maximum limit of {DSL_MAX_SIZE} bytes",
+                    )
+            except Exception as e:
+                logger.exception("Failed to fetch YAML from URL")
+                return SnippetImportInfo(
+                    id=import_id,
+                    status=ImportStatus.FAILED,
+                    error=f"Failed to fetch YAML from URL: {str(e)}",
+                )
+        elif mode == ImportMode.YAML_CONTENT:
+            if not yaml_content:
+                return SnippetImportInfo(
+                    id=import_id,
+                    status=ImportStatus.FAILED,
+                    error="yaml_content is required when import_mode is yaml-content",
+                )
+            content = yaml_content
+            if len(content) > DSL_MAX_SIZE:
+                return SnippetImportInfo(
+                    id=import_id,
+                    status=ImportStatus.FAILED,
+                    error=f"YAML content size exceeds maximum limit of {DSL_MAX_SIZE} bytes",
+                )
+
+        try:
+            # Parse YAML
+            data = yaml.safe_load(content)
+            if not isinstance(data, dict):
+                return SnippetImportInfo(
+                    id=import_id,
+                    status=ImportStatus.FAILED,
+                    error="Invalid YAML format: expected a dictionary",
+                )
+
+            # Validate and fix DSL version
+            if not data.get("version"):
+                data["version"] = "0.1.0"
+
+            # Strictly validate kind field
+            kind = data.get("kind")
+            if not kind:
+                return SnippetImportInfo(
+                    id=import_id,
+                    status=ImportStatus.FAILED,
+                    error="Missing 'kind' field in DSL. Expected 'kind: snippet'.",
+                )
+            if kind != "snippet":
+                return SnippetImportInfo(
+                    id=import_id,
+                    status=ImportStatus.FAILED,
+                    error=f"Invalid DSL kind: expected 'snippet', got '{kind}'. This DSL is for {kind}, not snippet.",
+                )
+
+            imported_version = data.get("version", "0.1.0")
+            if not isinstance(imported_version, str):
+                raise ValueError(f"Invalid version type, expected str, got {type(imported_version)}")
+            status = _check_version_compatibility(imported_version)
+
+            # Extract snippet data
+            snippet_data = data.get("snippet")
+            if not snippet_data:
+                return SnippetImportInfo(
+                    id=import_id,
+                    status=ImportStatus.FAILED,
+                    error="Missing snippet data in YAML content",
+                )
+
+            # Validate workflow nodes - check for forbidden node types
+            workflow_data = data.get("workflow", {})
+            if workflow_data:
+                graph = workflow_data.get("graph", {})
+                nodes = graph.get("nodes", [])
+                forbidden_nodes_found = []
+                for node in nodes:
+                    node_data = node.get("data", {})
+                    if not node_data:
+                        continue
+                    node_type = node_data.get("type", "")
+                    if node_type in SNIPPET_FORBIDDEN_NODE_TYPES:
+                        forbidden_nodes_found.append(node_type)
+                
+                if forbidden_nodes_found:
+                    forbidden_types_str = ", ".join(set(forbidden_nodes_found))
+                    return SnippetImportInfo(
+                        id=import_id,
+                        status=ImportStatus.FAILED,
+                        error=f"Snippet cannot contain the following node types: {forbidden_types_str}",
+                    )
+
+            # If snippet_id is provided, check if it exists
+            snippet = None
+            if snippet_id:
+                stmt = select(CustomizedSnippet).where(
+                    CustomizedSnippet.id == snippet_id,
+                    CustomizedSnippet.tenant_id == account.current_tenant_id,
+                )
+                snippet = self._session.scalar(stmt)
+
+                if not snippet:
+                    return SnippetImportInfo(
+                        id=import_id,
+                        status=ImportStatus.FAILED,
+                        error="Snippet not found",
+                    )
+
+            # If major version mismatch, store import info in Redis
+            if status == ImportStatus.PENDING:
+                pending_data = SnippetPendingData(
+                    import_mode=import_mode,
+                    yaml_content=content,
+                    name=name,
+                    description=description,
+                    snippet_id=snippet_id,
+                )
+                redis_client.setex(
+                    f"{IMPORT_INFO_REDIS_KEY_PREFIX}{import_id}",
+                    IMPORT_INFO_REDIS_EXPIRY,
+                    pending_data.model_dump_json(),
+                )
+
+                return SnippetImportInfo(
+                    id=import_id,
+                    status=status,
+                    snippet_id=snippet_id,
+                    imported_dsl_version=imported_version,
+                )
+
+            # Extract dependencies
+            dependencies = data.get("dependencies", [])
+            check_dependencies_pending_data = None
+            if dependencies:
+                check_dependencies_pending_data = [PluginDependency.model_validate(d) for d in dependencies]
+
+            # Create or update snippet
+            snippet = self._create_or_update_snippet(
+                snippet=snippet,
+                data=data,
+                account=account,
+                name=name,
+                description=description,
+                dependencies=check_dependencies_pending_data,
+            )
+
+            return SnippetImportInfo(
+                id=import_id,
+                status=status,
+                snippet_id=snippet.id,
+                imported_dsl_version=imported_version,
+            )
+
+        except yaml.YAMLError as e:
+            return SnippetImportInfo(
+                id=import_id,
+                status=ImportStatus.FAILED,
+                error=f"Invalid YAML format: {str(e)}",
+            )
+
+        except Exception as e:
+            logger.exception("Failed to import snippet")
+            return SnippetImportInfo(
+                id=import_id,
+                status=ImportStatus.FAILED,
+                error=str(e),
+            )
+
+    def confirm_import(self, *, import_id: str, account: Account) -> SnippetImportInfo:
+        """
+        Confirm an import that requires confirmation
+        """
+        redis_key = f"{IMPORT_INFO_REDIS_KEY_PREFIX}{import_id}"
+        pending_data = redis_client.get(redis_key)
+
+        if not pending_data:
+            return SnippetImportInfo(
+                id=import_id,
+                status=ImportStatus.FAILED,
+                error="Import information expired or does not exist",
+            )
+
+        try:
+            if not isinstance(pending_data, str | bytes):
+                return SnippetImportInfo(
+                    id=import_id,
+                    status=ImportStatus.FAILED,
+                    error="Invalid import information",
+                )
+
+            pending_data_str = pending_data.decode("utf-8") if isinstance(pending_data, bytes) else pending_data
+            pending = SnippetPendingData.model_validate_json(pending_data_str)
+
+            data = yaml.safe_load(pending.yaml_content)
+            if not isinstance(data, dict):
+                return SnippetImportInfo(
+                    id=import_id,
+                    status=ImportStatus.FAILED,
+                    error="Invalid YAML format: expected a dictionary",
+                )
+
+            snippet = None
+            if pending.snippet_id:
+                stmt = select(CustomizedSnippet).where(
+                    CustomizedSnippet.id == pending.snippet_id,
+                    CustomizedSnippet.tenant_id == account.current_tenant_id,
+                )
+                snippet = self._session.scalar(stmt)
+
+            snippet = self._create_or_update_snippet(
+                snippet=snippet,
+                data=data,
+                account=account,
+                name=pending.name,
+                description=pending.description,
+            )
+
+            redis_client.delete(redis_key)
+
+            return SnippetImportInfo(
+                id=import_id,
+                status=ImportStatus.COMPLETED,
+                snippet_id=snippet.id,
+                imported_dsl_version=data.get("version", "0.1.0"),
+            )
+
+        except Exception as e:
+            logger.exception("Failed to confirm import")
+            return SnippetImportInfo(
+                id=import_id,
+                status=ImportStatus.FAILED,
+                error=str(e),
+            )
+
+    def check_dependencies(self, snippet: CustomizedSnippet) -> CheckDependenciesResult:
+        """
+        Check dependencies for a snippet
+        """
+        snippet_service = SnippetService()
+        workflow = snippet_service.get_draft_workflow(snippet=snippet)
+        if not workflow:
+            return CheckDependenciesResult(leaked_dependencies=[])
+
+        dependencies = self._extract_dependencies_from_workflow(workflow)
+        leaked_dependencies = DependenciesAnalysisService.generate_dependencies(
+            tenant_id=snippet.tenant_id, dependencies=dependencies
+        )
+
+        return CheckDependenciesResult(leaked_dependencies=leaked_dependencies)
+
+    def _create_or_update_snippet(
+        self,
+        *,
+        snippet: CustomizedSnippet | None,
+        data: dict,
+        account: Account,
+        name: str | None = None,
+        description: str | None = None,
+        dependencies: list[PluginDependency] | None = None,
+    ) -> CustomizedSnippet:
+        """
+        Create or update snippet from DSL data
+        """
+        snippet_data = data.get("snippet", {})
+        workflow_data = data.get("workflow", {})
+
+        # Extract snippet info
+        snippet_name = name or snippet_data.get("name") or "Untitled Snippet"
+        snippet_description = description or snippet_data.get("description") or ""
+        snippet_type_str = snippet_data.get("type", "node")
+        try:
+            snippet_type = SnippetType(snippet_type_str)
+        except ValueError:
+            snippet_type = SnippetType.NODE
+
+        icon_info = snippet_data.get("icon_info", {})
+        input_fields = snippet_data.get("input_fields", [])
+
+        # Create or update snippet
+        if snippet:
+            # Update existing snippet
+            snippet.name = snippet_name
+            snippet.description = snippet_description
+            snippet.type = snippet_type.value
+            snippet.icon_info = icon_info or None
+            snippet.input_fields = json.dumps(input_fields) if input_fields else None
+            snippet.updated_by = account.id
+            snippet.updated_at = datetime.now(UTC).replace(tzinfo=None)
+        else:
+            # Create new snippet
+            snippet = CustomizedSnippet(
+                tenant_id=account.current_tenant_id,
+                name=snippet_name,
+                description=snippet_description,
+                type=snippet_type.value,
+                icon_info=icon_info or None,
+                input_fields=json.dumps(input_fields) if input_fields else None,
+                created_by=account.id,
+            )
+            self._session.add(snippet)
+            self._session.flush()
+
+        # Create or update draft workflow
+        if workflow_data:
+            graph = workflow_data.get("graph", {})
+
+            snippet_service = SnippetService()
+            # Get existing workflow hash if exists
+            existing_workflow = snippet_service.get_draft_workflow(snippet=snippet)
+            unique_hash = existing_workflow.unique_hash if existing_workflow else None
+
+            snippet_service.sync_draft_workflow(
+                snippet=snippet,
+                graph=graph,
+                unique_hash=unique_hash,
+                account=account,
+                input_fields=input_fields,
+            )
+
+        self._session.commit()
+        return snippet
+
+    def export_snippet_dsl(self, snippet: CustomizedSnippet, include_secret: bool = False) -> str:
+        """
+        Export snippet as DSL
+        :param snippet: CustomizedSnippet instance
+        :param include_secret: Whether include secret variable
+        :return: YAML string
+        """
+        snippet_service = SnippetService()
+        workflow = snippet_service.get_draft_workflow(snippet=snippet)
+        if not workflow:
+            raise ValueError("Missing draft workflow configuration, please check.")
+
+        icon_info = snippet.icon_info or {}
+        export_data = {
+            "version": CURRENT_DSL_VERSION,
+            "kind": "snippet",
+            "snippet": {
+                "name": snippet.name,
+                "description": snippet.description or "",
+                "type": snippet.type,
+                "icon_info": icon_info,
+                "input_fields": snippet.input_fields_list,
+            },
+        }
+
+        self._append_workflow_export_data(
+            export_data=export_data, snippet=snippet, workflow=workflow, include_secret=include_secret
+        )
+
+        return yaml.dump(export_data, allow_unicode=True)  # type: ignore
+
+    def _append_workflow_export_data(
+        self, *, export_data: dict, snippet: CustomizedSnippet, workflow: Workflow, include_secret: bool
+    ) -> None:
+        """
+        Append workflow export data
+        """
+        workflow_dict = workflow.to_dict(include_secret=include_secret)
+        # Filter workspace related data from nodes
+        workflow_dict["environment_variables"] = []
+        workflow_dict["conversation_variables"] = []
+
+        for node in workflow_dict.get("graph", {}).get("nodes", []):
+            node_data = node.get("data", {})
+            if not node_data:
+                continue
+            data_type = node_data.get("type", "")
+            if data_type == BuiltinNodeTypes.KNOWLEDGE_RETRIEVAL:
+                dataset_ids = node_data.get("dataset_ids", [])
+                node["data"]["dataset_ids"] = [
+                    self._encrypt_dataset_id(dataset_id=dataset_id, tenant_id=snippet.tenant_id)
+                    for dataset_id in dataset_ids
+                ]
+            # filter credential id from tool node
+            if not include_secret and data_type == BuiltinNodeTypes.TOOL:
+                node_data.pop("credential_id", None)
+            # filter credential id from agent node
+            if not include_secret and data_type == BuiltinNodeTypes.AGENT:
+                for tool in node_data.get("agent_parameters", {}).get("tools", {}).get("value", []):
+                    tool.pop("credential_id", None)
+
+        export_data["workflow"] = workflow_dict
+        dependencies = self._extract_dependencies_from_workflow(workflow)
+        export_data["dependencies"] = [
+            jsonable_encoder(d.model_dump())
+            for d in DependenciesAnalysisService.generate_dependencies(
+                tenant_id=snippet.tenant_id, dependencies=dependencies
+            )
+        ]
+
+    def _encrypt_dataset_id(self, *, dataset_id: str, tenant_id: str) -> str:
+        """
+        Encrypt dataset ID for export
+        """
+        # For now, just return the dataset_id as-is
+        # In the future, we might want to encrypt it
+        return dataset_id
+
+    def _extract_dependencies_from_workflow(self, workflow: Workflow) -> list[str]:
+        """
+        Extract dependencies from workflow
+        :param workflow: Workflow instance
+        :return: dependencies list format like ["langgenius/google"]
+        """
+        graph = workflow.graph_dict
+        dependencies = self._extract_dependencies_from_workflow_graph(graph)
+        return dependencies
+
+    def _extract_dependencies_from_workflow_graph(self, graph: Mapping) -> list[str]:
+        """
+        Extract dependencies from workflow graph
+        :param graph: Workflow graph
+        :return: dependencies list format like ["langgenius/google"]
+        """
+        dependencies = []
+        for node in graph.get("nodes", []):
+            node_data = node.get("data", {})
+            if not node_data:
+                continue
+            data_type = node_data.get("type", "")
+            if data_type == BuiltinNodeTypes.TOOL:
+                tool_config = node_data.get("tool_configurations", {})
+                provider_type = tool_config.get("provider_type")
+                provider_name = tool_config.get("provider")
+                if provider_type and provider_name:
+                    dependencies.append(f"{provider_name}/{provider_name}")
+            elif data_type == BuiltinNodeTypes.AGENT:
+                agent_parameters = node_data.get("agent_parameters", {})
+                tools = agent_parameters.get("tools", {}).get("value", [])
+                for tool in tools:
+                    provider_type = tool.get("provider_type")
+                    provider_name = tool.get("provider")
+                    if provider_type and provider_name:
+                        dependencies.append(f"{provider_name}/{provider_name}")
+
+        return dependencies

api/services/snippet_generate_service.py

@@ -0,0 +1,468 @@
+"""
+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.workflow.snippet_start import SNIPPET_VIRTUAL_START_NODE_ID
+from factories import file_factory
+from graphon.file.models import File
+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_NODE_ID
+
+    @classmethod
+    def _is_virtual_start_event(cls, message: Mapping[str, Any] | str) -> bool:
+        """
+        Return True when *message* is a snippet-only virtual Start node event.
+
+        The virtual Start node is injected purely for snippet execution and is
+        not part of the persisted draft graph. Filter its node lifecycle events
+        out of the SSE stream so the frontend only receives nodes that exist on
+        the canvas.
+        """
+        if not isinstance(message, Mapping):
+            return False
+
+        if message.get("event") not in {"node_started", "node_finished"}:
+            return False
+
+        data = message.get("data")
+        if not isinstance(data, Mapping):
+            return False
+
+        return data.get("node_id") == cls._VIRTUAL_START_NODE_ID
+
+    @classmethod
+    def _filter_virtual_start_events(
+        cls,
+        response: Mapping[str, Any] | Generator[Mapping[str, Any] | str, None, None],
+    ) -> Mapping[str, Any] | Generator[Mapping[str, Any] | str, None, None]:
+        """
+        Drop snippet virtual Start node lifecycle events from stream responses.
+
+        Blocking responses are returned unchanged because they never expose the
+        injected node as a standalone event payload.
+        """
+        if isinstance(response, Mapping):
+            return response
+
+        def _stream() -> Generator[Mapping[str, Any] | str, None, None]:
+            for message in response:
+                if cls._is_virtual_start_event(message):
+                    continue
+                yield message
+
+        return _stream()
+
+    @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)
+
+        response = 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,
+        )
+
+        return WorkflowAppGenerator.convert_to_event_stream(
+            cls._filter_virtual_start_events(response)
+        )
+
+    @classmethod
+    def run_published(
+        cls,
+        snippet: CustomizedSnippet,
+        user: Union[Account, EndUser],
+        args: Mapping[str, Any],
+        invoke_from: InvokeFrom,
+    ) -> Mapping[str, Any]:
+        """
+        Run a snippet's published workflow in non-streaming (blocking) mode.
+
+        Similar to :meth:`generate` but targets the published workflow instead
+        of the draft, and returns the raw blocking response without SSE
+        wrapping. Designed for programmatic callers that need direct workflow outputs.
+
+        :param snippet: CustomizedSnippet instance (must be published)
+        :param user: Account or EndUser initiating the run
+        :param args: Workflow inputs (must include "inputs" key)
+        :param invoke_from: Source of invocation
+        :return: Blocking response mapping with workflow outputs
+        :raises ValueError: If the snippet has no published workflow
+        """
+        snippet_service = SnippetService()
+        workflow = snippet_service.get_published_workflow(snippet)
+        if not workflow:
+            raise ValueError("No published workflow found for snippet")
+
+        # Inject a virtual Start node when the graph doesn't have one.
+        workflow = cls._ensure_start_node(workflow, snippet)
+
+        app_proxy = _SnippetAsApp(snippet)
+
+        response: Mapping[str, Any] = WorkflowAppGenerator().generate(
+            app_model=app_proxy,  # type: ignore[arg-type]
+            workflow=workflow,
+            user=user,
+            args=args,
+            invoke_from=invoke_from,
+            streaming=False,
+        )
+        return response
+
+    @classmethod
+    def ensure_start_node_for_worker(cls, workflow: Workflow, snippet: CustomizedSnippet) -> Workflow:
+        """Public wrapper for worker-thread start-node injection."""
+        return cls._ensure_start_node(workflow, snippet)
+
+    @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,675 @@
+import json
+import logging
+from collections.abc import Mapping, Sequence
+from datetime import UTC, datetime
+from typing import Any
+
+from sqlalchemy import delete, func, select
+from sqlalchemy.orm import Session, sessionmaker
+
+from core.workflow.node_factory import LATEST_VERSION, NODE_TYPE_CLASSES_MAPPING
+from extensions.ext_database import db
+from graphon.enums import BuiltinNodeTypes, NodeType
+from libs.infinite_scroll_pagination import InfiniteScrollPagination
+from models import Account, TagBinding
+from models.enums import WorkflowRunTriggeredFrom
+from models.snippet import CustomizedSnippet, SnippetType
+from models.workflow import (
+    Workflow,
+    WorkflowKind,
+    WorkflowNodeExecutionModel,
+    WorkflowRun,
+    WorkflowType,
+)
+from repositories.factory import DifyAPIRepositoryFactory
+from services.errors.app import IsDraftWorkflowError, WorkflowHashNotEqualError, WorkflowNotFoundError
+from services.tag_service import TagService
+from services.workflow_restore import apply_published_workflow_snapshot_to_draft
+
+logger = logging.getLogger(__name__)
+
+# Node types not allowed in snippet workflows (sync, publish, DSL import).
+SNIPPET_FORBIDDEN_NODE_TYPES: frozenset[str] = frozenset(
+    {
+        BuiltinNodeTypes.START,
+        BuiltinNodeTypes.HUMAN_INPUT,
+        BuiltinNodeTypes.KNOWLEDGE_RETRIEVAL,
+    }
+)
+
+
+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)
+
+    @staticmethod
+    def _snippet_kind_filter():
+        """Match snippet workflows by business kind."""
+        return Workflow.kind == WorkflowKind.SNIPPET.value
+
+    @staticmethod
+    def validate_snippet_graph_forbidden_nodes(graph: Mapping[str, Any]) -> None:
+        """Reject graphs that contain node types not allowed in snippets."""
+        nodes = graph.get("nodes") or []
+        disallowed: list[tuple[str, str]] = []
+        for node in nodes:
+            if not isinstance(node, dict):
+                continue
+            node_data = node.get("data") or {}
+            node_type = node_data.get("type")
+            if not isinstance(node_type, str):
+                continue
+            if node_type in SNIPPET_FORBIDDEN_NODE_TYPES:
+                node_id = node.get("id")
+                disallowed.append((str(node_id) if node_id is not None else "?", node_type))
+        if not disallowed:
+            return
+        detail = ", ".join(f"{nid}:{t}" for nid, t in disallowed)
+        raise ValueError(
+            "Snippet workflow cannot contain start, human-input, or knowledge-retrieval nodes. "
+            f"Found: {detail}"
+        )
+
+    # --- CRUD Operations ---
+
+    @staticmethod
+    def get_snippets(
+        *,
+        tenant_id: str,
+        page: int = 1,
+        limit: int = 20,
+        keyword: str | None = None,
+        is_published: bool | None = None,
+        creators: list[str] | None = None,
+        tag_ids: list[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
+        :param is_published: Optional filter by published status (True/False/None for all)
+        :param creators: Optional filter by creator account IDs
+        :param tag_ids: Optional filter by tag IDs
+        :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}%")
+            )
+
+        if is_published is not None:
+            stmt = stmt.where(CustomizedSnippet.is_published == is_published)
+
+        if creators:
+            stmt = stmt.where(CustomizedSnippet.created_by.in_(creators))
+
+        if tag_ids:
+            target_ids = TagService.get_target_ids_by_tag_ids("snippet", tenant_id, tag_ids)
+            if target_ids:
+                stmt = stmt.where(CustomizedSnippet.id.in_(target_ids))
+            else:
+                return [], 0, False
+
+        # 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,
+        input_fields: list[dict] | None,
+        account: Account,
+    ) -> CustomizedSnippet:
+        """
+        Create a new snippet.
+
+        :param tenant_id: Tenant ID
+        :param name: Snippet name
+        :param description: Snippet description
+        :param snippet_type: Type of snippet (node or group)
+        :param icon_info: Icon information
+        :param input_fields: Input field definitions
+        :param account: Creator account
+        :return: Created CustomizedSnippet
+        """
+        snippet = CustomizedSnippet(
+            tenant_id=tenant_id,
+            name=name,
+            description=description or "",
+            type=snippet_type.value,
+            icon_info=icon_info,
+            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:
+            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.execute(
+            delete(TagBinding).where(
+                TagBinding.tenant_id == snippet.tenant_id,
+                TagBinding.target_id == snippet.id,
+            )
+        )
+        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,
+                self._snippet_kind_filter(),
+                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,
+                self._snippet_kind_filter(),
+                Workflow.id == snippet.workflow_id,
+            )
+            .first()
+        )
+        return workflow
+
+    def get_published_workflow_by_id(self, snippet: CustomizedSnippet, workflow_id: str) -> Workflow | None:
+        """
+        Get a published workflow snapshot by ID for snippet history restore.
+
+        :param snippet: CustomizedSnippet instance
+        :param workflow_id: Workflow ID
+        :return: Published Workflow or None
+        :raises IsDraftWorkflowError: If the workflow ID points to a draft workflow
+        """
+        workflow = (
+            db.session.query(Workflow)
+            .where(
+                Workflow.tenant_id == snippet.tenant_id,
+                Workflow.app_id == snippet.id,
+                self._snippet_kind_filter(),
+                Workflow.id == workflow_id,
+            )
+            .first()
+        )
+        if not workflow:
+            return None
+        if workflow.version == Workflow.VERSION_DRAFT:
+            raise IsDraftWorkflowError("source workflow must be published")
+        return workflow
+
+    def sync_draft_workflow(
+        self,
+        *,
+        snippet: CustomizedSnippet,
+        graph: dict,
+        unique_hash: str | None,
+        account: Account,
+        input_fields: list[dict] | None = None,
+    ) -> Workflow:
+        """
+        Sync draft workflow for snippet.
+
+        Snippet workflows do not persist environment variables (always empty) or
+        conversation variables (always empty).
+
+        :param snippet: CustomizedSnippet instance
+        :param graph: Workflow graph configuration
+        :param unique_hash: Hash for conflict detection
+        :param account: Account making the change
+        :param input_fields: Input fields for snippet
+        :return: Synced Workflow
+        :raises WorkflowHashNotEqualError: If hash mismatch
+        """
+        SnippetService.validate_snippet_graph_forbidden_nodes(graph)
+
+        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.WORKFLOW.value,
+                kind=WorkflowKind.SNIPPET.value,
+                version="draft",
+                graph=json.dumps(graph),
+                created_by=account.id,
+                environment_variables=[],
+                conversation_variables=[],
+            )
+            db.session.add(workflow)
+            db.session.flush()
+        else:
+            # Update existing draft workflow
+            workflow.graph = json.dumps(graph)
+            workflow.type = WorkflowType.WORKFLOW.value
+            workflow.kind = WorkflowKind.SNIPPET
+            workflow.updated_by = account.id
+            workflow.updated_at = datetime.now(UTC).replace(tzinfo=None)
+            workflow.environment_variables = []
+            workflow.conversation_variables = []
+
+        # Update snippet's input_fields if provided
+        if input_fields is not None:
+            snippet.input_fields = json.dumps(input_fields)
+            snippet.updated_by = account.id
+            snippet.updated_at = datetime.now(UTC).replace(tzinfo=None)
+
+        db.session.commit()
+        return workflow
+
+    def restore_published_workflow_to_draft(
+        self,
+        *,
+        snippet: CustomizedSnippet,
+        workflow_id: str,
+        account: Account,
+    ) -> Workflow:
+        """
+        Restore a published snippet workflow snapshot into the draft workflow.
+
+        :param snippet: CustomizedSnippet instance
+        :param workflow_id: Published workflow ID
+        :param account: Account making the change
+        :return: Restored draft Workflow
+        :raises WorkflowNotFoundError: If the source workflow does not exist
+        :raises IsDraftWorkflowError: If the source workflow is a draft
+        :raises ValueError: If the restored graph is invalid for snippets
+        """
+        source_workflow = self.get_published_workflow_by_id(snippet=snippet, workflow_id=workflow_id)
+        if not source_workflow:
+            raise WorkflowNotFoundError("Workflow not found.")
+
+        SnippetService.validate_snippet_graph_forbidden_nodes(source_workflow.graph_dict)
+
+        draft_workflow = self.get_draft_workflow(snippet=snippet)
+        draft_workflow, is_new_draft = apply_published_workflow_snapshot_to_draft(
+            tenant_id=snippet.tenant_id,
+            app_id=snippet.id,
+            source_workflow=source_workflow,
+            draft_workflow=draft_workflow,
+            account=account,
+            updated_at_factory=lambda: datetime.now(UTC).replace(tzinfo=None),
+        )
+
+        if is_new_draft:
+            db.session.add(draft_workflow)
+
+        db.session.commit()
+        return draft_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,
+            self._snippet_kind_filter(),
+            Workflow.version == "draft",
+        )
+        draft_workflow = session.scalar(draft_workflow_stmt)
+        if not draft_workflow:
+            raise ValueError("No valid workflow found.")
+
+        SnippetService.validate_snippet_graph_forbidden_nodes(draft_workflow.graph_dict)
+
+        # Create new published workflow
+        workflow = Workflow.new(
+            tenant_id=snippet.tenant_id,
+            app_id=snippet.id,
+            type=WorkflowType.WORKFLOW.value,
+            version=str(datetime.now(UTC).replace(tzinfo=None)),
+            graph=draft_workflow.graph,
+            features=draft_workflow.features,
+            created_by=account.id,
+            environment_variables=[],
+            conversation_variables=[],
+            rag_pipeline_variables=draft_workflow.rag_pipeline_variables,
+            kind=WorkflowKind.SNIPPET.value,
+            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,
+                self._snippet_kind_filter(),
+                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)

api/services/tag_service.py

@@ -12,6 +12,7 @@ from extensions.ext_database import db
 from models.dataset import Dataset
 from models.enums import TagType
 from models.model import App, Tag, TagBinding
+from models.snippet import CustomizedSnippet
 
 
 class SaveTagPayload(BaseModel):
@@ -159,7 +160,14 @@ class TagService:
     @staticmethod
     def save_tag_binding(payload: TagBindingCreatePayload):
         TagService.check_target_exists(payload.type, payload.target_id)
-        for tag_id in payload.tag_ids:
+        valid_tag_ids = db.session.scalars(
+            select(Tag.id).where(
+                Tag.id.in_(payload.tag_ids),
+                Tag.tenant_id == current_user.current_tenant_id,
+                Tag.type == payload.type,
+            )
+        ).all()
+        for tag_id in valid_tag_ids:
             tag_binding = db.session.scalar(
                 select(TagBinding)
                 .where(TagBinding.tag_id == tag_id, TagBinding.target_id == payload.target_id)
@@ -186,6 +194,12 @@ class TagService:
                     TagBinding.target_id == payload.target_id,
                     TagBinding.tag_id.in_(payload.tag_ids),
                     TagBinding.tenant_id == current_user.current_tenant_id,
+                    TagBinding.tag_id.in_(
+                        select(Tag.id).where(
+                            Tag.tenant_id == current_user.current_tenant_id,
+                            Tag.type == payload.type,
+                        )
+                    ),
                 )
             ),
         )
@@ -209,5 +223,13 @@ class TagService:
             )
             if not app:
                 raise NotFound("App not found")
+        elif type == "snippet":
+            snippet = db.session.scalar(
+                select(CustomizedSnippet)
+                .where(CustomizedSnippet.tenant_id == current_user.current_tenant_id, CustomizedSnippet.id == target_id)
+                .limit(1)
+            )
+            if not snippet:
+                raise NotFound("Snippet not found")
         else:
             raise NotFound("Invalid binding type")

api/services/workflow_draft_variable_service.py

@@ -1,7 +1,7 @@
 import dataclasses
 import json
 import logging
-from collections.abc import Mapping, Sequence
+from collections.abc import Mapping, Sequence, Set
 from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime
 from enum import StrEnum
@@ -271,12 +271,20 @@ class WorkflowDraftVariableService:
         )
 
     def list_variables_without_values(
-        self, app_id: str, page: int, limit: int, user_id: str
+        self,
+        app_id: str,
+        page: int,
+        limit: int,
+        user_id: str,
+        *,
+        exclude_node_ids: Set[str] | None = None,
     ) -> WorkflowDraftVariableList:
         criteria = [
             WorkflowDraftVariable.app_id == app_id,
             WorkflowDraftVariable.user_id == user_id,
         ]
+        if exclude_node_ids:
+            criteria.append(WorkflowDraftVariable.node_id.notin_(list(exclude_node_ids)))
         total = None
         base_stmt = select(WorkflowDraftVariable).where(*criteria)
         if page == 1:

api/tests/unit_tests/controllers/console/snippets/test_snippet_list_query.py

@@ -0,0 +1,30 @@
+from werkzeug.datastructures import MultiDict
+
+from controllers.console.snippets.payloads import SnippetListQuery
+from controllers.console.workspace.snippets import _normalize_snippet_list_query_args
+
+
+def test_snippet_list_query_accepts_comma_separated_tag_ids() -> None:
+    first = "11111111-1111-1111-1111-111111111111"
+    second = "22222222-2222-2222-2222-222222222222"
+
+    query = SnippetListQuery.model_validate({"tag_ids": f"{first},{second}"})
+
+    assert query.tag_ids == [first, second]
+
+
+def test_normalize_snippet_list_query_accepts_indexed_tag_ids() -> None:
+    first = "11111111-1111-1111-1111-111111111111"
+    second = "22222222-2222-2222-2222-222222222222"
+
+    normalized = _normalize_snippet_list_query_args(
+        MultiDict(
+            [
+                ("tag_ids[1]", second),
+                ("tag_ids[0]", first),
+                ("keyword", "search"),
+            ]
+        )
+    )
+
+    assert normalized == {"keyword": "search", "tag_ids": [first, second]}

api/tests/unit_tests/controllers/console/snippets/test_snippet_workflow.py

@@ -0,0 +1,136 @@
+from __future__ import annotations
+
+from datetime import datetime
+from types import SimpleNamespace
+
+import pytest
+from werkzeug.exceptions import HTTPException, NotFound
+
+from controllers.console.snippets import snippet_workflow as snippet_workflow_module
+
+
+def _unwrap(func):
+    while hasattr(func, "__wrapped__"):
+        func = func.__wrapped__
+    return func
+
+
+def test_restore_published_snippet_workflow_to_draft_success(
+    app, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    workflow = SimpleNamespace(
+        unique_hash="restored-hash",
+        updated_at=None,
+        created_at=datetime(2024, 1, 1),
+    )
+    user = SimpleNamespace(id="account-1")
+    snippet = SimpleNamespace(id="snippet-1", tenant_id="tenant-1")
+
+    monkeypatch.setattr(snippet_workflow_module, "current_account_with_tenant", lambda: (user, "tenant-1"))
+    monkeypatch.setattr(
+        snippet_workflow_module,
+        "SnippetService",
+        lambda: SimpleNamespace(restore_published_workflow_to_draft=lambda **_kwargs: workflow),
+    )
+
+    api = snippet_workflow_module.SnippetDraftWorkflowRestoreApi()
+    handler = _unwrap(api.post)
+
+    with app.test_request_context(
+        "/snippets/snippet-1/workflows/published-workflow/restore",
+        method="POST",
+    ):
+        response = handler(api, snippet=snippet, workflow_id="published-workflow")
+
+    assert response["result"] == "success"
+    assert response["hash"] == "restored-hash"
+
+
+def test_restore_published_snippet_workflow_to_draft_not_found(
+    app, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    user = SimpleNamespace(id="account-1")
+    snippet = SimpleNamespace(id="snippet-1", tenant_id="tenant-1")
+
+    monkeypatch.setattr(snippet_workflow_module, "current_account_with_tenant", lambda: (user, "tenant-1"))
+    monkeypatch.setattr(
+        snippet_workflow_module,
+        "SnippetService",
+        lambda: SimpleNamespace(
+            restore_published_workflow_to_draft=lambda **_kwargs: (_ for _ in ()).throw(
+                snippet_workflow_module.WorkflowNotFoundError("Workflow not found")
+            )
+        ),
+    )
+
+    api = snippet_workflow_module.SnippetDraftWorkflowRestoreApi()
+    handler = _unwrap(api.post)
+
+    with app.test_request_context(
+        "/snippets/snippet-1/workflows/published-workflow/restore",
+        method="POST",
+    ):
+        with pytest.raises(NotFound):
+            handler(api, snippet=snippet, workflow_id="published-workflow")
+
+
+def test_restore_published_snippet_workflow_to_draft_returns_400_for_draft_source(
+    app, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    user = SimpleNamespace(id="account-1")
+    snippet = SimpleNamespace(id="snippet-1", tenant_id="tenant-1")
+
+    monkeypatch.setattr(snippet_workflow_module, "current_account_with_tenant", lambda: (user, "tenant-1"))
+    monkeypatch.setattr(
+        snippet_workflow_module,
+        "SnippetService",
+        lambda: SimpleNamespace(
+            restore_published_workflow_to_draft=lambda **_kwargs: (_ for _ in ()).throw(
+                snippet_workflow_module.IsDraftWorkflowError("source workflow must be published")
+            )
+        ),
+    )
+
+    api = snippet_workflow_module.SnippetDraftWorkflowRestoreApi()
+    handler = _unwrap(api.post)
+
+    with app.test_request_context(
+        "/snippets/snippet-1/workflows/draft-workflow/restore",
+        method="POST",
+    ):
+        with pytest.raises(HTTPException) as exc:
+            handler(api, snippet=snippet, workflow_id="draft-workflow")
+
+    assert exc.value.code == 400
+    assert exc.value.description == snippet_workflow_module.RESTORE_SOURCE_WORKFLOW_MUST_BE_PUBLISHED_MESSAGE
+
+
+def test_restore_published_snippet_workflow_to_draft_returns_400_for_invalid_graph(
+    app, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    user = SimpleNamespace(id="account-1")
+    snippet = SimpleNamespace(id="snippet-1", tenant_id="tenant-1")
+
+    monkeypatch.setattr(snippet_workflow_module, "current_account_with_tenant", lambda: (user, "tenant-1"))
+    monkeypatch.setattr(
+        snippet_workflow_module,
+        "SnippetService",
+        lambda: SimpleNamespace(
+            restore_published_workflow_to_draft=lambda **_kwargs: (_ for _ in ()).throw(
+                ValueError("invalid snippet workflow graph")
+            )
+        ),
+    )
+
+    api = snippet_workflow_module.SnippetDraftWorkflowRestoreApi()
+    handler = _unwrap(api.post)
+
+    with app.test_request_context(
+        "/snippets/snippet-1/workflows/published-workflow/restore",
+        method="POST",
+    ):
+        with pytest.raises(HTTPException) as exc:
+            handler(api, snippet=snippet, workflow_id="published-workflow")
+
+    assert exc.value.code == 400
+    assert exc.value.description == "invalid snippet workflow graph"

api/tests/unit_tests/controllers/console/tag/test_tags.py

@@ -101,6 +101,34 @@ class TestTagListApi:
         assert status == 200
         assert result == [{"id": "1", "name": "tag", "type": "knowledge", "binding_count": "1"}]
 
+    def test_get_snippet_tags(self, app: Flask):
+        api = TagListApi()
+        method = unwrap(api.get)
+
+        with app.test_request_context("/?type=snippet"):
+            with (
+                patch(
+                    "controllers.console.tag.tags.current_account_with_tenant",
+                    return_value=(MagicMock(), "tenant-1"),
+                ),
+                patch(
+                    "controllers.console.tag.tags.TagService.get_tags",
+                    return_value=[
+                        SimpleNamespace(
+                            id="1",
+                            name="snippet-tag",
+                            type=TagType.SNIPPET,
+                            binding_count=1,
+                        )
+                    ],
+                ) as get_tags_mock,
+            ):
+                result, status = method(api)
+
+        get_tags_mock.assert_called_once_with("snippet", "tenant-1", None)
+        assert status == 200
+        assert result == [{"id": "1", "name": "snippet-tag", "type": "snippet", "binding_count": "1"}]
+
     def test_post_success(self, app: Flask, admin_user, tag, payload_patch):
         api = TagListApi()
         method = unwrap(api.post)
@@ -235,6 +263,34 @@ class TestTagBindingCollectionApi:
         assert status == 200
         assert result["result"] == "success"
 
+    def test_create_snippet_binding_success(self, app: Flask, admin_user, payload_patch):
+        api = TagBindingCollectionApi()
+        method = unwrap(api.post)
+
+        payload = {
+            "tag_ids": ["tag-1"],
+            "target_id": "snippet-1",
+            "type": "snippet",
+        }
+
+        with app.test_request_context("/", json=payload):
+            with (
+                patch(
+                    "controllers.console.tag.tags.current_account_with_tenant",
+                    return_value=(admin_user, None),
+                ),
+                payload_patch(payload),
+                patch("controllers.console.tag.tags.TagService.save_tag_binding") as save_mock,
+            ):
+                result, status = method(api)
+
+        save_mock.assert_called_once()
+        binding_payload = save_mock.call_args.args[0]
+        assert binding_payload.type == TagType.SNIPPET
+        assert binding_payload.target_id == "snippet-1"
+        assert status == 200
+        assert result["result"] == "success"
+
     def test_create_forbidden(self, app: Flask, readonly_user, payload_patch):
         api = TagBindingCollectionApi()
         method = unwrap(api.post)

api/tests/unit_tests/core/workflow/test_snippet_start.py

@@ -0,0 +1,23 @@
+from core.workflow.snippet_start import (
+    LEGACY_START_NODE_ID,
+    SNIPPET_VIRTUAL_START_NODE_ID,
+    get_compatible_start_aliases,
+)
+
+
+def test_get_compatible_start_aliases_returns_legacy_start_for_snippet_virtual_start() -> None:
+    aliases = get_compatible_start_aliases(
+        workflow_kind="snippet",
+        root_node_id=SNIPPET_VIRTUAL_START_NODE_ID,
+    )
+
+    assert aliases == (LEGACY_START_NODE_ID,)
+
+
+def test_get_compatible_start_aliases_returns_empty_for_non_snippet_roots() -> None:
+    aliases = get_compatible_start_aliases(
+        workflow_kind="workflow",
+        root_node_id=SNIPPET_VIRTUAL_START_NODE_ID,
+    )
+
+    assert aliases == ()

api/tests/unit_tests/core/workflow/test_variable_pool.py

@@ -5,7 +5,7 @@ from collections import defaultdict
 import pytest
 
 from core.workflow.system_variables import build_system_variables, system_variables_to_mapping
-from core.workflow.variable_pool_initializer import add_variables_to_pool
+from core.workflow.variable_pool_initializer import add_node_inputs_to_pool, add_variables_to_pool
 from core.workflow.variable_prefixes import (
     CONVERSATION_VARIABLE_NODE_ID,
     ENVIRONMENT_VARIABLE_NODE_ID,
@@ -80,6 +80,25 @@ def test_get_file_attribute(pool, file):
     assert result is None
 
 
+def test_add_node_inputs_to_pool_stores_inputs_under_aliases():
+    pool = VariablePool()
+
+    add_node_inputs_to_pool(
+        pool,
+        node_id="__snippet_virtual_start__",
+        inputs={"query": "hello"},
+        aliases=("start", "__snippet_virtual_start__"),
+    )
+
+    primary_value = pool.get(["__snippet_virtual_start__", "query"])
+    alias_value = pool.get(["start", "query"])
+
+    assert primary_value is not None
+    assert primary_value.value == "hello"
+    assert alias_value is not None
+    assert alias_value.value == "hello"
+
+
 class TestVariablePool:
     def test_constructor(self):
         pool = VariablePool()

api/tests/unit_tests/services/test_snippet_dsl_service.py

@@ -0,0 +1,46 @@
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+from services.snippet_dsl_service import ImportStatus, SnippetDslService, SnippetPendingData
+
+
+def test_confirm_import_creates_snippet_from_pending_data(monkeypatch):
+    service = SnippetDslService(session=SimpleNamespace(scalar=Mock(return_value=None)))
+    account = SimpleNamespace(id="account-1", current_tenant_id="tenant-1")
+    snippet = SimpleNamespace(id="snippet-new")
+    yaml_content = """
+version: 9.0.0
+kind: snippet
+snippet:
+  name: From DSL
+  type: node
+workflow:
+  graph:
+    nodes: []
+    edges: []
+"""
+    pending = SnippetPendingData(
+        import_mode="yaml-content",
+        yaml_content=yaml_content,
+        name="Override name",
+        description="Override description",
+        snippet_id=None,
+    )
+    create_or_update = Mock(return_value=snippet)
+    monkeypatch.setattr(service, "_create_or_update_snippet", create_or_update)
+    monkeypatch.setattr("services.snippet_dsl_service.redis_client.get", Mock(return_value=pending.model_dump_json()))
+    redis_delete = Mock()
+    monkeypatch.setattr("services.snippet_dsl_service.redis_client.delete", redis_delete)
+
+    result = service.confirm_import(import_id="import-1", account=account)
+
+    assert result.status == ImportStatus.COMPLETED
+    assert result.snippet_id == "snippet-new"
+    assert result.imported_dsl_version == "9.0.0"
+    create_or_update.assert_called_once()
+    _, kwargs = create_or_update.call_args
+    assert kwargs["snippet"] is None
+    assert kwargs["account"] is account
+    assert kwargs["name"] == "Override name"
+    assert kwargs["description"] == "Override description"
+    redis_delete.assert_called_once_with("snippet_import_info:import-1")

api/tests/unit_tests/services/test_snippet_service.py

@@ -0,0 +1,149 @@
+from __future__ import annotations
+
+import json
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+
+from models.snippet import SnippetType
+from models.workflow import Workflow, WorkflowKind, WorkflowType
+from services.errors.app import WorkflowNotFoundError
+from services.snippet_service import SnippetService
+
+
+class _SessionWithoutNameLookup:
+    def __init__(self) -> None:
+        self.add = Mock()
+        self.commit = Mock()
+
+    def query(self, *args, **kwargs):
+        raise AssertionError("snippet name uniqueness lookup should not be used")
+
+
+def _create_workflow(*, workflow_id: str, version: str, graph: dict, features: dict) -> Workflow:
+    return Workflow(
+        id=workflow_id,
+        tenant_id="tenant-1",
+        app_id="snippet-1",
+        type=WorkflowType.WORKFLOW.value,
+        kind=WorkflowKind.SNIPPET.value,
+        version=version,
+        graph=json.dumps(graph),
+        features=json.dumps(features),
+        created_by="account-1",
+        environment_variables=[],
+        conversation_variables=[],
+        rag_pipeline_variables=[],
+    )
+
+
+def test_create_snippet_allows_duplicate_names(monkeypatch: pytest.MonkeyPatch) -> None:
+    session = _SessionWithoutNameLookup()
+    account = SimpleNamespace(id="account-1")
+
+    monkeypatch.setattr("services.snippet_service.db.session", session)
+
+    snippet = SnippetService.create_snippet(
+        tenant_id="tenant-1",
+        name="shared name",
+        description=None,
+        snippet_type=SnippetType.NODE,
+        icon_info=None,
+        input_fields=None,
+        account=account,
+    )
+
+    assert snippet.name == "shared name"
+    session.add.assert_called_once_with(snippet)
+    session.commit.assert_called_once()
+
+
+def test_update_snippet_allows_duplicate_names() -> None:
+    session = _SessionWithoutNameLookup()
+    snippet = SimpleNamespace(
+        id="snippet-1",
+        tenant_id="tenant-1",
+        name="old name",
+        description="",
+        icon_info=None,
+    )
+
+    result = SnippetService.update_snippet(
+        session=session,
+        snippet=snippet,
+        account_id="account-1",
+        data={"name": "shared name"},
+    )
+
+    assert result is snippet
+    assert snippet.name == "shared name"
+    session.add.assert_called_once_with(snippet)
+
+
+def test_restore_published_snippet_workflow_to_draft_copies_source_snapshot(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    snippet = SimpleNamespace(id="snippet-1", tenant_id="tenant-1")
+    account = SimpleNamespace(id="account-2")
+    source_graph = {"nodes": [{"id": "llm-1", "data": {"type": "llm"}}], "edges": []}
+    source_features = {"opening_statement": "hello"}
+    source_workflow = _create_workflow(
+        workflow_id="published-workflow",
+        version="2026-04-28 00:00:00",
+        graph=source_graph,
+        features=source_features,
+    )
+    draft_workflow = _create_workflow(
+        workflow_id="draft-workflow",
+        version=Workflow.VERSION_DRAFT,
+        graph={"nodes": [], "edges": []},
+        features={},
+    )
+    service = SnippetService.__new__(SnippetService)
+    session = SimpleNamespace(add=Mock(), commit=Mock())
+
+    monkeypatch.setattr(service, "get_published_workflow_by_id", Mock(return_value=source_workflow))
+    monkeypatch.setattr(service, "get_draft_workflow", Mock(return_value=draft_workflow))
+    monkeypatch.setattr("services.snippet_service.db.session", session)
+
+    result = service.restore_published_workflow_to_draft(
+        snippet=snippet,
+        workflow_id=source_workflow.id,
+        account=account,
+    )
+
+    assert result is draft_workflow
+    assert draft_workflow.graph_dict == source_graph
+    assert draft_workflow.features_dict == source_features
+    assert draft_workflow.updated_by == account.id
+    session.add.assert_not_called()
+    session.commit.assert_called_once()
+
+
+def test_restore_published_snippet_workflow_to_draft_raises_when_source_missing(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    snippet = SimpleNamespace(id="snippet-1", tenant_id="tenant-1")
+    account = SimpleNamespace(id="account-2")
+    service = SnippetService.__new__(SnippetService)
+
+    monkeypatch.setattr(service, "get_published_workflow_by_id", Mock(return_value=None))
+
+    with pytest.raises(WorkflowNotFoundError):
+        service.restore_published_workflow_to_draft(
+            snippet=snippet,
+            workflow_id="missing-workflow",
+            account=account,
+        )
+
+
+def test_delete_snippet_removes_tag_bindings() -> None:
+    snippet = SimpleNamespace(id="snippet-1", tenant_id="tenant-1")
+    session = SimpleNamespace(execute=Mock(), delete=Mock())
+
+    result = SnippetService.delete_snippet(session=session, snippet=snippet)
+
+    assert result is True
+    session.execute.assert_called_once()
+    session.delete.assert_called_once_with(snippet)