filter virtual start node event for snippet draft run.

Agent-Logs-Url: https://github.com/langgenius/dify/sessions/5a3defd9-7c29-4081-81c4-3506303d2a39

Co-authored-by: FFXN <[email protected]>

.agents/skills/frontend-testing/references/mocking.md

@@ -20,11 +20,11 @@
 ```typescript
 // ❌ WRONG: Don't mock base components
 vi.mock('@/app/components/base/loading', () => () => <div>Loading</div>)
-vi.mock('@/app/components/base/button', () => ({ children }: any) => <button>{children}</button>)
+vi.mock('@langgenius/dify-ui/button', () => ({ children }: any) => <button>{children}</button>)
 
 // ✅ CORRECT: Import and use real base components
 import Loading from '@/app/components/base/loading'
-import Button from '@/app/components/base/button'
+import { Button } from '@langgenius/dify-ui/button'
 // They will render normally in tests
 ```
 

.github/workflows/style.yml

@@ -77,8 +77,6 @@ jobs:
         with:
           files: |
             web/**
-            e2e/**
-            sdks/nodejs-client/**
             packages/**
             package.json
             pnpm-lock.yaml
@@ -97,14 +95,14 @@ jobs:
         id: eslint-cache-restore
         uses: actions/cache/restore@668228422ae6a00e4ad889ee87cd7109ec5666a7 # v5.0.4
         with:
-          path: .eslintcache
+          path: web/.eslintcache
-          key: ${{ runner.os }}-eslint-${{ hashFiles('pnpm-lock.yaml', 'eslint.config.mjs', 'web/eslint.config.mjs', 'web/eslint.constants.mjs', 'web/plugins/eslint/**') }}-${{ github.sha }}
+          key: ${{ runner.os }}-web-eslint-${{ hashFiles('web/package.json', 'pnpm-lock.yaml', 'web/eslint.config.mjs', 'web/eslint.constants.mjs', 'web/plugins/eslint/**') }}-${{ github.sha }}
           restore-keys: |
-            ${{ runner.os }}-eslint-${{ hashFiles('pnpm-lock.yaml', 'eslint.config.mjs', 'web/eslint.config.mjs', 'web/eslint.constants.mjs', 'web/plugins/eslint/**') }}-
+            ${{ runner.os }}-web-eslint-${{ hashFiles('web/package.json', 'pnpm-lock.yaml', 'web/eslint.config.mjs', 'web/eslint.constants.mjs', 'web/plugins/eslint/**') }}-
 
       - name: Web style check
         if: steps.changed-files.outputs.any_changed == 'true'
-        working-directory: .
+        working-directory: ./web
         run: vp run lint:ci
 
       - name: Web tsslint
@@ -114,7 +112,7 @@ jobs:
 
       - name: Web type check
         if: steps.changed-files.outputs.any_changed == 'true'
-        working-directory: .
+        working-directory: ./web
         run: vp run type-check
 
       - name: Web dead code check
@@ -126,7 +124,7 @@ jobs:
         if: steps.changed-files.outputs.any_changed == 'true' && success() && steps.eslint-cache-restore.outputs.cache-hit != 'true'
         uses: actions/cache/save@668228422ae6a00e4ad889ee87cd7109ec5666a7 # v5.0.4
         with:
-          path: .eslintcache
+          path: web/.eslintcache
           key: ${{ steps.eslint-cache-restore.outputs.cache-primary-key }}
 
   superlinter:

.gitignore

@@ -203,7 +203,6 @@ sdks/python-client/dify_client.egg-info
 
 .vscode/*
 !.vscode/launch.json.template
-!.vscode/settings.example.json
 !.vscode/README.md
 api/.vscode
 # vscode Code History Extension
@@ -243,5 +242,3 @@ scripts/stress-test/reports/
 
 # Code Agent Folder
 .qoder/*
-
-.eslintcache

.vite-hooks/pre-commit

@@ -56,9 +56,44 @@ if $api_modified; then
     fi
 fi
 
-if $skip_web_checks; then
+if $web_modified; then
-    echo "Git operation in progress, skipping web checks"
+    if $skip_web_checks; then
-    exit 0
+        echo "Git operation in progress, skipping web checks"
-fi
+        exit 0
+    fi
 
-vp staged
+    echo "Running ESLint on web module"
+
+    if git diff --cached --quiet -- 'web/**/*.ts' 'web/**/*.tsx'; then
+        web_ts_modified=false
+    else
+        ts_diff_status=$?
+        if [ $ts_diff_status -eq 1 ]; then
+            web_ts_modified=true
+        else
+            echo "Unable to determine staged TypeScript changes (git exit code: $ts_diff_status)."
+            exit $ts_diff_status
+        fi
+    fi
+
+    cd ./web || exit 1
+    pnpm exec vp staged
+
+    if $web_ts_modified; then
+        echo "Running TypeScript type-check:tsgo"
+        if ! npm run type-check:tsgo; then
+            echo "Type check failed. Please run 'npm run type-check:tsgo' to fix the errors."
+            exit 1
+        fi
+    else
+        echo "No staged TypeScript changes detected, skipping type-check:tsgo"
+    fi
+
+    echo "Running knip"
+    if ! npm run knip; then
+        echo "Knip check failed. Please run 'npm run knip' to fix the errors."
+        exit 1
+    fi
+
+    cd ../
+fi

api/configs/feature/__init__.py

@@ -1373,6 +1373,32 @@ class SandboxExpiredRecordsCleanConfig(BaseSettings):
     )
 
 
+class EvaluationConfig(BaseSettings):
+    """
+    Configuration for evaluation runtime
+    """
+
+    EVALUATION_FRAMEWORK: str = Field(
+        description="Evaluation framework to use (ragas/deepeval/none)",
+        default="none",
+    )
+
+    EVALUATION_MAX_CONCURRENT_RUNS: PositiveInt = Field(
+        description="Maximum number of concurrent evaluation runs per tenant",
+        default=3,
+    )
+
+    EVALUATION_MAX_DATASET_ROWS: PositiveInt = Field(
+        description="Maximum number of rows allowed in an evaluation dataset",
+        default=500,
+    )
+
+    EVALUATION_TASK_TIMEOUT: PositiveInt = Field(
+        description="Timeout in seconds for a single evaluation task",
+        default=3600,
+    )
+
+
 class FeatureConfig(
     # place the configs in alphabet order
     AppExecutionConfig,
@@ -1385,6 +1411,7 @@ class FeatureConfig(
     MarketplaceConfig,
     DataSetConfig,
     EndpointConfig,
+    EvaluationConfig,
     FileAccessConfig,
     FileUploadConfig,
     HttpConfig,

api/controllers/console/__init__.py

@@ -108,6 +108,9 @@ from .datasets.rag_pipeline import (
     rag_pipeline_workflow,
 )
 
+# Import evaluation controllers
+from .evaluation import evaluation
+
 # Import explore controllers
 from .explore import (
     banner,
@@ -117,8 +120,14 @@ from .explore import (
     saved_message,
     trial,
 )
+
+# Import snippet controllers
+from .snippets import snippet_workflow, snippet_workflow_draft_variable
 from .socketio import workflow as socketio_workflow  # pyright: ignore[reportUnusedImport]
 
+# Import snippet controllers
+from .snippets import snippet_workflow, snippet_workflow_draft_variable
+
 # Import tag controllers
 from .tag import tags
 
@@ -132,6 +141,7 @@ from .workspace import (
     model_providers,
     models,
     plugin,
+    snippets,
     tool_providers,
     trigger_providers,
     workspace,
@@ -169,6 +179,7 @@ __all__ = [
     "datasource_content_preview",
     "email_register",
     "endpoint",
+    "evaluation",
     "extension",
     "external",
     "feature",
@@ -203,7 +214,13 @@ __all__ = [
     "saved_message",
     "setup",
     "site",
+    "snippet_workflow",
+    "snippet_workflow_draft_variable",
+    "snippets",
     "socketio_workflow",
+    "snippet_workflow",
+    "snippet_workflow_draft_variable",
+    "snippets",
     "spec",
     "statistic",
     "tags",

api/controllers/console/app/app.py

@@ -5,6 +5,7 @@ from typing import Any, Literal
 
 from flask import request
 from flask_restx import Resource
+from graphon.enums import WorkflowExecutionStatus
 from pydantic import AliasChoices, BaseModel, Field, computed_field, field_validator
 from sqlalchemy import select
 from sqlalchemy.orm import Session
@@ -29,11 +30,11 @@ from core.rag.retrieval.retrieval_methods import RetrievalMethod
 from core.trigger.constants import TRIGGER_NODE_TYPES
 from extensions.ext_database import db
 from fields.base import ResponseModel
-from graphon.enums import WorkflowExecutionStatus
 from libs.helper import build_icon_url
 from libs.login import current_account_with_tenant, login_required
 from models import App, DatasetPermissionEnum, Workflow
 from models.model import IconType
+from models.workflow import resolve_workflow_kind
 from services.app_dsl_service import AppDslService
 from services.app_service import AppService
 from services.enterprise.enterprise_service import EnterpriseService
@@ -329,6 +330,8 @@ class AppPartial(ResponseModel):
     create_user_name: str | None = None
     author_name: str | None = None
     has_draft_trigger: bool | None = None
+    workflow_type: str | None = None
+    workflow_kind: str | None = None
 
     @computed_field(return_type=str | None)  # type: ignore
     @property
@@ -363,6 +366,8 @@ class AppDetail(ResponseModel):
     updated_by: str | None = None
     updated_at: int | None = None
     access_mode: str | None = None
+    workflow_type: str | None = None
+    workflow_kind: str | None = None
     tags: list[Tag] = Field(default_factory=list)
 
     @field_validator("created_at", "updated_at", mode="before")
@@ -505,6 +510,25 @@ class AppListApi(Resource):
         for app in app_pagination.items:
             app.has_draft_trigger = str(app.id) in draft_trigger_app_ids
 
+        workflow_ids = [str(app.workflow_id) for app in app_pagination.items if app.workflow_id]
+        workflow_info_map: dict[str, tuple[str, str]] = {}
+        if workflow_ids:
+            rows = db.session.execute(
+                select(Workflow.id, Workflow.type, Workflow.kind).where(Workflow.id.in_(workflow_ids))
+            ).all()
+            workflow_info_map = {
+                str(row.id): (
+                    row.type.value if hasattr(row.type, "value") else str(row.type),
+                    resolve_workflow_kind(row.kind).value,
+                )
+                for row in rows
+            }
+
+        for app in app_pagination.items:
+            workflow_info = workflow_info_map.get(str(app.workflow_id)) if app.workflow_id else None
+            app.workflow_type = workflow_info[0] if workflow_info else None
+            app.workflow_kind = workflow_info[1] if workflow_info else None
+
         pagination_model = AppPagination.model_validate(app_pagination, from_attributes=True)
         return pagination_model.model_dump(mode="json"), 200
 
@@ -551,6 +575,18 @@ class AppApi(Resource):
             app_setting = EnterpriseService.WebAppAuth.get_app_access_mode_by_id(app_id=str(app_model.id))
             app_model.access_mode = app_setting.access_mode
 
+        if app_model.workflow_id:
+            row = db.session.execute(
+                select(Workflow.type, Workflow.kind).where(Workflow.id == app_model.workflow_id)
+            ).first()
+            app_model.workflow_type = (
+                (row.type.value if hasattr(row.type, "value") else str(row.type)) if row else None
+            )
+            app_model.workflow_kind = resolve_workflow_kind(row.kind).value if row else None
+        else:
+            app_model.workflow_type = None
+            app_model.workflow_kind = None
+
         response_model = AppDetailWithSite.model_validate(app_model, from_attributes=True)
         return response_model.model_dump(mode="json")
 

api/controllers/console/app/audio.py

@@ -2,6 +2,7 @@ import logging
 
 from flask import request
 from flask_restx import Resource, fields
+from graphon.model_runtime.errors.invoke import InvokeError
 from pydantic import BaseModel, Field
 from werkzeug.exceptions import InternalServerError
 
@@ -22,7 +23,6 @@ from controllers.console.app.error import (
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
-from graphon.model_runtime.errors.invoke import InvokeError
 from libs.login import login_required
 from models import App, AppMode
 from services.audio_service import AudioService

api/controllers/console/app/completion.py

@@ -3,6 +3,7 @@ from typing import Any, Literal
 
 from flask import request
 from flask_restx import Resource
+from graphon.model_runtime.errors.invoke import InvokeError
 from pydantic import BaseModel, Field, field_validator
 from werkzeug.exceptions import InternalServerError, NotFound
 
@@ -26,7 +27,6 @@ from core.errors.error import (
     QuotaExceededError,
 )
 from core.helper.trace_id_helper import get_external_trace_id
-from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from libs.helper import uuid_value
 from libs.login import current_user, login_required

api/controllers/console/app/generator.py

@@ -1,6 +1,7 @@
 from collections.abc import Sequence
 
 from flask_restx import Resource
+from graphon.model_runtime.errors.invoke import InvokeError
 from pydantic import BaseModel, Field
 
 from controllers.console import console_ns
@@ -19,7 +20,6 @@ from core.helper.code_executor.python3.python3_code_provider import Python3CodeP
 from core.llm_generator.entities import RuleCodeGeneratePayload, RuleGeneratePayload, RuleStructuredOutputPayload
 from core.llm_generator.llm_generator import LLMGenerator
 from extensions.ext_database import db
-from graphon.model_runtime.errors.invoke import InvokeError
 from libs.login import current_account_with_tenant, login_required
 from models import App
 from services.workflow_service import WorkflowService

api/controllers/console/app/mcp_server.py

@@ -18,6 +18,12 @@ from models.enums import AppMCPServerStatus
 from models.model import AppMCPServer
 
 
+def _to_timestamp(value: datetime | int | None) -> int | None:
+    if isinstance(value, datetime):
+        return int(value.timestamp())
+    return value
+
+
 class MCPServerCreatePayload(BaseModel):
     description: str | None = Field(default=None, description="Server description")
     parameters: dict[str, Any] = Field(..., description="Server parameters configuration")
@@ -30,25 +36,19 @@ class MCPServerUpdatePayload(BaseModel):
     status: str | None = Field(default=None, description="Server status")
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 class AppMCPServerResponse(ResponseModel):
     id: str
     name: str
     server_code: str
     description: str
-    status: AppMCPServerStatus
+    status: str
     parameters: dict[str, Any] | list[Any] | str
     created_at: int | None = None
     updated_at: int | None = None
 
     @field_validator("parameters", mode="before")
     @classmethod
-    def _normalize_parameters(cls, value: Any) -> Any:
+    def _parse_json_string(cls, value: Any) -> Any:
         if isinstance(value, str):
             try:
                 return json.loads(value)
@@ -70,9 +70,7 @@ class AppMCPServerController(Resource):
     @console_ns.doc("get_app_mcp_server")
     @console_ns.doc(description="Get MCP server configuration for an application")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.response(
+    @console_ns.response(200, "Server configuration", console_ns.models[AppMCPServerResponse.__name__])
-        200, "MCP server configuration retrieved successfully", console_ns.models[AppMCPServerResponse.__name__]
-    )
     @login_required
     @account_initialization_required
     @setup_required
@@ -87,9 +85,7 @@ class AppMCPServerController(Resource):
     @console_ns.doc(description="Create MCP server configuration for an application")
     @console_ns.doc(params={"app_id": "Application ID"})
     @console_ns.expect(console_ns.models[MCPServerCreatePayload.__name__])
-    @console_ns.response(
+    @console_ns.response(200, "Server created", console_ns.models[AppMCPServerResponse.__name__])
-        201, "MCP server configuration created successfully", console_ns.models[AppMCPServerResponse.__name__]
-    )
     @console_ns.response(403, "Insufficient permissions")
     @account_initialization_required
     @get_app_model
@@ -115,15 +111,13 @@ class AppMCPServerController(Resource):
         )
         db.session.add(server)
         db.session.commit()
-        return AppMCPServerResponse.model_validate(server, from_attributes=True).model_dump(mode="json"), 201
+        return AppMCPServerResponse.model_validate(server, from_attributes=True).model_dump(mode="json")
 
     @console_ns.doc("update_app_mcp_server")
     @console_ns.doc(description="Update MCP server configuration for an application")
     @console_ns.doc(params={"app_id": "Application ID"})
     @console_ns.expect(console_ns.models[MCPServerUpdatePayload.__name__])
-    @console_ns.response(
+    @console_ns.response(200, "Server updated", console_ns.models[AppMCPServerResponse.__name__])
-        200, "MCP server configuration updated successfully", console_ns.models[AppMCPServerResponse.__name__]
-    )
     @console_ns.response(403, "Insufficient permissions")
     @console_ns.response(404, "Server not found")
     @get_app_model
@@ -160,7 +154,7 @@ class AppMCPServerRefreshController(Resource):
     @console_ns.doc("refresh_app_mcp_server")
     @console_ns.doc(description="Refresh MCP server configuration and regenerate server code")
     @console_ns.doc(params={"server_id": "Server ID"})
-    @console_ns.response(200, "MCP server refreshed successfully", console_ns.models[AppMCPServerResponse.__name__])
+    @console_ns.response(200, "Server refreshed", console_ns.models[AppMCPServerResponse.__name__])
     @console_ns.response(403, "Insufficient permissions")
     @console_ns.response(404, "Server not found")
     @setup_required

api/controllers/console/app/workflow.py

@@ -1,16 +1,21 @@
 import json
 import logging
 from collections.abc import Sequence
-from typing import Any
+from typing import Any, Literal
 
 from flask import abort, request
 from flask_restx import Resource, fields, marshal, marshal_with
+from graphon.enums import NodeType
+from graphon.file import File
+from graphon.file import helpers as file_helpers
+from graphon.graph_engine.manager import GraphEngineManager
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 from pydantic import BaseModel, Field, ValidationError, field_validator
 from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import BadRequest, Forbidden, InternalServerError, NotFound
 
 import services
-from controllers.common.controller_schemas import DefaultBlockConfigQuery, WorkflowListQuery, WorkflowUpdatePayload
+from controllers.common.controller_schemas import DefaultBlockConfigQuery
 from controllers.console import console_ns
 from controllers.console.app.error import ConversationCompletedError, DraftWorkflowNotExist, DraftWorkflowNotSync
 from controllers.console.app.workflow_run import workflow_run_node_execution_model
@@ -37,18 +42,13 @@ from factories import file_factory, variable_factory
 from fields.member_fields import simple_account_fields
 from fields.online_user_fields import online_user_list_fields
 from fields.workflow_fields import workflow_fields, workflow_pagination_fields
-from graphon.enums import NodeType
-from graphon.file import File
-from graphon.file import helpers as file_helpers
-from graphon.graph_engine.manager import GraphEngineManager
-from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs import helper
 from libs.datetime_utils import naive_utc_now
 from libs.helper import TimestampField, uuid_value
 from libs.login import current_account_with_tenant, login_required
 from models import App
 from models.model import AppMode
-from models.workflow import Workflow
+from models.workflow import Workflow, WorkflowKind
 from repositories.workflow_collaboration_repository import WORKFLOW_ONLINE_USERS_PREFIX
 from services.app_generate_service import AppGenerateService
 from services.errors.app import IsDraftWorkflowError, WorkflowHashNotEqualError, WorkflowNotFoundError
@@ -154,6 +154,23 @@ class ConvertToWorkflowPayload(BaseModel):
     icon_background: str | None = None
 
 
+class WorkflowListQuery(BaseModel):
+    page: int = Field(default=1, ge=1, le=99999)
+    limit: int = Field(default=10, ge=1, le=100)
+    user_id: str | None = None
+    named_only: bool = False
+    keyword: str | None = Field(default=None, max_length=255)
+
+
+class WorkflowUpdatePayload(BaseModel):
+    marked_name: str | None = Field(default=None, max_length=20)
+    marked_comment: str | None = Field(default=None, max_length=100)
+
+
+class WorkflowTypeConvertQuery(BaseModel):
+    target_type: Literal["workflow", "evaluation"]
+
+
 class WorkflowFeaturesPayload(BaseModel):
     features: dict[str, Any] = Field(..., description="Workflow feature configuration")
 
@@ -185,6 +202,7 @@ reg(DefaultBlockConfigQuery)
 reg(ConvertToWorkflowPayload)
 reg(WorkflowListQuery)
 reg(WorkflowUpdatePayload)
+reg(WorkflowTypeConvertQuery)
 reg(WorkflowFeaturesPayload)
 reg(WorkflowOnlineUsersQuery)
 reg(DraftWorkflowTriggerRunPayload)
@@ -859,6 +877,54 @@ class PublishedWorkflowApi(Resource):
         }
 
 
+@console_ns.route("/apps/<uuid:app_id>/workflows/publish/evaluation")
+class EvaluationPublishedWorkflowApi(Resource):
+    @console_ns.doc("publish_evaluation_workflow")
+    @console_ns.doc(description="Publish draft workflow as evaluation workflow")
+    @console_ns.doc(params={"app_id": "Application ID"})
+    @console_ns.expect(console_ns.models[PublishWorkflowPayload.__name__])
+    @console_ns.response(200, "Evaluation workflow published successfully")
+    @console_ns.response(400, "Invalid workflow or unsupported node type")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    @edit_permission_required
+    def post(self, app_model: App):
+        """
+        Publish draft workflow as evaluation workflow.
+
+        Evaluation workflows cannot include trigger or human-input nodes.
+        """
+        current_user, _ = current_account_with_tenant()
+        args = PublishWorkflowPayload.model_validate(console_ns.payload or {})
+
+        workflow_service = WorkflowService()
+        with Session(db.engine) as session:
+            workflow = workflow_service.publish_evaluation_workflow(
+                session=session,
+                app_model=app_model,
+                account=current_user,
+                marked_name=args.marked_name or "",
+                marked_comment=args.marked_comment or "",
+            )
+
+            # Keep workflow_id aligned with the latest published workflow.
+            app_model_in_session = session.get(App, app_model.id)
+            if app_model_in_session:
+                app_model_in_session.workflow_id = workflow.id
+                app_model_in_session.updated_by = current_user.id
+                app_model_in_session.updated_at = naive_utc_now()
+
+            workflow_created_at = TimestampField().format(workflow.created_at)
+            session.commit()
+
+        return {
+            "result": "success",
+            "created_at": workflow_created_at,
+        }
+
+
 @console_ns.route("/apps/<uuid:app_id>/workflows/default-workflow-block-configs")
 class DefaultBlockConfigsApi(Resource):
     @console_ns.doc("get_default_block_configs")
@@ -1056,6 +1122,52 @@ class DraftWorkflowRestoreApi(Resource):
         }
 
 
+@console_ns.route("/apps/<uuid:app_id>/workflows/convert-type")
+class WorkflowTypeConvertApi(Resource):
+    @console_ns.doc("convert_published_workflow_type")
+    @console_ns.doc(description="Convert current effective published workflow type in-place")
+    @console_ns.doc(params={"app_id": "Application ID"})
+    @console_ns.expect(console_ns.models[WorkflowTypeConvertQuery.__name__])
+    @console_ns.response(200, "Workflow type converted successfully")
+    @console_ns.response(400, "Invalid workflow type or unsupported workflow graph")
+    @console_ns.response(404, "Workflow not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    @edit_permission_required
+    def post(self, app_model: App):
+        current_user, _ = current_account_with_tenant()
+        args = WorkflowTypeConvertQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        target_type = WorkflowKind.EVALUATION if args.target_type == "evaluation" else WorkflowKind.STANDARD
+
+        workflow_service = WorkflowService()
+        with Session(db.engine) as session:
+            try:
+                workflow = workflow_service.convert_published_workflow_type(
+                    session=session,
+                    app_model=app_model,
+                    target_type=target_type,
+                    account=current_user,
+                )
+            except WorkflowNotFoundError as exc:
+                raise NotFound(str(exc)) from exc
+            except IsDraftWorkflowError as exc:
+                raise BadRequest(str(exc)) from exc
+            except ValueError as exc:
+                raise BadRequest(str(exc)) from exc
+
+            session.commit()
+
+        return {
+            "result": "success",
+            "workflow_id": workflow.id,
+            "type": workflow.type.value,
+            "kind": workflow.kind_or_standard,
+            "updated_at": TimestampField().format(workflow.updated_at or workflow.created_at),
+        }
+
+
 @console_ns.route("/apps/<uuid:app_id>/workflows/<string:workflow_id>")
 class WorkflowByIdApi(Resource):
     @console_ns.doc("update_workflow_by_id")

api/controllers/console/app/workflow_app_log.py

@@ -4,6 +4,7 @@ from typing import Any
 from dateutil.parser import isoparse
 from flask import request
 from flask_restx import Resource
+from graphon.enums import WorkflowExecutionStatus
 from pydantic import BaseModel, Field, field_validator
 from sqlalchemy.orm import sessionmaker
 
@@ -15,7 +16,6 @@ from extensions.ext_database import db
 from fields.base import ResponseModel
 from fields.end_user_fields import SimpleEndUser
 from fields.member_fields import SimpleAccount
-from graphon.enums import WorkflowExecutionStatus
 from libs.login import login_required
 from models import App
 from models.model import AppMode

api/controllers/console/app/workflow_run.py

@@ -3,6 +3,8 @@ from typing import Literal, TypedDict, cast
 
 from flask import request
 from flask_restx import Resource, fields, marshal_with
+from graphon.entities.pause_reason import HumanInputRequired
+from graphon.enums import WorkflowExecutionStatus
 from pydantic import BaseModel, Field, field_validator
 from sqlalchemy import select
 from sqlalchemy.orm import sessionmaker
@@ -26,8 +28,6 @@ from fields.workflow_run_fields import (
     workflow_run_node_execution_list_fields,
     workflow_run_pagination_fields,
 )
-from graphon.entities.pause_reason import HumanInputRequired
-from graphon.enums import WorkflowExecutionStatus
 from libs.archive_storage import ArchiveStorageNotConfiguredError, get_archive_storage
 from libs.custom_inputs import time_duration
 from libs.helper import uuid_value

api/controllers/console/auth/oauth_server.py

@@ -5,11 +5,11 @@ from typing import Concatenate
 from flask import jsonify, request
 from flask.typing import ResponseReturnValue
 from flask_restx import Resource
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 from pydantic import BaseModel
 from werkzeug.exceptions import BadRequest, NotFound
 
 from controllers.console.wraps import account_initialization_required, setup_required
-from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs.login import current_account_with_tenant, login_required
 from models import Account
 from models.model import OAuthProviderApp

api/controllers/console/billing/billing.py

@@ -1,4 +1,6 @@
 import base64
+import json
+from datetime import UTC, datetime, timedelta
 from typing import Literal
 
 from flask import request
@@ -10,6 +12,7 @@ from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, only_edition_cloud, setup_required
 from enums.cloud_plan import CloudPlan
+from extensions.ext_redis import redis_client
 from libs.login import current_account_with_tenant, login_required
 from services.billing_service import BillingService
 
@@ -77,3 +80,39 @@ class PartnerTenants(Resource):
             raise BadRequest("Invalid partner information")
 
         return BillingService.sync_partner_tenants_bindings(current_user.id, decoded_partner_key, click_id)
+
+
+_DEBUG_KEY = "billing:debug"
+_DEBUG_TTL = timedelta(days=7)
+
+
+class DebugDataPayload(BaseModel):
+    type: str = Field(..., min_length=1, description="Data type key")
+    data: str = Field(..., min_length=1, description="Data value to append")
+
+
+@console_ns.route("/billing/debug/data")
+class DebugData(Resource):
+    def post(self):
+        body = DebugDataPayload.model_validate(request.get_json(force=True))
+        item = json.dumps({
+            "type": body.type,
+            "data": body.data,
+            "createTime": datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ"),
+        })
+        redis_client.lpush(_DEBUG_KEY, item)
+        redis_client.expire(_DEBUG_KEY, _DEBUG_TTL)
+        return {"result": "ok"}, 201
+
+    def get(self):
+        recent = request.args.get("recent", 10, type=int)
+        items = redis_client.lrange(_DEBUG_KEY, 0, recent - 1)
+        return {
+            "data": [
+                json.loads(item.decode("utf-8") if isinstance(item, bytes) else item) for item in items
+            ]
+        }
+
+    def delete(self):
+        redis_client.delete(_DEBUG_KEY)
+        return {"result": "ok"}

api/controllers/console/datasets/datasets.py

@@ -1,10 +1,14 @@
+import json
 from typing import Any, cast
+from urllib.parse import quote
 
-from flask import request
+from flask import Response, request
 from flask_restx import Resource, fields, marshal, marshal_with
+from graphon.model_runtime.entities.model_entities import ModelType
 from pydantic import BaseModel, Field, field_validator
 from sqlalchemy import func, select
-from werkzeug.exceptions import Forbidden, NotFound
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import BadRequest, Forbidden, NotFound
 
 import services
 from configs import dify_config
@@ -21,6 +25,7 @@ from controllers.console.wraps import (
     setup_required,
 )
 from core.errors.error import LLMBadRequestError, ProviderTokenNotInitError
+from core.evaluation.entities.evaluation_entity import EvaluationCategory, EvaluationConfigData, EvaluationRunRequest
 from core.indexing_runner import IndexingRunner
 from core.plugin.impl.model_runtime_factory import create_plugin_provider_manager
 from core.rag.datasource.vdb.vector_type import VectorType
@@ -29,6 +34,7 @@ from core.rag.extractor.entity.extract_setting import ExtractSetting, NotionInfo
 from core.rag.index_processor.constant.index_type import IndexTechniqueType
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
 from extensions.ext_database import db
+from extensions.ext_storage import storage
 from fields.app_fields import app_detail_kernel_fields, related_app_list
 from fields.dataset_fields import (
     content_fields,
@@ -48,14 +54,20 @@ from fields.dataset_fields import (
     weighted_score_fields,
 )
 from fields.document_fields import document_status_fields
-from graphon.model_runtime.entities.model_entities import ModelType
 from libs.login import current_account_with_tenant, login_required
-from models import ApiToken, Dataset, Document, DocumentSegment, UploadFile
+from models import ApiToken, Dataset, Document, DocumentSegment, EvaluationRun, EvaluationTargetType, UploadFile
 from models.dataset import DatasetPermission, DatasetPermissionEnum
 from models.enums import ApiTokenType, SegmentStatus
 from models.provider_ids import ModelProviderID
 from services.api_token_service import ApiTokenCache
 from services.dataset_service import DatasetPermissionService, DatasetService, DocumentService
+from services.errors.evaluation import (
+    EvaluationDatasetInvalidError,
+    EvaluationFrameworkNotConfiguredError,
+    EvaluationMaxConcurrentRunsError,
+    EvaluationNotFoundError,
+)
+from services.evaluation_service import EvaluationService
 
 # Register models for flask_restx to avoid dict type issues in Swagger
 dataset_base_model = get_or_create_model("DatasetBase", dataset_fields)
@@ -983,3 +995,432 @@ class DatasetAutoDisableLogApi(Resource):
         if dataset is None:
             raise NotFound("Dataset not found.")
         return DatasetService.get_dataset_auto_disable_logs(dataset_id_str), 200
+
+
+# ---- Knowledge Base Retrieval Evaluation ----
+
+
+def _serialize_dataset_evaluation_run(run: EvaluationRun) -> dict[str, Any]:
+    return {
+        "id": run.id,
+        "tenant_id": run.tenant_id,
+        "target_type": run.target_type,
+        "target_id": run.target_id,
+        "evaluation_config_id": run.evaluation_config_id,
+        "status": run.status,
+        "dataset_file_id": run.dataset_file_id,
+        "result_file_id": run.result_file_id,
+        "total_items": run.total_items,
+        "completed_items": run.completed_items,
+        "failed_items": run.failed_items,
+        "progress": run.progress,
+        "metrics_summary": json.loads(run.metrics_summary) if run.metrics_summary else {},
+        "error": run.error,
+        "created_by": run.created_by,
+        "started_at": int(run.started_at.timestamp()) if run.started_at else None,
+        "completed_at": int(run.completed_at.timestamp()) if run.completed_at else None,
+        "created_at": int(run.created_at.timestamp()) if run.created_at else None,
+    }
+
+
+def _serialize_dataset_evaluation_run_item(item: Any) -> dict[str, Any]:
+    return {
+        "id": item.id,
+        "item_index": item.item_index,
+        "inputs": item.inputs_dict,
+        "expected_output": item.expected_output,
+        "actual_output": item.actual_output,
+        "metrics": item.metrics_list,
+        "judgment": item.judgment_dict,
+        "metadata": item.metadata_dict,
+        "error": item.error,
+        "overall_score": item.overall_score,
+    }
+
+
+@console_ns.route("/datasets/<uuid:dataset_id>/evaluation/template/download")
+class DatasetEvaluationTemplateDownloadApi(Resource):
+    @console_ns.doc("download_dataset_evaluation_template")
+    @console_ns.response(200, "Template file streamed as XLSX attachment")
+    @console_ns.response(403, "Permission denied")
+    @console_ns.response(404, "Dataset not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, dataset_id):
+        """Download evaluation dataset template for knowledge base retrieval."""
+        current_user, _ = current_account_with_tenant()
+        dataset_id_str = str(dataset_id)
+        dataset = DatasetService.get_dataset(dataset_id_str)
+        if dataset is None:
+            raise NotFound("Dataset not found.")
+        try:
+            DatasetService.check_dataset_permission(dataset, current_user)
+        except services.errors.account.NoPermissionError as e:
+            raise Forbidden(str(e))
+
+        xlsx_content, filename = EvaluationService.generate_retrieval_dataset_template()
+        encoded_filename = quote(filename)
+        response = Response(
+            xlsx_content,
+            mimetype="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+        )
+        response.headers["Content-Disposition"] = f"attachment; filename*=UTF-8''{encoded_filename}"
+        response.headers["Content-Length"] = str(len(xlsx_content))
+        return response
+
+
+@console_ns.route("/datasets/<uuid:dataset_id>/evaluation")
+class DatasetEvaluationDetailApi(Resource):
+    @console_ns.doc("get_dataset_evaluation_config")
+    @console_ns.response(200, "Evaluation configuration retrieved")
+    @console_ns.response(403, "Permission denied")
+    @console_ns.response(404, "Dataset not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, dataset_id):
+        """Get evaluation configuration for the knowledge base."""
+        current_user, current_tenant_id = current_account_with_tenant()
+        dataset_id_str = str(dataset_id)
+        dataset = DatasetService.get_dataset(dataset_id_str)
+        if dataset is None:
+            raise NotFound("Dataset not found.")
+        try:
+            DatasetService.check_dataset_permission(dataset, current_user)
+        except services.errors.account.NoPermissionError as e:
+            raise Forbidden(str(e))
+
+        with Session(db.engine, expire_on_commit=False) as session:
+            config = EvaluationService.get_evaluation_config(
+                session, current_tenant_id, "dataset", dataset_id_str
+            )
+
+        if config is None:
+            return {
+                "evaluation_model": None,
+                "evaluation_model_provider": None,
+                "default_metrics": None,
+                "customized_metrics": None,
+                "judgment_config": None,
+            }
+
+        return {
+            "evaluation_model": config.evaluation_model,
+            "evaluation_model_provider": config.evaluation_model_provider,
+            "default_metrics": config.default_metrics_list,
+            "customized_metrics": config.customized_metrics_dict,
+            "judgment_config": config.judgment_config_dict,
+        }
+
+    @console_ns.doc("save_dataset_evaluation_config")
+    @console_ns.response(200, "Evaluation configuration saved")
+    @console_ns.response(403, "Permission denied")
+    @console_ns.response(404, "Dataset not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def put(self, dataset_id):
+        """Save evaluation configuration for the knowledge base."""
+        current_user, current_tenant_id = current_account_with_tenant()
+        dataset_id_str = str(dataset_id)
+        dataset = DatasetService.get_dataset(dataset_id_str)
+        if dataset is None:
+            raise NotFound("Dataset not found.")
+        try:
+            DatasetService.check_dataset_permission(dataset, current_user)
+        except services.errors.account.NoPermissionError as e:
+            raise Forbidden(str(e))
+
+        body = request.get_json(force=True)
+        try:
+            config_data = EvaluationConfigData.model_validate(body)
+        except Exception as e:
+            raise BadRequest(f"Invalid request body: {e}")
+
+        with Session(db.engine, expire_on_commit=False) as session:
+            config = EvaluationService.save_evaluation_config(
+                session=session,
+                tenant_id=current_tenant_id,
+                target_type="dataset",
+                target_id=dataset_id_str,
+                account_id=str(current_user.id),
+                data=config_data,
+            )
+
+        return {
+            "evaluation_model": config.evaluation_model,
+            "evaluation_model_provider": config.evaluation_model_provider,
+            "default_metrics": config.default_metrics_list,
+            "customized_metrics": config.customized_metrics_dict,
+            "judgment_config": config.judgment_config_dict,
+        }
+
+
+@console_ns.route("/datasets/<uuid:dataset_id>/evaluation/run")
+class DatasetEvaluationRunApi(Resource):
+    @console_ns.doc("start_dataset_evaluation_run")
+    @console_ns.response(200, "Evaluation run started")
+    @console_ns.response(400, "Invalid request")
+    @console_ns.response(403, "Permission denied")
+    @console_ns.response(404, "Dataset not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, dataset_id):
+        """Start an evaluation run for the knowledge base retrieval."""
+        current_user, current_tenant_id = current_account_with_tenant()
+        dataset_id_str = str(dataset_id)
+        dataset = DatasetService.get_dataset(dataset_id_str)
+        if dataset is None:
+            raise NotFound("Dataset not found.")
+        try:
+            DatasetService.check_dataset_permission(dataset, current_user)
+        except services.errors.account.NoPermissionError as e:
+            raise Forbidden(str(e))
+
+        body = request.get_json(force=True)
+        if not body:
+            raise BadRequest("Request body is required.")
+
+        try:
+            run_request = EvaluationRunRequest.model_validate(body)
+        except Exception as e:
+            raise BadRequest(f"Invalid request body: {e}")
+
+        upload_file = (
+            db.session.query(UploadFile).filter_by(id=run_request.file_id, tenant_id=current_tenant_id).first()
+        )
+        if not upload_file:
+            raise NotFound("Dataset file not found.")
+
+        try:
+            dataset_content = storage.load_once(upload_file.key)
+        except Exception:
+            raise BadRequest("Failed to read dataset file.")
+
+        if not dataset_content:
+            raise BadRequest("Dataset file is empty.")
+
+        try:
+            with Session(db.engine, expire_on_commit=False) as session:
+                evaluation_run = EvaluationService.start_evaluation_run(
+                    session=session,
+                    tenant_id=current_tenant_id,
+                    target_type=EvaluationTargetType.KNOWLEDGE_BASE,
+                    target_id=dataset_id_str,
+                    account_id=str(current_user.id),
+                    dataset_file_content=dataset_content,
+                    run_request=run_request,
+                )
+                return _serialize_dataset_evaluation_run(evaluation_run), 200
+        except EvaluationFrameworkNotConfiguredError as e:
+            return {"message": str(e.description)}, 400
+        except EvaluationNotFoundError as e:
+            return {"message": str(e.description)}, 404
+        except EvaluationMaxConcurrentRunsError as e:
+            return {"message": str(e.description)}, 429
+        except EvaluationDatasetInvalidError as e:
+            return {"message": str(e.description)}, 400
+
+
+@console_ns.route("/datasets/<uuid:dataset_id>/evaluation/logs")
+class DatasetEvaluationLogsApi(Resource):
+    @console_ns.doc("get_dataset_evaluation_logs")
+    @console_ns.response(200, "Evaluation logs retrieved")
+    @console_ns.response(403, "Permission denied")
+    @console_ns.response(404, "Dataset not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, dataset_id):
+        """Get evaluation run history for the knowledge base."""
+        current_user, current_tenant_id = current_account_with_tenant()
+        dataset_id_str = str(dataset_id)
+        dataset = DatasetService.get_dataset(dataset_id_str)
+        if dataset is None:
+            raise NotFound("Dataset not found.")
+        try:
+            DatasetService.check_dataset_permission(dataset, current_user)
+        except services.errors.account.NoPermissionError as e:
+            raise Forbidden(str(e))
+
+        page = request.args.get("page", 1, type=int)
+        page_size = request.args.get("page_size", 20, type=int)
+
+        with Session(db.engine, expire_on_commit=False) as session:
+            runs, total = EvaluationService.get_evaluation_runs(
+                session=session,
+                tenant_id=current_tenant_id,
+                target_type="dataset",
+                target_id=dataset_id_str,
+                page=page,
+                page_size=page_size,
+            )
+
+        return {
+            "data": [_serialize_dataset_evaluation_run(run) for run in runs],
+            "total": total,
+            "page": page,
+            "page_size": page_size,
+        }
+
+
+@console_ns.route("/datasets/<uuid:dataset_id>/evaluation/runs/<uuid:run_id>")
+class DatasetEvaluationRunDetailApi(Resource):
+    @console_ns.doc("get_dataset_evaluation_run_detail")
+    @console_ns.response(200, "Evaluation run detail retrieved")
+    @console_ns.response(403, "Permission denied")
+    @console_ns.response(404, "Dataset or run not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, dataset_id, run_id):
+        """Get evaluation run detail including per-item results."""
+        current_user, current_tenant_id = current_account_with_tenant()
+        dataset_id_str = str(dataset_id)
+        dataset = DatasetService.get_dataset(dataset_id_str)
+        if dataset is None:
+            raise NotFound("Dataset not found.")
+        try:
+            DatasetService.check_dataset_permission(dataset, current_user)
+        except services.errors.account.NoPermissionError as e:
+            raise Forbidden(str(e))
+
+        run_id_str = str(run_id)
+        page = request.args.get("page", 1, type=int)
+        page_size = request.args.get("page_size", 50, type=int)
+
+        try:
+            with Session(db.engine, expire_on_commit=False) as session:
+                run = EvaluationService.get_evaluation_run_detail(
+                    session=session,
+                    tenant_id=current_tenant_id,
+                    run_id=run_id_str,
+                )
+                items, total_items = EvaluationService.get_evaluation_run_items(
+                    session=session,
+                    run_id=run_id_str,
+                    page=page,
+                    page_size=page_size,
+                )
+                return {
+                    "run": _serialize_dataset_evaluation_run(run),
+                    "items": {
+                        "data": [_serialize_dataset_evaluation_run_item(item) for item in items],
+                        "total": total_items,
+                        "page": page,
+                        "page_size": page_size,
+                    },
+                }
+        except EvaluationNotFoundError as e:
+            return {"message": str(e.description)}, 404
+
+
+@console_ns.route("/datasets/<uuid:dataset_id>/evaluation/runs/<uuid:run_id>/cancel")
+class DatasetEvaluationRunCancelApi(Resource):
+    @console_ns.doc("cancel_dataset_evaluation_run")
+    @console_ns.response(200, "Evaluation run cancelled")
+    @console_ns.response(403, "Permission denied")
+    @console_ns.response(404, "Dataset or run not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, dataset_id, run_id):
+        """Cancel a running knowledge base evaluation."""
+        current_user, current_tenant_id = current_account_with_tenant()
+        dataset_id_str = str(dataset_id)
+        dataset = DatasetService.get_dataset(dataset_id_str)
+        if dataset is None:
+            raise NotFound("Dataset not found.")
+        try:
+            DatasetService.check_dataset_permission(dataset, current_user)
+        except services.errors.account.NoPermissionError as e:
+            raise Forbidden(str(e))
+
+        run_id_str = str(run_id)
+        try:
+            with Session(db.engine, expire_on_commit=False) as session:
+                run = EvaluationService.cancel_evaluation_run(
+                    session=session,
+                    tenant_id=current_tenant_id,
+                    run_id=run_id_str,
+                )
+                return _serialize_dataset_evaluation_run(run)
+        except EvaluationNotFoundError as e:
+            return {"message": str(e.description)}, 404
+        except ValueError as e:
+            return {"message": str(e)}, 400
+
+
+@console_ns.route("/datasets/<uuid:dataset_id>/evaluation/metrics")
+class DatasetEvaluationMetricsApi(Resource):
+    @console_ns.doc("get_dataset_evaluation_metrics")
+    @console_ns.response(200, "Available retrieval metrics retrieved")
+    @console_ns.response(403, "Permission denied")
+    @console_ns.response(404, "Dataset not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, dataset_id):
+        """Get available evaluation metrics for knowledge base retrieval."""
+        current_user, _ = current_account_with_tenant()
+        dataset_id_str = str(dataset_id)
+        dataset = DatasetService.get_dataset(dataset_id_str)
+        if dataset is None:
+            raise NotFound("Dataset not found.")
+        try:
+            DatasetService.check_dataset_permission(dataset, current_user)
+        except services.errors.account.NoPermissionError as e:
+            raise Forbidden(str(e))
+
+        return {
+            "metrics": EvaluationService.get_supported_metrics(EvaluationCategory.KNOWLEDGE_BASE)
+        }
+
+
+@console_ns.route("/datasets/<uuid:dataset_id>/evaluation/files/<uuid:file_id>")
+class DatasetEvaluationFileDownloadApi(Resource):
+    @console_ns.doc("download_dataset_evaluation_file")
+    @console_ns.response(200, "File download URL generated")
+    @console_ns.response(403, "Permission denied")
+    @console_ns.response(404, "Dataset or file not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, dataset_id, file_id):
+        """Download evaluation test file or result file for the knowledge base."""
+        from core.workflow.file import helpers as file_helpers
+
+        current_user, current_tenant_id = current_account_with_tenant()
+        dataset_id_str = str(dataset_id)
+        dataset = DatasetService.get_dataset(dataset_id_str)
+        if dataset is None:
+            raise NotFound("Dataset not found.")
+        try:
+            DatasetService.check_dataset_permission(dataset, current_user)
+        except services.errors.account.NoPermissionError as e:
+            raise Forbidden(str(e))
+
+        file_id_str = str(file_id)
+        with Session(db.engine, expire_on_commit=False) as session:
+            stmt = select(UploadFile).where(
+                UploadFile.id == file_id_str,
+                UploadFile.tenant_id == current_tenant_id,
+            )
+            upload_file = session.execute(stmt).scalar_one_or_none()
+
+        if not upload_file:
+            raise NotFound("File not found.")
+
+        download_url = file_helpers.get_signed_file_url(upload_file_id=upload_file.id, as_attachment=True)
+
+        return {
+            "id": upload_file.id,
+            "name": upload_file.name,
+            "size": upload_file.size,
+            "extension": upload_file.extension,
+            "mime_type": upload_file.mime_type,
+            "created_at": int(upload_file.created_at.timestamp()) if upload_file.created_at else None,
+            "download_url": download_url,
+        }

api/controllers/console/datasets/datasets_document.py

@@ -3,19 +3,20 @@ import logging
 from argparse import ArgumentTypeError
 from collections.abc import Sequence
 from contextlib import ExitStack
-from datetime import datetime
 from typing import Any, Literal, cast
 
 import sqlalchemy as sa
 from flask import request, send_file
-from flask_restx import Resource, marshal
+from flask_restx import Resource, fields, marshal, marshal_with
-from pydantic import BaseModel, Field, field_validator
+from graphon.model_runtime.entities.model_entities import ModelType
+from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
+from pydantic import BaseModel, Field
 from sqlalchemy import asc, desc, func, select
 from werkzeug.exceptions import Forbidden, NotFound
 
 import services
 from controllers.common.controller_schemas import DocumentBatchDownloadZipPayload
-from controllers.common.schema import register_schema_models
+from controllers.common.schema import get_or_create_model, register_schema_models
 from controllers.console import console_ns
 from core.errors.error import (
     LLMBadRequestError,
@@ -30,14 +31,14 @@ from core.rag.extractor.entity.datasource_type import DatasourceType
 from core.rag.extractor.entity.extract_setting import ExtractSetting, NotionInfo, WebsiteInfo
 from core.rag.index_processor.constant.index_type import IndexTechniqueType
 from extensions.ext_database import db
-from fields.base import ResponseModel
+from fields.dataset_fields import dataset_fields
 from fields.document_fields import (
+    dataset_and_document_fields,
     document_fields,
+    document_metadata_fields,
     document_status_fields,
     document_with_segments_fields,
 )
-from graphon.model_runtime.entities.model_entities import ModelType
-from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
 from libs.datetime_utils import naive_utc_now
 from libs.login import current_account_with_tenant, login_required
 from models import DatasetProcessRule, Document, DocumentSegment, UploadFile
@@ -71,100 +72,27 @@ from ..wraps import (
 logger = logging.getLogger(__name__)
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
+# Register models for flask_restx to avoid dict type issues in Swagger
-    if isinstance(value, datetime):
+dataset_model = get_or_create_model("Dataset", dataset_fields)
-        return int(value.timestamp())
-    return value
 
+document_metadata_model = get_or_create_model("DocumentMetadata", document_metadata_fields)
 
-def _normalize_enum(value: Any) -> Any:
+document_fields_copy = document_fields.copy()
-    if isinstance(value, str) or value is None:
+document_fields_copy["doc_metadata"] = fields.List(
-        return value
+    fields.Nested(document_metadata_model), attribute="doc_metadata_details"
-    return getattr(value, "value", value)
+)
+document_model = get_or_create_model("Document", document_fields_copy)
 
+document_with_segments_fields_copy = document_with_segments_fields.copy()
+document_with_segments_fields_copy["doc_metadata"] = fields.List(
+    fields.Nested(document_metadata_model), attribute="doc_metadata_details"
+)
+document_with_segments_model = get_or_create_model("DocumentWithSegments", document_with_segments_fields_copy)
 
-class DatasetResponse(ResponseModel):
+dataset_and_document_fields_copy = dataset_and_document_fields.copy()
-    id: str
+dataset_and_document_fields_copy["dataset"] = fields.Nested(dataset_model)
-    name: str
+dataset_and_document_fields_copy["documents"] = fields.List(fields.Nested(document_model))
-    description: str | None = None
+dataset_and_document_model = get_or_create_model("DatasetAndDocument", dataset_and_document_fields_copy)
-    permission: str | None = None
-    data_source_type: str | None = None
-    indexing_technique: str | None = None
-    created_by: str | None = None
-    created_at: int | None = None
-
-    @field_validator("data_source_type", "indexing_technique", mode="before")
-    @classmethod
-    def _normalize_enum_fields(cls, value: Any) -> Any:
-        return _normalize_enum(value)
-
-    @field_validator("created_at", mode="before")
-    @classmethod
-    def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
-
-
-class DocumentMetadataResponse(ResponseModel):
-    id: str
-    name: str
-    type: str
-    value: str | None = None
-
-
-class DocumentResponse(ResponseModel):
-    id: str
-    position: int | None = None
-    data_source_type: str | None = None
-    data_source_info: Any = Field(default=None, validation_alias="data_source_info_dict")
-    data_source_detail_dict: Any = None
-    dataset_process_rule_id: str | None = None
-    name: str
-    created_from: str | None = None
-    created_by: str | None = None
-    created_at: int | None = None
-    tokens: int | None = None
-    indexing_status: str | None = None
-    error: str | None = None
-    enabled: bool | None = None
-    disabled_at: int | None = None
-    disabled_by: str | None = None
-    archived: bool | None = None
-    display_status: str | None = None
-    word_count: int | None = None
-    hit_count: int | None = None
-    doc_form: str | None = None
-    doc_metadata: list[DocumentMetadataResponse] = Field(default_factory=list, validation_alias="doc_metadata_details")
-    summary_index_status: str | None = None
-    need_summary: bool | None = None
-
-    @field_validator("data_source_type", "indexing_status", "display_status", "doc_form", mode="before")
-    @classmethod
-    def _normalize_enum_fields(cls, value: Any) -> Any:
-        return _normalize_enum(value)
-
-    @field_validator("doc_metadata", mode="before")
-    @classmethod
-    def _normalize_doc_metadata(cls, value: Any) -> list[Any]:
-        if value is None:
-            return []
-        return value
-
-    @field_validator("created_at", "disabled_at", mode="before")
-    @classmethod
-    def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
-
-
-class DocumentWithSegmentsResponse(DocumentResponse):
-    process_rule_dict: Any = None
-    completed_segments: int | None = None
-    total_segments: int | None = None
-
-
-class DatasetAndDocumentResponse(ResponseModel):
-    dataset: DatasetResponse
-    documents: list[DocumentResponse]
-    batch: str
 
 
 class DocumentRetryPayload(BaseModel):
@@ -179,11 +107,6 @@ class GenerateSummaryPayload(BaseModel):
     document_list: list[str]
 
 
-class DocumentMetadataUpdatePayload(BaseModel):
-    doc_type: str | None = None
-    doc_metadata: Any = None
-
-
 class DocumentDatasetListParam(BaseModel):
     page: int = Field(1, title="Page", description="Page number.")
     limit: int = Field(20, title="Limit", description="Page size.")
@@ -201,13 +124,7 @@ register_schema_models(
     DocumentRetryPayload,
     DocumentRenamePayload,
     GenerateSummaryPayload,
-    DocumentMetadataUpdatePayload,
     DocumentBatchDownloadZipPayload,
-    DatasetResponse,
-    DocumentMetadataResponse,
-    DocumentResponse,
-    DocumentWithSegmentsResponse,
-    DatasetAndDocumentResponse,
 )
 
 
@@ -440,10 +357,10 @@ class DatasetDocumentListApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @marshal_with(dataset_and_document_model)
     @cloud_edition_billing_resource_check("vector_space")
     @cloud_edition_billing_rate_limit_check("knowledge")
     @console_ns.expect(console_ns.models[KnowledgeConfig.__name__])
-    @console_ns.response(200, "Documents created successfully", console_ns.models[DatasetAndDocumentResponse.__name__])
     def post(self, dataset_id):
         current_user, _ = current_account_with_tenant()
         dataset_id = str(dataset_id)
@@ -481,9 +398,7 @@ class DatasetDocumentListApi(Resource):
         except ModelCurrentlyNotSupportError:
             raise ProviderModelCurrentlyNotSupportError()
 
-        return DatasetAndDocumentResponse.model_validate(
+        return {"dataset": dataset, "documents": documents, "batch": batch}
-            {"dataset": dataset, "documents": documents, "batch": batch}, from_attributes=True
-        ).model_dump(mode="json")
 
     @setup_required
     @login_required
@@ -511,13 +426,12 @@ class DatasetInitApi(Resource):
     @console_ns.doc("init_dataset")
     @console_ns.doc(description="Initialize dataset with documents")
     @console_ns.expect(console_ns.models[KnowledgeConfig.__name__])
-    @console_ns.response(
+    @console_ns.response(201, "Dataset initialized successfully", dataset_and_document_model)
-        201, "Dataset initialized successfully", console_ns.models[DatasetAndDocumentResponse.__name__]
-    )
     @console_ns.response(400, "Invalid request parameters")
     @setup_required
     @login_required
     @account_initialization_required
+    @marshal_with(dataset_and_document_model)
     @cloud_edition_billing_resource_check("vector_space")
     @cloud_edition_billing_rate_limit_check("knowledge")
     def post(self):
@@ -565,9 +479,9 @@ class DatasetInitApi(Resource):
         except ModelCurrentlyNotSupportError:
             raise ProviderModelCurrentlyNotSupportError()
 
-        return DatasetAndDocumentResponse.model_validate(
+        response = {"dataset": dataset, "documents": documents, "batch": batch}
-            {"dataset": dataset, "documents": documents, "batch": batch}, from_attributes=True
+
-        ).model_dump(mode="json")
+        return response
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/indexing-estimate")
@@ -1074,7 +988,15 @@ class DocumentMetadataApi(DocumentResource):
     @console_ns.doc("update_document_metadata")
     @console_ns.doc(description="Update document metadata")
     @console_ns.doc(params={"dataset_id": "Dataset ID", "document_id": "Document ID"})
-    @console_ns.expect(console_ns.models[DocumentMetadataUpdatePayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "UpdateDocumentMetadataRequest",
+            {
+                "doc_type": fields.String(description="Document type"),
+                "doc_metadata": fields.Raw(description="Document metadata"),
+            },
+        )
+    )
     @console_ns.response(200, "Document metadata updated successfully")
     @console_ns.response(404, "Document not found")
     @console_ns.response(403, "Permission denied")
@@ -1087,10 +1009,10 @@ class DocumentMetadataApi(DocumentResource):
         document_id = str(document_id)
         document = self.get_document(dataset_id, document_id)
 
-        req_data = DocumentMetadataUpdatePayload.model_validate(request.get_json() or {})
+        req_data = request.get_json()
 
-        doc_type = req_data.doc_type
+        doc_type = req_data.get("doc_type")
-        doc_metadata = req_data.doc_metadata
+        doc_metadata = req_data.get("doc_metadata")
 
         # The role of the current user in the ta table must be admin, owner, dataset_operator, or editor
         if not current_user.is_dataset_editor:
@@ -1272,7 +1194,7 @@ class DocumentRenameApi(DocumentResource):
     @setup_required
     @login_required
     @account_initialization_required
-    @console_ns.response(200, "Document renamed successfully", console_ns.models[DocumentResponse.__name__])
+    @marshal_with(document_model)
     @console_ns.expect(console_ns.models[DocumentRenamePayload.__name__])
     def post(self, dataset_id, document_id):
         # The role of the current user in the ta table must be admin, owner, editor, or dataset_operator
@@ -1290,7 +1212,7 @@ class DocumentRenameApi(DocumentResource):
         except services.errors.document.DocumentIndexingError:
             raise DocumentIndexingError("Cannot delete document during indexing.")
 
-        return DocumentResponse.model_validate(document, from_attributes=True).model_dump(mode="json")
+        return document
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/website-sync")

api/controllers/console/datasets/datasets_segments.py

@@ -2,6 +2,7 @@ import uuid
 
 from flask import request
 from flask_restx import Resource, marshal
+from graphon.model_runtime.entities.model_entities import ModelType
 from pydantic import BaseModel, Field
 from sqlalchemy import String, cast, func, or_, select
 from sqlalchemy.dialects.postgresql import JSONB
@@ -31,7 +32,6 @@ from core.rag.index_processor.constant.index_type import IndexTechniqueType
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from fields.segment_fields import child_chunk_fields, segment_fields
-from graphon.model_runtime.entities.model_entities import ModelType
 from libs.helper import escape_like_pattern
 from libs.login import current_account_with_tenant, login_required
 from models.dataset import ChildChunk, DocumentSegment

api/controllers/console/datasets/hit_testing_base.py

@@ -2,6 +2,7 @@ import logging
 from typing import Any
 
 from flask_restx import marshal
+from graphon.model_runtime.errors.invoke import InvokeError
 from pydantic import BaseModel, Field
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
 
@@ -20,7 +21,6 @@ from core.errors.error import (
     QuotaExceededError,
 )
 from fields.hit_testing_fields import hit_testing_record_fields
-from graphon.model_runtime.errors.invoke import InvokeError
 from libs.login import current_user
 from models.account import Account
 from services.dataset_service import DatasetService

api/controllers/console/datasets/rag_pipeline/datasource_auth.py

@@ -2,6 +2,8 @@ from typing import Any
 
 from flask import make_response, redirect, request
 from flask_restx import Resource
+from graphon.model_runtime.errors.validate import CredentialsValidateFailedError
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 from pydantic import BaseModel, Field
 from werkzeug.exceptions import Forbidden, NotFound
 
@@ -10,8 +12,6 @@ from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
 from core.plugin.impl.oauth import OAuthHandler
-from graphon.model_runtime.errors.validate import CredentialsValidateFailedError
-from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs.login import current_account_with_tenant, login_required
 from models.provider_ids import DatasourceProviderID
 from services.datasource_provider_service import DatasourceProviderService

api/controllers/console/datasets/rag_pipeline/rag_pipeline_draft_variable.py

@@ -4,6 +4,7 @@ from typing import Any, NoReturn
 
 from flask import Response, request
 from flask_restx import Resource, marshal, marshal_with
+from graphon.variables.types import SegmentType
 from pydantic import BaseModel, Field
 from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import Forbidden
@@ -27,7 +28,6 @@ from core.workflow.variable_prefixes import CONVERSATION_VARIABLE_NODE_ID, SYSTE
 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 import Account
 from models.dataset import Pipeline

api/controllers/console/datasets/rag_pipeline/rag_pipeline_workflow.py

@@ -4,6 +4,7 @@ from typing import Any, Literal, cast
 
 from flask import abort, request
 from flask_restx import Resource, marshal_with  # type: ignore
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 from pydantic import BaseModel, Field, ValidationError
 from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import BadRequest, Forbidden, InternalServerError, NotFound
@@ -40,7 +41,6 @@ from core.app.apps.pipeline.pipeline_generator import PipelineGenerator
 from core.app.entities.app_invoke_entities import InvokeFrom
 from extensions.ext_database import db
 from factories import variable_factory
-from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs import helper
 from libs.helper import TimestampField, UUIDStrOrEmpty
 from libs.login import current_account_with_tenant, current_user, login_required

api/controllers/console/evaluation/__init__.py

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

api/controllers/console/evaluation/evaluation.py

@@ -0,0 +1,871 @@
+from __future__ import annotations
+
+import logging
+from collections.abc import Callable
+from functools import wraps
+from typing import TYPE_CHECKING, ParamSpec, TypeVar, Union
+from urllib.parse import quote
+
+from flask import Response, request
+from flask_restx import Resource, fields, marshal
+from pydantic import BaseModel
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import BadRequest, Forbidden, NotFound
+
+from controllers.common.schema import register_schema_models
+from controllers.console import console_ns
+from controllers.console.app.workflow import WorkflowListQuery
+from controllers.console.wraps import (
+    account_initialization_required,
+    edit_permission_required,
+    setup_required,
+)
+from core.evaluation.entities.evaluation_entity import EvaluationCategory, EvaluationConfigData, EvaluationRunRequest
+from extensions.ext_database import db
+from extensions.ext_storage import storage
+from fields.member_fields import simple_account_fields
+from graphon.file import helpers as file_helpers
+from libs.helper import TimestampField
+from libs.login import current_account_with_tenant, login_required
+from models import App, Dataset
+from models.model import UploadFile
+from models.snippet import CustomizedSnippet
+from services.errors.evaluation import (
+    EvaluationDatasetInvalidError,
+    EvaluationFrameworkNotConfiguredError,
+    EvaluationMaxConcurrentRunsError,
+    EvaluationNotFoundError,
+)
+from services.evaluation_service import EvaluationService
+from services.workflow_service import WorkflowService
+
+if TYPE_CHECKING:
+    from models.evaluation import EvaluationRun, EvaluationRunItem
+
+logger = logging.getLogger(__name__)
+
+P = ParamSpec("P")
+R = TypeVar("R")
+
+# Valid evaluation target types
+EVALUATE_TARGET_TYPES = {"app", "snippets"}
+
+
+class VersionQuery(BaseModel):
+    """Query parameters for version endpoint."""
+
+    version: str
+
+
+register_schema_models(
+    console_ns,
+    VersionQuery,
+)
+
+
+# Response field definitions
+file_info_fields = {
+    "id": fields.String,
+    "name": fields.String,
+}
+
+evaluation_log_fields = {
+    "created_at": TimestampField,
+    "created_by": fields.String,
+    "test_file": fields.Nested(
+        console_ns.model(
+            "EvaluationTestFile",
+            file_info_fields,
+        )
+    ),
+    "result_file": fields.Nested(
+        console_ns.model(
+            "EvaluationResultFile",
+            file_info_fields,
+        ),
+        allow_null=True,
+    ),
+    "version": fields.String,
+}
+
+evaluation_log_list_model = console_ns.model(
+    "EvaluationLogList",
+    {
+        "data": fields.List(fields.Nested(console_ns.model("EvaluationLog", evaluation_log_fields))),
+    },
+)
+
+evaluation_default_metric_node_info_fields = {
+    "node_id": fields.String,
+    "type": fields.String,
+    "title": fields.String,
+}
+evaluation_default_metric_item_fields = {
+    "metric": fields.String,
+    "value_type": fields.String,
+    "node_info_list": fields.List(
+        fields.Nested(
+            console_ns.model("EvaluationDefaultMetricNodeInfo", evaluation_default_metric_node_info_fields),
+        ),
+    ),
+}
+
+customized_metrics_fields = {
+    "evaluation_workflow_id": fields.String,
+    "input_fields": fields.Raw,
+    "output_fields": fields.Raw,
+}
+
+judgment_condition_fields = {
+    "variable_selector": fields.List(fields.String),
+    "comparison_operator": fields.String,
+    "value": fields.String,
+}
+
+judgment_config_fields = {
+    "logical_operator": fields.String,
+    "conditions": fields.List(fields.Nested(console_ns.model("JudgmentCondition", judgment_condition_fields))),
+}
+
+evaluation_detail_fields = {
+    "evaluation_model": fields.String,
+    "evaluation_model_provider": fields.String,
+    "default_metrics": fields.List(
+        fields.Nested(console_ns.model("EvaluationDefaultMetricItem_Detail", evaluation_default_metric_item_fields)),
+        allow_null=True,
+    ),
+    "customized_metrics": fields.Nested(
+        console_ns.model("EvaluationCustomizedMetrics", customized_metrics_fields),
+        allow_null=True,
+    ),
+    "judgment_config": fields.Nested(
+        console_ns.model("EvaluationJudgmentConfig", judgment_config_fields),
+        allow_null=True,
+    ),
+}
+
+evaluation_detail_model = console_ns.model("EvaluationDetail", evaluation_detail_fields)
+
+available_evaluation_workflow_list_fields = {
+    "id": fields.String,
+    "app_id": fields.String,
+    "app_name": fields.String,
+    "type": fields.String,
+    "kind": fields.String,
+    "version": fields.String,
+    "marked_name": fields.String,
+    "marked_comment": fields.String,
+    "hash": fields.String,
+    "created_by": fields.Nested(simple_account_fields),
+    "created_at": TimestampField,
+    "updated_by": fields.Nested(simple_account_fields, allow_null=True),
+    "updated_at": TimestampField,
+}
+
+available_evaluation_workflow_pagination_fields = {
+    "items": fields.List(fields.Nested(available_evaluation_workflow_list_fields)),
+    "page": fields.Integer,
+    "limit": fields.Integer,
+    "has_more": fields.Boolean,
+}
+
+available_evaluation_workflow_pagination_model = console_ns.model(
+    "AvailableEvaluationWorkflowPagination",
+    available_evaluation_workflow_pagination_fields,
+)
+
+evaluation_default_metrics_response_model = console_ns.model(
+    "EvaluationDefaultMetricsResponse",
+    {
+        "default_metrics": fields.List(
+            fields.Nested(console_ns.model("EvaluationDefaultMetricItem", evaluation_default_metric_item_fields)),
+        ),
+    },
+)
+
+
+def get_evaluation_target(view_func: Callable[P, R]):
+    """
+    Decorator to resolve polymorphic evaluation target (app or snippet).
+
+    Validates the target_type parameter and fetches the corresponding
+    model (App or CustomizedSnippet) with tenant isolation.
+    """
+
+    @wraps(view_func)
+    def decorated_view(*args: P.args, **kwargs: P.kwargs):
+        target_type = kwargs.get("evaluate_target_type")
+        target_id = kwargs.get("evaluate_target_id")
+
+        if target_type not in EVALUATE_TARGET_TYPES:
+            raise NotFound(f"Invalid evaluation target type: {target_type}")
+
+        _, current_tenant_id = current_account_with_tenant()
+
+        target_id = str(target_id)
+
+        # Remove path parameters
+        del kwargs["evaluate_target_type"]
+        del kwargs["evaluate_target_id"]
+
+        target: Union[App, CustomizedSnippet, Dataset] | None = None
+
+        if target_type == "app":
+            target = db.session.query(App).where(App.id == target_id, App.tenant_id == current_tenant_id).first()
+        elif target_type == "snippets":
+            target = (
+                db.session.query(CustomizedSnippet)
+                .where(CustomizedSnippet.id == target_id, CustomizedSnippet.tenant_id == current_tenant_id)
+                .first()
+            )
+        elif target_type == "knowledge":
+            target = (db.session.query(Dataset)
+                      .where(Dataset.id == target_id, Dataset.tenant_id == current_tenant_id)
+                      .first())
+
+        if not target:
+            raise NotFound(f"{str(target_type)} not found")
+
+        kwargs["target"] = target
+        kwargs["target_type"] = target_type
+
+        return view_func(*args, **kwargs)
+
+    return decorated_view
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/dataset-template/download")
+class EvaluationDatasetTemplateDownloadApi(Resource):
+    @console_ns.doc("download_evaluation_dataset_template")
+    @console_ns.response(200, "Template file streamed as XLSX attachment")
+    @console_ns.response(400, "Invalid target type or excluded app mode")
+    @console_ns.response(404, "Target not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    @edit_permission_required
+    def post(self, target: Union[App, CustomizedSnippet], target_type: str):
+        """
+        Download evaluation dataset template.
+
+        Generates an XLSX template based on the target's input parameters
+        and streams it directly as a file attachment.
+        """
+        try:
+            xlsx_content, filename = EvaluationService.generate_dataset_template(
+                target=target,
+                target_type=target_type,
+            )
+        except ValueError as e:
+            return {"message": str(e)}, 400
+
+        encoded_filename = quote(filename)
+        response = Response(
+            xlsx_content,
+            mimetype="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+        )
+        response.headers["Content-Disposition"] = f"attachment; filename*=UTF-8''{encoded_filename}"
+        response.headers["Content-Length"] = str(len(xlsx_content))
+        return response
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation")
+class EvaluationDetailApi(Resource):
+    @console_ns.doc("get_evaluation_detail")
+    @console_ns.response(200, "Evaluation details retrieved successfully", evaluation_detail_model)
+    @console_ns.response(404, "Target not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    def get(self, target: Union[App, CustomizedSnippet], target_type: str):
+        """
+        Get evaluation configuration for the target.
+
+        Returns evaluation configuration including model settings,
+        metrics config, and judgement conditions.
+        """
+        _, current_tenant_id = current_account_with_tenant()
+
+        with Session(db.engine, expire_on_commit=False) as session:
+            config = EvaluationService.get_evaluation_config(session, current_tenant_id, target_type, str(target.id))
+
+        if config is None:
+            return {
+                "evaluation_model": None,
+                "evaluation_model_provider": None,
+                "default_metrics": None,
+                "customized_metrics": None,
+                "judgment_config": None,
+            }
+
+        return {
+            "evaluation_model": config.evaluation_model,
+            "evaluation_model_provider": config.evaluation_model_provider,
+            "default_metrics": config.default_metrics_list,
+            "customized_metrics": config.customized_metrics_dict,
+            "judgment_config": config.judgment_config_dict,
+        }
+
+    @console_ns.doc("save_evaluation_detail")
+    @console_ns.response(200, "Evaluation configuration saved successfully")
+    @console_ns.response(404, "Target not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    @edit_permission_required
+    def put(self, target: Union[App, CustomizedSnippet], target_type: str):
+        """
+        Save evaluation configuration for the target.
+        """
+        current_account, current_tenant_id = current_account_with_tenant()
+        body = request.get_json(force=True)
+
+        try:
+            config_data = EvaluationConfigData.model_validate(body)
+        except Exception as e:
+            raise BadRequest(f"Invalid request body: {e}")
+
+        with Session(db.engine, expire_on_commit=False) as session:
+            config = EvaluationService.save_evaluation_config(
+                session=session,
+                tenant_id=current_tenant_id,
+                target_type=target_type,
+                target_id=str(target.id),
+                account_id=str(current_account.id),
+                data=config_data,
+            )
+
+        return {
+            "evaluation_model": config.evaluation_model,
+            "evaluation_model_provider": config.evaluation_model_provider,
+            "default_metrics": config.default_metrics_list,
+            "customized_metrics": config.customized_metrics_dict,
+            "judgment_config": config.judgment_config_dict,
+        }
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation/logs")
+class EvaluationLogsApi(Resource):
+    @console_ns.doc("get_evaluation_logs")
+    @console_ns.response(200, "Evaluation logs retrieved successfully")
+    @console_ns.response(404, "Target not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    def get(self, target: Union[App, CustomizedSnippet], target_type: str):
+        """
+        Get evaluation run history for the target.
+
+        Returns a paginated list of evaluation runs.
+        """
+        _, current_tenant_id = current_account_with_tenant()
+        page = request.args.get("page", 1, type=int)
+        page_size = request.args.get("page_size", 20, type=int)
+
+        with Session(db.engine, expire_on_commit=False) as session:
+            runs, total = EvaluationService.get_evaluation_runs(
+                session=session,
+                tenant_id=current_tenant_id,
+                target_type=target_type,
+                target_id=str(target.id),
+                page=page,
+                page_size=page_size,
+            )
+
+        return {
+            "data": [_serialize_evaluation_run(run) for run in runs],
+            "total": total,
+            "page": page,
+            "page_size": page_size,
+        }
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation/run")
+class EvaluationRunApi(Resource):
+    @console_ns.doc("start_evaluation_run")
+    @console_ns.response(200, "Evaluation run started")
+    @console_ns.response(400, "Invalid request")
+    @console_ns.response(404, "Target not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    @edit_permission_required
+    def post(self, target: Union[App, CustomizedSnippet, Dataset], target_type: str):
+        """
+        Start an evaluation run.
+
+        Expects JSON body with:
+        - file_id: uploaded dataset file ID
+        - evaluation_model: evaluation model name
+        - evaluation_model_provider: evaluation model provider
+        - default_metrics: list of default metric objects
+        - customized_metrics: customized metrics object (optional)
+        - judgment_config: judgment conditions config (optional)
+        """
+        current_account, current_tenant_id = current_account_with_tenant()
+
+        body = request.get_json(force=True)
+        if not body:
+            raise BadRequest("Request body is required.")
+
+        # Validate and parse request body
+        try:
+            run_request = EvaluationRunRequest.model_validate(body)
+        except Exception as e:
+            raise BadRequest(f"Invalid request body: {e}")
+
+        # Load dataset file
+        upload_file = (
+            db.session.query(UploadFile).filter_by(id=run_request.file_id, tenant_id=current_tenant_id).first()
+        )
+        if not upload_file:
+            raise NotFound("Dataset file not found.")
+
+        try:
+            dataset_content = storage.load_once(upload_file.key)
+        except Exception:
+            raise BadRequest("Failed to read dataset file.")
+
+        if not dataset_content:
+            raise BadRequest("Dataset file is empty.")
+
+        try:
+            with Session(db.engine, expire_on_commit=False) as session:
+                evaluation_run = EvaluationService.start_evaluation_run(
+                    session=session,
+                    tenant_id=current_tenant_id,
+                    target_type=target_type,
+                    target_id=str(target.id),
+                    account_id=str(current_account.id),
+                    dataset_file_content=dataset_content,
+                    run_request=run_request,
+                )
+                return _serialize_evaluation_run(evaluation_run), 200
+        except EvaluationFrameworkNotConfiguredError as e:
+            return {"message": str(e.description)}, 400
+        except EvaluationNotFoundError as e:
+            return {"message": str(e.description)}, 404
+        except EvaluationMaxConcurrentRunsError as e:
+            return {"message": str(e.description)}, 429
+        except EvaluationDatasetInvalidError as e:
+            return {"message": str(e.description)}, 400
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation/runs/<uuid:run_id>")
+class EvaluationRunDetailApi(Resource):
+    @console_ns.doc("get_evaluation_run_detail")
+    @console_ns.response(200, "Evaluation run detail retrieved")
+    @console_ns.response(404, "Run not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    def get(self, target: Union[App, CustomizedSnippet], target_type: str, run_id: str):
+        """
+        Get evaluation run detail including items.
+        """
+        _, current_tenant_id = current_account_with_tenant()
+        run_id = str(run_id)
+        page = request.args.get("page", 1, type=int)
+        page_size = request.args.get("page_size", 50, type=int)
+
+        try:
+            with Session(db.engine, expire_on_commit=False) as session:
+                run = EvaluationService.get_evaluation_run_detail(
+                    session=session,
+                    tenant_id=current_tenant_id,
+                    run_id=run_id,
+                )
+                items, total_items = EvaluationService.get_evaluation_run_items(
+                    session=session,
+                    run_id=run_id,
+                    page=page,
+                    page_size=page_size,
+                )
+
+                return {
+                    "run": _serialize_evaluation_run(run),
+                    "items": {
+                        "data": [_serialize_evaluation_run_item(item) for item in items],
+                        "total": total_items,
+                        "page": page,
+                        "page_size": page_size,
+                    },
+                }
+        except EvaluationNotFoundError as e:
+            return {"message": str(e.description)}, 404
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation/runs/<uuid:run_id>/cancel")
+class EvaluationRunCancelApi(Resource):
+    @console_ns.doc("cancel_evaluation_run")
+    @console_ns.response(200, "Evaluation run cancelled")
+    @console_ns.response(404, "Run not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    @edit_permission_required
+    def post(self, target: Union[App, CustomizedSnippet], target_type: str, run_id: str):
+        """Cancel a running evaluation."""
+        _, current_tenant_id = current_account_with_tenant()
+        run_id = str(run_id)
+
+        try:
+            with Session(db.engine, expire_on_commit=False) as session:
+                run = EvaluationService.cancel_evaluation_run(
+                    session=session,
+                    tenant_id=current_tenant_id,
+                    run_id=run_id,
+                )
+                return _serialize_evaluation_run(run)
+        except EvaluationNotFoundError as e:
+            return {"message": str(e.description)}, 404
+        except ValueError as e:
+            return {"message": str(e)}, 400
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation/metrics")
+class EvaluationMetricsApi(Resource):
+    @console_ns.doc("get_evaluation_metrics")
+    @console_ns.response(200, "Available metrics retrieved")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    def get(self, target: Union[App, CustomizedSnippet], target_type: str):
+        """
+        Get available evaluation metrics for the current framework.
+        """
+        result = {}
+        for category in EvaluationCategory:
+            result[category.value] = EvaluationService.get_supported_metrics(category)
+        return {"metrics": result}
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation/default-metrics")
+class EvaluationDefaultMetricsApi(Resource):
+    @console_ns.doc(
+        "get_evaluation_default_metrics_with_nodes",
+        description=(
+            "List default metrics supported by the current evaluation framework with matching nodes "
+            "from the target's published workflow only (draft is ignored)."
+        ),
+    )
+    @console_ns.response(
+        200,
+        "Default metrics and node candidates for the published workflow",
+        evaluation_default_metrics_response_model,
+    )
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    def get(self, target: Union[App, CustomizedSnippet], target_type: str):
+        default_metrics = EvaluationService.get_default_metrics_with_nodes_for_published_target(
+            target=target,
+            target_type=target_type,
+        )
+        return {"default_metrics": [m.model_dump() for m in default_metrics]}
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation/node-info")
+class EvaluationNodeInfoApi(Resource):
+    @console_ns.doc("get_evaluation_node_info")
+    @console_ns.response(200, "Node info grouped by metric")
+    @console_ns.response(404, "Target not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    def post(self, target: Union[App, CustomizedSnippet], target_type: str):
+        """Return workflow/snippet node info grouped by requested metrics.
+
+        Request body (JSON):
+            - metrics: list[str] | None  – metric names to query; omit or pass
+              an empty list to get all nodes under key ``"all"``.
+
+        Response:
+            ``{metric_or_all: [{"node_id": ..., "type": ..., "title": ...}, ...]}``
+        """
+        body = request.get_json(silent=True) or {}
+        metrics: list[str] | None = body.get("metrics") or None
+
+        result = EvaluationService.get_nodes_for_metrics(
+            target=target,
+            target_type=target_type,
+            metrics=metrics,
+        )
+        return result
+
+
+@console_ns.route("/evaluation/available-metrics")
+class EvaluationAvailableMetricsApi(Resource):
+    @console_ns.doc("get_available_evaluation_metrics")
+    @console_ns.response(200, "Available metrics list")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self):
+        """Return the centrally-defined list of evaluation metrics."""
+        return {"metrics": EvaluationService.get_available_metrics()}
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation/files/<uuid:file_id>")
+class EvaluationFileDownloadApi(Resource):
+    @console_ns.doc("download_evaluation_file")
+    @console_ns.response(200, "File download URL generated successfully")
+    @console_ns.response(404, "Target or file not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    def get(self, target: Union[App, CustomizedSnippet], target_type: str, file_id: str):
+        """
+        Download evaluation test file or result file.
+
+        Looks up the specified file, verifies it belongs to the same tenant,
+        and returns file info and download URL.
+        """
+        file_id = str(file_id)
+        _, current_tenant_id = current_account_with_tenant()
+
+        with Session(db.engine, expire_on_commit=False) as session:
+            stmt = select(UploadFile).where(
+                UploadFile.id == file_id,
+                UploadFile.tenant_id == current_tenant_id,
+            )
+            upload_file = session.execute(stmt).scalar_one_or_none()
+
+        if not upload_file:
+            raise NotFound("File not found")
+
+        download_url = file_helpers.get_signed_file_url(upload_file_id=upload_file.id, as_attachment=True)
+
+        return {
+            "id": upload_file.id,
+            "name": upload_file.name,
+            "size": upload_file.size,
+            "extension": upload_file.extension,
+            "mime_type": upload_file.mime_type,
+            "created_at": int(upload_file.created_at.timestamp()) if upload_file.created_at else None,
+            "download_url": download_url,
+        }
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation/version")
+class EvaluationVersionApi(Resource):
+    @console_ns.doc("get_evaluation_version_detail")
+    @console_ns.expect(console_ns.models.get(VersionQuery.__name__))
+    @console_ns.response(200, "Version details retrieved successfully")
+    @console_ns.response(404, "Target or version not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    def get(self, target: Union[App, CustomizedSnippet], target_type: str):
+        """
+        Get evaluation target version details.
+
+        Returns the workflow graph for the specified version.
+        """
+        version = request.args.get("version")
+
+        if not version:
+            return {"message": "version parameter is required"}, 400
+
+        graph = {}
+        if target_type == "snippets" and isinstance(target, CustomizedSnippet):
+            graph = target.graph_dict
+
+        return {
+            "graph": graph,
+        }
+
+
+@console_ns.route("/workspaces/current/available-evaluation-workflows")
+class AvailableEvaluationWorkflowsApi(Resource):
+    @console_ns.expect(console_ns.models[WorkflowListQuery.__name__])
+    @console_ns.doc("list_available_evaluation_workflows")
+    @console_ns.doc(description="List published evaluation workflows in the current workspace (all apps)")
+    @console_ns.response(
+        200,
+        "Available evaluation workflows retrieved",
+        available_evaluation_workflow_pagination_model,
+    )
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def get(self):
+        """List published evaluation-type workflows for the current tenant (cross-app)."""
+        current_user, current_tenant_id = current_account_with_tenant()
+
+        args = WorkflowListQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        page = args.page
+        limit = args.limit
+        user_id = args.user_id
+        named_only = args.named_only
+        keyword = args.keyword
+
+        if user_id and user_id != current_user.id:
+            raise Forbidden()
+
+        workflow_service = WorkflowService()
+        with Session(db.engine) as session:
+            workflows, has_more = workflow_service.list_published_evaluation_workflows(
+                session=session,
+                tenant_id=current_tenant_id,
+                page=page,
+                limit=limit,
+                user_id=user_id,
+                named_only=named_only,
+                keyword=keyword,
+            )
+
+            app_ids = {w.app_id for w in workflows}
+            if app_ids:
+                apps = session.scalars(select(App).where(App.id.in_(app_ids))).all()
+                app_names = {a.id: a.name for a in apps}
+            else:
+                app_names = {}
+
+        items = []
+        for wf in workflows:
+            items.append(
+                {
+                    "id": wf.id,
+                    "app_id": wf.app_id,
+                    "app_name": app_names.get(wf.app_id, ""),
+                    "type": wf.type.value,
+                    "kind": wf.kind_or_standard,
+                    "version": wf.version,
+                    "marked_name": wf.marked_name,
+                    "marked_comment": wf.marked_comment,
+                    "hash": wf.unique_hash,
+                    "created_by": wf.created_by_account,
+                    "created_at": wf.created_at,
+                    "updated_by": wf.updated_by_account,
+                    "updated_at": wf.updated_at,
+                }
+            )
+
+        return (
+            marshal(
+                {"items": items, "page": page, "limit": limit, "has_more": has_more},
+                available_evaluation_workflow_pagination_fields,
+            ),
+            200,
+        )
+
+
+@console_ns.route("/workspaces/current/evaluation-workflows/<string:workflow_id>/associated-targets")
+class EvaluationWorkflowAssociatedTargetsApi(Resource):
+    @console_ns.doc("list_evaluation_workflow_associated_targets")
+    @console_ns.doc(
+        description="List targets (apps / snippets / knowledge bases) that use the given workflow as customized metrics"
+    )
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def get(self, workflow_id: str):
+        """Return all evaluation targets that reference this workflow as customized metrics."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        with Session(db.engine) as session:
+            configs = EvaluationService.list_targets_by_customized_workflow(
+                session=session,
+                tenant_id=current_tenant_id,
+                customized_workflow_id=workflow_id,
+            )
+
+            target_ids_by_type: dict[str, list[str]] = {}
+            for cfg in configs:
+                target_ids_by_type.setdefault(cfg.target_type, []).append(cfg.target_id)
+
+            app_names: dict[str, str] = {}
+            if "app" in target_ids_by_type:
+                apps = session.scalars(select(App).where(App.id.in_(target_ids_by_type["app"]))).all()
+                app_names = {a.id: a.name for a in apps}
+
+            snippet_names: dict[str, str] = {}
+            if "snippets" in target_ids_by_type:
+                snippets = session.scalars(
+                    select(CustomizedSnippet).where(CustomizedSnippet.id.in_(target_ids_by_type["snippets"]))
+                ).all()
+                snippet_names = {s.id: s.name for s in snippets}
+
+            dataset_names: dict[str, str] = {}
+            if "knowledge_base" in target_ids_by_type:
+                datasets = session.scalars(
+                    select(Dataset).where(Dataset.id.in_(target_ids_by_type["knowledge_base"]))
+                ).all()
+                dataset_names = {d.id: d.name for d in datasets}
+
+        items = []
+        for cfg in configs:
+            name = ""
+            if cfg.target_type == "app":
+                name = app_names.get(cfg.target_id, "")
+            elif cfg.target_type == "snippets":
+                name = snippet_names.get(cfg.target_id, "")
+            elif cfg.target_type == "knowledge_base":
+                name = dataset_names.get(cfg.target_id, "")
+
+            items.append(
+                {
+                    "target_type": cfg.target_type,
+                    "target_id": cfg.target_id,
+                    "target_name": name,
+                }
+            )
+
+        return {"items": items}, 200
+
+
+# ---- Serialization Helpers ----
+
+
+def _serialize_evaluation_run(run: EvaluationRun) -> dict[str, object]:
+    return {
+        "id": run.id,
+        "tenant_id": run.tenant_id,
+        "target_type": run.target_type,
+        "target_id": run.target_id,
+        "evaluation_config_id": run.evaluation_config_id,
+        "status": run.status,
+        "dataset_file_id": run.dataset_file_id,
+        "result_file_id": run.result_file_id,
+        "total_items": run.total_items,
+        "completed_items": run.completed_items,
+        "failed_items": run.failed_items,
+        "progress": run.progress,
+        "metrics_summary": run.metrics_summary_dict,
+        "error": run.error,
+        "created_by": run.created_by,
+        "started_at": int(run.started_at.timestamp()) if run.started_at else None,
+        "completed_at": int(run.completed_at.timestamp()) if run.completed_at else None,
+        "created_at": int(run.created_at.timestamp()) if run.created_at else None,
+    }
+
+
+def _serialize_evaluation_run_item(item: EvaluationRunItem) -> dict[str, object]:
+    return {
+        "id": item.id,
+        "item_index": item.item_index,
+        "inputs": item.inputs_dict,
+        "expected_output": item.expected_output,
+        "actual_output": item.actual_output,
+        "metrics": item.metrics_list,
+        "judgment": item.judgment_dict,
+        "metadata": item.metadata_dict,
+        "error": item.error,
+        "overall_score": item.overall_score,
+    }

api/controllers/console/explore/audio.py

@@ -1,6 +1,7 @@
 import logging
 
 from flask import request
+from graphon.model_runtime.errors.invoke import InvokeError
 from werkzeug.exceptions import InternalServerError
 
 import services
@@ -19,7 +20,6 @@ from controllers.console.app.error import (
 )
 from controllers.console.explore.wraps import InstalledAppResource
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
-from graphon.model_runtime.errors.invoke import InvokeError
 from services.audio_service import AudioService
 from services.errors.audio import (
     AudioTooLargeServiceError,

api/controllers/console/explore/completion.py

@@ -2,6 +2,7 @@ import logging
 from typing import Any, Literal
 from uuid import UUID
 
+from graphon.model_runtime.errors.invoke import InvokeError
 from pydantic import BaseModel, Field, field_validator
 from werkzeug.exceptions import InternalServerError, NotFound
 
@@ -25,7 +26,6 @@ from core.errors.error import (
     QuotaExceededError,
 )
 from extensions.ext_database import db
-from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from libs.datetime_utils import naive_utc_now
 from libs.login import current_user

api/controllers/console/explore/message.py

@@ -2,6 +2,7 @@ import logging
 from typing import Literal
 
 from flask import request
+from graphon.model_runtime.errors.invoke import InvokeError
 from pydantic import BaseModel, TypeAdapter
 from werkzeug.exceptions import InternalServerError, NotFound
 
@@ -24,7 +25,6 @@ from core.app.entities.app_invoke_entities import InvokeFrom
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
 from fields.conversation_fields import ResultResponse
 from fields.message_fields import MessageInfiniteScrollPagination, MessageListItem, SuggestedQuestionsResponse
-from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from libs.login import current_account_with_tenant
 from models.enums import FeedbackRating

api/controllers/console/explore/trial.py

@@ -3,6 +3,8 @@ from typing import Any, Literal, cast
 
 from flask import request
 from flask_restx import Resource, fields, marshal, marshal_with
+from graphon.graph_engine.manager import GraphEngineManager
+from graphon.model_runtime.errors.invoke import InvokeError
 from pydantic import BaseModel
 from sqlalchemy import select
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
@@ -59,8 +61,6 @@ from fields.workflow_fields import (
     workflow_fields,
     workflow_partial_fields,
 )
-from graphon.graph_engine.manager import GraphEngineManager
-from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from libs.helper import uuid_value
 from libs.login import current_user

api/controllers/console/explore/workflow.py

@@ -1,5 +1,7 @@
 import logging
 
+from graphon.graph_engine.manager import GraphEngineManager
+from graphon.model_runtime.errors.invoke import InvokeError
 from werkzeug.exceptions import InternalServerError
 
 from controllers.common.controller_schemas import WorkflowRunPayload
@@ -21,8 +23,6 @@ from core.errors.error import (
     QuotaExceededError,
 )
 from extensions.ext_redis import redis_client
-from graphon.graph_engine.manager import GraphEngineManager
-from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from libs.login import current_account_with_tenant
 from models.model import AppMode, InstalledApp

api/controllers/console/remote_files.py

@@ -2,6 +2,7 @@ import urllib.parse
 
 import httpx
 from flask_restx import Resource
+from graphon.file import helpers as file_helpers
 from pydantic import BaseModel, Field
 
 import services
@@ -15,7 +16,6 @@ from controllers.console import console_ns
 from core.helper import ssrf_proxy
 from extensions.ext_database import db
 from fields.file_fields import FileWithSignedUrl, RemoteFileInfo
-from graphon.file import helpers as file_helpers
 from libs.login import current_account_with_tenant, login_required
 from services.file_service import FileService
 

api/controllers/console/snippets/payloads.py

@@ -0,0 +1,142 @@
+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")
+
+    @field_validator("creators", mode="before")
+    @classmethod
+    def parse_creators(cls, value: object) -> list[str] | None:
+        """Normalize creators filter from query string or list input."""
+        if value is None:
+            return None
+        if isinstance(value, str):
+            return [creator.strip() for creator in value.split(",") if creator.strip()] or None
+        if isinstance(value, list):
+            return [str(creator).strip() for creator in value if str(creator).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,579 @@
+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, fields, marshal, marshal_with
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import InternalServerError, NotFound
+
+from controllers.common.schema import register_schema_models
+from controllers.console import console_ns
+from controllers.console.app.error import DraftWorkflowNotExist, DraftWorkflowNotSync
+from controllers.console.app.workflow import workflow_model, workflow_pagination_model
+from controllers.console.app.workflow_run import (
+    workflow_run_detail_model,
+    workflow_run_node_execution_list_model,
+    workflow_run_node_execution_model,
+    workflow_run_pagination_model,
+)
+from controllers.console.snippets.payloads import (
+    PublishWorkflowPayload,
+    SnippetDraftNodeRunPayload,
+    SnippetDraftRunPayload,
+    SnippetDraftSyncPayload,
+    SnippetIterationNodeRunPayload,
+    SnippetLoopNodeRunPayload,
+    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 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 WorkflowHashNotEqualError
+from services.snippet_generate_service import SnippetGenerateService
+from services.snippet_service import SnippetService
+
+logger = logging.getLogger(__name__)
+
+P = ParamSpec("P")
+R = TypeVar("R")
+
+# Register Pydantic models with Swagger
+register_schema_models(
+    console_ns,
+    SnippetDraftSyncPayload,
+    SnippetDraftNodeRunPayload,
+    SnippetDraftRunPayload,
+    SnippetIterationNodeRunPayload,
+    SnippetLoopNodeRunPayload,
+    SnippetWorkflowListQuery,
+    WorkflowRunQuery,
+    PublishWorkflowPayload,
+)
+
+
+snippet_workflow_model = console_ns.clone("SnippetWorkflow", workflow_model, {
+    "input_fields": fields.Raw(default=[]),
+})
+
+
+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", snippet_workflow_model)
+    @console_ns.response(404, "Snippet or draft workflow not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    @marshal_with(snippet_workflow_model)
+    def get(self, snippet: CustomizedSnippet):
+        """Get draft workflow for snippet."""
+        snippet_service = SnippetService()
+        workflow = snippet_service.get_draft_workflow(snippet=snippet)
+
+        if not workflow:
+            raise DraftWorkflowNotExist()
+
+        db.session.expunge(workflow)
+        workflow.conversation_variables = []
+        workflow.input_fields = snippet.input_fields_list
+        return workflow
+
+    @console_ns.doc("sync_snippet_draft_workflow")
+    @console_ns.expect(console_ns.models.get(SnippetDraftSyncPayload.__name__))
+    @console_ns.response(200, "Draft workflow synced successfully")
+    @console_ns.response(400, "Hash mismatch")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet):
+        """Sync draft workflow for snippet."""
+        current_user, _ = current_account_with_tenant()
+
+        payload = SnippetDraftSyncPayload.model_validate(console_ns.payload or {})
+
+        try:
+            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", snippet_workflow_model)
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    @marshal_with(snippet_workflow_model)
+    def get(self, snippet: CustomizedSnippet):
+        """Get published workflow for snippet."""
+        if not snippet.is_published:
+            return None
+
+        snippet_service = SnippetService()
+        workflow = snippet_service.get_published_workflow(snippet=snippet)
+
+        if workflow:
+            workflow.input_fields = snippet.input_fields_list
+
+        return workflow
+
+    @console_ns.doc("publish_snippet_workflow")
+    @console_ns.expect(console_ns.models.get(PublishWorkflowPayload.__name__))
+    @console_ns.response(200, "Workflow published successfully")
+    @console_ns.response(400, "No draft workflow found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet):
+        """Publish snippet workflow."""
+        current_user, _ = current_account_with_tenant()
+        snippet_service = SnippetService()
+
+        with Session(db.engine) as session:
+            snippet = session.merge(snippet)
+            try:
+                workflow = snippet_service.publish_workflow(
+                    session=session,
+                    snippet=snippet,
+                    account=current_user,
+                )
+                workflow_created_at = TimestampField().format(workflow.created_at)
+                session.commit()
+            except ValueError as e:
+                return {"message": str(e)}, 400
+
+        return {
+            "result": "success",
+            "created_at": workflow_created_at,
+        }
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/default-workflow-block-configs")
+class SnippetDefaultBlockConfigsApi(Resource):
+    @console_ns.doc("get_snippet_default_block_configs")
+    @console_ns.response(200, "Default block configs retrieved successfully")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def get(self, snippet: CustomizedSnippet):
+        """Get default block configurations for snippet workflow."""
+        snippet_service = SnippetService()
+        return snippet_service.get_default_block_configs()
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/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", workflow_pagination_model)
+    @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,
+            )
+            serialized_workflows = marshal(workflows, workflow_model)
+
+        return {
+            "items": serialized_workflows,
+            "page": args.page,
+            "limit": args.limit,
+            "has_more": has_more,
+        }
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflow-runs")
+class SnippetWorkflowRunsApi(Resource):
+    @console_ns.doc("list_snippet_workflow_runs")
+    @console_ns.response(200, "Workflow runs retrieved successfully", workflow_run_pagination_model)
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @marshal_with(workflow_run_pagination_model)
+    def get(self, snippet: CustomizedSnippet):
+        """List workflow runs for snippet."""
+        query = WorkflowRunQuery.model_validate(
+            {
+                "last_id": request.args.get("last_id"),
+                "limit": request.args.get("limit", type=int, default=20),
+            }
+        )
+        args = {
+            "last_id": query.last_id,
+            "limit": query.limit,
+        }
+
+        snippet_service = SnippetService()
+        result = snippet_service.get_snippet_workflow_runs(snippet=snippet, args=args)
+
+        return result
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflow-runs/<uuid:run_id>")
+class SnippetWorkflowRunDetailApi(Resource):
+    @console_ns.doc("get_snippet_workflow_run_detail")
+    @console_ns.response(200, "Workflow run detail retrieved successfully", workflow_run_detail_model)
+    @console_ns.response(404, "Workflow run not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @marshal_with(workflow_run_detail_model)
+    def get(self, snippet: CustomizedSnippet, run_id):
+        """Get workflow run detail for snippet."""
+        run_id = str(run_id)
+
+        snippet_service = SnippetService()
+        workflow_run = snippet_service.get_snippet_workflow_run(snippet=snippet, run_id=run_id)
+
+        if not workflow_run:
+            raise NotFound("Workflow run not found")
+
+        return workflow_run
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflow-runs/<uuid:run_id>/node-executions")
+class SnippetWorkflowRunNodeExecutionsApi(Resource):
+    @console_ns.doc("list_snippet_workflow_run_node_executions")
+    @console_ns.response(200, "Node executions retrieved successfully", workflow_run_node_execution_list_model)
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @marshal_with(workflow_run_node_execution_list_model)
+    def get(self, snippet: CustomizedSnippet, run_id):
+        """List node executions for a workflow run."""
+        run_id = str(run_id)
+
+        snippet_service = SnippetService()
+        node_executions = snippet_service.get_snippet_workflow_run_node_executions(
+            snippet=snippet,
+            run_id=run_id,
+        )
+
+        return {"data": node_executions}
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/nodes/<string:node_id>/run")
+class SnippetDraftNodeRunApi(Resource):
+    @console_ns.doc("run_snippet_draft_node")
+    @console_ns.doc(description="Run a single node in snippet draft workflow (single-step debugging)")
+    @console_ns.doc(params={"snippet_id": "Snippet ID", "node_id": "Node ID"})
+    @console_ns.expect(console_ns.models.get(SnippetDraftNodeRunPayload.__name__))
+    @console_ns.response(200, "Node run completed successfully", workflow_run_node_execution_model)
+    @console_ns.response(404, "Snippet or draft workflow not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @marshal_with(workflow_run_node_execution_model)
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet, node_id: str):
+        """
+        Run a single node in snippet draft workflow.
+
+        Executes a specific node with provided inputs for single-step debugging.
+        Returns the node execution result including status, outputs, and timing.
+        """
+        current_user, _ = current_account_with_tenant()
+        payload = SnippetDraftNodeRunPayload.model_validate(console_ns.payload or {})
+
+        user_inputs = payload.inputs
+
+        # Get draft workflow for file parsing
+        snippet_service = SnippetService()
+        draft_workflow = snippet_service.get_draft_workflow(snippet=snippet)
+        if not draft_workflow:
+            raise NotFound("Draft workflow not found")
+
+        files = SnippetGenerateService.parse_files(draft_workflow, payload.files)
+
+        workflow_node_execution = SnippetGenerateService.run_draft_node(
+            snippet=snippet,
+            node_id=node_id,
+            user_inputs=user_inputs,
+            account=current_user,
+            query=payload.query,
+            files=files,
+        )
+
+        return workflow_node_execution
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/nodes/<string:node_id>/last-run")
+class SnippetDraftNodeLastRunApi(Resource):
+    @console_ns.doc("get_snippet_draft_node_last_run")
+    @console_ns.doc(description="Get last run result for a node in snippet draft workflow")
+    @console_ns.doc(params={"snippet_id": "Snippet ID", "node_id": "Node ID"})
+    @console_ns.response(200, "Node last run retrieved successfully", workflow_run_node_execution_model)
+    @console_ns.response(404, "Snippet, draft workflow, or node last run not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @marshal_with(workflow_run_node_execution_model)
+    def get(self, snippet: CustomizedSnippet, node_id: str):
+        """
+        Get the last run result for a specific node in snippet draft workflow.
+
+        Returns the most recent execution record for the given node,
+        including status, inputs, outputs, and timing information.
+        """
+        snippet_service = SnippetService()
+        draft_workflow = snippet_service.get_draft_workflow(snippet=snippet)
+        if not draft_workflow:
+            raise NotFound("Draft workflow not found")
+
+        node_exec = snippet_service.get_snippet_node_last_run(
+            snippet=snippet,
+            workflow=draft_workflow,
+            node_id=node_id,
+        )
+        if node_exec is None:
+            raise NotFound("Node last run not found")
+
+        return node_exec
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/iteration/nodes/<string:node_id>/run")
+class SnippetDraftRunIterationNodeApi(Resource):
+    @console_ns.doc("run_snippet_draft_iteration_node")
+    @console_ns.doc(description="Run draft workflow iteration node for snippet")
+    @console_ns.doc(params={"snippet_id": "Snippet ID", "node_id": "Node ID"})
+    @console_ns.expect(console_ns.models.get(SnippetIterationNodeRunPayload.__name__))
+    @console_ns.response(200, "Iteration node run started successfully (SSE stream)")
+    @console_ns.response(404, "Snippet or draft workflow not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet, node_id: str):
+        """
+        Run a draft workflow iteration node for snippet.
+
+        Iteration nodes execute their internal sub-graph multiple times over an input list.
+        Returns an SSE event stream with iteration progress and results.
+        """
+        current_user, _ = current_account_with_tenant()
+        args = SnippetIterationNodeRunPayload.model_validate(console_ns.payload or {}).model_dump(exclude_none=True)
+
+        try:
+            response = SnippetGenerateService.generate_single_iteration(
+                snippet=snippet, user=current_user, node_id=node_id, args=args, streaming=True
+            )
+
+            return helper.compact_generate_response(response)
+        except ValueError as e:
+            raise e
+        except Exception:
+            logger.exception("internal server error.")
+            raise InternalServerError()
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/loop/nodes/<string:node_id>/run")
+class SnippetDraftRunLoopNodeApi(Resource):
+    @console_ns.doc("run_snippet_draft_loop_node")
+    @console_ns.doc(description="Run draft workflow loop node for snippet")
+    @console_ns.doc(params={"snippet_id": "Snippet ID", "node_id": "Node ID"})
+    @console_ns.expect(console_ns.models.get(SnippetLoopNodeRunPayload.__name__))
+    @console_ns.response(200, "Loop node run started successfully (SSE stream)")
+    @console_ns.response(404, "Snippet or draft workflow not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet, node_id: str):
+        """
+        Run a draft workflow loop node for snippet.
+
+        Loop nodes execute their internal sub-graph repeatedly until a condition is met.
+        Returns an SSE event stream with loop progress and results.
+        """
+        current_user, _ = current_account_with_tenant()
+        args = SnippetLoopNodeRunPayload.model_validate(console_ns.payload or {})
+
+        try:
+            response = SnippetGenerateService.generate_single_loop(
+                snippet=snippet, user=current_user, node_id=node_id, args=args, streaming=True
+            )
+
+            return helper.compact_generate_response(response)
+        except ValueError as e:
+            raise e
+        except Exception:
+            logger.exception("internal server error.")
+            raise InternalServerError()
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/run")
+class SnippetDraftWorkflowRunApi(Resource):
+    @console_ns.doc("run_snippet_draft_workflow")
+    @console_ns.expect(console_ns.models.get(SnippetDraftRunPayload.__name__))
+    @console_ns.response(200, "Draft workflow run started successfully (SSE stream)")
+    @console_ns.response(404, "Snippet or draft workflow not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet):
+        """
+        Run draft workflow for snippet.
+
+        Executes the snippet's draft workflow with the provided inputs
+        and returns an SSE event stream with execution progress and results.
+        """
+        current_user, _ = current_account_with_tenant()
+
+        payload = SnippetDraftRunPayload.model_validate(console_ns.payload or {})
+        args = payload.model_dump(exclude_none=True)
+
+        try:
+            response = SnippetGenerateService.generate(
+                snippet=snippet,
+                user=current_user,
+                args=args,
+                invoke_from=InvokeFrom.DEBUGGER,
+                streaming=True,
+            )
+
+            return helper.compact_generate_response(response)
+        except ValueError as e:
+            raise e
+        except Exception:
+            logger.exception("internal server error.")
+            raise InternalServerError()
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflow-runs/tasks/<string:task_id>/stop")
+class SnippetWorkflowTaskStopApi(Resource):
+    @console_ns.doc("stop_snippet_workflow_task")
+    @console_ns.response(200, "Task stopped successfully")
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet, task_id: str):
+        """
+        Stop a running snippet workflow task.
+
+        Uses both the legacy stop flag mechanism and the graph engine
+        command channel for backward compatibility.
+        """
+        # Stop using both mechanisms for backward compatibility
+        # Legacy stop flag mechanism (without user check)
+        AppQueueManager.set_stop_flag_no_user_check(task_id)
+
+        # New graph engine command channel mechanism
+        GraphEngineManager(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/workspace/agent_providers.py

@@ -1,8 +1,8 @@
 from flask_restx import Resource, fields
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, setup_required
-from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs.login import current_account_with_tenant, login_required
 from services.agent_service import AgentService
 

api/controllers/console/workspace/endpoint.py

@@ -2,13 +2,13 @@ from typing import Any
 
 from flask import request
 from flask_restx import Resource
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 from pydantic import BaseModel, Field
 
 from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, is_admin_or_owner_required, setup_required
 from core.plugin.impl.exc import PluginPermissionDeniedError
-from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs.login import current_account_with_tenant, login_required
 from services.plugin.endpoint_service import EndpointService
 

api/controllers/console/workspace/load_balancing_config.py

@@ -1,12 +1,12 @@
 from flask_restx import Resource
+from graphon.model_runtime.entities.model_entities import ModelType
+from graphon.model_runtime.errors.validate import CredentialsValidateFailedError
 from pydantic import BaseModel
 from werkzeug.exceptions import Forbidden
 
 from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, setup_required
-from graphon.model_runtime.entities.model_entities import ModelType
-from graphon.model_runtime.errors.validate import CredentialsValidateFailedError
 from libs.login import current_account_with_tenant, login_required
 from models import TenantAccountRole
 from services.model_load_balancing_service import ModelLoadBalancingService

api/controllers/console/workspace/model_providers.py

@@ -3,13 +3,13 @@ from typing import Any, Literal
 
 from flask import request, send_file
 from flask_restx import Resource
+from graphon.model_runtime.entities.model_entities import ModelType
+from graphon.model_runtime.errors.validate import CredentialsValidateFailedError
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 from pydantic import BaseModel, Field, field_validator
 
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, is_admin_or_owner_required, setup_required
-from graphon.model_runtime.entities.model_entities import ModelType
-from graphon.model_runtime.errors.validate import CredentialsValidateFailedError
-from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs.helper import uuid_value
 from libs.login import current_account_with_tenant, login_required
 from services.billing_service import BillingService

api/controllers/console/workspace/models.py

@@ -3,14 +3,14 @@ from typing import Any, cast
 
 from flask import request
 from flask_restx import Resource
+from graphon.model_runtime.entities.model_entities import ModelType
+from graphon.model_runtime.errors.validate import CredentialsValidateFailedError
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 from pydantic import BaseModel, Field, field_validator
 
 from controllers.common.schema import register_enum_models, register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, is_admin_or_owner_required, setup_required
-from graphon.model_runtime.entities.model_entities import ModelType
-from graphon.model_runtime.errors.validate import CredentialsValidateFailedError
-from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs.helper import uuid_value
 from libs.login import current_account_with_tenant, login_required
 from services.model_load_balancing_service import ModelLoadBalancingService

api/controllers/console/workspace/plugin.py

@@ -4,6 +4,7 @@ from typing import Any, Literal
 
 from flask import request, send_file
 from flask_restx import Resource
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 from pydantic import BaseModel, Field
 from werkzeug.datastructures import FileStorage
 from werkzeug.exceptions import Forbidden
@@ -14,7 +15,6 @@ from controllers.console import console_ns
 from controllers.console.workspace import plugin_permission_required
 from controllers.console.wraps import account_initialization_required, is_admin_or_owner_required, setup_required
 from core.plugin.impl.exc import PluginDaemonClientSideError
-from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs.login import current_account_with_tenant, login_required
 from models.account import TenantPluginAutoUpgradeStrategy, TenantPluginPermission
 from services.plugin.plugin_auto_upgrade_service import PluginAutoUpgradeService

api/controllers/console/workspace/snippets.py

@@ -0,0 +1,380 @@
+import logging
+from urllib.parse import quote
+
+from flask import Response, request
+from flask_restx import Resource, marshal
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import NotFound
+
+from controllers.common.schema import register_schema_models
+from controllers.console import console_ns
+from controllers.console.snippets.payloads import (
+    CreateSnippetPayload,
+    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__)
+
+# 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_params = request.args.to_dict()
+        query = SnippetListQuery.model_validate(query_params)
+
+        snippets, total, has_more = SnippetService.get_snippets(
+            tenant_id=current_tenant_id,
+            page=query.page,
+            limit=query.limit,
+            keyword=query.keyword,
+            is_published=query.is_published,
+            creators=query.creators,
+        )
+
+        return {
+            "data": marshal(snippets, snippet_list_fields),
+            "page": query.page,
+            "limit": query.limit,
+            "total": total,
+            "has_more": has_more,
+        }, 200
+
+    @console_ns.doc("create_customized_snippet")
+    @console_ns.expect(console_ns.models.get(CreateSnippetPayload.__name__))
+    @console_ns.response(201, "Snippet created successfully", snippet_model)
+    @console_ns.response(400, "Invalid request or name already exists")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def post(self):
+        """Create a new customized snippet."""
+        current_user, current_tenant_id = current_account_with_tenant()
+
+        payload = CreateSnippetPayload.model_validate(console_ns.payload or {})
+
+        try:
+            snippet_type = SnippetType(payload.type)
+        except ValueError:
+            snippet_type = SnippetType.NODE
+
+        try:
+            snippet = SnippetService.create_snippet(
+                tenant_id=current_tenant_id,
+                name=payload.name,
+                description=payload.description,
+                snippet_type=snippet_type,
+                icon_info=payload.icon_info.model_dump() if payload.icon_info else None,
+                input_fields=[f.model_dump() for f in payload.input_fields] if payload.input_fields else None,
+                account=current_user,
+            )
+        except ValueError as e:
+            return {"message": str(e)}, 400
+
+        return marshal(snippet, snippet_fields), 201
+
+
+@console_ns.route("/workspaces/current/customized-snippets/<uuid:snippet_id>")
+class CustomizedSnippetDetailApi(Resource):
+    @console_ns.doc("get_customized_snippet")
+    @console_ns.response(200, "Snippet retrieved successfully", snippet_model)
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, snippet_id: str):
+        """Get customized snippet details."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        snippet = SnippetService.get_snippet_by_id(
+            snippet_id=str(snippet_id),
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        return marshal(snippet, snippet_fields), 200
+
+    @console_ns.doc("update_customized_snippet")
+    @console_ns.expect(console_ns.models.get(UpdateSnippetPayload.__name__))
+    @console_ns.response(200, "Snippet updated successfully", snippet_model)
+    @console_ns.response(400, "Invalid request or name already exists")
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def patch(self, snippet_id: str):
+        """Update customized snippet."""
+        current_user, current_tenant_id = current_account_with_tenant()
+
+        snippet = SnippetService.get_snippet_by_id(
+            snippet_id=str(snippet_id),
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        payload = UpdateSnippetPayload.model_validate(console_ns.payload or {})
+        update_data = payload.model_dump(exclude_unset=True)
+
+        if "icon_info" in update_data and update_data["icon_info"] is not None:
+            update_data["icon_info"] = payload.icon_info.model_dump() if payload.icon_info else None
+
+        if not update_data:
+            return {"message": "No valid fields to update"}, 400
+
+        try:
+            with Session(db.engine, expire_on_commit=False) as session:
+                snippet = session.merge(snippet)
+                snippet = SnippetService.update_snippet(
+                    session=session,
+                    snippet=snippet,
+                    account_id=current_user.id,
+                    data=update_data,
+                )
+                session.commit()
+        except ValueError as e:
+            return {"message": str(e)}, 400
+
+        return marshal(snippet, snippet_fields), 200
+
+    @console_ns.doc("delete_customized_snippet")
+    @console_ns.response(204, "Snippet deleted successfully")
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def delete(self, snippet_id: str):
+        """Delete customized snippet."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        snippet = SnippetService.get_snippet_by_id(
+            snippet_id=str(snippet_id),
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        with Session(db.engine) as session:
+            snippet = session.merge(snippet)
+            SnippetService.delete_snippet(
+                session=session,
+                snippet=snippet,
+            )
+            session.commit()
+
+        return "", 204
+
+
+@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/controllers/console/workspace/tool_providers.py

@@ -5,6 +5,7 @@ from urllib.parse import urlparse
 
 from flask import make_response, redirect, request, send_file
 from flask_restx import Resource
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 from pydantic import BaseModel, Field, HttpUrl, field_validator, model_validator
 from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import Forbidden
@@ -27,7 +28,6 @@ from core.plugin.entities.plugin_daemon import CredentialType
 from core.plugin.impl.oauth import OAuthHandler
 from core.tools.entities.tool_entities import ApiProviderSchemaType, WorkflowToolParameterConfiguration
 from extensions.ext_database import db
-from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs.helper import alphanumeric, uuid_value
 from libs.login import current_account_with_tenant, login_required
 from models.provider_ids import ToolProviderID

api/controllers/console/workspace/trigger_providers.py

@@ -3,6 +3,7 @@ from typing import Any
 
 from flask import make_response, redirect, request
 from flask_restx import Resource
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 from pydantic import BaseModel, model_validator
 from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import BadRequest, Forbidden
@@ -15,7 +16,6 @@ from core.plugin.impl.oauth import OAuthHandler
 from core.trigger.entities.entities import SubscriptionBuilderUpdater
 from core.trigger.trigger_manager import TriggerManager
 from extensions.ext_database import db
-from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs.login import current_user, login_required
 from models.account import Account
 from models.provider_ids import TriggerProviderID

api/controllers/inner_api/plugin/plugin.py

@@ -1,4 +1,5 @@
 from flask_restx import Resource
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 
 from controllers.console.wraps import setup_required
 from controllers.inner_api import inner_api_ns
@@ -29,7 +30,6 @@ from core.plugin.entities.request import (
 )
 from core.tools.entities.tool_entities import ToolProviderType
 from core.tools.signature import get_signed_file_url_for_plugin
-from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs.helper import length_prefixed_response
 from models import Account, Tenant
 from models.model import EndUser

api/controllers/inner_api/plugin/wraps.py

@@ -20,13 +20,10 @@ class TenantUserPayload(BaseModel):
 
 def get_user(tenant_id: str, user_id: str | None) -> EndUser:
     """
-    Get current user.
+    Get current user
 
     NOTE: user_id is not trusted, it could be maliciously set to any value.
-    As a result, it could only be considered as an end user id. Even when a
+    As a result, it could only be considered as an end user id.
-    concrete end-user ID is supplied, lookups must stay tenant-scoped so one
-    tenant cannot bind another tenant's user record into the plugin request
-    context.
     """
     if not user_id:
         user_id = DefaultEndUserSessionID.DEFAULT_SESSION_ID
@@ -45,14 +42,7 @@ def get_user(tenant_id: str, user_id: str | None) -> EndUser:
                     .limit(1)
                 )
             else:
-                user_model = session.scalar(
+                user_model = session.get(EndUser, user_id)
-                    select(EndUser)
-                    .where(
-                        EndUser.id == user_id,
-                        EndUser.tenant_id == tenant_id,
-                    )
-                    .limit(1)
-                )
 
             if not user_model:
                 user_model = EndUser(

api/controllers/mcp/mcp.py

@@ -2,6 +2,7 @@ from typing import Any, Union
 
 from flask import Response
 from flask_restx import Resource
+from graphon.variables.input_entities import VariableEntity, VariableEntityType
 from pydantic import BaseModel, Field, ValidationError
 from sqlalchemy import select
 from sqlalchemy.orm import Session, sessionmaker
@@ -11,7 +12,6 @@ from controllers.mcp import mcp_ns
 from core.mcp import types as mcp_types
 from core.mcp.server.streamable_http import handle_mcp_request
 from extensions.ext_database import db
-from graphon.variables.input_entities import VariableEntity, VariableEntityType
 from libs import helper
 from models.enums import AppMCPServerStatus
 from models.model import App, AppMCPServer, AppMode, EndUser

api/controllers/service_api/app/audio.py

@@ -2,6 +2,7 @@ import logging
 
 from flask import request
 from flask_restx import Resource
+from graphon.model_runtime.errors.invoke import InvokeError
 from werkzeug.exceptions import InternalServerError
 
 import services
@@ -21,7 +22,6 @@ from controllers.service_api.app.error import (
 )
 from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate_app_token
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
-from graphon.model_runtime.errors.invoke import InvokeError
 from models.model import App, EndUser
 from services.audio_service import AudioService
 from services.errors.audio import (

api/controllers/service_api/app/completion.py

@@ -4,6 +4,7 @@ from uuid import UUID
 
 from flask import request
 from flask_restx import Resource
+from graphon.model_runtime.errors.invoke import InvokeError
 from pydantic import BaseModel, Field, field_validator
 from werkzeug.exceptions import BadRequest, InternalServerError, NotFound
 
@@ -28,7 +29,6 @@ from core.errors.error import (
     QuotaExceededError,
 )
 from core.helper.trace_id_helper import get_external_trace_id
-from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from libs.helper import UUIDStrOrEmpty
 from models.model import App, AppMode, EndUser

api/controllers/service_api/app/conversation.py

@@ -3,6 +3,7 @@ from typing import Any, Literal
 
 from flask import request
 from flask_restx import Resource
+from graphon.variables.types import SegmentType
 from pydantic import BaseModel, Field, TypeAdapter, field_validator
 from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import BadRequest, NotFound
@@ -21,7 +22,6 @@ from fields.conversation_fields import (
     ConversationInfiniteScrollPagination,
     SimpleConversation,
 )
-from graphon.variables.types import SegmentType
 from libs.helper import UUIDStrOrEmpty
 from models.model import App, AppMode, EndUser
 from services.conversation_service import ConversationService

api/controllers/service_api/dataset/dataset.py

@@ -2,6 +2,7 @@ from typing import Any, Literal, cast
 
 from flask import request
 from flask_restx import marshal
+from graphon.model_runtime.entities.model_entities import ModelType
 from pydantic import BaseModel, Field, TypeAdapter, field_validator
 from werkzeug.exceptions import Forbidden, NotFound
 
@@ -18,7 +19,6 @@ from core.plugin.impl.model_runtime_factory import create_plugin_provider_manage
 from core.rag.index_processor.constant.index_type import IndexTechniqueType
 from fields.dataset_fields import dataset_detail_fields
 from fields.tag_fields import DataSetTag
-from graphon.model_runtime.entities.model_entities import ModelType
 from libs.login import current_user
 from models.account import Account
 from models.dataset import DatasetPermissionEnum

api/controllers/service_api/dataset/segment.py

@@ -2,6 +2,7 @@ from typing import Any
 
 from flask import request
 from flask_restx import marshal
+from graphon.model_runtime.entities.model_entities import ModelType
 from pydantic import BaseModel, Field
 from sqlalchemy import select
 from werkzeug.exceptions import NotFound
@@ -22,7 +23,6 @@ from core.model_manager import ModelManager
 from core.rag.index_processor.constant.index_type import IndexTechniqueType
 from extensions.ext_database import db
 from fields.segment_fields import child_chunk_fields, segment_fields
-from graphon.model_runtime.entities.model_entities import ModelType
 from libs.login import current_account_with_tenant
 from models.dataset import Dataset
 from services.dataset_service import DatasetService, DocumentService, SegmentService

api/controllers/service_api/workspace/models.py

@@ -1,9 +1,9 @@
 from flask_login import current_user
 from flask_restx import Resource
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 
 from controllers.service_api import service_api_ns
 from controllers.service_api.wraps import validate_dataset_token
-from graphon.model_runtime.utils.encoders import jsonable_encoder
 from services.model_provider_service import ModelProviderService
 
 

api/controllers/web/audio.py

@@ -2,6 +2,7 @@ import logging
 
 from flask import request
 from flask_restx import fields, marshal_with
+from graphon.model_runtime.errors.invoke import InvokeError
 from pydantic import field_validator
 from werkzeug.exceptions import InternalServerError
 
@@ -21,7 +22,6 @@ from controllers.web.error import (
 )
 from controllers.web.wraps import WebApiResource
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
-from graphon.model_runtime.errors.invoke import InvokeError
 from libs.helper import uuid_value
 from models.model import App
 from services.audio_service import AudioService

api/controllers/web/completion.py

@@ -1,6 +1,7 @@
 import logging
 from typing import Any, Literal
 
+from graphon.model_runtime.errors.invoke import InvokeError
 from pydantic import BaseModel, Field, field_validator
 from werkzeug.exceptions import InternalServerError, NotFound
 
@@ -25,7 +26,6 @@ from core.errors.error import (
     ProviderTokenNotInitError,
     QuotaExceededError,
 )
-from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from libs.helper import uuid_value
 from models.model import AppMode

api/controllers/web/message.py

@@ -2,6 +2,7 @@ import logging
 from typing import Literal
 
 from flask import request
+from graphon.model_runtime.errors.invoke import InvokeError
 from pydantic import BaseModel, Field, TypeAdapter
 from werkzeug.exceptions import InternalServerError, NotFound
 
@@ -23,7 +24,6 @@ from core.app.entities.app_invoke_entities import InvokeFrom
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
 from fields.conversation_fields import ResultResponse
 from fields.message_fields import SuggestedQuestionsResponse, WebMessageInfiniteScrollPagination, WebMessageListItem
-from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from models.enums import FeedbackRating
 from models.model import AppMode

api/controllers/web/remote_files.py

@@ -1,6 +1,7 @@
 import urllib.parse
 
 import httpx
+from graphon.file import helpers as file_helpers
 from pydantic import BaseModel, Field, HttpUrl
 
 import services
@@ -13,7 +14,6 @@ from controllers.common.errors import (
 from core.helper import ssrf_proxy
 from extensions.ext_database import db
 from fields.file_fields import FileWithSignedUrl, RemoteFileInfo
-from graphon.file import helpers as file_helpers
 from services.file_service import FileService
 
 from ..common.schema import register_schema_models

api/controllers/web/workflow.py

@@ -1,5 +1,7 @@
 import logging
 
+from graphon.graph_engine.manager import GraphEngineManager
+from graphon.model_runtime.errors.invoke import InvokeError
 from werkzeug.exceptions import InternalServerError
 
 from controllers.common.controller_schemas import WorkflowRunPayload
@@ -22,8 +24,6 @@ from core.errors.error import (
     QuotaExceededError,
 )
 from extensions.ext_redis import redis_client
-from graphon.graph_engine.manager import GraphEngineManager
-from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from models.model import App, AppMode, EndUser
 from services.app_generate_service import AppGenerateService

api/core/agent/cot_agent_runner.py

@@ -4,6 +4,15 @@ from abc import ABC, abstractmethod
 from collections.abc import Generator, Mapping, Sequence
 from typing import Any, TypedDict
 
+from graphon.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk, LLMResultChunkDelta, LLMUsage
+from graphon.model_runtime.entities.message_entities import (
+    AssistantPromptMessage,
+    PromptMessage,
+    PromptMessageTool,
+    ToolPromptMessage,
+    UserPromptMessage,
+)
+
 from core.agent.base_agent_runner import BaseAgentRunner
 from core.agent.entities import AgentScratchpadUnit
 from core.agent.errors import AgentMaxIterationError
@@ -15,14 +24,6 @@ from core.prompt.agent_history_prompt_transform import AgentHistoryPromptTransfo
 from core.tools.__base.tool import Tool
 from core.tools.entities.tool_entities import ToolInvokeMeta
 from core.tools.tool_engine import ToolEngine
-from graphon.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk, LLMResultChunkDelta, LLMUsage
-from graphon.model_runtime.entities.message_entities import (
-    AssistantPromptMessage,
-    PromptMessage,
-    PromptMessageTool,
-    ToolPromptMessage,
-    UserPromptMessage,
-)
 from models.model import Message
 
 logger = logging.getLogger(__name__)

api/core/agent/cot_chat_agent_runner.py

@@ -1,6 +1,5 @@
 import json
 
-from core.agent.cot_agent_runner import CotAgentRunner
 from graphon.file import file_manager
 from graphon.model_runtime.entities import (
     AssistantPromptMessage,
@@ -12,6 +11,8 @@ from graphon.model_runtime.entities import (
 from graphon.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
 from graphon.model_runtime.utils.encoders import jsonable_encoder
 
+from core.agent.cot_agent_runner import CotAgentRunner
+
 
 class CotChatAgentRunner(CotAgentRunner):
     def _organize_system_prompt(self) -> SystemPromptMessage:

api/core/agent/cot_completion_agent_runner.py

@@ -1,6 +1,5 @@
 import json
 
-from core.agent.cot_agent_runner import CotAgentRunner
 from graphon.model_runtime.entities.message_entities import (
     AssistantPromptMessage,
     PromptMessage,
@@ -9,6 +8,8 @@ from graphon.model_runtime.entities.message_entities import (
 )
 from graphon.model_runtime.utils.encoders import jsonable_encoder
 
+from core.agent.cot_agent_runner import CotAgentRunner
+
 
 class CotCompletionAgentRunner(CotAgentRunner):
     def _organize_instruction_prompt(self) -> str:

api/core/agent/fc_agent_runner.py

@@ -4,13 +4,6 @@ from collections.abc import Generator
 from copy import deepcopy
 from typing import Any, Union
 
-from core.agent.base_agent_runner import BaseAgentRunner
-from core.agent.errors import AgentMaxIterationError
-from core.app.apps.base_app_queue_manager import PublishFrom
-from core.app.entities.queue_entities import QueueAgentThoughtEvent, QueueMessageEndEvent, QueueMessageFileEvent
-from core.prompt.agent_history_prompt_transform import AgentHistoryPromptTransform
-from core.tools.entities.tool_entities import ToolInvokeMeta
-from core.tools.tool_engine import ToolEngine
 from graphon.file import file_manager
 from graphon.model_runtime.entities import (
     AssistantPromptMessage,
@@ -26,6 +19,14 @@ from graphon.model_runtime.entities import (
     UserPromptMessage,
 )
 from graphon.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
+
+from core.agent.base_agent_runner import BaseAgentRunner
+from core.agent.errors import AgentMaxIterationError
+from core.app.apps.base_app_queue_manager import PublishFrom
+from core.app.entities.queue_entities import QueueAgentThoughtEvent, QueueMessageEndEvent, QueueMessageFileEvent
+from core.prompt.agent_history_prompt_transform import AgentHistoryPromptTransform
+from core.tools.entities.tool_entities import ToolInvokeMeta
+from core.tools.tool_engine import ToolEngine
 from models.model import Message
 
 logger = logging.getLogger(__name__)

api/core/agent/output_parser/cot_output_parser.py

@@ -3,9 +3,10 @@ import re
 from collections.abc import Generator
 from typing import Any, Union
 
-from core.agent.entities import AgentScratchpadUnit
 from graphon.model_runtime.entities.llm_entities import LLMResultChunk
 
+from core.agent.entities import AgentScratchpadUnit
+
 
 class CotAgentOutputParser:
     @classmethod

api/core/app/app_config/easy_ui_based_app/model_config/manager.py

@@ -1,9 +1,10 @@
 from collections.abc import Mapping
 from typing import Any
 
+from graphon.model_runtime.entities.model_entities import ModelPropertyKey, ModelType
+
 from core.app.app_config.entities import ModelConfigEntity
 from core.plugin.impl.model_runtime_factory import create_plugin_model_assembly
-from graphon.model_runtime.entities.model_entities import ModelPropertyKey, ModelType
 from models.model import AppModelConfigDict
 from models.provider_ids import ModelProviderID
 

api/core/app/app_config/easy_ui_based_app/prompt_template/manager.py

@@ -1,5 +1,7 @@
 from typing import Any
 
+from graphon.model_runtime.entities.message_entities import PromptMessageRole
+
 from core.app.app_config.entities import (
     AdvancedChatMessageEntity,
     AdvancedChatPromptTemplateEntity,
@@ -7,7 +9,6 @@ from core.app.app_config.entities import (
     PromptTemplateEntity,
 )
 from core.prompt.simple_prompt_transform import ModelMode
-from graphon.model_runtime.entities.message_entities import PromptMessageRole
 from models.model import AppMode, AppModelConfigDict
 
 

api/core/app/app_config/entities.py

@@ -1,14 +1,14 @@
 from enum import StrEnum, auto
 from typing import Any, Literal
 
-from pydantic import BaseModel, Field
-
-from core.rag.data_post_processor.data_post_processor import RerankingModelDict, WeightsDict
-from core.rag.entities import MetadataFilteringCondition
 from graphon.file import FileUploadConfig
 from graphon.model_runtime.entities.llm_entities import LLMMode
 from graphon.model_runtime.entities.message_entities import PromptMessageRole
 from graphon.variables.input_entities import VariableEntity as WorkflowVariableEntity
+from pydantic import BaseModel, Field
+
+from core.rag.data_post_processor.data_post_processor import RerankingModelDict, WeightsDict
+from core.rag.entities import MetadataFilteringCondition
 from models.model import AppMode
 
 

api/core/app/app_config/features/file_upload/manager.py

@@ -1,9 +1,10 @@
 from collections.abc import Mapping
 from typing import Any
 
-from constants import DEFAULT_FILE_NUMBER_LIMITS
 from graphon.file import FileUploadConfig
 
+from constants import DEFAULT_FILE_NUMBER_LIMITS
+
 
 class FileUploadConfigManager:
     @classmethod

api/core/app/app_config/workflow_ui_based_app/variables/manager.py

@@ -1,7 +1,8 @@
 import re
 
-from core.app.app_config.entities import RagPipelineVariableEntity
 from graphon.variables.input_entities import VariableEntity
+
+from core.app.app_config.entities import RagPipelineVariableEntity
 from models.workflow import Workflow
 
 

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

@@ -18,6 +18,11 @@ from constants import UUID_NIL
 
 if TYPE_CHECKING:
     from controllers.console.app.workflow import LoopNodeRunPayload
+from graphon.graph_engine.layers import GraphEngineLayer
+from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
+from graphon.runtime import GraphRuntimeState
+from graphon.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader
+
 from core.app.app_config.features.file_upload.manager import FileUploadConfigManager
 from core.app.apps.advanced_chat.app_config_manager import AdvancedChatAppConfigManager
 from core.app.apps.advanced_chat.app_runner import AdvancedChatAppRunner
@@ -43,10 +48,6 @@ from core.repositories import DifyCoreRepositoryFactory
 from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
 from extensions.ext_database import db
 from factories import file_factory
-from graphon.graph_engine.layers import GraphEngineLayer
-from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
-from graphon.runtime import GraphRuntimeState
-from graphon.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader
 from libs.flask_utils import preserve_flask_contexts
 from models import Account, App, Conversation, EndUser, Message, Workflow, WorkflowNodeExecutionTriggeredFrom
 from models.enums import WorkflowRunTriggeredFrom

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

@@ -3,6 +3,12 @@ import time
 from collections.abc import Mapping, Sequence
 from typing import Any, cast
 
+from graphon.enums import WorkflowType
+from graphon.graph_engine.command_channels import RedisChannel
+from graphon.graph_engine.layers import GraphEngineLayer
+from graphon.runtime import GraphRuntimeState, VariablePool
+from graphon.variable_loader import VariableLoader
+from graphon.variables.variables import Variable
 from sqlalchemy import select
 from sqlalchemy.orm import Session, sessionmaker
 
@@ -37,12 +43,6 @@ from core.workflow.workflow_entry import WorkflowEntry
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from extensions.otel import WorkflowAppRunnerHandler, trace_span
-from graphon.enums import WorkflowType
-from graphon.graph_engine.command_channels import RedisChannel
-from graphon.graph_engine.layers import GraphEngineLayer
-from graphon.runtime import GraphRuntimeState, VariablePool
-from graphon.variable_loader import VariableLoader
-from graphon.variables.variables import Variable
 from models import Workflow
 from models.model import App, Conversation, Message, MessageAnnotation
 from models.workflow import ConversationVariable

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

@@ -6,6 +6,7 @@ from collections.abc import Generator, Mapping
 from typing import Any, Literal, overload
 
 from flask import Flask, current_app
+from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
 from pydantic import ValidationError
 
 from configs import dify_config
@@ -23,7 +24,6 @@ from core.app.entities.app_invoke_entities import AgentChatAppGenerateEntity, In
 from core.ops.ops_trace_manager import TraceQueueManager
 from extensions.ext_database import db
 from factories import file_factory
-from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
 from libs.flask_utils import preserve_flask_contexts
 from models import Account, App, EndUser
 from services.conversation_service import ConversationService

api/core/app/apps/base_app_generate_response_converter.py

@@ -3,10 +3,11 @@ from abc import ABC, abstractmethod
 from collections.abc import Generator, Mapping
 from typing import Any, Union
 
+from graphon.model_runtime.errors.invoke import InvokeError
+
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.app.entities.task_entities import AppBlockingResponse, AppStreamResponse
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
-from graphon.model_runtime.errors.invoke import InvokeError
 
 logger = logging.getLogger(__name__)
 

api/core/app/apps/base_app_queue_manager.py

@@ -7,6 +7,7 @@ from enum import IntEnum, auto
 from typing import Any
 
 from cachetools import TTLCache, cachedmethod
+from graphon.runtime import GraphRuntimeState
 from redis.exceptions import RedisError
 from sqlalchemy.orm import DeclarativeMeta
 
@@ -21,7 +22,6 @@ from core.app.entities.queue_entities import (
     WorkflowQueueMessage,
 )
 from extensions.ext_redis import redis_client
-from graphon.runtime import GraphRuntimeState
 
 logger = logging.getLogger(__name__)
 

api/core/app/apps/base_app_runner.py

@@ -5,6 +5,17 @@ from collections.abc import Generator, Mapping, Sequence
 from mimetypes import guess_extension
 from typing import TYPE_CHECKING, Any, Union
 
+from graphon.file import FileTransferMethod, FileType
+from graphon.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk, LLMResultChunkDelta, LLMUsage
+from graphon.model_runtime.entities.message_entities import (
+    AssistantPromptMessage,
+    ImagePromptMessageContent,
+    PromptMessage,
+    TextPromptMessageContent,
+)
+from graphon.model_runtime.entities.model_entities import ModelPropertyKey
+from graphon.model_runtime.errors.invoke import InvokeBadRequestError
+
 from core.app.app_config.entities import ExternalDataVariableEntity, PromptTemplateEntity
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
 from core.app.entities.app_invoke_entities import (
@@ -30,16 +41,6 @@ from core.prompt.entities.advanced_prompt_entities import ChatModelMessage, Comp
 from core.prompt.simple_prompt_transform import ModelMode, SimplePromptTransform
 from core.tools.tool_file_manager import ToolFileManager
 from extensions.ext_database import db
-from graphon.file import FileTransferMethod, FileType
-from graphon.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk, LLMResultChunkDelta, LLMUsage
-from graphon.model_runtime.entities.message_entities import (
-    AssistantPromptMessage,
-    ImagePromptMessageContent,
-    PromptMessage,
-    TextPromptMessageContent,
-)
-from graphon.model_runtime.entities.model_entities import ModelPropertyKey
-from graphon.model_runtime.errors.invoke import InvokeBadRequestError
 from models.enums import CreatorUserRole, MessageFileBelongsTo
 from models.model import App, AppMode, Message, MessageAnnotation, MessageFile
 

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

@@ -6,6 +6,7 @@ from collections.abc import Generator, Mapping
 from typing import Any, Literal, overload
 
 from flask import Flask, copy_current_request_context, current_app
+from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
 from pydantic import ValidationError
 
 from configs import dify_config
@@ -23,7 +24,6 @@ from core.app.entities.app_invoke_entities import ChatAppGenerateEntity, InvokeF
 from core.ops.ops_trace_manager import TraceQueueManager
 from extensions.ext_database import db
 from factories import file_factory
-from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
 from models import Account
 from models.model import App, EndUser
 from services.conversation_service import ConversationService

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

@@ -1,6 +1,8 @@
 import logging
 from typing import cast
 
+from graphon.file import File
+from graphon.model_runtime.entities.message_entities import ImagePromptMessageContent
 from sqlalchemy import select
 
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
@@ -16,8 +18,6 @@ from core.model_manager import ModelInstance
 from core.moderation.base import ModerationError
 from core.rag.retrieval.dataset_retrieval import DatasetRetrieval
 from extensions.ext_database import db
-from graphon.file import File
-from graphon.model_runtime.entities.message_entities import ImagePromptMessageContent
 from models.model import App, Conversation, Message
 
 logger = logging.getLogger(__name__)

api/core/app/apps/common/graph_runtime_state_support.py

@@ -4,9 +4,10 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING
 
-from core.workflow.system_variables import SystemVariableKey, get_system_text
 from graphon.runtime import GraphRuntimeState
 
+from core.workflow.system_variables import SystemVariableKey, get_system_text
+
 if TYPE_CHECKING:
     from core.app.task_pipeline.based_generate_task_pipeline import BasedGenerateTaskPipeline
 

api/core/app/apps/common/workflow_response_converter.py

@@ -6,6 +6,19 @@ from dataclasses import dataclass
 from datetime import datetime
 from typing import Any, NewType, TypedDict, Union
 
+from graphon.entities import WorkflowStartReason
+from graphon.entities.pause_reason import HumanInputRequired
+from graphon.enums import (
+    BuiltinNodeTypes,
+    WorkflowExecutionStatus,
+    WorkflowNodeExecutionMetadataKey,
+    WorkflowNodeExecutionStatus,
+)
+from graphon.file import FILE_MODEL_IDENTITY, File
+from graphon.runtime import GraphRuntimeState
+from graphon.variables.segments import ArrayFileSegment, FileSegment, Segment
+from graphon.variables.variables import Variable
+from graphon.workflow_type_encoder import WorkflowRuntimeTypeConverter
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
@@ -55,19 +68,6 @@ from core.workflow.human_input_forms import load_form_tokens_by_form_id
 from core.workflow.system_variables import SystemVariableKey, system_variables_to_mapping
 from core.workflow.workflow_entry import WorkflowEntry
 from extensions.ext_database import db
-from graphon.entities import WorkflowStartReason
-from graphon.entities.pause_reason import HumanInputRequired
-from graphon.enums import (
-    BuiltinNodeTypes,
-    WorkflowExecutionStatus,
-    WorkflowNodeExecutionMetadataKey,
-    WorkflowNodeExecutionStatus,
-)
-from graphon.file import FILE_MODEL_IDENTITY, File
-from graphon.runtime import GraphRuntimeState
-from graphon.variables.segments import ArrayFileSegment, FileSegment, Segment
-from graphon.variables.variables import Variable
-from graphon.workflow_type_encoder import WorkflowRuntimeTypeConverter
 from libs.datetime_utils import naive_utc_now
 from models import Account, EndUser
 from models.human_input import HumanInputForm

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

@@ -6,6 +6,7 @@ from collections.abc import Generator, Mapping
 from typing import Any, Literal, overload
 
 from flask import Flask, copy_current_request_context, current_app
+from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
 from pydantic import ValidationError
 from sqlalchemy import select
 
@@ -23,7 +24,6 @@ from core.app.entities.app_invoke_entities import CompletionAppGenerateEntity, I
 from core.ops.ops_trace_manager import TraceQueueManager
 from extensions.ext_database import db
 from factories import file_factory
-from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
 from models import Account, App, EndUser, Message
 from services.errors.app import MoreLikeThisDisabledError
 from services.errors.message import MessageNotExistsError

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

@@ -1,6 +1,8 @@
 import logging
 from typing import cast
 
+from graphon.file import File
+from graphon.model_runtime.entities.message_entities import ImagePromptMessageContent
 from sqlalchemy import select
 
 from core.app.apps.base_app_queue_manager import AppQueueManager
@@ -14,8 +16,6 @@ from core.model_manager import ModelInstance
 from core.moderation.base import ModerationError
 from core.rag.retrieval.dataset_retrieval import DatasetRetrieval
 from extensions.ext_database import db
-from graphon.file import File
-from graphon.model_runtime.entities.message_entities import ImagePromptMessageContent
 from models.model import App, Message
 
 logger = logging.getLogger(__name__)

api/core/app/apps/pipeline/pipeline_generator.py

@@ -10,6 +10,8 @@ from collections.abc import Generator, Mapping
 from typing import Any, Literal, cast, overload
 
 from flask import Flask, current_app
+from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
+from graphon.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader
 from pydantic import ValidationError
 from sqlalchemy import select
 from sqlalchemy.orm import Session, sessionmaker
@@ -41,8 +43,6 @@ from core.repositories.factory import (
     WorkflowNodeExecutionRepository,
 )
 from extensions.ext_database import db
-from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
-from graphon.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader
 from libs.flask_utils import preserve_flask_contexts
 from models import Account, EndUser, Workflow, WorkflowNodeExecutionTriggeredFrom
 from models.dataset import Document, DocumentPipelineExecutionLog, Pipeline

api/core/app/apps/pipeline/pipeline_runner.py

@@ -2,6 +2,12 @@ import logging
 import time
 from typing import cast
 
+from graphon.enums import WorkflowType
+from graphon.graph import Graph
+from graphon.graph_events import GraphEngineEvent, GraphRunFailedEvent
+from graphon.runtime import GraphRuntimeState, VariablePool
+from graphon.variable_loader import VariableLoader
+from graphon.variables.variables import RAGPipelineVariable, RAGPipelineVariableInput
 from sqlalchemy import select
 
 from core.app.apps.base_app_queue_manager import AppQueueManager
@@ -20,12 +26,6 @@ from core.workflow.system_variables import build_bootstrap_variables, build_syst
 from core.workflow.variable_pool_initializer import add_node_inputs_to_pool, add_variables_to_pool
 from core.workflow.workflow_entry import WorkflowEntry
 from extensions.ext_database import db
-from graphon.enums import WorkflowType
-from graphon.graph import Graph
-from graphon.graph_events import GraphEngineEvent, GraphRunFailedEvent
-from graphon.runtime import GraphRuntimeState, VariablePool
-from graphon.variable_loader import VariableLoader
-from graphon.variables.variables import RAGPipelineVariable, RAGPipelineVariableInput
 from models.dataset import Document, Pipeline
 from models.model import EndUser
 from models.workflow import Workflow

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

@@ -8,9 +8,13 @@ from collections.abc import Generator, Mapping, Sequence
 from typing import TYPE_CHECKING, Any, Literal, overload
 
 from flask import Flask, current_app
+from graphon.graph_engine.layers import GraphEngineLayer
+from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
+from graphon.runtime import GraphRuntimeState
+from graphon.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader
 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
@@ -34,10 +38,6 @@ from core.repositories import DifyCoreRepositoryFactory
 from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
 from extensions.ext_database import db
 from factories import file_factory
-from graphon.graph_engine.layers import GraphEngineLayer
-from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
-from graphon.runtime import GraphRuntimeState
-from graphon.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader
 from libs.flask_utils import preserve_flask_contexts
 from models.account import Account
 from models.enums import WorkflowRunTriggeredFrom
@@ -54,6 +54,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))
@@ -557,6 +576,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

@@ -3,6 +3,12 @@ import time
 from collections.abc import Sequence
 from typing import cast
 
+from graphon.enums import WorkflowType
+from graphon.graph_engine.command_channels import RedisChannel
+from graphon.graph_engine.layers import GraphEngineLayer
+from graphon.runtime import GraphRuntimeState, VariablePool
+from graphon.variable_loader import VariableLoader
+
 from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.apps.workflow.app_config_manager import WorkflowAppConfig
 from core.app.apps.workflow_app_runner import WorkflowBasedAppRunner
@@ -15,11 +21,6 @@ from core.workflow.variable_pool_initializer import add_node_inputs_to_pool, add
 from core.workflow.workflow_entry import WorkflowEntry
 from extensions.ext_redis import redis_client
 from extensions.otel import WorkflowAppRunnerHandler, trace_span
-from graphon.enums import WorkflowType
-from graphon.graph_engine.command_channels import RedisChannel
-from graphon.graph_engine.layers import GraphEngineLayer
-from graphon.runtime import GraphRuntimeState, VariablePool
-from graphon.variable_loader import VariableLoader
 from libs.datetime_utils import naive_utc_now
 from models.workflow import Workflow
 

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

@@ -4,6 +4,9 @@ from collections.abc import Callable, Generator
 from contextlib import contextmanager
 from typing import Union
 
+from graphon.entities import WorkflowStartReason
+from graphon.enums import WorkflowExecutionStatus
+from graphon.runtime import GraphRuntimeState
 from sqlalchemy.orm import Session, sessionmaker
 
 from constants.tts_auto_play_timeout import TTS_AUTO_PLAY_TIMEOUT, TTS_AUTO_PLAY_YIELD_CPU_TIME
@@ -58,9 +61,6 @@ from core.base.tts import AppGeneratorTTSPublisher, AudioTrunk
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.workflow.system_variables import build_system_variables
 from extensions.ext_database import db
-from graphon.entities import WorkflowStartReason
-from graphon.enums import WorkflowExecutionStatus
-from graphon.runtime import GraphRuntimeState
 from models import Account
 from models.enums import CreatorUserRole
 from models.model import EndUser

api/core/app/apps/workflow_app_runner.py

@@ -3,6 +3,39 @@ import time
 from collections.abc import Mapping, Sequence
 from typing import Any, cast
 
+from graphon.entities.graph_config import NodeConfigDictAdapter
+from graphon.entities.pause_reason import HumanInputRequired
+from graphon.graph import Graph
+from graphon.graph_engine.layers import GraphEngineLayer
+from graphon.graph_events import (
+    GraphEngineEvent,
+    GraphRunAbortedEvent,
+    GraphRunFailedEvent,
+    GraphRunPartialSucceededEvent,
+    GraphRunPausedEvent,
+    GraphRunStartedEvent,
+    GraphRunSucceededEvent,
+    NodeRunAgentLogEvent,
+    NodeRunExceptionEvent,
+    NodeRunFailedEvent,
+    NodeRunHumanInputFormFilledEvent,
+    NodeRunHumanInputFormTimeoutEvent,
+    NodeRunIterationFailedEvent,
+    NodeRunIterationNextEvent,
+    NodeRunIterationStartedEvent,
+    NodeRunIterationSucceededEvent,
+    NodeRunLoopFailedEvent,
+    NodeRunLoopNextEvent,
+    NodeRunLoopStartedEvent,
+    NodeRunLoopSucceededEvent,
+    NodeRunRetrieverResourceEvent,
+    NodeRunRetryEvent,
+    NodeRunStartedEvent,
+    NodeRunStreamChunkEvent,
+    NodeRunSucceededEvent,
+)
+from graphon.runtime import GraphRuntimeState, VariablePool
+from graphon.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader, load_into_variable_pool
 from pydantic import ValidationError
 
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
@@ -49,39 +82,6 @@ from core.workflow.system_variables import (
 from core.workflow.variable_pool_initializer import add_variables_to_pool
 from core.workflow.workflow_entry import WorkflowEntry
 from core.workflow.workflow_run_outputs import project_node_outputs_for_workflow_run
-from graphon.entities.graph_config import NodeConfigDictAdapter
-from graphon.entities.pause_reason import HumanInputRequired
-from graphon.graph import Graph
-from graphon.graph_engine.layers import GraphEngineLayer
-from graphon.graph_events import (
-    GraphEngineEvent,
-    GraphRunAbortedEvent,
-    GraphRunFailedEvent,
-    GraphRunPartialSucceededEvent,
-    GraphRunPausedEvent,
-    GraphRunStartedEvent,
-    GraphRunSucceededEvent,
-    NodeRunAgentLogEvent,
-    NodeRunExceptionEvent,
-    NodeRunFailedEvent,
-    NodeRunHumanInputFormFilledEvent,
-    NodeRunHumanInputFormTimeoutEvent,
-    NodeRunIterationFailedEvent,
-    NodeRunIterationNextEvent,
-    NodeRunIterationStartedEvent,
-    NodeRunIterationSucceededEvent,
-    NodeRunLoopFailedEvent,
-    NodeRunLoopNextEvent,
-    NodeRunLoopStartedEvent,
-    NodeRunLoopSucceededEvent,
-    NodeRunRetrieverResourceEvent,
-    NodeRunRetryEvent,
-    NodeRunStartedEvent,
-    NodeRunStreamChunkEvent,
-    NodeRunSucceededEvent,
-)
-from graphon.runtime import GraphRuntimeState, VariablePool
-from graphon.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader, load_into_variable_pool
 from models.workflow import Workflow
 from tasks.mail_human_input_delivery_task import dispatch_human_input_email_task
 

api/core/app/entities/app_invoke_entities.py

@@ -2,13 +2,13 @@ from collections.abc import Mapping, Sequence
 from enum import StrEnum
 from typing import TYPE_CHECKING, Any
 
+from graphon.file import File, FileUploadConfig
+from graphon.model_runtime.entities.model_entities import AIModelEntity
 from pydantic import BaseModel, ConfigDict, Field, ValidationInfo, field_validator
 
 from constants import UUID_NIL
 from core.app.app_config.entities import EasyUIBasedAppConfig, WorkflowUIBasedAppConfig
 from core.entities.provider_configuration import ProviderModelBundle
-from graphon.file import File, FileUploadConfig
-from graphon.model_runtime.entities.model_entities import AIModelEntity
 
 if TYPE_CHECKING:
     from core.ops.ops_trace_manager import TraceQueueManager

api/core/app/entities/queue_entities.py

@@ -3,14 +3,14 @@ from datetime import datetime
 from enum import StrEnum, auto
 from typing import Any
 
-from pydantic import BaseModel, ConfigDict, Field
-
-from core.app.entities.agent_strategy import AgentStrategyInfo
-from core.rag.entities import RetrievalSourceMetadata
 from graphon.entities import WorkflowStartReason
 from graphon.entities.pause_reason import PauseReason
 from graphon.enums import NodeType, WorkflowNodeExecutionMetadataKey
 from graphon.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk
+from pydantic import BaseModel, ConfigDict, Field
+
+from core.app.entities.agent_strategy import AgentStrategyInfo
+from core.rag.entities import RetrievalSourceMetadata
 
 
 class QueueEvent(StrEnum):

api/core/app/entities/task_entities.py

@@ -2,14 +2,14 @@ from collections.abc import Mapping, Sequence
 from enum import StrEnum
 from typing import Any
 
-from pydantic import BaseModel, ConfigDict, Field
-
-from core.app.entities.agent_strategy import AgentStrategyInfo
-from core.rag.entities import RetrievalSourceMetadata
 from graphon.entities import WorkflowStartReason
 from graphon.enums import WorkflowExecutionStatus, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
 from graphon.model_runtime.entities.llm_entities import LLMResult, LLMUsage
 from graphon.nodes.human_input.entities import FormInput, UserAction
+from pydantic import BaseModel, ConfigDict, Field
+
+from core.app.entities.agent_strategy import AgentStrategyInfo
+from core.rag.entities import RetrievalSourceMetadata
 
 
 class AnnotationReplyAccount(BaseModel):

api/core/app/features/hosting_moderation/hosting_moderation.py

@@ -1,8 +1,9 @@
 import logging
 
+from graphon.model_runtime.entities.message_entities import PromptMessage
+
 from core.app.entities.app_invoke_entities import EasyUIBasedAppGenerateEntity
 from core.helper import moderation
-from graphon.model_runtime.entities.message_entities import PromptMessage
 
 logger = logging.getLogger(__name__)
 

api/core/app/layers/pause_state_persist_layer.py

@@ -1,14 +1,14 @@
 from dataclasses import dataclass
 from typing import Annotated, Literal, Self
 
+from graphon.graph_engine.layers import GraphEngineLayer
+from graphon.graph_events import GraphEngineEvent, GraphRunPausedEvent
 from pydantic import BaseModel, Field
 from sqlalchemy import Engine
 from sqlalchemy.orm import Session, sessionmaker
 
 from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, WorkflowAppGenerateEntity
 from core.workflow.system_variables import SystemVariableKey, get_system_text
-from graphon.graph_engine.layers import GraphEngineLayer
-from graphon.graph_events import GraphEngineEvent, GraphRunPausedEvent
 from models.model import AppMode
 from repositories.api_workflow_run_repository import APIWorkflowRunRepository
 from repositories.factory import DifyAPIRepositoryFactory

api/core/app/layers/timeslice_layer.py

@@ -3,10 +3,10 @@ import uuid
 from typing import ClassVar
 
 from apscheduler.schedulers.background import BackgroundScheduler  # type: ignore
-
 from graphon.graph_engine.entities.commands import CommandType, GraphEngineCommand
 from graphon.graph_engine.layers import GraphEngineLayer
 from graphon.graph_events import GraphEngineEvent
+
 from services.workflow.entities import WorkflowScheduleCFSPlanEntity
 from services.workflow.scheduler import CFSPlanScheduler, SchedulerCommand
 

api/core/app/layers/trigger_post_layer.py

@@ -2,12 +2,12 @@ import logging
 from datetime import UTC, datetime
 from typing import Any, ClassVar
 
+from graphon.graph_engine.layers import GraphEngineLayer
+from graphon.graph_events import GraphEngineEvent, GraphRunFailedEvent, GraphRunPausedEvent, GraphRunSucceededEvent
 from pydantic import TypeAdapter
 
 from core.db.session_factory import session_factory
 from core.workflow.system_variables import SystemVariableKey, get_system_text
-from graphon.graph_engine.layers import GraphEngineLayer
-from graphon.graph_events import GraphEngineEvent, GraphRunFailedEvent, GraphRunPausedEvent, GraphRunSucceededEvent
 from models.enums import WorkflowTriggerStatus
 from repositories.sqlalchemy_workflow_trigger_log_repository import SQLAlchemyWorkflowTriggerLogRepository
 from tasks.workflow_cfs_scheduler.cfs_scheduler import AsyncWorkflowCFSPlanEntity

api/core/app/llm/model_access.py

@@ -2,15 +2,16 @@ from __future__ import annotations
 
 from typing import Any
 
+from graphon.model_runtime.entities.model_entities import ModelType
+from graphon.nodes.llm.entities import ModelConfig
+from graphon.nodes.llm.exc import LLMModeRequiredError, ModelNotExistError
+from graphon.nodes.llm.protocols import CredentialsProvider
+
 from core.app.entities.app_invoke_entities import DifyRunContext, ModelConfigWithCredentialsEntity
 from core.errors.error import ProviderTokenNotInitError
 from core.model_manager import ModelInstance, ModelManager
 from core.plugin.impl.model_runtime_factory import create_plugin_provider_manager
 from core.provider_manager import ProviderManager
-from graphon.model_runtime.entities.model_entities import ModelType
-from graphon.nodes.llm.entities import ModelConfig
-from graphon.nodes.llm.exc import LLMModeRequiredError, ModelNotExistError
-from graphon.nodes.llm.protocols import CredentialsProvider
 
 
 class DifyCredentialsProvider:

api/core/app/llm/quota.py

@@ -1,3 +1,4 @@
+from graphon.model_runtime.entities.llm_entities import LLMUsage
 from sqlalchemy import update
 from sqlalchemy.orm import sessionmaker
 
@@ -7,7 +8,6 @@ from core.entities.provider_entities import ProviderQuotaType, QuotaUnit
 from core.errors.error import QuotaExceededError
 from core.model_manager import ModelInstance
 from extensions.ext_database import db
-from graphon.model_runtime.entities.llm_entities import LLMUsage
 from libs.datetime_utils import naive_utc_now
 from models.provider import Provider, ProviderType
 from models.provider_ids import ModelProviderID

api/core/app/task_pipeline/based_generate_task_pipeline.py

@@ -1,6 +1,7 @@
 import logging
 import time
 
+from graphon.model_runtime.errors.invoke import InvokeAuthorizationError, InvokeError
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
@@ -17,7 +18,6 @@ from core.app.entities.task_entities import (
 )
 from core.errors.error import QuotaExceededError
 from core.moderation.output_moderation import ModerationRule, OutputModeration
-from graphon.model_runtime.errors.invoke import InvokeAuthorizationError, InvokeError
 from models.enums import MessageStatus
 from models.model import Message
 

api/core/app/workflow/layers/llm_quota.py

@@ -7,16 +7,17 @@ This layer centralizes model-quota deduction outside node implementations.
 import logging
 from typing import TYPE_CHECKING, cast, final, override
 
-from core.app.entities.app_invoke_entities import DIFY_RUN_CONTEXT_KEY, DifyRunContext
-from core.app.llm import deduct_llm_quota, ensure_llm_quota_available
-from core.errors.error import QuotaExceededError
-from core.model_manager import ModelInstance
 from graphon.enums import BuiltinNodeTypes
 from graphon.graph_engine.entities.commands import AbortCommand, CommandType
 from graphon.graph_engine.layers import GraphEngineLayer
 from graphon.graph_events import GraphEngineEvent, GraphNodeEventBase, NodeRunSucceededEvent
 from graphon.nodes.base.node import Node
 
+from core.app.entities.app_invoke_entities import DIFY_RUN_CONTEXT_KEY, DifyRunContext
+from core.app.llm import deduct_llm_quota, ensure_llm_quota_available
+from core.errors.error import QuotaExceededError
+from core.model_manager import ModelInstance
+
 if TYPE_CHECKING:
     from graphon.nodes.llm.node import LLMNode
     from graphon.nodes.parameter_extractor.parameter_extractor_node import ParameterExtractorNode

api/core/app/workflow/layers/persistence.py

@@ -14,13 +14,6 @@ from dataclasses import dataclass
 from datetime import datetime
 from typing import Any, Union
 
-from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, WorkflowAppGenerateEntity
-from core.ops.entities.trace_entity import TraceTaskName
-from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
-from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
-from core.workflow.system_variables import SystemVariableKey
-from core.workflow.variable_prefixes import SYSTEM_VARIABLE_NODE_ID
-from core.workflow.workflow_run_outputs import project_node_outputs_for_workflow_run
 from graphon.entities import WorkflowExecution, WorkflowNodeExecution
 from graphon.enums import (
     WorkflowExecutionStatus,
@@ -45,6 +38,14 @@ from graphon.graph_events import (
     NodeRunSucceededEvent,
 )
 from graphon.node_events import NodeRunResult
+
+from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, WorkflowAppGenerateEntity
+from core.ops.entities.trace_entity import TraceTaskName
+from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
+from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
+from core.workflow.system_variables import SystemVariableKey
+from core.workflow.variable_prefixes import SYSTEM_VARIABLE_NODE_ID
+from core.workflow.workflow_run_outputs import project_node_outputs_for_workflow_run
 from libs.datetime_utils import naive_utc_now
 
 

api/core/base/tts/app_generator_tts_publisher.py

@@ -6,6 +6,9 @@ import re
 import threading
 from collections.abc import Iterable
 
+from graphon.model_runtime.entities.message_entities import TextPromptMessageContent
+from graphon.model_runtime.entities.model_entities import ModelType
+
 from core.app.entities.queue_entities import (
     MessageQueueMessage,
     QueueAgentMessageEvent,
@@ -15,8 +18,6 @@ from core.app.entities.queue_entities import (
     WorkflowQueueMessage,
 )
 from core.model_manager import ModelInstance, ModelManager
-from graphon.model_runtime.entities.message_entities import TextPromptMessageContent
-from graphon.model_runtime.entities.model_entities import ModelType
 
 
 class AudioTrunk:

api/core/datasource/datasource_manager.py

@@ -3,6 +3,9 @@ from collections.abc import Generator
 from threading import Lock
 from typing import Any, cast
 
+from graphon.enums import WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
+from graphon.file import File, FileTransferMethod, FileType, get_file_type_by_mime_type
+from graphon.node_events import NodeRunResult, StreamChunkEvent, StreamCompletedEvent
 from sqlalchemy import select
 
 import contexts
@@ -28,9 +31,6 @@ from core.plugin.impl.datasource import PluginDatasourceManager
 from core.workflow.file_reference import build_file_reference
 from core.workflow.nodes.datasource.entities import DatasourceParameter, OnlineDriveDownloadFileParam
 from factories import file_factory
-from graphon.enums import WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
-from graphon.file import File, FileTransferMethod, FileType, get_file_type_by_mime_type
-from graphon.node_events import NodeRunResult, StreamChunkEvent, StreamCompletedEvent
 from models.model import UploadFile
 from models.tools import ToolFile
 from services.datasource_provider_service import DatasourceProviderService

api/core/datasource/entities/api_entities.py

@@ -1,10 +1,10 @@
 from typing import Any, Literal, TypedDict
 
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 from pydantic import BaseModel, Field, field_validator
 
 from core.datasource.entities.datasource_entities import DatasourceParameter
 from core.tools.entities.common_entities import I18nObject, I18nObjectDict
-from graphon.model_runtime.utils.encoders import jsonable_encoder
 
 
 class DatasourceApiEntity(BaseModel):