feat: allow duplicate snippet names and enhance snippet import handling

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

.github/workflows/pyrefly-type-coverage-comment.yml

@@ -63,8 +63,8 @@ jobs:
         id: render
         run: |
           comment_body="$(uv run --directory api python libs/pyrefly_type_coverage.py \
-            --base "$GITHUB_WORKSPACE/base_report.json" \
-            < "$GITHUB_WORKSPACE/pr_report.json")"
+            --base base_report.json \
+            < pr_report.json)"
 
           {
             echo "### Pyrefly Type Coverage"

.github/workflows/pyrefly-type-coverage.yml

@@ -65,9 +65,6 @@ jobs:
           # Save structured data for the fork-PR comment workflow
           cp /tmp/pyrefly_report_pr.json pr_report.json
           cp /tmp/pyrefly_report_base.json base_report.json
-          # Keep fork-PR comments correct while the trusted workflow_run job is
-          # still using the default-branch renderer, which resolves --base from api/.
-          cp /tmp/pyrefly_report_base.json api/base_report.json
 
       - name: Save PR number
         run: |
@@ -80,7 +77,6 @@ jobs:
           path: |
             pr_report.json
             base_report.json
-            api/base_report.json
             pr_number.txt
 
       - name: Comment PR with type coverage

.github/workflows/style.yml

@@ -47,10 +47,6 @@ jobs:
         if: steps.changed-files.outputs.any_changed == 'true'
         run: uv run --directory api --dev lint-imports
 
-      - name: Run Response Contract Linter
-        if: steps.changed-files.outputs.any_changed == 'true'
-        run: uv run --project api --dev python api/dev/lint_response_contracts.py --fail-on-mismatch
-
       - name: Run Type Checks
         if: steps.changed-files.outputs.any_changed == 'true'
         run: make type-check-core

Makefile

@@ -75,19 +75,13 @@ check:
 	@echo "✅ Code check complete"
 
 lint:
-	@echo "🔧 Running ruff format, check with fixes, response contract lint, import linter, and dotenv-linter..."
+	@echo "🔧 Running ruff format, check with fixes, import linter, and dotenv-linter..."
 	@uv run --project api --dev ruff format ./api
 	@uv run --project api --dev ruff check --fix ./api
-	@$(MAKE) api-contract-lint
 	@uv run --directory api --dev lint-imports
 	@uv run --project api --dev dotenv-linter ./api/.env.example ./web/.env.example
 	@echo "✅ Linting complete"
 
-api-contract-lint:
-	@echo "🔎 Linting Flask response contracts..."
-	@uv run --project api --dev python api/dev/lint_response_contracts.py
-	@echo "✅ Response contract lint complete"
-
 type-check:
 	@echo "📝 Running type checks (pyrefly + mypy)..."
 	@./dev/pyrefly-check-local $(PATH_TO_CHECK)
@@ -197,7 +191,6 @@ help:
 	@echo "  make format         - Format code with ruff"
 	@echo "  make check          - Check code with ruff"
 	@echo "  make lint           - Format, fix, and lint code (ruff, imports, dotenv)"
-	@echo "  make api-contract-lint - Check Flask response docs against returned schemas"
 	@echo "  make type-check     - Run type checks (pyrefly, mypy)"
 	@echo "  make type-check-core - Run core type checks (pyrefly, mypy)"
 	@echo "  make test           - Run backend unit tests (or TARGET_TESTS=./api/tests/<target_tests>)"
@@ -211,4 +204,4 @@ help:
 	@echo "  make build-push-all - Build and push all Docker images"
 
 # Phony targets
-.PHONY: build-web build-api push-web push-api build-all push-all build-push-all dev-setup prepare-docker prepare-web prepare-api dev-clean help format check lint api-contract-lint type-check test test-all
+.PHONY: build-web build-api push-web push-api build-all push-all build-push-all dev-setup prepare-docker prepare-web prepare-api dev-clean help format check lint type-check test test-all

api/.env.example

@@ -657,7 +657,6 @@ PLUGIN_REMOTE_INSTALL_PORT=5003
 PLUGIN_REMOTE_INSTALL_HOST=localhost
 PLUGIN_MAX_PACKAGE_SIZE=15728640
 PLUGIN_MODEL_SCHEMA_CACHE_TTL=3600
-PLUGIN_MODEL_PROVIDERS_CACHE_TTL=86400
 INNER_API_KEY_FOR_PLUGIN=QaHbTe77CtuXmsfyhR7+vRjI/+XbV1AaFy691iy+kGDv2Jvy0/eAh8Y1
 
 # Marketplace configuration

api/AGENTS.md

@@ -195,7 +195,6 @@ Before opening a PR / submitting:
 - Controllers: parse input via Pydantic, invoke services, return serialised responses; no business logic.
 - Services: coordinate repositories, providers, background tasks; keep side effects explicit.
 - Document non-obvious behaviour with concise docstrings and comments.
-- For `204 No Content` responses, return an empty body only; never return a dict, model, or other payload.
 - For Flask-RESTX controller request, query, and response schemas, follow `controllers/API_SCHEMA_GUIDE.md`.
   In short: use Pydantic models, document GET query params with `query_params_from_model(...)`, register response
   DTOs with `register_response_schema_models(...)`, serialize response DTOs with `dump_response(...)`,

api/clients/agent_backend/request_builder.py

@@ -49,7 +49,6 @@ class AgentBackendModelConfig(BaseModel):
     model: str
     user_id: str | None = None
     credentials: dict[str, DifyPluginCredentialValue] = Field(default_factory=dict)
-    model_settings: dict[str, JsonValue] = Field(default_factory=dict)
 
     model_config: ClassVar[ConfigDict] = ConfigDict(extra="forbid")
 
@@ -139,7 +138,6 @@ class AgentBackendRunRequestBuilder:
                         model_provider=run_input.model.model_provider,
                         model=run_input.model.model,
                         credentials=run_input.model.credentials,
-                        model_settings=run_input.model.model_settings or None,
                     ),
                 ),
             ]

api/commands/plugin.py

@@ -11,7 +11,6 @@ from configs import dify_config
 from core.helper import encrypter
 from core.plugin.entities.plugin_daemon import CredentialType
 from core.plugin.impl.plugin import PluginInstaller
-from core.plugin.plugin_service import PluginService
 from core.tools.utils.system_encryption import encrypt_system_params
 from extensions.ext_database import db
 from models import Tenant
@@ -21,6 +20,7 @@ from models.source import DataSourceApiKeyAuthBinding, DataSourceOauthBinding
 from models.tools import ToolOAuthSystemClient
 from services.plugin.data_migration import PluginDataMigration
 from services.plugin.plugin_migration import PluginMigration
+from services.plugin.plugin_service import PluginService
 
 logger = logging.getLogger(__name__)
 

api/configs/extra/__init__.py

@@ -1,4 +1,3 @@
-from configs.extra.agent_backend_config import AgentBackendConfig
 from configs.extra.archive_config import ArchiveStorageConfig
 from configs.extra.notion_config import NotionConfig
 from configs.extra.sentry_config import SentryConfig
@@ -6,7 +5,6 @@ from configs.extra.sentry_config import SentryConfig
 
 class ExtraServiceConfig(
     # place the configs in alphabet order
-    AgentBackendConfig,
     ArchiveStorageConfig,
     NotionConfig,
     SentryConfig,

api/configs/extra/agent_backend_config.py

@@ -1,23 +0,0 @@
-from pydantic import Field
-from pydantic_settings import BaseSettings
-
-
-class AgentBackendConfig(BaseSettings):
-    """
-    Configuration settings for the Agent backend runtime integration.
-    """
-
-    AGENT_BACKEND_BASE_URL: str | None = Field(
-        description="Base URL for the Dify Agent backend service.",
-        default=None,
-    )
-
-    AGENT_BACKEND_USE_FAKE: bool = Field(
-        description="Use the deterministic in-process fake Agent backend client.",
-        default=False,
-    )
-
-    AGENT_BACKEND_FAKE_SCENARIO: str = Field(
-        description="Scenario used by the fake Agent backend client.",
-        default="success",
-    )

api/configs/feature/__init__.py

@@ -265,11 +265,6 @@ class PluginConfig(BaseSettings):
         default=60 * 60,
     )
 
-    PLUGIN_MODEL_PROVIDERS_CACHE_TTL: PositiveInt = Field(
-        description="TTL in seconds for caching tenant plugin model providers in Redis",
-        default=60 * 60 * 24,
-    )
-
     PLUGIN_MAX_FILE_SIZE: PositiveInt = Field(
         description="Maximum allowed size (bytes) for plugin-generated files",
         default=50 * 1024 * 1024,

api/configs/remote_settings_sources/apollo/__init__.py

@@ -1,5 +1,5 @@
 from collections.abc import Mapping
-from typing import Any, override
+from typing import Any
 
 from pydantic import Field
 from pydantic.fields import FieldInfo
@@ -48,7 +48,6 @@ class ApolloSettingsSource(RemoteSettingsSource):
         self.namespace = configs["APOLLO_NAMESPACE"]
         self.remote_configs = self.client.get_all_dicts(self.namespace)
 
-    @override
     def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]:
         if not isinstance(self.remote_configs, dict):
             raise ValueError(f"remote configs is not dict, but {type(self.remote_configs)}")

api/configs/remote_settings_sources/nacos/__init__.py

@@ -1,7 +1,7 @@
 import logging
 import os
 from collections.abc import Mapping
-from typing import Any, override
+from typing import Any
 
 from pydantic.fields import FieldInfo
 
@@ -41,7 +41,6 @@ class NacosSettingsSource(RemoteSettingsSource):
         except Exception as e:
             raise RuntimeError(f"Failed to parse config: {e}")
 
-    @override
     def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]:
         field_value = self.remote_configs.get(field_name)
         if field_value is None:

api/context/execution_context.py

@@ -10,7 +10,7 @@ import threading
 from abc import ABC, abstractmethod
 from collections.abc import Callable, Generator
 from contextlib import AbstractContextManager, contextmanager
-from typing import Any, Protocol, final, override, runtime_checkable
+from typing import Any, Protocol, final, runtime_checkable
 
 from pydantic import BaseModel
 
@@ -133,12 +133,10 @@ class NullAppContext(AppContext):
         self._config = config or {}
         self._extensions: dict[str, Any] = {}
 
-    @override
     def get_config(self, key: str, default: Any = None) -> Any:
         """Get configuration value by key."""
         return self._config.get(key, default)
 
-    @override
     def get_extension(self, name: str) -> Any:
         """Get extension by name."""
         return self._extensions.get(name)
@@ -148,7 +146,6 @@ class NullAppContext(AppContext):
         self._extensions[name] = extension
 
     @contextmanager
-    @override
     def enter(self) -> Generator[None, None, None]:
         """Enter null context (no-op)."""
         yield

api/context/flask_app_context.py

@@ -6,7 +6,7 @@ import contextvars
 import threading
 from collections.abc import Generator
 from contextlib import contextmanager
-from typing import Any, final, override
+from typing import Any, final
 
 from flask import Flask, current_app, g
 
@@ -30,18 +30,15 @@ class FlaskAppContext(AppContext):
         """
         self._flask_app = flask_app
 
-    @override
     def get_config(self, key: str, default: Any = None) -> Any:
         """Get configuration value from Flask app config."""
         return self._flask_app.config.get(key, default)
 
-    @override
     def get_extension(self, name: str) -> Any:
         """Get Flask extension by name."""
         return self._flask_app.extensions.get(name)
 
     @contextmanager
-    @override
     def enter(self) -> Generator[None, None, None]:
         """Enter Flask app context."""
         with self._flask_app.app_context():

api/controllers/common/helpers.py

@@ -36,24 +36,6 @@ class FileInfo(BaseModel):
     size: int
 
 
-def decode_remote_url(url: str, query_string: bytes | str = b"") -> str:
-    decoded_url = urllib.parse.unquote(url)
-    if isinstance(query_string, bytes):
-        raw_query = query_string.decode()
-    else:
-        raw_query = query_string
-    if not raw_query:
-        return decoded_url
-
-    if decoded_url.endswith(("?", "&")):
-        separator = ""
-    elif urllib.parse.urlsplit(decoded_url).query:
-        separator = "&"
-    else:
-        separator = "?"
-    return f"{decoded_url}{separator}{raw_query}"
-
-
 def guess_file_info_from_response(response: httpx.Response):
     url = str(response.url)
     # Try to extract filename from URL

api/controllers/console/__init__.py

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

api/controllers/console/apikey.py

@@ -146,7 +146,7 @@ class BaseApiKeyResource(Resource):
         db.session.execute(delete(ApiToken).where(ApiToken.id == api_key_id))
         db.session.commit()
 
-        return "", 204
+        return {"result": "success"}, 204
 
 
 @console_ns.route("/apps/<uuid:resource_id>/api-keys")

api/controllers/console/app/annotation.py

@@ -269,12 +269,12 @@ class AnnotationApi(Resource):
                     "message": "annotation_ids are required if the parameter is provided.",
                 }, 400
 
-            AppAnnotationService.delete_app_annotations_in_batch(str(app_id), annotation_ids)
-            return "", 204
+            result = AppAnnotationService.delete_app_annotations_in_batch(str(app_id), annotation_ids)
+            return result, 204
         # If no annotation_ids are provided, handle clearing all annotations
         else:
             AppAnnotationService.clear_all_annotations(str(app_id))
-            return "", 204
+            return {"result": "success"}, 204
 
 
 @console_ns.route("/apps/<uuid:app_id>/annotations/export")
@@ -335,7 +335,7 @@ class AnnotationUpdateDeleteApi(Resource):
     @edit_permission_required
     def delete(self, app_id: UUID, annotation_id: UUID):
         AppAnnotationService.delete_app_annotation(str(app_id), str(annotation_id))
-        return "", 204
+        return {"result": "success"}, 204
 
 
 @console_ns.route("/apps/<uuid:app_id>/annotations/batch-import")

api/controllers/console/app/app.py

@@ -633,7 +633,7 @@ class AppApi(Resource):
         app_service = AppService()
         app_service.delete_app(app_model)
 
-        return "", 204
+        return {"result": "success"}, 204
 
 
 @console_ns.route("/apps/<uuid:app_id>/copy")

api/controllers/console/app/conversation.py

@@ -29,6 +29,9 @@ from fields.conversation_fields import (
 from fields.conversation_fields import (
     ConversationWithSummaryPagination as ConversationWithSummaryPaginationResponse,
 )
+from fields.conversation_fields import (
+    ResultResponse,
+)
 from libs.datetime_utils import naive_utc_now, parse_time_range
 from libs.login import current_account_with_tenant, login_required
 from models import Conversation, EndUser, Message, MessageAnnotation
@@ -74,6 +77,7 @@ register_schema_models(
     ConversationMessageDetailResponse,
     ConversationWithSummaryPaginationResponse,
     ConversationDetailResponse,
+    ResultResponse,
     CompletionConversationQuery,
     ChatConversationQuery,
 )
@@ -190,7 +194,7 @@ class CompletionConversationDetailApi(Resource):
         except ConversationNotExistsError:
             raise NotFound("Conversation Not Exists.")
 
-        return "", 204
+        return ResultResponse(result="success").model_dump(mode="json"), 204
 
 
 @console_ns.route("/apps/<uuid:app_id>/chat-conversations")
@@ -343,7 +347,7 @@ class ChatConversationDetailApi(Resource):
         except ConversationNotExistsError:
             raise NotFound("Conversation Not Exists.")
 
-        return "", 204
+        return ResultResponse(result="success").model_dump(mode="json"), 204
 
 
 def _get_conversation(app_model, conversation_id):

api/controllers/console/app/ops_trace.py

@@ -128,6 +128,6 @@ class TraceAppConfigApi(Resource):
             result = OpsService.delete_tracing_app_config(app_id=app_model.id, tracing_provider=args.tracing_provider)
             if not result:
                 raise TracingConfigNotExist()
-            return "", 204
+            return {"result": "success"}, 204
         except Exception as e:
             raise BadRequest(str(e))

api/controllers/console/app/workflow_comment.py

@@ -311,7 +311,7 @@ class WorkflowCommentDetailApi(Resource):
             user_id=current_user.id,
         )
 
-        return "", 204
+        return {"result": "success"}, 204
 
 
 @console_ns.route("/apps/<uuid:app_id>/workflow/comments/<string:comment_id>/resolve")
@@ -431,7 +431,7 @@ class WorkflowCommentReplyDetailApi(Resource):
             user_id=current_user.id,
         )
 
-        return "", 204
+        return {"result": "success"}, 204
 
 
 @console_ns.route("/apps/<uuid:app_id>/workflow/comments/mention-users")

api/controllers/console/auth/data_source_bearer_auth.py

@@ -93,4 +93,4 @@ class ApiKeyAuthDataSourceBindingDelete(Resource):
 
         ApiKeyAuthService.delete_provider_auth(current_tenant_id, binding_id)
 
-        return "", 204
+        return {"result": "success"}, 204

api/controllers/console/datasets/datasets.py

@@ -1,16 +1,15 @@
-from datetime import datetime
-from typing import Any
+from typing import Any, cast
 
 from flask import request
-from flask_restx import Resource
-from pydantic import BaseModel, Field, field_validator, model_validator
+from flask_restx import Resource, fields, marshal, marshal_with
+from pydantic import BaseModel, Field, field_validator
 from sqlalchemy import func, select
 from werkzeug.exceptions import Forbidden, NotFound
 
 import services
 from configs import dify_config
 from controllers.common.fields import ApiBaseUrlResponse, SimpleResultResponse, UsageCheckResponse
-from controllers.common.schema import query_params_from_model, register_response_schema_models, register_schema_models
+from controllers.common.schema import get_or_create_model, register_response_schema_models, register_schema_models
 from controllers.console import console_ns
 from controllers.console.apikey import ApiKeyItem, ApiKeyList
 from controllers.console.app.error import ProviderNotInitializeError
@@ -31,10 +30,26 @@ 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 fields.base import ResponseModel
-from fields.dataset_fields import DatasetDetailResponse
+from fields.app_fields import app_detail_kernel_fields, related_app_list
+from fields.dataset_fields import (
+    content_fields,
+    dataset_detail_fields,
+    dataset_fields,
+    dataset_query_detail_fields,
+    dataset_retrieval_model_fields,
+    doc_metadata_fields,
+    external_knowledge_info_fields,
+    external_retrieval_model_fields,
+    file_info_fields,
+    icon_info_fields,
+    keyword_setting_fields,
+    reranking_model_fields,
+    tag_fields,
+    vector_setting_fields,
+    weighted_score_fields,
+)
+from fields.document_fields import document_status_fields
 from graphon.model_runtime.entities.model_entities import ModelType
-from libs.helper import build_icon_url, dump_response, to_timestamp
 from libs.login import current_account_with_tenant, login_required
 from libs.url_utils import normalize_api_base_url
 from models import ApiToken, Dataset, Document, DocumentSegment, UploadFile
@@ -46,6 +61,58 @@ from services.dataset_service import DatasetPermissionService, DatasetService, D
 
 register_response_schema_models(console_ns, ApiBaseUrlResponse, SimpleResultResponse, UsageCheckResponse)
 
+# Register models for flask_restx to avoid dict type issues in Swagger
+dataset_base_model = get_or_create_model("DatasetBase", dataset_fields)
+
+tag_model = get_or_create_model("Tag", tag_fields)
+
+keyword_setting_model = get_or_create_model("DatasetKeywordSetting", keyword_setting_fields)
+vector_setting_model = get_or_create_model("DatasetVectorSetting", vector_setting_fields)
+
+weighted_score_fields_copy = weighted_score_fields.copy()
+weighted_score_fields_copy["keyword_setting"] = fields.Nested(keyword_setting_model)
+weighted_score_fields_copy["vector_setting"] = fields.Nested(vector_setting_model)
+weighted_score_model = get_or_create_model("DatasetWeightedScore", weighted_score_fields_copy)
+
+reranking_model = get_or_create_model("DatasetRerankingModel", reranking_model_fields)
+
+dataset_retrieval_model_fields_copy = dataset_retrieval_model_fields.copy()
+dataset_retrieval_model_fields_copy["reranking_model"] = fields.Nested(reranking_model)
+dataset_retrieval_model_fields_copy["weights"] = fields.Nested(weighted_score_model, allow_null=True)
+dataset_retrieval_model = get_or_create_model("DatasetRetrievalModel", dataset_retrieval_model_fields_copy)
+
+external_knowledge_info_model = get_or_create_model("ExternalKnowledgeInfo", external_knowledge_info_fields)
+
+external_retrieval_model = get_or_create_model("ExternalRetrievalModel", external_retrieval_model_fields)
+
+doc_metadata_model = get_or_create_model("DatasetDocMetadata", doc_metadata_fields)
+
+icon_info_model = get_or_create_model("DatasetIconInfo", icon_info_fields)
+
+dataset_detail_fields_copy = dataset_detail_fields.copy()
+dataset_detail_fields_copy["retrieval_model_dict"] = fields.Nested(dataset_retrieval_model)
+dataset_detail_fields_copy["tags"] = fields.List(fields.Nested(tag_model))
+dataset_detail_fields_copy["external_knowledge_info"] = fields.Nested(external_knowledge_info_model)
+dataset_detail_fields_copy["external_retrieval_model"] = fields.Nested(external_retrieval_model, allow_null=True)
+dataset_detail_fields_copy["doc_metadata"] = fields.List(fields.Nested(doc_metadata_model))
+dataset_detail_fields_copy["icon_info"] = fields.Nested(icon_info_model)
+dataset_detail_model = get_or_create_model("DatasetDetail", dataset_detail_fields_copy)
+
+file_info_model = get_or_create_model("DatasetFileInfo", file_info_fields)
+
+content_fields_copy = content_fields.copy()
+content_fields_copy["file_info"] = fields.Nested(file_info_model, allow_null=True)
+content_model = get_or_create_model("DatasetContent", content_fields_copy)
+
+dataset_query_detail_fields_copy = dataset_query_detail_fields.copy()
+dataset_query_detail_fields_copy["queries"] = fields.Nested(content_model)
+dataset_query_detail_model = get_or_create_model("DatasetQueryDetail", dataset_query_detail_fields_copy)
+
+app_detail_kernel_model = get_or_create_model("AppDetailKernel", app_detail_kernel_fields)
+related_app_list_copy = related_app_list.copy()
+related_app_list_copy["data"] = fields.List(fields.Nested(app_detail_kernel_model))
+related_app_list_model = get_or_create_model("RelatedAppList", related_app_list_copy)
+
 
 def _validate_indexing_technique(value: str | None) -> str | None:
     if value is None:
@@ -141,165 +208,9 @@ class ConsoleDatasetListQuery(BaseModel):
     tag_ids: list[str] = Field(default_factory=list, description="Filter by tag IDs")
 
 
-class DatasetListItemResponse(DatasetDetailResponse):
-    partial_member_list: list[str]
-
-
-class DatasetListResponse(ResponseModel):
-    data: list[DatasetListItemResponse]
-    has_more: bool
-    limit: int
-    total: int
-    page: int
-
-
-class DatasetDetailWithPartialMembersResponse(DatasetDetailResponse):
-    partial_member_list: list[str] | None = None
-
-
-class DatasetQueryFileInfoResponse(ResponseModel):
-    id: str
-    name: str
-    size: int
-    extension: str
-    mime_type: str
-    source_url: str
-
-
-class DatasetQueryContentResponse(ResponseModel):
-    content_type: str
-    content: str
-    file_info: DatasetQueryFileInfoResponse | None = None
-
-
-class DatasetQueryDetailResponse(ResponseModel):
-    id: str
-    queries: list[DatasetQueryContentResponse]
-    source: str
-    source_app_id: str | None
-    created_by_role: str
-    created_by: str
-    created_at: int
-
-    @field_validator("created_at", mode="before")
-    @classmethod
-    def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        return to_timestamp(value)
-
-
-class DatasetQueryListResponse(ResponseModel):
-    data: list[DatasetQueryDetailResponse]
-    has_more: bool
-    limit: int
-    total: int
-    page: int
-
-
-class RelatedAppResponse(ResponseModel):
-    id: str
-    name: str
-    description: str
-    mode: str = Field(validation_alias="mode_compatible_with_agent")
-    icon_type: str | None
-    icon: str | None
-    icon_background: str | None
-    icon_url: str | None = None
-
-    @model_validator(mode="after")
-    def _set_icon_url(self) -> "RelatedAppResponse":
-        self.icon_url = self.icon_url or build_icon_url(self.icon_type, self.icon)
-        return self
-
-
-class RelatedAppListResponse(ResponseModel):
-    data: list[RelatedAppResponse]
-    total: int
-
-
-class DocumentStatusResponse(ResponseModel):
-    id: str
-    indexing_status: str
-    processing_started_at: int | None
-    parsing_completed_at: int | None
-    cleaning_completed_at: int | None
-    splitting_completed_at: int | None
-    completed_at: int | None
-    paused_at: int | None
-    error: str | None
-    stopped_at: int | None
-    completed_segments: int | None = None
-    total_segments: int | None = None
-
-    @field_validator(
-        "processing_started_at",
-        "parsing_completed_at",
-        "cleaning_completed_at",
-        "splitting_completed_at",
-        "completed_at",
-        "paused_at",
-        "stopped_at",
-        mode="before",
-    )
-    @classmethod
-    def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return to_timestamp(value)
-
-
-class DocumentStatusListResponse(ResponseModel):
-    data: list[DocumentStatusResponse]
-
-
-class ErrorDocsResponse(DocumentStatusListResponse):
-    total: int
-
-
-class IndexingEstimatePreviewItemResponse(ResponseModel):
-    content: str
-    child_chunks: list[str] | None = None
-    summary: str | None = None
-
-
-class IndexingEstimateQaPreviewItemResponse(ResponseModel):
-    question: str
-    answer: str
-
-
-class IndexingEstimateResponse(ResponseModel):
-    total_segments: int
-    preview: list[IndexingEstimatePreviewItemResponse]
-    qa_preview: list[IndexingEstimateQaPreviewItemResponse] | None = None
-
-
-class RetrievalSettingResponse(ResponseModel):
-    retrieval_method: list[str]
-
-
-class PartialMemberListResponse(ResponseModel):
-    data: list[str]
-
-
-class AutoDisableLogsResponse(ResponseModel):
-    document_ids: list[str]
-    count: int
-
-
 register_schema_models(
     console_ns, DatasetCreatePayload, DatasetUpdatePayload, IndexingEstimatePayload, ConsoleDatasetListQuery
 )
-register_response_schema_models(
-    console_ns,
-    DatasetDetailResponse,
-    DatasetDetailWithPartialMembersResponse,
-    DatasetListResponse,
-    DatasetQueryListResponse,
-    IndexingEstimateResponse,
-    RelatedAppListResponse,
-    DocumentStatusListResponse,
-    ErrorDocsResponse,
-    RetrievalSettingResponse,
-    PartialMemberListResponse,
-    AutoDisableLogsResponse,
-)
 
 
 def _get_retrieval_methods_by_vector_type(vector_type: str | None, is_mock: bool = False) -> dict[str, list[str]]:
@@ -382,8 +293,17 @@ def _get_retrieval_methods_by_vector_type(vector_type: str | None, is_mock: bool
 class DatasetListApi(Resource):
     @console_ns.doc("get_datasets")
     @console_ns.doc(description="Get list of datasets")
-    @console_ns.doc(params=query_params_from_model(ConsoleDatasetListQuery))
-    @console_ns.response(200, "Datasets retrieved successfully", console_ns.models[DatasetListResponse.__name__])
+    @console_ns.doc(
+        params={
+            "page": "Page number (default: 1)",
+            "limit": "Number of items per page (default: 20)",
+            "ids": "Filter by dataset IDs (list)",
+            "keyword": "Search keyword",
+            "tag_ids": "Filter by tag IDs (list)",
+            "include_all": "Include all datasets (default: false)",
+        }
+    )
+    @console_ns.response(200, "Datasets retrieved successfully")
     @setup_required
     @login_required
     @account_initialization_required
@@ -422,7 +342,7 @@ class DatasetListApi(Resource):
         for embedding_model in embedding_models:
             model_names.append(f"{embedding_model.model}:{embedding_model.provider.provider}")
 
-        data = [dump_response(DatasetDetailResponse, dataset) for dataset in datasets]
+        data = cast(list[dict[str, Any]], marshal(datasets, dataset_detail_fields))
         dataset_ids = [item["id"] for item in data if item.get("permission") == "partial_members"]
         partial_members_map: dict[str, list[str]] = {}
         if dataset_ids:
@@ -459,12 +379,12 @@ class DatasetListApi(Resource):
             "total": total,
             "page": query.page,
         }
-        return dump_response(DatasetListResponse, response), 200
+        return response, 200
 
     @console_ns.doc("create_dataset")
     @console_ns.doc(description="Create a new dataset")
     @console_ns.expect(console_ns.models[DatasetCreatePayload.__name__])
-    @console_ns.response(201, "Dataset created successfully", console_ns.models[DatasetDetailResponse.__name__])
+    @console_ns.response(201, "Dataset created successfully")
     @console_ns.response(400, "Invalid request parameters")
     @setup_required
     @login_required
@@ -493,7 +413,7 @@ class DatasetListApi(Resource):
         except services.errors.dataset.DatasetNameDuplicateError:
             raise DatasetNameDuplicateError()
 
-        return dump_response(DatasetDetailResponse, dataset), 201
+        return marshal(dataset, dataset_detail_fields), 201
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>")
@@ -501,11 +421,7 @@ class DatasetApi(Resource):
     @console_ns.doc("get_dataset")
     @console_ns.doc(description="Get dataset details")
     @console_ns.doc(params={"dataset_id": "Dataset ID"})
-    @console_ns.response(
-        200,
-        "Dataset retrieved successfully",
-        console_ns.models[DatasetDetailWithPartialMembersResponse.__name__],
-    )
+    @console_ns.response(200, "Dataset retrieved successfully", dataset_detail_model)
     @console_ns.response(404, "Dataset not found")
     @console_ns.response(403, "Permission denied")
     @setup_required
@@ -521,7 +437,7 @@ class DatasetApi(Resource):
             DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
-        data = dump_response(DatasetDetailResponse, dataset)
+        data = cast(dict[str, Any], marshal(dataset, dataset_detail_fields))
         if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
             if dataset.embedding_model_provider:
                 provider_id = ModelProviderID(dataset.embedding_model_provider)
@@ -554,11 +470,7 @@ class DatasetApi(Resource):
     @console_ns.doc("update_dataset")
     @console_ns.doc(description="Update dataset details")
     @console_ns.expect(console_ns.models[DatasetUpdatePayload.__name__])
-    @console_ns.response(
-        200,
-        "Dataset updated successfully",
-        console_ns.models[DatasetDetailWithPartialMembersResponse.__name__],
-    )
+    @console_ns.response(200, "Dataset updated successfully", dataset_detail_model)
     @console_ns.response(404, "Dataset not found")
     @console_ns.response(403, "Permission denied")
     @setup_required
@@ -594,7 +506,7 @@ class DatasetApi(Resource):
         if dataset is None:
             raise NotFound("Dataset not found.")
 
-        result_data = dump_response(DatasetDetailResponse, dataset)
+        result_data = cast(dict[str, Any], marshal(dataset, dataset_detail_fields))
         tenant_id = current_tenant_id
 
         if payload.partial_member_list is not None and payload.permission == DatasetPermissionEnum.PARTIAL_TEAM:
@@ -623,7 +535,7 @@ class DatasetApi(Resource):
         try:
             if DatasetService.delete_dataset(dataset_id_str, current_user):
                 DatasetPermissionService.clear_partial_member_list(dataset_id_str)
-                return "", 204
+                return {"result": "success"}, 204
             else:
                 raise NotFound("Dataset not found.")
         except services.errors.dataset.DatasetInUseError:
@@ -655,11 +567,7 @@ class DatasetQueryApi(Resource):
     @console_ns.doc("get_dataset_queries")
     @console_ns.doc(description="Get dataset query history")
     @console_ns.doc(params={"dataset_id": "Dataset ID"})
-    @console_ns.response(
-        200,
-        "Query history retrieved successfully",
-        console_ns.models[DatasetQueryListResponse.__name__],
-    )
+    @console_ns.response(200, "Query history retrieved successfully", dataset_query_detail_model)
     @setup_required
     @login_required
     @account_initialization_required
@@ -681,24 +589,20 @@ class DatasetQueryApi(Resource):
         dataset_queries, total = DatasetService.get_dataset_queries(dataset_id=dataset.id, page=page, per_page=limit)
 
         response = {
-            "data": dataset_queries,
+            "data": marshal(dataset_queries, dataset_query_detail_model),
             "has_more": len(dataset_queries) == limit,
             "limit": limit,
             "total": total,
             "page": page,
         }
-        return dump_response(DatasetQueryListResponse, response), 200
+        return response, 200
 
 
 @console_ns.route("/datasets/indexing-estimate")
 class DatasetIndexingEstimateApi(Resource):
     @console_ns.doc("estimate_dataset_indexing")
     @console_ns.doc(description="Estimate dataset indexing cost")
-    @console_ns.response(
-        200,
-        "Indexing estimate calculated successfully",
-        console_ns.models[IndexingEstimateResponse.__name__],
-    )
+    @console_ns.response(200, "Indexing estimate calculated successfully")
     @setup_required
     @login_required
     @account_initialization_required
@@ -795,14 +699,11 @@ class DatasetRelatedAppListApi(Resource):
     @console_ns.doc("get_dataset_related_apps")
     @console_ns.doc(description="Get applications related to dataset")
     @console_ns.doc(params={"dataset_id": "Dataset ID"})
-    @console_ns.response(
-        200,
-        "Related apps retrieved successfully",
-        console_ns.models[RelatedAppListResponse.__name__],
-    )
+    @console_ns.response(200, "Related apps retrieved successfully", related_app_list_model)
     @setup_required
     @login_required
     @account_initialization_required
+    @marshal_with(related_app_list_model)
     def get(self, dataset_id):
         current_user, _ = current_account_with_tenant()
         dataset_id_str = str(dataset_id)
@@ -823,7 +724,7 @@ class DatasetRelatedAppListApi(Resource):
             if app_model:
                 related_apps.append(app_model)
 
-        return dump_response(RelatedAppListResponse, {"data": related_apps, "total": len(related_apps)}), 200
+        return {"data": related_apps, "total": len(related_apps)}, 200
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/indexing-status")
@@ -831,11 +732,7 @@ class DatasetIndexingStatusApi(Resource):
     @console_ns.doc("get_dataset_indexing_status")
     @console_ns.doc(description="Get dataset indexing status")
     @console_ns.doc(params={"dataset_id": "Dataset ID"})
-    @console_ns.response(
-        200,
-        "Indexing status retrieved successfully",
-        console_ns.models[DocumentStatusListResponse.__name__],
-    )
+    @console_ns.response(200, "Indexing status retrieved successfully")
     @setup_required
     @login_required
     @account_initialization_required
@@ -881,8 +778,9 @@ class DatasetIndexingStatusApi(Resource):
                 "completed_segments": completed_segments,
                 "total_segments": total_segments,
             }
-            documents_status.append(document_dict)
-        return dump_response(DocumentStatusListResponse, {"data": documents_status}), 200
+            documents_status.append(marshal(document_dict, document_status_fields))
+        data = {"data": documents_status}
+        return data, 200
 
 
 @console_ns.route("/datasets/api-keys")
@@ -975,7 +873,7 @@ class DatasetApiDeleteApi(Resource):
         db.session.delete(key)
         db.session.commit()
 
-        return "", 204
+        return {"result": "success"}, 204
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/api-keys/<string:status>")
@@ -1009,18 +907,13 @@ class DatasetApiBaseUrlApi(Resource):
 class DatasetRetrievalSettingApi(Resource):
     @console_ns.doc("get_dataset_retrieval_setting")
     @console_ns.doc(description="Get dataset retrieval settings")
-    @console_ns.response(
-        200, "Retrieval settings retrieved successfully", console_ns.models[RetrievalSettingResponse.__name__]
-    )
+    @console_ns.response(200, "Retrieval settings retrieved successfully")
     @setup_required
     @login_required
     @account_initialization_required
     def get(self):
         vector_type = dify_config.VECTOR_STORE
-        return dump_response(
-            RetrievalSettingResponse,
-            _get_retrieval_methods_by_vector_type(vector_type, is_mock=False),
-        )
+        return _get_retrieval_methods_by_vector_type(vector_type, is_mock=False)
 
 
 @console_ns.route("/datasets/retrieval-setting/<string:vector_type>")
@@ -1028,19 +921,12 @@ class DatasetRetrievalSettingMockApi(Resource):
     @console_ns.doc("get_dataset_retrieval_setting_mock")
     @console_ns.doc(description="Get mock dataset retrieval settings by vector type")
     @console_ns.doc(params={"vector_type": "Vector store type"})
-    @console_ns.response(
-        200,
-        "Mock retrieval settings retrieved successfully",
-        console_ns.models[RetrievalSettingResponse.__name__],
-    )
+    @console_ns.response(200, "Mock retrieval settings retrieved successfully")
     @setup_required
     @login_required
     @account_initialization_required
     def get(self, vector_type):
-        return dump_response(
-            RetrievalSettingResponse,
-            _get_retrieval_methods_by_vector_type(vector_type, is_mock=True),
-        )
+        return _get_retrieval_methods_by_vector_type(vector_type, is_mock=True)
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/error-docs")
@@ -1048,7 +934,7 @@ class DatasetErrorDocs(Resource):
     @console_ns.doc("get_dataset_error_docs")
     @console_ns.doc(description="Get dataset error documents")
     @console_ns.doc(params={"dataset_id": "Dataset ID"})
-    @console_ns.response(200, "Error documents retrieved successfully", console_ns.models[ErrorDocsResponse.__name__])
+    @console_ns.response(200, "Error documents retrieved successfully")
     @console_ns.response(404, "Dataset not found")
     @setup_required
     @login_required
@@ -1060,7 +946,7 @@ class DatasetErrorDocs(Resource):
             raise NotFound("Dataset not found.")
         results = DocumentService.get_error_documents_by_dataset_id(dataset_id_str)
 
-        return dump_response(ErrorDocsResponse, {"data": results, "total": len(results)}), 200
+        return {"data": [marshal(item, document_status_fields) for item in results], "total": len(results)}, 200
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/permission-part-users")
@@ -1068,11 +954,7 @@ class DatasetPermissionUserListApi(Resource):
     @console_ns.doc("get_dataset_permission_users")
     @console_ns.doc(description="Get dataset permission user list")
     @console_ns.doc(params={"dataset_id": "Dataset ID"})
-    @console_ns.response(
-        200,
-        "Permission users retrieved successfully",
-        console_ns.models[PartialMemberListResponse.__name__],
-    )
+    @console_ns.response(200, "Permission users retrieved successfully")
     @console_ns.response(404, "Dataset not found")
     @console_ns.response(403, "Permission denied")
     @setup_required
@@ -1091,7 +973,9 @@ class DatasetPermissionUserListApi(Resource):
 
         partial_members_list = DatasetPermissionService.get_dataset_partial_member_list(dataset_id_str)
 
-        return dump_response(PartialMemberListResponse, {"data": partial_members_list}), 200
+        return {
+            "data": partial_members_list,
+        }, 200
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/auto-disable-logs")
@@ -1099,11 +983,7 @@ class DatasetAutoDisableLogApi(Resource):
     @console_ns.doc("get_dataset_auto_disable_logs")
     @console_ns.doc(description="Get dataset auto disable logs")
     @console_ns.doc(params={"dataset_id": "Dataset ID"})
-    @console_ns.response(
-        200,
-        "Auto disable logs retrieved successfully",
-        console_ns.models[AutoDisableLogsResponse.__name__],
-    )
+    @console_ns.response(200, "Auto disable logs retrieved successfully")
     @console_ns.response(404, "Dataset not found")
     @setup_required
     @login_required
@@ -1113,4 +993,4 @@ class DatasetAutoDisableLogApi(Resource):
         dataset = DatasetService.get_dataset(dataset_id_str)
         if dataset is None:
             raise NotFound("Dataset not found.")
-        return dump_response(AutoDisableLogsResponse, DatasetService.get_dataset_auto_disable_logs(dataset_id_str)), 200
+        return DatasetService.get_dataset_auto_disable_logs(dataset_id_str), 200

api/controllers/console/datasets/datasets_document.py

@@ -504,7 +504,7 @@ class DatasetDocumentListApi(Resource):
         except services.errors.document.DocumentIndexingError:
             raise DocumentIndexingError("Cannot delete document during indexing.")
 
-        return "", 204
+        return {"result": "success"}, 204
 
 
 @console_ns.route("/datasets/init")
@@ -966,7 +966,7 @@ class DocumentApi(DocumentResource):
         except services.errors.document.DocumentIndexingError:
             raise DocumentIndexingError("Cannot delete document during indexing.")
 
-        return "", 204
+        return {"result": "success"}, 204
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/download")
@@ -1204,7 +1204,7 @@ class DocumentPauseApi(DocumentResource):
         except services.errors.document.DocumentIndexingError:
             raise DocumentIndexingError("Cannot pause completed document.")
 
-        return "", 204
+        return {"result": "success"}, 204
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/processing/resume")
@@ -1236,7 +1236,7 @@ class DocumentRecoverApi(DocumentResource):
         except services.errors.document.DocumentIndexingError:
             raise DocumentIndexingError("Document is not in paused status.")
 
-        return "", 204
+        return {"result": "success"}, 204
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/retry")
@@ -1279,7 +1279,7 @@ class DocumentRetryApi(DocumentResource):
         # retry document
         DocumentService.retry_document(dataset_id, retry_documents)
 
-        return "", 204
+        return {"result": "success"}, 204
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/rename")

api/controllers/console/datasets/datasets_segments.py

@@ -251,7 +251,7 @@ class DatasetDocumentSegmentListApi(Resource):
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
         SegmentService.delete_segments(segment_ids, document, dataset)
-        return "", 204
+        return {"result": "success"}, 204
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/segment/<string:action>")
@@ -467,7 +467,7 @@ class DatasetDocumentSegmentUpdateApi(Resource):
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
         SegmentService.delete_segment(segment, document, dataset)
-        return "", 204
+        return {"result": "success"}, 204
 
 
 @console_ns.route(
@@ -754,7 +754,7 @@ class ChildChunkUpdateApi(Resource):
             SegmentService.delete_child_chunk(child_chunk, dataset)
         except ChildChunkDeleteIndexServiceError as e:
             raise ChildChunkDeleteIndexError(str(e))
-        return "", 204
+        return {"result": "success"}, 204
 
     @setup_required
     @login_required

api/controllers/console/datasets/external.py

@@ -218,7 +218,7 @@ class ExternalApiTemplateApi(Resource):
             raise Forbidden()
 
         ExternalDatasetService.delete_external_knowledge_api(current_tenant_id, external_knowledge_api_id)
-        return "", 204
+        return {"result": "success"}, 204
 
 
 @console_ns.route("/datasets/external-knowledge-api/<uuid:external_knowledge_api_id>/use-check")

api/controllers/console/datasets/metadata.py

@@ -1,18 +1,14 @@
 from typing import Literal
 
-from flask_restx import Resource
+from flask_restx import Resource, marshal_with
 from werkzeug.exceptions import NotFound
 
 from controllers.common.controller_schemas import MetadataUpdatePayload
+from controllers.common.fields import SimpleResultResponse
 from controllers.common.schema import register_response_schema_models, register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, enterprise_license_required, setup_required
-from fields.dataset_fields import (
-    DatasetMetadataBuiltInFieldsResponse,
-    DatasetMetadataListResponse,
-    DatasetMetadataResponse,
-)
-from libs.helper import dump_response
+from fields.dataset_fields import dataset_metadata_fields
 from libs.login import current_account_with_tenant, login_required
 from services.dataset_service import DatasetService
 from services.entities.knowledge_entities.knowledge_entities import (
@@ -26,12 +22,7 @@ from services.metadata_service import MetadataService
 register_schema_models(
     console_ns, MetadataArgs, MetadataOperationData, MetadataUpdatePayload, DocumentMetadataOperation, MetadataDetail
 )
-register_response_schema_models(
-    console_ns,
-    DatasetMetadataBuiltInFieldsResponse,
-    DatasetMetadataListResponse,
-    DatasetMetadataResponse,
-)
+register_response_schema_models(console_ns, SimpleResultResponse)
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/metadata")
@@ -40,7 +31,7 @@ class DatasetMetadataCreateApi(Resource):
     @login_required
     @account_initialization_required
     @enterprise_license_required
-    @console_ns.response(201, "Metadata created successfully", console_ns.models[DatasetMetadataResponse.__name__])
+    @marshal_with(dataset_metadata_fields)
     @console_ns.expect(console_ns.models[MetadataArgs.__name__])
     def post(self, dataset_id):
         current_user, _ = current_account_with_tenant()
@@ -53,22 +44,18 @@ class DatasetMetadataCreateApi(Resource):
         DatasetService.check_dataset_permission(dataset, current_user)
 
         metadata = MetadataService.create_metadata(dataset_id_str, metadata_args)
-        return dump_response(DatasetMetadataResponse, metadata), 201
+        return metadata, 201
 
     @setup_required
     @login_required
     @account_initialization_required
     @enterprise_license_required
-    @console_ns.response(
-        200, "Metadata retrieved successfully", console_ns.models[DatasetMetadataListResponse.__name__]
-    )
     def get(self, dataset_id):
         dataset_id_str = str(dataset_id)
         dataset = DatasetService.get_dataset(dataset_id_str)
         if dataset is None:
             raise NotFound("Dataset not found.")
-        metadata = MetadataService.get_dataset_metadatas(dataset)
-        return dump_response(DatasetMetadataListResponse, metadata), 200
+        return MetadataService.get_dataset_metadatas(dataset), 200
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/metadata/<uuid:metadata_id>")
@@ -77,7 +64,7 @@ class DatasetMetadataApi(Resource):
     @login_required
     @account_initialization_required
     @enterprise_license_required
-    @console_ns.response(200, "Metadata updated successfully", console_ns.models[DatasetMetadataResponse.__name__])
+    @marshal_with(dataset_metadata_fields)
     @console_ns.expect(console_ns.models[MetadataUpdatePayload.__name__])
     def patch(self, dataset_id, metadata_id):
         current_user, _ = current_account_with_tenant()
@@ -92,7 +79,7 @@ class DatasetMetadataApi(Resource):
         DatasetService.check_dataset_permission(dataset, current_user)
 
         metadata = MetadataService.update_metadata_name(dataset_id_str, metadata_id_str, name)
-        return dump_response(DatasetMetadataResponse, metadata), 200
+        return metadata, 200
 
     @setup_required
     @login_required
@@ -109,8 +96,7 @@ class DatasetMetadataApi(Resource):
         DatasetService.check_dataset_permission(dataset, current_user)
 
         MetadataService.delete_metadata(dataset_id_str, metadata_id_str)
-        # Frontend callers only await success and invalidate metadata caches; no response body is consumed.
-        return "", 204
+        return {"result": "success"}, 204
 
 
 @console_ns.route("/datasets/metadata/built-in")
@@ -119,14 +105,9 @@ class DatasetMetadataBuiltInFieldApi(Resource):
     @login_required
     @account_initialization_required
     @enterprise_license_required
-    @console_ns.response(
-        200,
-        "Built-in fields retrieved successfully",
-        console_ns.models[DatasetMetadataBuiltInFieldsResponse.__name__],
-    )
     def get(self):
         built_in_fields = MetadataService.get_built_in_fields()
-        return dump_response(DatasetMetadataBuiltInFieldsResponse, {"fields": built_in_fields}), 200
+        return {"fields": built_in_fields}, 200
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/metadata/built-in/<string:action>")
@@ -135,7 +116,7 @@ class DatasetMetadataBuiltInFieldActionApi(Resource):
     @login_required
     @account_initialization_required
     @enterprise_license_required
-    @console_ns.response(204, "Action completed successfully")
+    @console_ns.response(200, "Success", console_ns.models[SimpleResultResponse.__name__])
     def post(self, dataset_id, action: Literal["enable", "disable"]):
         current_user, _ = current_account_with_tenant()
         dataset_id_str = str(dataset_id)
@@ -149,8 +130,7 @@ class DatasetMetadataBuiltInFieldActionApi(Resource):
                 MetadataService.enable_built_in_field(dataset)
             case "disable":
                 MetadataService.disable_built_in_field(dataset)
-        # Frontend callers only await success and invalidate metadata caches; no response body is consumed.
-        return "", 204
+        return {"result": "success"}, 200
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/documents/metadata")
@@ -160,10 +140,7 @@ class DocumentMetadataEditApi(Resource):
     @account_initialization_required
     @enterprise_license_required
     @console_ns.expect(console_ns.models[MetadataOperationData.__name__])
-    @console_ns.response(
-        204,
-        "Documents metadata updated successfully",
-    )
+    @console_ns.response(200, "Success", console_ns.models[SimpleResultResponse.__name__])
     def post(self, dataset_id):
         current_user, _ = current_account_with_tenant()
         dataset_id_str = str(dataset_id)
@@ -176,5 +153,4 @@ class DocumentMetadataEditApi(Resource):
 
         MetadataService.update_documents_metadata(dataset, metadata_args)
 
-        # Frontend callers only await success and invalidate caches; no response body is consumed.
-        return "", 204
+        return {"result": "success"}, 200

api/controllers/console/explore/conversation.py

@@ -105,7 +105,7 @@ class ConversationApi(InstalledAppResource):
         except ConversationNotExistsError:
             raise NotFound("Conversation Not Exists.")
 
-        return "", 204
+        return ResultResponse(result="success").model_dump(mode="json"), 204
 
 
 @console_ns.route(

api/controllers/console/explore/installed_app.py

@@ -270,7 +270,7 @@ class InstalledAppApi(InstalledAppResource):
         db.session.delete(installed_app)
         db.session.commit()
 
-        return "", 204
+        return {"result": "success", "message": "App uninstalled successfully"}, 204
 
     @console_ns.response(200, "Success", console_ns.models[SimpleResultMessageResponse.__name__])
     def patch(self, installed_app):

api/controllers/console/explore/saved_message.py

@@ -76,4 +76,4 @@ class SavedMessageApi(InstalledAppResource):
 
         SavedMessageService.delete(app_model, current_user, message_id)
 
-        return "", 204
+        return ResultResponse(result="success").model_dump(mode="json"), 204

api/controllers/console/extension.py

@@ -204,4 +204,4 @@ class APIBasedExtensionDetailAPI(Resource):
 
         APIBasedExtensionService.delete(extension_data_from_db)
 
-        return "", 204
+        return {"result": "success"}, 204

api/controllers/console/feature.py

@@ -3,12 +3,12 @@ from werkzeug.exceptions import Unauthorized
 
 from controllers.common.schema import register_response_schema_models
 from libs.login import current_account_with_tenant, current_user, login_required
-from services.feature_service import FeatureModel, FeatureService, LimitationModel, SystemFeatureModel
+from services.feature_service import FeatureModel, FeatureService, SystemFeatureModel
 
 from . import console_ns
 from .wraps import account_initialization_required, cloud_utm_record, setup_required
 
-register_response_schema_models(console_ns, FeatureModel, LimitationModel, SystemFeatureModel)
+register_response_schema_models(console_ns, FeatureModel, SystemFeatureModel)
 
 
 @console_ns.route("/features")
@@ -28,32 +28,7 @@ class FeatureApi(Resource):
         """Get feature configuration for current tenant"""
         _, current_tenant_id = current_account_with_tenant()
 
-        payload = FeatureService.get_features(
-            current_tenant_id,
-            exclude_vector_space=True,
-        ).model_dump()
-        payload.pop("vector_space", None)
-        return payload
-
-
-@console_ns.route("/features/vector-space")
-class FeatureVectorSpaceApi(Resource):
-    @console_ns.doc("get_tenant_feature_vector_space")
-    @console_ns.doc(description="Get vector-space usage and limit for current tenant")
-    @console_ns.response(
-        200,
-        "Success",
-        console_ns.models[LimitationModel.__name__],
-    )
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @cloud_utm_record
-    def get(self):
-        """Get vector-space usage and limit for current tenant"""
-        _, current_tenant_id = current_account_with_tenant()
-
-        return FeatureService.get_vector_space(current_tenant_id).model_dump()
+        return FeatureService.get_features(current_tenant_id).model_dump()
 
 
 @console_ns.route("/system-features")

api/controllers/console/remote_files.py

@@ -1,5 +1,6 @@
+import urllib.parse
+
 import httpx
-from flask import request
 from flask_restx import Resource
 from pydantic import BaseModel, Field
 
@@ -33,7 +34,7 @@ class GetRemoteFileInfo(Resource):
     @console_ns.response(200, "Success", console_ns.models[RemoteFileInfo.__name__])
     @login_required
     def get(self, url: str):
-        decoded_url = helpers.decode_remote_url(url, request.query_string)
+        decoded_url = urllib.parse.unquote(url)
         resp = ssrf_proxy.head(decoded_url)
         if resp.status_code != httpx.codes.OK:
             resp = ssrf_proxy.get(decoded_url, timeout=3)

api/controllers/console/snippets/payloads.py

@@ -0,0 +1,160 @@
+import uuid
+from typing import Any, Literal
+
+from pydantic import BaseModel, Field, field_validator
+
+
+class SnippetListQuery(BaseModel):
+    """Query parameters for listing snippets."""
+
+    page: int = Field(default=1, ge=1, le=99999)
+    limit: int = Field(default=20, ge=1, le=100)
+    keyword: str | None = None
+    is_published: bool | None = Field(default=None, description="Filter by published status")
+    creators: list[str] | None = Field(default=None, description="Filter by creator account IDs")
+    tag_ids: list[str] | None = Field(default=None, description="Filter by tag IDs")
+
+    @field_validator("creators", mode="before")
+    @classmethod
+    def parse_creators(cls, value: object) -> list[str] | None:
+        """Normalize creators filter from query string or list input."""
+        return cls._normalize_string_list(value)
+
+    @field_validator("tag_ids", mode="before")
+    @classmethod
+    def parse_tag_ids(cls, value: object) -> list[str] | None:
+        """Normalize and validate tag IDs from query string or list input."""
+        items = cls._normalize_string_list(value)
+        if not items:
+            return None
+        try:
+            return [str(uuid.UUID(item)) for item in items]
+        except ValueError as exc:
+            raise ValueError("Invalid UUID format in tag_ids.") from exc
+
+    @staticmethod
+    def _normalize_string_list(value: object) -> list[str] | None:
+        if value is None:
+            return None
+        if isinstance(value, str):
+            return [item.strip() for item in value.split(",") if item.strip()] or None
+        if isinstance(value, list):
+            return [str(item).strip() for item in value if str(item).strip()] or None
+        return None
+
+
+class IconInfo(BaseModel):
+    """Icon information model."""
+
+    icon: str | None = None
+    icon_type: Literal["emoji", "image"] | None = None
+    icon_background: str | None = None
+    icon_url: str | None = None
+
+
+class InputFieldDefinition(BaseModel):
+    """Input field definition for snippet parameters."""
+
+    default: str | None = None
+    hint: bool | None = None
+    label: str | None = None
+    max_length: int | None = None
+    options: list[str] | None = None
+    placeholder: str | None = None
+    required: bool | None = None
+    type: str | None = None  # e.g., "text-input"
+
+
+class CreateSnippetPayload(BaseModel):
+    """Payload for creating a new snippet."""
+
+    name: str = Field(..., min_length=1, max_length=255)
+    description: str | None = Field(default=None, max_length=2000)
+    type: Literal["node", "group"] = "node"
+    icon_info: IconInfo | None = None
+    graph: dict[str, Any] | None = None
+    input_fields: list[InputFieldDefinition] | None = Field(default_factory=list)
+
+
+class UpdateSnippetPayload(BaseModel):
+    """Payload for updating a snippet."""
+
+    name: str | None = Field(default=None, min_length=1, max_length=255)
+    description: str | None = Field(default=None, max_length=2000)
+    icon_info: IconInfo | None = None
+
+
+class SnippetDraftSyncPayload(BaseModel):
+    """Payload for syncing snippet draft workflow."""
+
+    graph: dict[str, Any]
+    hash: str | None = None
+    conversation_variables: list[dict[str, Any]] | None = Field(
+        default=None,
+        description="Ignored. Snippet workflows do not persist conversation variables.",
+    )
+    input_fields: list[dict[str, Any]] | None = None
+
+
+class SnippetWorkflowListQuery(BaseModel):
+    """Query parameters for listing snippet published workflows."""
+
+    page: int = Field(default=1, ge=1, le=99999)
+    limit: int = Field(default=10, ge=1, le=100)
+
+
+class WorkflowRunQuery(BaseModel):
+    """Query parameters for workflow runs."""
+
+    last_id: str | None = None
+    limit: int = Field(default=20, ge=1, le=100)
+
+
+class SnippetDraftRunPayload(BaseModel):
+    """Payload for running snippet draft workflow."""
+
+    inputs: dict[str, Any]
+    files: list[dict[str, Any]] | None = None
+
+
+class SnippetDraftNodeRunPayload(BaseModel):
+    """Payload for running a single node in snippet draft workflow."""
+
+    inputs: dict[str, Any]
+    query: str = ""
+    files: list[dict[str, Any]] | None = None
+
+
+class SnippetIterationNodeRunPayload(BaseModel):
+    """Payload for running an iteration node in snippet draft workflow."""
+
+    inputs: dict[str, Any] | None = None
+
+
+class SnippetLoopNodeRunPayload(BaseModel):
+    """Payload for running a loop node in snippet draft workflow."""
+
+    inputs: dict[str, Any] | None = None
+
+
+class PublishWorkflowPayload(BaseModel):
+    """Payload for publishing snippet workflow."""
+
+    knowledge_base_setting: dict[str, Any] | None = None
+
+
+class SnippetImportPayload(BaseModel):
+    """Payload for importing snippet from DSL."""
+
+    mode: str = Field(..., description="Import mode: yaml-content or yaml-url")
+    yaml_content: str | None = Field(default=None, description="YAML content (required for yaml-content mode)")
+    yaml_url: str | None = Field(default=None, description="YAML URL (required for yaml-url mode)")
+    name: str | None = Field(default=None, description="Override snippet name")
+    description: str | None = Field(default=None, description="Override snippet description")
+    snippet_id: str | None = Field(default=None, description="Snippet ID to update (optional)")
+
+
+class IncludeSecretQuery(BaseModel):
+    """Query parameter for including secret variables in export."""
+
+    include_secret: str = Field(default="false", description="Whether to include secret variables")

api/controllers/console/snippets/snippet_workflow.py

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

api/controllers/console/snippets/snippet_workflow_draft_variable.py

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

api/controllers/console/tag/tags.py

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

api/controllers/console/workspace/model_providers.py

@@ -194,7 +194,7 @@ class ModelProviderCredentialApi(Resource):
             tenant_id=current_tenant_id, provider=provider, credential_id=args.credential_id
         )
 
-        return "", 204
+        return {"result": "success"}, 204
 
 
 @console_ns.route("/workspaces/current/model-providers/<path:provider>/credentials/switch")

api/controllers/console/workspace/models.py

@@ -259,7 +259,7 @@ class ModelProviderModelApi(Resource):
             tenant_id=tenant_id, provider=provider, model=args.model, model_type=args.model_type
         )
 
-        return "", 204
+        return {"result": "success"}, 204
 
 
 @console_ns.route("/workspaces/current/model-providers/<path:provider>/models/credentials")
@@ -395,7 +395,7 @@ class ModelProviderModelCredentialApi(Resource):
             credential_id=args.credential_id,
         )
 
-        return "", 204
+        return {"result": "success"}, 204
 
 
 @console_ns.route("/workspaces/current/model-providers/<path:provider>/models/credentials/switch")

api/controllers/console/workspace/plugin.py

@@ -15,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 core.plugin.plugin_service import PluginService
 from fields.base import ResponseModel
 from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs.login import current_account_with_tenant, login_required
@@ -23,6 +22,7 @@ from models.account import TenantPluginAutoUpgradeStrategy, TenantPluginPermissi
 from services.plugin.plugin_auto_upgrade_service import PluginAutoUpgradeService
 from services.plugin.plugin_parameter_service import PluginParameterService
 from services.plugin.plugin_permission_service import PluginPermissionService
+from services.plugin.plugin_service import PluginService
 
 
 class ParserList(BaseModel):

api/controllers/console/workspace/snippets.py

@@ -0,0 +1,404 @@
+import logging
+import re
+from urllib.parse import quote
+
+from flask import Response, request
+from flask_restx import Resource, marshal
+from sqlalchemy.orm import Session
+from werkzeug.datastructures import MultiDict
+from werkzeug.exceptions import NotFound
+
+from controllers.common.schema import register_schema_models
+from controllers.console import console_ns
+from controllers.console.snippets.payloads import (
+    CreateSnippetPayload,
+    IncludeSecretQuery,
+    SnippetImportPayload,
+    SnippetListQuery,
+    UpdateSnippetPayload,
+)
+from controllers.console.wraps import (
+    account_initialization_required,
+    edit_permission_required,
+    setup_required,
+)
+from extensions.ext_database import db
+from fields.snippet_fields import snippet_fields, snippet_list_fields, snippet_pagination_fields
+from libs.login import current_account_with_tenant, login_required
+from models.snippet import SnippetType
+from services.app_dsl_service import ImportStatus
+from services.snippet_dsl_service import SnippetDslService
+from services.snippet_service import SnippetService
+
+logger = logging.getLogger(__name__)
+_TAG_IDS_BRACKET_PATTERN = re.compile(r"^tag_ids\[(\d+)\]$")
+
+
+def _normalize_snippet_list_query_args(query_args: MultiDict[str, str]) -> dict[str, str | list[str]]:
+    normalized: dict[str, str | list[str]] = {}
+    indexed_tag_ids: list[tuple[int, str]] = []
+
+    for key in query_args:
+        match = _TAG_IDS_BRACKET_PATTERN.fullmatch(key)
+        if match:
+            indexed_tag_ids.extend((int(match.group(1)), value) for value in query_args.getlist(key))
+            continue
+
+        value = query_args.get(key)
+        if value is not None:
+            normalized[key] = value
+
+    if indexed_tag_ids:
+        normalized["tag_ids"] = [value for _, value in sorted(indexed_tag_ids)]
+
+    return normalized
+
+
+# Register Pydantic models with Swagger
+register_schema_models(
+    console_ns,
+    SnippetListQuery,
+    CreateSnippetPayload,
+    UpdateSnippetPayload,
+    SnippetImportPayload,
+    IncludeSecretQuery,
+)
+
+# Create namespace models for marshaling
+snippet_model = console_ns.model("Snippet", snippet_fields)
+snippet_list_model = console_ns.model("SnippetList", snippet_list_fields)
+snippet_pagination_model = console_ns.model("SnippetPagination", snippet_pagination_fields)
+
+
+@console_ns.route("/workspaces/current/customized-snippets")
+class CustomizedSnippetsApi(Resource):
+    @console_ns.doc("list_customized_snippets")
+    @console_ns.expect(console_ns.models.get(SnippetListQuery.__name__))
+    @console_ns.response(200, "Snippets retrieved successfully", snippet_pagination_model)
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self):
+        """List customized snippets with pagination and search."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        query = SnippetListQuery.model_validate(_normalize_snippet_list_query_args(request.args))
+
+        snippets, total, has_more = SnippetService.get_snippets(
+            tenant_id=current_tenant_id,
+            page=query.page,
+            limit=query.limit,
+            keyword=query.keyword,
+            is_published=query.is_published,
+            creators=query.creators,
+            tag_ids=query.tag_ids,
+        )
+
+        return {
+            "data": marshal(snippets, snippet_list_fields),
+            "page": query.page,
+            "limit": query.limit,
+            "total": total,
+            "has_more": has_more,
+        }, 200
+
+    @console_ns.doc("create_customized_snippet")
+    @console_ns.expect(console_ns.models.get(CreateSnippetPayload.__name__))
+    @console_ns.response(201, "Snippet created successfully", snippet_model)
+    @console_ns.response(400, "Invalid request")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def post(self):
+        """Create a new customized snippet."""
+        current_user, current_tenant_id = current_account_with_tenant()
+
+        payload = CreateSnippetPayload.model_validate(console_ns.payload or {})
+
+        try:
+            snippet_type = SnippetType(payload.type)
+        except ValueError:
+            snippet_type = SnippetType.NODE
+
+        try:
+            snippet = SnippetService.create_snippet(
+                tenant_id=current_tenant_id,
+                name=payload.name,
+                description=payload.description,
+                snippet_type=snippet_type,
+                icon_info=payload.icon_info.model_dump() if payload.icon_info else None,
+                input_fields=[f.model_dump() for f in payload.input_fields] if payload.input_fields else None,
+                account=current_user,
+            )
+        except ValueError as e:
+            return {"message": str(e)}, 400
+
+        return marshal(snippet, snippet_fields), 201
+
+
+@console_ns.route("/workspaces/current/customized-snippets/<uuid:snippet_id>")
+class CustomizedSnippetDetailApi(Resource):
+    @console_ns.doc("get_customized_snippet")
+    @console_ns.response(200, "Snippet retrieved successfully", snippet_model)
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, snippet_id: str):
+        """Get customized snippet details."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        snippet = SnippetService.get_snippet_by_id(
+            snippet_id=str(snippet_id),
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        return marshal(snippet, snippet_fields), 200
+
+    @console_ns.doc("update_customized_snippet")
+    @console_ns.expect(console_ns.models.get(UpdateSnippetPayload.__name__))
+    @console_ns.response(200, "Snippet updated successfully", snippet_model)
+    @console_ns.response(400, "Invalid request")
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def patch(self, snippet_id: str):
+        """Update customized snippet."""
+        current_user, current_tenant_id = current_account_with_tenant()
+
+        snippet = SnippetService.get_snippet_by_id(
+            snippet_id=str(snippet_id),
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        payload = UpdateSnippetPayload.model_validate(console_ns.payload or {})
+        update_data = payload.model_dump(exclude_unset=True)
+
+        if "icon_info" in update_data and update_data["icon_info"] is not None:
+            update_data["icon_info"] = payload.icon_info.model_dump() if payload.icon_info else None
+
+        if not update_data:
+            return {"message": "No valid fields to update"}, 400
+
+        try:
+            with Session(db.engine, expire_on_commit=False) as session:
+                snippet = session.merge(snippet)
+                snippet = SnippetService.update_snippet(
+                    session=session,
+                    snippet=snippet,
+                    account_id=current_user.id,
+                    data=update_data,
+                )
+                session.commit()
+        except ValueError as e:
+            return {"message": str(e)}, 400
+
+        return marshal(snippet, snippet_fields), 200
+
+    @console_ns.doc("delete_customized_snippet")
+    @console_ns.response(204, "Snippet deleted successfully")
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def delete(self, snippet_id: str):
+        """Delete customized snippet."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        snippet = SnippetService.get_snippet_by_id(
+            snippet_id=str(snippet_id),
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        with Session(db.engine) as session:
+            snippet = session.merge(snippet)
+            SnippetService.delete_snippet(
+                session=session,
+                snippet=snippet,
+            )
+            session.commit()
+
+        return "", 204
+
+
+@console_ns.route("/workspaces/current/customized-snippets/<uuid:snippet_id>/export")
+class CustomizedSnippetExportApi(Resource):
+    @console_ns.doc("export_customized_snippet")
+    @console_ns.doc(description="Export snippet configuration as DSL")
+    @console_ns.doc(params={"snippet_id": "Snippet ID to export"})
+    @console_ns.response(200, "Snippet exported successfully")
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def get(self, snippet_id: str):
+        """Export snippet as DSL."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        snippet = SnippetService.get_snippet_by_id(
+            snippet_id=str(snippet_id),
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        # Get include_secret parameter
+        query = IncludeSecretQuery.model_validate(request.args.to_dict())
+
+        with Session(db.engine) as session:
+            export_service = SnippetDslService(session)
+            result = export_service.export_snippet_dsl(snippet=snippet, include_secret=query.include_secret == "true")
+
+        # Set filename with .snippet extension
+        filename = f"{snippet.name}.snippet"
+        encoded_filename = quote(filename)
+        
+        response = Response(
+            result,
+            mimetype="application/x-yaml",
+        )
+        response.headers["Content-Disposition"] = f"attachment; filename*=UTF-8''{encoded_filename}"
+        response.headers["Content-Type"] = "application/x-yaml"
+        
+        return response
+
+
+@console_ns.route("/workspaces/current/customized-snippets/imports")
+class CustomizedSnippetImportApi(Resource):
+    @console_ns.doc("import_customized_snippet")
+    @console_ns.doc(description="Import snippet from DSL")
+    @console_ns.expect(console_ns.models.get(SnippetImportPayload.__name__))
+    @console_ns.response(200, "Snippet imported successfully")
+    @console_ns.response(202, "Import pending confirmation")
+    @console_ns.response(400, "Import failed")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def post(self):
+        """Import snippet from DSL."""
+        current_user, _ = current_account_with_tenant()
+        payload = SnippetImportPayload.model_validate(console_ns.payload or {})
+
+        with Session(db.engine) as session:
+            import_service = SnippetDslService(session)
+            result = import_service.import_snippet(
+                account=current_user,
+                import_mode=payload.mode,
+                yaml_content=payload.yaml_content,
+                yaml_url=payload.yaml_url,
+                snippet_id=payload.snippet_id,
+                name=payload.name,
+                description=payload.description,
+            )
+            session.commit()
+
+        # Return appropriate status code based on result
+        status = result.status
+        if status == ImportStatus.FAILED:
+            return result.model_dump(mode="json"), 400
+        elif status == ImportStatus.PENDING:
+            return result.model_dump(mode="json"), 202
+        return result.model_dump(mode="json"), 200
+
+
+@console_ns.route("/workspaces/current/customized-snippets/imports/<string:import_id>/confirm")
+class CustomizedSnippetImportConfirmApi(Resource):
+    @console_ns.doc("confirm_snippet_import")
+    @console_ns.doc(description="Confirm a pending snippet import")
+    @console_ns.doc(params={"import_id": "Import ID to confirm"})
+    @console_ns.response(200, "Import confirmed successfully")
+    @console_ns.response(400, "Import failed")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def post(self, import_id: str):
+        """Confirm a pending snippet import."""
+        current_user, _ = current_account_with_tenant()
+
+        with Session(db.engine) as session:
+            import_service = SnippetDslService(session)
+            result = import_service.confirm_import(import_id=import_id, account=current_user)
+            session.commit()
+
+        if result.status == ImportStatus.FAILED:
+            return result.model_dump(mode="json"), 400
+        return result.model_dump(mode="json"), 200
+
+
+@console_ns.route("/workspaces/current/customized-snippets/<uuid:snippet_id>/check-dependencies")
+class CustomizedSnippetCheckDependenciesApi(Resource):
+    @console_ns.doc("check_snippet_dependencies")
+    @console_ns.doc(description="Check dependencies for a snippet")
+    @console_ns.doc(params={"snippet_id": "Snippet ID"})
+    @console_ns.response(200, "Dependencies checked successfully")
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def get(self, snippet_id: str):
+        """Check dependencies for a snippet."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        snippet = SnippetService.get_snippet_by_id(
+            snippet_id=str(snippet_id),
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        with Session(db.engine) as session:
+            import_service = SnippetDslService(session)
+            result = import_service.check_dependencies(snippet=snippet)
+
+        return result.model_dump(mode="json"), 200
+
+
+@console_ns.route("/workspaces/current/customized-snippets/<uuid:snippet_id>/use-count/increment")
+class CustomizedSnippetUseCountIncrementApi(Resource):
+    @console_ns.doc("increment_snippet_use_count")
+    @console_ns.doc(description="Increment snippet use count by 1")
+    @console_ns.doc(params={"snippet_id": "Snippet ID"})
+    @console_ns.response(200, "Use count incremented successfully")
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def post(self, snippet_id: str):
+        """Increment snippet use count when it is inserted into a workflow."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        snippet = SnippetService.get_snippet_by_id(
+            snippet_id=str(snippet_id),
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        with Session(db.engine) as session:
+            snippet = session.merge(snippet)
+            SnippetService.increment_use_count(session=session, snippet=snippet)
+            session.commit()
+            session.refresh(snippet)
+
+        return {"result": "success", "use_count": snippet.use_count}, 200

api/controllers/service_api/app/workflow.py

@@ -1,7 +1,7 @@
 import logging
 from collections.abc import Mapping
 from datetime import datetime
-from typing import Literal, override
+from typing import Literal
 
 from dateutil.parser import isoparse
 from flask import request
@@ -76,13 +76,11 @@ def _enum_value(value):
 
 
 class WorkflowRunStatusField(fields.Raw):
-    @override
     def output(self, key, obj: WorkflowRun, **kwargs):
         return _enum_value(obj.status)
 
 
 class WorkflowRunOutputsField(fields.Raw):
-    @override
     def output(self, key, obj: WorkflowRun, **kwargs):
         status = _enum_value(obj.status)
         if status == WorkflowExecutionStatus.PAUSED.value:

api/controllers/service_api/dataset/dataset.py

@@ -1,17 +1,13 @@
-from typing import Any, Literal
+from typing import Any, Literal, cast
 
 from flask import request
-from pydantic import BaseModel, ConfigDict, Field, RootModel, field_validator, model_validator
+from flask_restx import marshal
+from pydantic import BaseModel, Field, TypeAdapter, field_validator, model_validator
 from werkzeug.exceptions import Forbidden, NotFound
 
 import services
 from controllers.common.fields import SimpleResultResponse
-from controllers.common.schema import (
-    query_params_from_model,
-    register_enum_models,
-    register_response_schema_models,
-    register_schema_models,
-)
+from controllers.common.schema import register_enum_models, register_response_schema_models, register_schema_models
 from controllers.console.wraps import edit_permission_required
 from controllers.service_api import service_api_ns
 from controllers.service_api.dataset.error import DatasetInUseError, DatasetNameDuplicateError, InvalidActionError
@@ -21,10 +17,9 @@ from controllers.service_api.wraps import (
 )
 from core.plugin.impl.model_runtime_factory import create_plugin_provider_manager
 from core.rag.index_processor.constant.index_type import IndexTechniqueType
-from fields.base import ResponseModel
-from fields.dataset_fields import DatasetDetailResponse
+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.helper import dump_response
 from libs.login import current_user
 from models.account import Account
 from models.dataset import DatasetPermissionEnum
@@ -124,21 +119,6 @@ class TagUnbindingPayload(BaseModel):
         return self
 
 
-class KnowledgeTagResponse(ResponseModel):
-    model_config = ConfigDict(coerce_numbers_to_str=True)
-
-    id: str
-    name: str
-    type: str
-    # TODO: The public Service API docs expose binding_count as string|null.
-    # Keep matching the old RESTX fields.String coercion until that contract is intentionally migrated.
-    binding_count: str | None = None
-
-
-class KnowledgeTagListResponse(RootModel[list[KnowledgeTagResponse]]):
-    pass
-
-
 class DatasetListQuery(BaseModel):
     page: int = Field(default=1, description="Page number")
     limit: int = Field(default=20, description="Number of items per page")
@@ -147,29 +127,6 @@ class DatasetListQuery(BaseModel):
     tag_ids: list[str] = Field(default_factory=list, description="Filter by tag IDs")
 
 
-class DatasetDetailWithPartialMembersResponse(DatasetDetailResponse):
-    partial_member_list: list[str] | None = None
-
-
-# todo: duplicate code, but the partial_member_list has different nullability
-class DatasetListResponse(ResponseModel):
-    data: list[DatasetDetailResponse]
-    has_more: bool
-    limit: int
-    total: int
-    page: int
-
-
-class DatasetBoundTagResponse(ResponseModel):
-    id: str
-    name: str
-
-
-class DatasetBoundTagListResponse(ResponseModel):
-    data: list[DatasetBoundTagResponse]
-    total: int
-
-
 register_schema_models(
     service_api_ns,
     DatasetCreatePayload,
@@ -180,17 +137,9 @@ register_schema_models(
     TagBindingPayload,
     TagUnbindingPayload,
     DatasetListQuery,
+    DataSetTag,
 )
-register_response_schema_models(
-    service_api_ns,
-    SimpleResultResponse,
-    KnowledgeTagResponse,
-    KnowledgeTagListResponse,
-    DatasetDetailResponse,
-    DatasetDetailWithPartialMembersResponse,
-    DatasetListResponse,
-    DatasetBoundTagListResponse,
-)
+register_response_schema_models(service_api_ns, SimpleResultResponse)
 
 
 @service_api_ns.route("/datasets")
@@ -205,18 +154,9 @@ class DatasetListApi(DatasetApiResource):
             401: "Unauthorized - invalid API token",
         }
     )
-    @service_api_ns.doc(params=query_params_from_model(DatasetListQuery))
-    @service_api_ns.response(
-        200,
-        "Datasets retrieved successfully",
-        service_api_ns.models[DatasetListResponse.__name__],
-    )
     def get(self, tenant_id):
         """Resource for getting datasets."""
-        query_params: dict[str, str | list[str]] = dict(request.args.to_dict())
-        if "tag_ids" in request.args:
-            query_params["tag_ids"] = request.args.getlist("tag_ids")
-        query = DatasetListQuery.model_validate(query_params)
+        query = DatasetListQuery.model_validate(request.args.to_dict())
         # provider = request.args.get("provider", default="vendor")
 
         datasets, total = DatasetService.get_datasets(
@@ -235,17 +175,17 @@ class DatasetListApi(DatasetApiResource):
         for embedding_model in embedding_models:
             model_names.append(f"{embedding_model.model}:{embedding_model.provider.provider}")
 
-        data = [dump_response(DatasetDetailResponse, dataset) for dataset in datasets]
+        data = marshal(datasets, dataset_detail_fields)
         for item in data:
             if item["indexing_technique"] == IndexTechniqueType.HIGH_QUALITY and item["embedding_model_provider"]:
                 item["embedding_model_provider"] = str(ModelProviderID(item["embedding_model_provider"]))
                 item_model = f"{item['embedding_model']}:{item['embedding_model_provider']}"
                 if item_model in model_names:
-                    item["embedding_available"] = True
+                    item["embedding_available"] = True  # type: ignore
                 else:
-                    item["embedding_available"] = False
+                    item["embedding_available"] = False  # type: ignore
             else:
-                item["embedding_available"] = True
+                item["embedding_available"] = True  # type: ignore
         response = {
             "data": data,
             "has_more": len(datasets) == query.limit,
@@ -253,7 +193,7 @@ class DatasetListApi(DatasetApiResource):
             "total": total,
             "page": query.page,
         }
-        return dump_response(DatasetListResponse, response), 200
+        return response, 200
 
     @service_api_ns.expect(service_api_ns.models[DatasetCreatePayload.__name__])
     @service_api_ns.doc("create_dataset")
@@ -265,11 +205,6 @@ class DatasetListApi(DatasetApiResource):
             400: "Bad request - invalid parameters",
         }
     )
-    @service_api_ns.response(
-        200,
-        "Dataset created successfully",
-        service_api_ns.models[DatasetDetailResponse.__name__],
-    )
     @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
     def post(self, tenant_id):
         """Resource for creating datasets."""
@@ -313,7 +248,7 @@ class DatasetListApi(DatasetApiResource):
         except services.errors.dataset.DatasetNameDuplicateError:
             raise DatasetNameDuplicateError()
 
-        return dump_response(DatasetDetailResponse, dataset), 200
+        return marshal(dataset, dataset_detail_fields), 200
 
 
 @service_api_ns.route("/datasets/<uuid:dataset_id>")
@@ -331,11 +266,6 @@ class DatasetApi(DatasetApiResource):
             404: "Dataset not found",
         }
     )
-    @service_api_ns.response(
-        200,
-        "Dataset retrieved successfully",
-        service_api_ns.models[DatasetDetailWithPartialMembersResponse.__name__],
-    )
     def get(self, _, dataset_id):
         dataset_id_str = str(dataset_id)
         dataset = DatasetService.get_dataset(dataset_id_str)
@@ -345,7 +275,7 @@ class DatasetApi(DatasetApiResource):
             DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
-        data = dump_response(DatasetDetailResponse, dataset)
+        data = cast(dict[str, Any], marshal(dataset, dataset_detail_fields))
         # check embedding setting
         assert isinstance(current_user, Account)
         cid = current_user.current_tenant_id
@@ -377,13 +307,7 @@ class DatasetApi(DatasetApiResource):
             part_users_list = DatasetPermissionService.get_dataset_partial_member_list(dataset_id_str)
             data.update({"partial_member_list": part_users_list})
 
-        return (
-            DatasetDetailWithPartialMembersResponse.model_validate(data).model_dump(
-                mode="json",
-                exclude={"partial_member_list"} if "partial_member_list" not in data else set(),
-            ),
-            200,
-        )
+        return data, 200
 
     @service_api_ns.expect(service_api_ns.models[DatasetUpdatePayload.__name__])
     @service_api_ns.doc("update_dataset")
@@ -397,11 +321,6 @@ class DatasetApi(DatasetApiResource):
             404: "Dataset not found",
         }
     )
-    @service_api_ns.response(
-        200,
-        "Dataset updated successfully",
-        service_api_ns.models[DatasetDetailWithPartialMembersResponse.__name__],
-    )
     @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
     def patch(self, _, dataset_id):
         dataset_id_str = str(dataset_id)
@@ -452,7 +371,7 @@ class DatasetApi(DatasetApiResource):
         if dataset is None:
             raise NotFound("Dataset not found.")
 
-        result_data = dump_response(DatasetDetailResponse, dataset)
+        result_data = cast(dict[str, Any], marshal(dataset, dataset_detail_fields))
         assert isinstance(current_user, Account)
         tenant_id = current_user.current_tenant_id
 
@@ -465,7 +384,7 @@ class DatasetApi(DatasetApiResource):
         partial_member_list = DatasetPermissionService.get_dataset_partial_member_list(dataset_id_str)
         result_data.update({"partial_member_list": partial_member_list})
 
-        return DatasetDetailWithPartialMembersResponse.model_validate(result_data).model_dump(mode="json"), 200
+        return result_data, 200
 
     @service_api_ns.doc("delete_dataset")
     @service_api_ns.doc(description="Delete a dataset")
@@ -578,7 +497,7 @@ class DocumentStatusApi(DatasetApiResource):
         except ValueError as e:
             raise InvalidActionError(str(e))
 
-        return dump_response(SimpleResultResponse, {"result": "success"}), 200
+        return {"result": "success"}, 200
 
 
 @service_api_ns.route("/datasets/tags")
@@ -591,18 +510,14 @@ class DatasetTagsApi(DatasetApiResource):
             401: "Unauthorized - invalid API token",
         }
     )
-    @service_api_ns.response(
-        200,
-        "Tags retrieved successfully",
-        service_api_ns.models[KnowledgeTagListResponse.__name__],
-    )
     def get(self, _):
         """Get all knowledge type tags."""
         assert isinstance(current_user, Account)
         cid = current_user.current_tenant_id
         assert cid is not None
         tags = TagService.get_tags("knowledge", cid)
-        return dump_response(KnowledgeTagListResponse, tags), 200
+        tag_models = TypeAdapter(list[DataSetTag]).validate_python(tags, from_attributes=True)
+        return [tag.model_dump(mode="json") for tag in tag_models], 200
 
     @service_api_ns.expect(service_api_ns.models[TagCreatePayload.__name__])
     @service_api_ns.doc("create_dataset_tag")
@@ -614,11 +529,6 @@ class DatasetTagsApi(DatasetApiResource):
             403: "Forbidden - insufficient permissions",
         }
     )
-    @service_api_ns.response(
-        200,
-        "Tag created successfully",
-        service_api_ns.models[KnowledgeTagResponse.__name__],
-    )
     def post(self, _):
         """Add a knowledge type tag."""
         assert isinstance(current_user, Account)
@@ -628,10 +538,9 @@ class DatasetTagsApi(DatasetApiResource):
         payload = TagCreatePayload.model_validate(service_api_ns.payload or {})
         tag = TagService.save_tags(SaveTagPayload(name=payload.name, type=TagType.KNOWLEDGE))
 
-        response = dump_response(
-            KnowledgeTagResponse,
-            {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": 0},
-        )
+        response = DataSetTag.model_validate(
+            {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": 0}
+        ).model_dump(mode="json")
         return response, 200
 
     @service_api_ns.expect(service_api_ns.models[TagUpdatePayload.__name__])
@@ -644,11 +553,6 @@ class DatasetTagsApi(DatasetApiResource):
             403: "Forbidden - insufficient permissions",
         }
     )
-    @service_api_ns.response(
-        200,
-        "Tag updated successfully",
-        service_api_ns.models[KnowledgeTagResponse.__name__],
-    )
     def patch(self, _):
         assert isinstance(current_user, Account)
         if not (current_user.has_edit_permission or current_user.is_dataset_editor):
@@ -660,10 +564,9 @@ class DatasetTagsApi(DatasetApiResource):
 
         binding_count = TagService.get_tag_binding_count(tag_id)
 
-        response = dump_response(
-            KnowledgeTagResponse,
-            {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": binding_count},
-        )
+        response = DataSetTag.model_validate(
+            {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": binding_count}
+        ).model_dump(mode="json")
         return response, 200
 
     @service_api_ns.expect(service_api_ns.models[TagDeletePayload.__name__])
@@ -748,11 +651,6 @@ class DatasetTagsBindingStatusApi(DatasetApiResource):
             401: "Unauthorized - invalid API token",
         }
     )
-    @service_api_ns.response(
-        200,
-        "Tags retrieved successfully",
-        service_api_ns.models[DatasetBoundTagListResponse.__name__],
-    )
     def get(self, _, *args, **kwargs):
         """Get all knowledge type tags."""
         dataset_id = kwargs.get("dataset_id")
@@ -760,4 +658,5 @@ class DatasetTagsBindingStatusApi(DatasetApiResource):
         assert current_user.current_tenant_id is not None
         tags = TagService.get_tags_by_target_id("knowledge", current_user.current_tenant_id, str(dataset_id))
         tags_list = [{"id": tag.id, "name": tag.name} for tag in tags]
-        return dump_response(DatasetBoundTagListResponse, {"data": tags_list, "total": len(tags)}), 200
+        response = {"data": tags_list, "total": len(tags)}
+        return response, 200

api/controllers/service_api/dataset/metadata.py

@@ -1,19 +1,15 @@
 from typing import Literal
 
 from flask_login import current_user
+from flask_restx import marshal
 from werkzeug.exceptions import NotFound
 
 from controllers.common.controller_schemas import MetadataUpdatePayload
+from controllers.common.fields import SimpleResultResponse
 from controllers.common.schema import register_response_schema_models, register_schema_model, register_schema_models
 from controllers.service_api import service_api_ns
 from controllers.service_api.wraps import DatasetApiResource, cloud_edition_billing_rate_limit_check
-from fields.dataset_fields import (
-    DatasetMetadataActionResponse,
-    DatasetMetadataBuiltInFieldsResponse,
-    DatasetMetadataListResponse,
-    DatasetMetadataResponse,
-)
-from libs.helper import dump_response
+from fields.dataset_fields import dataset_metadata_fields
 from services.dataset_service import DatasetService
 from services.entities.knowledge_entities.knowledge_entities import (
     DocumentMetadataOperation,
@@ -31,13 +27,7 @@ register_schema_models(
     DocumentMetadataOperation,
     MetadataOperationData,
 )
-register_response_schema_models(
-    service_api_ns,
-    DatasetMetadataActionResponse,
-    DatasetMetadataBuiltInFieldsResponse,
-    DatasetMetadataListResponse,
-    DatasetMetadataResponse,
-)
+register_response_schema_models(service_api_ns, SimpleResultResponse)
 
 
 @service_api_ns.route("/datasets/<uuid:dataset_id>/metadata")
@@ -53,9 +43,6 @@ class DatasetMetadataCreateServiceApi(DatasetApiResource):
             404: "Dataset not found",
         }
     )
-    @service_api_ns.response(
-        201, "Metadata created successfully", service_api_ns.models[DatasetMetadataResponse.__name__]
-    )
     @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
     def post(self, tenant_id, dataset_id):
         """Create metadata for a dataset."""
@@ -68,7 +55,7 @@ class DatasetMetadataCreateServiceApi(DatasetApiResource):
         DatasetService.check_dataset_permission(dataset, current_user)
 
         metadata = MetadataService.create_metadata(dataset_id_str, metadata_args)
-        return dump_response(DatasetMetadataResponse, metadata), 201
+        return marshal(metadata, dataset_metadata_fields), 201
 
     @service_api_ns.doc("get_dataset_metadata")
     @service_api_ns.doc(description="Get all metadata for a dataset")
@@ -80,17 +67,13 @@ class DatasetMetadataCreateServiceApi(DatasetApiResource):
             404: "Dataset not found",
         }
     )
-    @service_api_ns.response(
-        200, "Metadata retrieved successfully", service_api_ns.models[DatasetMetadataListResponse.__name__]
-    )
     def get(self, tenant_id, dataset_id):
         """Get all metadata for a dataset."""
         dataset_id_str = str(dataset_id)
         dataset = DatasetService.get_dataset(dataset_id_str)
         if dataset is None:
             raise NotFound("Dataset not found.")
-        metadata = MetadataService.get_dataset_metadatas(dataset)
-        return dump_response(DatasetMetadataListResponse, metadata), 200
+        return MetadataService.get_dataset_metadatas(dataset), 200
 
 
 @service_api_ns.route("/datasets/<uuid:dataset_id>/metadata/<uuid:metadata_id>")
@@ -106,9 +89,6 @@ class DatasetMetadataServiceApi(DatasetApiResource):
             404: "Dataset or metadata not found",
         }
     )
-    @service_api_ns.response(
-        200, "Metadata updated successfully", service_api_ns.models[DatasetMetadataResponse.__name__]
-    )
     @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
     def patch(self, tenant_id, dataset_id, metadata_id):
         """Update metadata name."""
@@ -122,7 +102,7 @@ class DatasetMetadataServiceApi(DatasetApiResource):
         DatasetService.check_dataset_permission(dataset, current_user)
 
         metadata = MetadataService.update_metadata_name(dataset_id_str, metadata_id_str, payload.name)
-        return dump_response(DatasetMetadataResponse, metadata), 200
+        return marshal(metadata, dataset_metadata_fields), 200
 
     @service_api_ns.doc("delete_dataset_metadata")
     @service_api_ns.doc(description="Delete metadata")
@@ -134,7 +114,6 @@ class DatasetMetadataServiceApi(DatasetApiResource):
             404: "Dataset or metadata not found",
         }
     )
-    @service_api_ns.response(204, "Metadata deleted successfully")
     @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
     def delete(self, tenant_id, dataset_id, metadata_id):
         """Delete metadata."""
@@ -159,15 +138,10 @@ class DatasetMetadataBuiltInFieldServiceApi(DatasetApiResource):
             401: "Unauthorized - invalid API token",
         }
     )
-    @service_api_ns.response(
-        200,
-        "Built-in fields retrieved successfully",
-        service_api_ns.models[DatasetMetadataBuiltInFieldsResponse.__name__],
-    )
     def get(self, tenant_id, dataset_id):
         """Get all built-in metadata fields."""
         built_in_fields = MetadataService.get_built_in_fields()
-        return dump_response(DatasetMetadataBuiltInFieldsResponse, {"fields": built_in_fields}), 200
+        return {"fields": built_in_fields}, 200
 
 
 @service_api_ns.route("/datasets/<uuid:dataset_id>/metadata/built-in/<string:action>")
@@ -183,7 +157,9 @@ class DatasetMetadataBuiltInFieldActionServiceApi(DatasetApiResource):
         }
     )
     @service_api_ns.response(
-        200, "Action completed successfully", service_api_ns.models[DatasetMetadataActionResponse.__name__]
+        200,
+        "Action completed successfully",
+        service_api_ns.models[SimpleResultResponse.__name__],
     )
     @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
     def post(self, tenant_id, dataset_id, action: Literal["enable", "disable"]):
@@ -199,7 +175,7 @@ class DatasetMetadataBuiltInFieldActionServiceApi(DatasetApiResource):
                 MetadataService.enable_built_in_field(dataset)
             case "disable":
                 MetadataService.disable_built_in_field(dataset)
-        return dump_response(DatasetMetadataActionResponse, {"result": "success"}), 200
+        return {"result": "success"}, 200
 
 
 @service_api_ns.route("/datasets/<uuid:dataset_id>/documents/metadata")
@@ -218,7 +194,7 @@ class DocumentMetadataEditServiceApi(DatasetApiResource):
     @service_api_ns.response(
         200,
         "Documents metadata updated successfully",
-        service_api_ns.models[DatasetMetadataActionResponse.__name__],
+        service_api_ns.models[SimpleResultResponse.__name__],
     )
     @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
     def post(self, tenant_id, dataset_id):
@@ -233,4 +209,4 @@ class DocumentMetadataEditServiceApi(DatasetApiResource):
 
         MetadataService.update_documents_metadata(dataset, metadata_args)
 
-        return dump_response(DatasetMetadataActionResponse, {"result": "success"}), 200
+        return {"result": "success"}, 200

api/controllers/service_api/index.py

@@ -11,7 +11,7 @@ register_response_schema_models(service_api_ns, IndexInfoResponse)
 @service_api_ns.route("/")
 class IndexApi(Resource):
     @service_api_ns.response(200, "Success", service_api_ns.models[IndexInfoResponse.__name__])
-    def get(self) -> dict[str, str]:
+    def get(self):
         return {
             "welcome": "Dify OpenAPI",
             "api_version": "v1",

api/controllers/web/conversation.py

@@ -136,7 +136,7 @@ class ConversationApi(WebApiResource):
             ConversationService.delete(app_model, conversation_id, end_user)
         except ConversationNotExistsError:
             raise NotFound("Conversation Not Exists.")
-        return "", 204
+        return ResultResponse(result="success").model_dump(mode="json"), 204
 
 
 @web_ns.route("/conversations/<uuid:c_id>/name")

api/controllers/web/remote_files.py

@@ -1,5 +1,6 @@
+import urllib.parse
+
 import httpx
-from flask import request
 from pydantic import BaseModel, Field, HttpUrl
 
 import services
@@ -58,7 +59,7 @@ class RemoteFileInfoApi(WebApiResource):
         Raises:
             HTTPException: If the remote file cannot be accessed
         """
-        decoded_url = helpers.decode_remote_url(url, request.query_string)
+        decoded_url = urllib.parse.unquote(url)
         resp = ssrf_proxy.head(decoded_url)
         if resp.status_code != httpx.codes.OK:
             # failed back to get method

api/controllers/web/saved_message.py

@@ -112,4 +112,4 @@ class SavedMessageApi(WebApiResource):
 
         SavedMessageService.delete(app_model, end_user, message_id)
 
-        return "", 204
+        return ResultResponse(result="success").model_dump(mode="json"), 204

api/core/agent/cot_chat_agent_runner.py

@@ -1,5 +1,4 @@
 import json
-from typing import override
 
 from core.agent.cot_agent_runner import CotAgentRunner
 from graphon.file import file_manager
@@ -67,7 +66,6 @@ class CotChatAgentRunner(CotAgentRunner):
 
         return prompt_messages
 
-    @override
     def _organize_prompt_messages(self) -> list[PromptMessage]:
         """
         Organize

api/core/agent/cot_completion_agent_runner.py

@@ -1,5 +1,4 @@
 import json
-from typing import override
 
 from core.agent.cot_agent_runner import CotAgentRunner
 from graphon.model_runtime.entities.message_entities import (
@@ -52,7 +51,6 @@ class CotCompletionAgentRunner(CotAgentRunner):
 
         return historic_prompt
 
-    @override
     def _organize_prompt_messages(self) -> list[PromptMessage]:
         """
         Organize prompt messages

api/core/agent/strategy/plugin.py

@@ -1,5 +1,5 @@
 from collections.abc import Generator, Sequence
-from typing import Any, override
+from typing import Any
 
 from core.agent.entities import AgentInvokeMessage
 from core.agent.plugin_entities import AgentStrategyEntity, AgentStrategyParameter
@@ -23,7 +23,6 @@ class PluginAgentStrategy(BaseAgentStrategy):
         self.declaration = declaration
         self.meta_version = meta_version
 
-    @override
     def get_parameters(self) -> Sequence[AgentStrategyParameter]:
         return self.declaration.parameters
 
@@ -35,7 +34,6 @@ class PluginAgentStrategy(BaseAgentStrategy):
             params[parameter.name] = parameter.init_frontend_parameter(params.get(parameter.name))
         return params
 
-    @override
     def _invoke(
         self,
         params: dict[str, Any],

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

@@ -55,7 +55,6 @@ from libs.flask_utils import preserve_flask_contexts
 from models import Account, App, Conversation, EndUser, Message, Workflow, WorkflowNodeExecutionTriggeredFrom
 from models.enums import WorkflowRunTriggeredFrom
 from services.conversation_service import ConversationService
-from services.errors.conversation import ConversationNotExistsError
 from services.workflow_draft_variable_service import (
     DraftVarLoader,
     WorkflowDraftVariableService,
@@ -146,15 +145,9 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         conversation = None
         conversation_id = args.get("conversation_id")
         if conversation_id:
-            try:
-                conversation = ConversationService.get_conversation(
-                    app_model=app_model, conversation_id=conversation_id, user=user
-                )
-            except ConversationNotExistsError:
-                if invoke_from == InvokeFrom.SERVICE_API:
-                    conversation = None
-                else:
-                    raise
+            conversation = ConversationService.get_conversation(
+                app_model=app_model, conversation_id=conversation_id, user=user
+            )
 
         # parse files
         # TODO(QuantumGhost): Move file parsing logic to the API controller layer

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

@@ -1,5 +1,5 @@
 from collections.abc import Generator
-from typing import Any, cast, override
+from typing import Any, cast
 
 from core.app.apps.base_app_generate_response_converter import AppGenerateResponseConverter
 from core.app.entities.task_entities import (
@@ -20,7 +20,6 @@ class AdvancedChatAppGenerateResponseConverter(
     AppGenerateResponseConverter[ChatbotAppBlockingResponse | AdvancedChatPausedBlockingResponse]
 ):
     @classmethod
-    @override
     def convert_blocking_full_response(
         cls, blocking_response: ChatbotAppBlockingResponse | AdvancedChatPausedBlockingResponse
     ) -> dict[str, Any]:
@@ -60,7 +59,6 @@ class AdvancedChatAppGenerateResponseConverter(
         return response
 
     @classmethod
-    @override
     def convert_blocking_simple_response(
         cls, blocking_response: ChatbotAppBlockingResponse | AdvancedChatPausedBlockingResponse
     ) -> dict[str, Any]:
@@ -78,7 +76,6 @@ class AdvancedChatAppGenerateResponseConverter(
         return response
 
     @classmethod
-    @override
     def convert_stream_full_response(
         cls, stream_response: Generator[AppStreamResponse, None, None]
     ) -> Generator[dict[str, Any] | str, Any, None]:
@@ -110,7 +107,6 @@ class AdvancedChatAppGenerateResponseConverter(
             yield response_chunk
 
     @classmethod
-    @override
     def convert_stream_simple_response(
         cls, stream_response: Generator[AppStreamResponse, None, None]
     ) -> Generator[dict[str, Any] | str, Any, None]:

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

@@ -1,5 +1,5 @@
 from collections.abc import Generator
-from typing import Any, cast, override
+from typing import Any, cast
 
 from pydantic import JsonValue
 
@@ -16,7 +16,6 @@ from core.app.entities.task_entities import (
 
 class AgentChatAppGenerateResponseConverter(AppGenerateResponseConverter[ChatbotAppBlockingResponse]):
     @classmethod
-    @override
     def convert_blocking_full_response(cls, blocking_response: ChatbotAppBlockingResponse):
         """
         Convert blocking full response.
@@ -38,7 +37,6 @@ class AgentChatAppGenerateResponseConverter(AppGenerateResponseConverter[Chatbot
         return response
 
     @classmethod
-    @override
     def convert_blocking_simple_response(cls, blocking_response: ChatbotAppBlockingResponse):
         """
         Convert blocking simple response.
@@ -56,7 +54,6 @@ class AgentChatAppGenerateResponseConverter(AppGenerateResponseConverter[Chatbot
         return response
 
     @classmethod
-    @override
     def convert_stream_full_response(
         cls, stream_response: Generator[AppStreamResponse, None, None]
     ) -> Generator[dict[str, Any] | str, None, None]:
@@ -88,7 +85,6 @@ class AgentChatAppGenerateResponseConverter(AppGenerateResponseConverter[Chatbot
             yield response_chunk
 
     @classmethod
-    @override
     def convert_stream_simple_response(
         cls, stream_response: Generator[AppStreamResponse, None, None]
     ) -> Generator[dict[str, Any] | str, None, None]:

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

@@ -1,5 +1,5 @@
 from collections.abc import Generator
-from typing import Any, cast, override
+from typing import Any, cast
 
 from pydantic import JsonValue
 
@@ -16,7 +16,6 @@ from core.app.entities.task_entities import (
 
 class ChatAppGenerateResponseConverter(AppGenerateResponseConverter[ChatbotAppBlockingResponse]):
     @classmethod
-    @override
     def convert_blocking_full_response(cls, blocking_response: ChatbotAppBlockingResponse):
         """
         Convert blocking full response.
@@ -38,7 +37,6 @@ class ChatAppGenerateResponseConverter(AppGenerateResponseConverter[ChatbotAppBl
         return response
 
     @classmethod
-    @override
     def convert_blocking_simple_response(cls, blocking_response: ChatbotAppBlockingResponse):
         """
         Convert blocking simple response.
@@ -56,7 +54,6 @@ class ChatAppGenerateResponseConverter(AppGenerateResponseConverter[ChatbotAppBl
         return response
 
     @classmethod
-    @override
     def convert_stream_full_response(
         cls, stream_response: Generator[AppStreamResponse, None, None]
     ) -> Generator[dict[str, Any] | str, None, None]:
@@ -88,7 +85,6 @@ class ChatAppGenerateResponseConverter(AppGenerateResponseConverter[ChatbotAppBl
             yield response_chunk
 
     @classmethod
-    @override
     def convert_stream_simple_response(
         cls, stream_response: Generator[AppStreamResponse, None, None]
     ) -> Generator[dict[str, Any] | str, None, None]:

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

@@ -1,5 +1,5 @@
 from collections.abc import Generator
-from typing import Any, cast, override
+from typing import Any, cast
 
 from pydantic import JsonValue
 
@@ -16,7 +16,6 @@ from core.app.entities.task_entities import (
 
 class CompletionAppGenerateResponseConverter(AppGenerateResponseConverter[CompletionAppBlockingResponse]):
     @classmethod
-    @override
     def convert_blocking_full_response(cls, blocking_response: CompletionAppBlockingResponse):
         """
         Convert blocking full response.
@@ -37,7 +36,6 @@ class CompletionAppGenerateResponseConverter(AppGenerateResponseConverter[Comple
         return response
 
     @classmethod
-    @override
     def convert_blocking_simple_response(cls, blocking_response: CompletionAppBlockingResponse):
         """
         Convert blocking simple response.
@@ -55,7 +53,6 @@ class CompletionAppGenerateResponseConverter(AppGenerateResponseConverter[Comple
         return response
 
     @classmethod
-    @override
     def convert_stream_full_response(
         cls, stream_response: Generator[AppStreamResponse, None, None]
     ) -> Generator[dict[str, Any] | str, None, None]:
@@ -86,7 +83,6 @@ class CompletionAppGenerateResponseConverter(AppGenerateResponseConverter[Comple
             yield response_chunk
 
     @classmethod
-    @override
     def convert_stream_simple_response(
         cls, stream_response: Generator[AppStreamResponse, None, None]
     ) -> Generator[dict[str, Any] | str, None, None]:

api/core/app/apps/draft_variable_saver.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import abc
 from collections.abc import Mapping
-from typing import Any, Protocol, override
+from typing import Any, Protocol
 
 from graphon.enums import NodeType
 
@@ -29,6 +29,5 @@ class DraftVariableSaverFactory(Protocol):
 
 
 class NoopDraftVariableSaver(DraftVariableSaver):
-    @override
     def save(self, process_data: Mapping[str, Any] | None, outputs: Mapping[str, Any] | None) -> None:
         return None

api/core/app/apps/message_based_app_queue_manager.py

@@ -1,5 +1,3 @@
-from typing import override
-
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
 from core.app.apps.exc import GenerateTaskStoppedError
 from core.app.entities.app_invoke_entities import InvokeFrom
@@ -23,7 +21,6 @@ class MessageBasedAppQueueManager(AppQueueManager):
         self._app_mode = app_mode
         self._message_id = str(message_id)
 
-    @override
     def _publish(self, event: AppQueueEvent, pub_from: PublishFrom):
         """
         Publish event to queue

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

@@ -1,5 +1,3 @@
-from typing import override
-
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
 from core.app.apps.exc import GenerateTaskStoppedError
 from core.app.entities.app_invoke_entities import InvokeFrom
@@ -21,7 +19,6 @@ class PipelineQueueManager(AppQueueManager):
 
         self._app_mode = app_mode
 
-    @override
     def _publish(self, event: AppQueueEvent, pub_from: PublishFrom) -> None:
         """
         Publish event to queue

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

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

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

@@ -1,5 +1,3 @@
-from typing import override
-
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
 from core.app.apps.exc import GenerateTaskStoppedError
 from core.app.entities.app_invoke_entities import InvokeFrom
@@ -21,7 +19,6 @@ class WorkflowAppQueueManager(AppQueueManager):
 
         self._app_mode = app_mode
 
-    @override
     def _publish(self, event: AppQueueEvent, pub_from: PublishFrom):
         """
         Publish event to queue

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

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

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

@@ -1,5 +1,5 @@
 from collections.abc import Generator
-from typing import Any, cast, override
+from typing import Any, cast
 
 from core.app.apps.base_app_generate_response_converter import AppGenerateResponseConverter
 from core.app.entities.task_entities import (
@@ -18,7 +18,6 @@ class WorkflowAppGenerateResponseConverter(
     AppGenerateResponseConverter[WorkflowAppBlockingResponse | WorkflowAppPausedBlockingResponse]
 ):
     @classmethod
-    @override
     def convert_blocking_full_response(
         cls, blocking_response: WorkflowAppBlockingResponse | WorkflowAppPausedBlockingResponse
     ) -> dict[str, Any]:
@@ -30,7 +29,6 @@ class WorkflowAppGenerateResponseConverter(
         return dict(blocking_response.model_dump())
 
     @classmethod
-    @override
     def convert_blocking_simple_response(
         cls, blocking_response: WorkflowAppBlockingResponse | WorkflowAppPausedBlockingResponse
     ) -> dict[str, Any]:
@@ -42,7 +40,6 @@ class WorkflowAppGenerateResponseConverter(
         return cls.convert_blocking_full_response(blocking_response)
 
     @classmethod
-    @override
     def convert_stream_full_response(
         cls, stream_response: Generator[AppStreamResponse, None, None]
     ) -> Generator[dict[str, Any] | str, None, None]:
@@ -76,7 +73,6 @@ class WorkflowAppGenerateResponseConverter(
             yield response_chunk
 
     @classmethod
-    @override
     def convert_stream_simple_response(
         cls, stream_response: Generator[AppStreamResponse, None, None]
     ) -> Generator[dict[str, Any] | str, None, None]:

api/core/app/file_access/controller.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 from collections.abc import Callable
-from typing import override
 
 from sqlalchemy import and_, or_, select
 from sqlalchemy.orm import Session
@@ -32,11 +31,9 @@ class DatabaseFileAccessController(FileAccessControllerProtocol):
     ) -> None:
         self._scope_getter = scope_getter
 
-    @override
     def current_scope(self) -> FileAccessScope | None:
         return self._scope_getter()
 
-    @override
     def apply_upload_file_filters(
         self,
         stmt: Select[tuple[UploadFile]],
@@ -65,7 +62,6 @@ class DatabaseFileAccessController(FileAccessControllerProtocol):
             )
         )
 
-    @override
     def apply_tool_file_filters(
         self,
         stmt: Select[tuple[ToolFile]],
@@ -82,7 +78,6 @@ class DatabaseFileAccessController(FileAccessControllerProtocol):
 
         return scoped_stmt.where(ToolFile.user_id == resolved_scope.user_id)
 
-    @override
     def get_upload_file(
         self,
         *,
@@ -100,7 +95,6 @@ class DatabaseFileAccessController(FileAccessControllerProtocol):
         )
         return session.scalar(stmt)
 
-    @override
     def get_tool_file(
         self,
         *,

api/core/app/layers/conversation_variable_persist_layer.py

@@ -8,7 +8,6 @@ scope updates that matter to chat applications.
 """
 
 import logging
-from typing import override
 
 from core.workflow.system_variables import SystemVariableKey, get_system_text
 from core.workflow.variable_prefixes import CONVERSATION_VARIABLE_NODE_ID
@@ -24,11 +23,9 @@ class ConversationVariablePersistenceLayer(GraphEngineLayer):
         super().__init__()
         self._conversation_variable_updater = conversation_variable_updater
 
-    @override
     def on_graph_start(self) -> None:
         pass
 
-    @override
     def on_event(self, event: GraphEngineEvent) -> None:
         if not isinstance(event, NodeRunVariableUpdatedEvent):
             return
@@ -47,6 +44,5 @@ class ConversationVariablePersistenceLayer(GraphEngineLayer):
 
         self._conversation_variable_updater.update(conversation_id=conversation_id, variable=event.variable)
 
-    @override
     def on_graph_end(self, error: Exception | None) -> None:
         pass

api/core/app/layers/pause_state_persist_layer.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from typing import Annotated, Literal, Self, override
+from typing import Annotated, Literal, Self
 
 from pydantic import BaseModel, Field
 from sqlalchemy import Engine
@@ -83,7 +83,6 @@ class PauseStatePersistenceLayer(GraphEngineLayer):
     def _get_repo(self) -> APIWorkflowRunRepository:
         return DifyAPIRepositoryFactory.create_api_workflow_run_repository(self._session_maker)
 
-    @override
     def on_graph_start(self) -> None:
         """
         Called when graph execution starts.
@@ -93,7 +92,6 @@ class PauseStatePersistenceLayer(GraphEngineLayer):
         """
         pass
 
-    @override
     def on_event(self, event: GraphEngineEvent) -> None:
         """
         Called for every event emitted by the engine.
@@ -134,7 +132,6 @@ class PauseStatePersistenceLayer(GraphEngineLayer):
             pause_reasons=event.reasons,
         )
 
-    @override
     def on_graph_end(self, error: Exception | None) -> None:
         """
         Called when graph execution ends.

api/core/app/layers/suspend_layer.py

@@ -1,5 +1,3 @@
-from typing import override
-
 from graphon.graph_engine.layers import GraphEngineLayer
 from graphon.graph_events import GraphEngineEvent, GraphRunPausedEvent
 
@@ -11,11 +9,9 @@ class SuspendLayer(GraphEngineLayer):
         super().__init__()
         self._paused = False
 
-    @override
     def on_graph_start(self):
         self._paused = False
 
-    @override
     def on_event(self, event: GraphEngineEvent):
         """
         Handle the paused event, stash runtime state into storage and wait for resume.
@@ -23,7 +19,6 @@ class SuspendLayer(GraphEngineLayer):
         if isinstance(event, GraphRunPausedEvent):
             self._paused = True
 
-    @override
     def on_graph_end(self, error: Exception | None):
         """ """
         self._paused = False

api/core/app/layers/timeslice_layer.py

@@ -1,6 +1,6 @@
 import logging
 import uuid
-from typing import ClassVar, override
+from typing import ClassVar
 
 from apscheduler.schedulers.background import BackgroundScheduler  # type: ignore
 
@@ -63,7 +63,6 @@ class TimeSliceLayer(GraphEngineLayer):
         except Exception:
             logger.exception("scheduler error during check if the workflow need to be suspended")
 
-    @override
     def on_graph_start(self):
         """
         Start timer to check if the workflow need to be suspended.
@@ -79,11 +78,9 @@ class TimeSliceLayer(GraphEngineLayer):
                 id=self.schedule_id,
             )
 
-    @override
     def on_event(self, event: GraphEngineEvent):
         pass
 
-    @override
     def on_graph_end(self, error: Exception | None) -> None:
         self.stopped = True
         # remove the scheduler

api/core/app/layers/trigger_post_layer.py

@@ -1,6 +1,6 @@
 import logging
 from datetime import UTC, datetime
-from typing import Any, ClassVar, override
+from typing import Any, ClassVar
 
 from pydantic import TypeAdapter
 
@@ -37,11 +37,9 @@ class TriggerPostLayer(GraphEngineLayer):
         self.start_time = start_time
         self.cfs_plan_scheduler_entity = cfs_plan_scheduler_entity
 
-    @override
     def on_graph_start(self):
         pass
 
-    @override
     def on_event(self, event: GraphEngineEvent):
         """
         Update trigger log with success or failure.
@@ -84,6 +82,5 @@ class TriggerPostLayer(GraphEngineLayer):
                 repo.update(trigger_log)
                 session.commit()
 
-    @override
     def on_graph_end(self, error: Exception | None) -> None:
         pass

api/core/app/workflow/file_runtime.py

@@ -7,7 +7,7 @@ import os
 import time
 import urllib.parse
 from collections.abc import Generator
-from typing import TYPE_CHECKING, Literal, override
+from typing import TYPE_CHECKING, Literal
 
 from configs import dify_config
 from core.app.file_access import DatabaseFileAccessController, FileAccessControllerProtocol
@@ -40,19 +40,15 @@ class DifyWorkflowFileRuntime(WorkflowFileRuntimeProtocol):
         self._file_access_controller = file_access_controller
 
     @property
-    @override
     def multimodal_send_format(self) -> str:
         return dify_config.MULTIMODAL_SEND_FORMAT
 
-    @override
     def http_get(self, url: str, *, follow_redirects: bool = True) -> HttpResponseProtocol:
         return graphon_ssrf_proxy.get(url, follow_redirects=follow_redirects)
 
-    @override
     def storage_load(self, path: str, *, stream: bool = False) -> bytes | Generator:
         return storage.load(path, stream=stream)
 
-    @override
     def load_file_bytes(self, *, file: File) -> bytes:
         storage_key = self._resolve_storage_key(file=file)
         data = storage.load(storage_key, stream=False)
@@ -60,7 +56,6 @@ class DifyWorkflowFileRuntime(WorkflowFileRuntimeProtocol):
             raise ValueError(f"file {storage_key} is not a bytes object")
         return data
 
-    @override
     def resolve_file_url(self, *, file: File, for_external: bool = True) -> str | None:
         if file.transfer_method == FileTransferMethod.REMOTE_URL:
             return file.remote_url
@@ -91,7 +86,6 @@ class DifyWorkflowFileRuntime(WorkflowFileRuntimeProtocol):
             )
         return None
 
-    @override
     def resolve_upload_file_url(
         self,
         *,
@@ -107,12 +101,10 @@ class DifyWorkflowFileRuntime(WorkflowFileRuntimeProtocol):
             query["as_attachment"] = "true"
         return f"{url}?{urllib.parse.urlencode(query)}"
 
-    @override
     def resolve_tool_file_url(self, *, tool_file_id: str, extension: str, for_external: bool = True) -> str:
         self._assert_tool_file_access(tool_file_id=tool_file_id)
         return sign_tool_file(tool_file_id=tool_file_id, extension=extension, for_external=for_external)
 
-    @override
     def verify_preview_signature(
         self,
         *,

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

@@ -12,7 +12,7 @@ state.
 from collections.abc import Mapping
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Any, Union, override
+from typing import Any, Union
 
 from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, WorkflowAppGenerateEntity
 from core.helper.trace_id_helper import ParentTraceContext
@@ -98,14 +98,12 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
     # ------------------------------------------------------------------
     # GraphEngineLayer lifecycle
     # ------------------------------------------------------------------
-    @override
     def on_graph_start(self) -> None:
         self._workflow_execution = None
         self._node_execution_cache.clear()
         self._node_snapshots.clear()
         self._node_sequence = 0
 
-    @override
     def on_event(self, event: GraphEngineEvent) -> None:
         match event:
             case GraphRunStartedEvent():
@@ -133,7 +131,6 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
             case NodeRunPauseRequestedEvent():
                 self._handle_node_pause_requested(event)
 
-    @override
     def on_graph_end(self, error: Exception | None) -> None:
         return
 

api/core/datasource/local_file/local_file_plugin.py

@@ -1,5 +1,3 @@
-from typing import override
-
 from core.datasource.__base.datasource_plugin import DatasourcePlugin
 from core.datasource.__base.datasource_runtime import DatasourceRuntime
 from core.datasource.entities.datasource_entities import (
@@ -24,10 +22,8 @@ class LocalFileDatasourcePlugin(DatasourcePlugin):
         self.tenant_id = tenant_id
         self.plugin_unique_identifier = plugin_unique_identifier
 
-    @override
     def datasource_provider_type(self) -> str:
         return DatasourceProviderType.LOCAL_FILE
 
-    @override
     def get_icon_url(self, tenant_id: str) -> str:
         return self.icon

api/core/datasource/local_file/local_file_provider.py

@@ -1,4 +1,4 @@
-from typing import Any, override
+from typing import Any
 
 from core.datasource.__base.datasource_provider import DatasourcePluginProviderController
 from core.datasource.__base.datasource_runtime import DatasourceRuntime
@@ -19,14 +19,12 @@ class LocalFileDatasourcePluginProviderController(DatasourcePluginProviderContro
         self.plugin_unique_identifier = plugin_unique_identifier
 
     @property
-    @override
     def provider_type(self) -> DatasourceProviderType:
         """
         returns the type of the provider
         """
         return DatasourceProviderType.LOCAL_FILE
 
-    @override
     def _validate_credentials(self, user_id: str, credentials: dict[str, Any]) -> None:
         """
         validate the credentials of the provider

api/core/datasource/online_document/online_document_plugin.py

@@ -1,5 +1,5 @@
 from collections.abc import Generator
-from typing import Any, override
+from typing import Any
 
 from core.datasource.__base.datasource_plugin import DatasourcePlugin
 from core.datasource.__base.datasource_runtime import DatasourceRuntime
@@ -67,6 +67,5 @@ class OnlineDocumentDatasourcePlugin(DatasourcePlugin):
             provider_type=provider_type,
         )
 
-    @override
     def datasource_provider_type(self) -> str:
         return DatasourceProviderType.ONLINE_DOCUMENT

api/core/datasource/online_document/online_document_provider.py

@@ -1,5 +1,3 @@
-from typing import override
-
 from core.datasource.__base.datasource_provider import DatasourcePluginProviderController
 from core.datasource.__base.datasource_runtime import DatasourceRuntime
 from core.datasource.entities.datasource_entities import DatasourceProviderEntityWithPlugin, DatasourceProviderType
@@ -19,7 +17,6 @@ class OnlineDocumentDatasourcePluginProviderController(DatasourcePluginProviderC
         self.plugin_unique_identifier = plugin_unique_identifier
 
     @property
-    @override
     def provider_type(self) -> DatasourceProviderType:
         """
         returns the type of the provider

api/core/datasource/online_drive/online_drive_plugin.py

@@ -1,5 +1,4 @@
 from collections.abc import Generator
-from typing import override
 
 from core.datasource.__base.datasource_plugin import DatasourcePlugin
 from core.datasource.__base.datasource_runtime import DatasourceRuntime
@@ -68,6 +67,5 @@ class OnlineDriveDatasourcePlugin(DatasourcePlugin):
             provider_type=provider_type,
         )
 
-    @override
     def datasource_provider_type(self) -> str:
         return DatasourceProviderType.ONLINE_DRIVE

api/core/datasource/online_drive/online_drive_provider.py

@@ -1,5 +1,3 @@
-from typing import override
-
 from core.datasource.__base.datasource_provider import DatasourcePluginProviderController
 from core.datasource.__base.datasource_runtime import DatasourceRuntime
 from core.datasource.entities.datasource_entities import DatasourceProviderEntityWithPlugin, DatasourceProviderType
@@ -19,7 +17,6 @@ class OnlineDriveDatasourcePluginProviderController(DatasourcePluginProviderCont
         self.plugin_unique_identifier = plugin_unique_identifier
 
     @property
-    @override
     def provider_type(self) -> DatasourceProviderType:
         """
         returns the type of the provider

api/core/datasource/website_crawl/website_crawl_plugin.py

@@ -1,5 +1,5 @@
 from collections.abc import Generator, Mapping
-from typing import Any, override
+from typing import Any
 
 from core.datasource.__base.datasource_plugin import DatasourcePlugin
 from core.datasource.__base.datasource_runtime import DatasourceRuntime
@@ -47,6 +47,5 @@ class WebsiteCrawlDatasourcePlugin(DatasourcePlugin):
             provider_type=provider_type,
         )
 
-    @override
     def datasource_provider_type(self) -> str:
         return DatasourceProviderType.WEBSITE_CRAWL

api/core/datasource/website_crawl/website_crawl_provider.py

@@ -1,5 +1,3 @@
-from typing import override
-
 from core.datasource.__base.datasource_provider import DatasourcePluginProviderController
 from core.datasource.__base.datasource_runtime import DatasourceRuntime
 from core.datasource.entities.datasource_entities import DatasourceProviderEntityWithPlugin, DatasourceProviderType
@@ -23,7 +21,6 @@ class WebsiteCrawlDatasourcePluginProviderController(DatasourcePluginProviderCon
         self.plugin_unique_identifier = plugin_unique_identifier
 
     @property
-    @override
     def provider_type(self) -> DatasourceProviderType:
         """
         returns the type of the provider

api/core/entities/provider_configuration.py

@@ -6,7 +6,7 @@ import re
 from collections import defaultdict
 from collections.abc import Iterator, Sequence
 from json import JSONDecodeError
-from typing import Any, override
+from typing import Any
 
 from pydantic import BaseModel, ConfigDict, Field, PrivateAttr, model_validator
 from sqlalchemy import func, select
@@ -1889,7 +1889,6 @@ class ProviderConfigurations(BaseModel):
             key = str(ModelProviderID(key))
         return key in self.configurations
 
-    @override
     def __iter__(self):
         # Return an iterator of (key, value) tuples to match BaseModel's __iter__
         yield from self.configurations.items()

api/core/external_data_tool/api/api.py

@@ -1,5 +1,5 @@
 from collections.abc import Mapping
-from typing import Any, TypedDict, override
+from typing import Any, TypedDict
 
 from sqlalchemy import select
 
@@ -29,7 +29,6 @@ class ApiExternalDataTool(ExternalDataTool):
     """the unique name of external data tool"""
 
     @classmethod
-    @override
     def validate_config(cls, tenant_id: str, config: dict[str, Any]):
         """
         Validate the incoming form config data.
@@ -51,7 +50,6 @@ class ApiExternalDataTool(ExternalDataTool):
         if not api_based_extension:
             raise ValueError("api_based_extension_id is invalid")
 
-    @override
     def query(self, inputs: Mapping[str, Any], query: str | None = None) -> str:
         """
         Query the external data tool.

api/core/helper/code_executor/javascript/javascript_code_provider.py

@@ -1,5 +1,4 @@
 from textwrap import dedent
-from typing import override
 
 from core.helper.code_executor.code_executor import CodeLanguage
 from core.helper.code_executor.code_node_provider import CodeNodeProvider
@@ -7,12 +6,10 @@ from core.helper.code_executor.code_node_provider import CodeNodeProvider
 
 class JavascriptCodeProvider(CodeNodeProvider):
     @staticmethod
-    @override
     def get_language() -> str:
         return CodeLanguage.JAVASCRIPT
 
     @classmethod
-    @override
     def get_default_code(cls) -> str:
         return dedent(
             """

api/core/helper/code_executor/javascript/javascript_transformer.py

@@ -1,12 +1,10 @@
 from textwrap import dedent
-from typing import override
 
 from core.helper.code_executor.template_transformer import TemplateTransformer
 
 
 class NodeJsTemplateTransformer(TemplateTransformer):
     @classmethod
-    @override
     def get_runner_script(cls) -> str:
         runner_script = dedent(f"""            {cls._code_placeholder}
 

api/core/helper/code_executor/jinja2/jinja2_transformer.py

@@ -1,6 +1,6 @@
 from collections.abc import Mapping
 from textwrap import dedent
-from typing import Any, override
+from typing import Any
 
 from core.helper.code_executor.template_transformer import TemplateTransformer
 
@@ -10,7 +10,6 @@ class Jinja2TemplateTransformer(TemplateTransformer):
     _template_b64_placeholder: str = "{{template_b64}}"
 
     @classmethod
-    @override
     def transform_response(cls, response: str):
         """
         Transform response to dict
@@ -20,7 +19,6 @@ class Jinja2TemplateTransformer(TemplateTransformer):
         return {"result": cls.extract_result_str_from_response(response)}
 
     @classmethod
-    @override
     def assemble_runner_script(cls, code: str, inputs: Mapping[str, Any]) -> str:
         """
         Override base class to use base64 encoding for template code.
@@ -36,7 +34,6 @@ class Jinja2TemplateTransformer(TemplateTransformer):
         return script
 
     @classmethod
-    @override
     def get_runner_script(cls) -> str:
         runner_script = dedent(f"""
             import jinja2
@@ -64,7 +61,6 @@ class Jinja2TemplateTransformer(TemplateTransformer):
         return runner_script
 
     @classmethod
-    @override
     def get_preload_script(cls) -> str:
         preload_script = dedent("""
             import jinja2

api/core/helper/code_executor/python3/python3_code_provider.py

@@ -1,5 +1,4 @@
 from textwrap import dedent
-from typing import override
 
 from core.helper.code_executor.code_executor import CodeLanguage
 from core.helper.code_executor.code_node_provider import CodeNodeProvider
@@ -7,12 +6,10 @@ from core.helper.code_executor.code_node_provider import CodeNodeProvider
 
 class Python3CodeProvider(CodeNodeProvider):
     @staticmethod
-    @override
     def get_language() -> str:
         return CodeLanguage.PYTHON3
 
     @classmethod
-    @override
     def get_default_code(cls) -> str:
         return dedent(
             """

api/core/helper/code_executor/python3/python3_transformer.py

@@ -1,12 +1,10 @@
 from textwrap import dedent
-from typing import override
 
 from core.helper.code_executor.template_transformer import TemplateTransformer
 
 
 class Python3TemplateTransformer(TemplateTransformer):
     @classmethod
-    @override
     def get_runner_script(cls) -> str:
         runner_script = dedent(f"""            {cls._code_placeholder}
 

api/core/helper/provider_cache.py

@@ -1,7 +1,7 @@
 import json
 from abc import ABC, abstractmethod
 from json import JSONDecodeError
-from typing import Any, override
+from typing import Any
 
 from extensions.ext_redis import redis_client
 
@@ -47,7 +47,6 @@ class SingletonProviderCredentialsCache(ProviderCredentialsCache):
             provider_identity=provider_identity,
         )
 
-    @override
     def _generate_cache_key(self, **kwargs) -> str:
         tenant_id = kwargs["tenant_id"]
         provider_type = kwargs["provider_type"]
@@ -62,7 +61,6 @@ class ToolProviderCredentialsCache(ProviderCredentialsCache):
     def __init__(self, tenant_id: str, provider: str, credential_id: str):
         super().__init__(tenant_id=tenant_id, provider=provider, credential_id=credential_id)
 
-    @override
     def _generate_cache_key(self, **kwargs) -> str:
         tenant_id = kwargs["tenant_id"]
         provider = kwargs["provider"]

api/core/helper/ssrf_proxy.py

@@ -43,16 +43,13 @@ request_error = httpx.RequestError
 max_retries_exceeded_error = MaxRetriesExceededError
 
 
-def _create_proxy_mounts(verify: bool) -> dict[str, httpx.HTTPTransport]:
-    """Build per-scheme proxy transports with the same TLS policy as the SSRF client."""
+def _create_proxy_mounts() -> dict[str, httpx.HTTPTransport]:
     return {
         "http://": httpx.HTTPTransport(
             proxy=dify_config.SSRF_PROXY_HTTP_URL,
-            verify=verify,
         ),
         "https://": httpx.HTTPTransport(
             proxy=dify_config.SSRF_PROXY_HTTPS_URL,
-            verify=verify,
         ),
     }
 
@@ -67,7 +64,7 @@ def _build_ssrf_client(verify: bool) -> httpx.Client:
 
     if dify_config.SSRF_PROXY_HTTP_URL and dify_config.SSRF_PROXY_HTTPS_URL:
         return httpx.Client(
-            mounts=_create_proxy_mounts(verify=verify),
+            mounts=_create_proxy_mounts(),
             verify=verify,
             limits=_SSRF_CLIENT_LIMITS,
         )

api/core/logging/filters.py

@@ -2,7 +2,6 @@
 
 import contextlib
 import logging
-from typing import override
 
 import flask
 
@@ -16,7 +15,6 @@ class TraceContextFilter(logging.Filter):
     Integrates with OpenTelemetry when available, falls back to ContextVar-based trace_id.
     """
 
-    @override
     def filter(self, record: logging.LogRecord) -> bool:
         # Get trace context from OpenTelemetry
         trace_id, span_id = self._get_otel_context()
@@ -56,7 +54,6 @@ class IdentityContextFilter(logging.Filter):
     Extracts tenant_id, user_id, and user_type from Flask-Login current_user.
     """
 
-    @override
     def filter(self, record: logging.LogRecord) -> bool:
         identity = self._extract_identity()
         record.tenant_id = identity.get("tenant_id", "")

api/core/logging/structured_formatter.py

@@ -3,7 +3,7 @@
 import logging
 import traceback
 from datetime import UTC, datetime
-from typing import Any, NotRequired, TypedDict, override
+from typing import Any, NotRequired, TypedDict
 
 import orjson
 
@@ -58,7 +58,6 @@ class StructuredJSONFormatter(logging.Formatter):
         super().__init__()
         self._service_name = service_name or dify_config.APPLICATION_NAME
 
-    @override
     def format(self, record: logging.LogRecord) -> str:
         log_dict = self._build_log_dict(record)
         try:

api/core/mcp/auth_client.py

@@ -7,7 +7,7 @@ authentication failures and retries operations after refreshing tokens.
 
 import logging
 from collections.abc import Callable
-from typing import Any, override
+from typing import Any
 
 from sqlalchemy.orm import Session
 
@@ -159,7 +159,6 @@ class MCPClientWithAuthRetry(MCPClient):
             # Reset retry flag after operation completes
             self._has_retried = False
 
-    @override
     def __enter__(self):
         """Enter the context manager with retry support."""
 
@@ -169,7 +168,6 @@ class MCPClientWithAuthRetry(MCPClient):
 
         return self._execute_with_retry(initialize_with_retry)
 
-    @override
     def list_tools(self) -> list[Tool]:
         """
         List available tools from the MCP server with auth retry.
@@ -182,7 +180,6 @@ class MCPClientWithAuthRetry(MCPClient):
         """
         return self._execute_with_retry(super().list_tools)
 
-    @override
     def invoke_tool(self, tool_name: str, tool_args: dict[str, Any]) -> CallToolResult:
         """
         Invoke a tool on the MCP server with auth retry.

api/core/mcp/session/client_session.py

@@ -1,6 +1,6 @@
 import queue
 from datetime import timedelta
-from typing import Any, Protocol, override
+from typing import Any, Protocol
 
 from pydantic import AnyUrl, TypeAdapter
 
@@ -159,7 +159,6 @@ class ClientSession(
             types.EmptyResult,
         )
 
-    @override
     def send_progress_notification(self, progress_token: str | int, progress: float, total: float | None = None):
         """Send a progress notification."""
         self.send_notification(
@@ -327,7 +326,6 @@ class ClientSession(
             )
         )
 
-    @override
     def _received_request(self, responder: RequestResponder[types.ServerRequest, types.ClientResult]):
         ctx = RequestContext[ClientSession, Any](
             request_id=responder.request_id,
@@ -353,7 +351,6 @@ class ClientSession(
                 with responder:
                     return responder.respond(types.ClientResult(root=types.EmptyResult()))
 
-    @override
     def _handle_incoming(
         self,
         req: RequestResponder[types.ServerRequest, types.ClientResult] | types.ServerNotification | Exception,
@@ -361,7 +358,6 @@ class ClientSession(
         """Handle incoming messages by forwarding to the message handler."""
         self._message_handler(req)
 
-    @override
     def _received_notification(self, notification: types.ServerNotification):
         """Handle notifications from the server."""
         # Process specific notification types

api/core/moderation/api/api.py

@@ -1,4 +1,4 @@
-from typing import Any, override
+from typing import Any
 
 from pydantic import BaseModel, Field
 from sqlalchemy import select
@@ -25,7 +25,6 @@ class ApiModeration(Moderation):
     name: str = "api"
 
     @classmethod
-    @override
     def validate_config(cls, tenant_id: str, config: dict[str, Any]):
         """
         Validate the incoming form config data.
@@ -44,7 +43,6 @@ class ApiModeration(Moderation):
         if not extension:
             raise ValueError("API-based Extension not found. Please check it again.")
 
-    @override
     def moderation_for_inputs(self, inputs: dict[str, Any], query: str = "") -> ModerationInputsResult:
         flagged = False
         preset_response = ""
@@ -61,7 +59,6 @@ class ApiModeration(Moderation):
             flagged=flagged, action=ModerationAction.DIRECT_OUTPUT, preset_response=preset_response
         )
 
-    @override
     def moderation_for_outputs(self, text: str) -> ModerationOutputsResult:
         flagged = False
         preset_response = ""

api/core/moderation/keywords/keywords.py

@@ -1,5 +1,5 @@
 from collections.abc import Sequence
-from typing import Any, override
+from typing import Any
 
 from core.moderation.base import Moderation, ModerationAction, ModerationInputsResult, ModerationOutputsResult
 
@@ -8,7 +8,6 @@ class KeywordsModeration(Moderation):
     name: str = "keywords"
 
     @classmethod
-    @override
     def validate_config(cls, tenant_id: str, config: dict[str, Any]):
         """
         Validate the incoming form config data.
@@ -29,7 +28,6 @@ class KeywordsModeration(Moderation):
         if len(keywords_row_len) > 100:
             raise ValueError("the number of rows for the keywords must be less than 100")
 
-    @override
     def moderation_for_inputs(self, inputs: dict[str, Any], query: str = "") -> ModerationInputsResult:
         flagged = False
         preset_response = ""
@@ -51,7 +49,6 @@ class KeywordsModeration(Moderation):
             flagged=flagged, action=ModerationAction.DIRECT_OUTPUT, preset_response=preset_response
         )
 
-    @override
     def moderation_for_outputs(self, text: str) -> ModerationOutputsResult:
         flagged = False
         preset_response = ""

api/core/moderation/openai_moderation/openai_moderation.py

@@ -1,4 +1,4 @@
-from typing import Any, override
+from typing import Any
 
 from core.model_manager import ModelManager
 from core.moderation.base import Moderation, ModerationAction, ModerationInputsResult, ModerationOutputsResult
@@ -9,7 +9,6 @@ class OpenAIModeration(Moderation):
     name: str = "openai_moderation"
 
     @classmethod
-    @override
     def validate_config(cls, tenant_id: str, config: dict[str, Any]):
         """
         Validate the incoming form config data.
@@ -20,7 +19,6 @@ class OpenAIModeration(Moderation):
         """
         cls._validate_inputs_and_outputs_config(config, True)
 
-    @override
     def moderation_for_inputs(self, inputs: dict[str, Any], query: str = "") -> ModerationInputsResult:
         flagged = False
         preset_response = ""
@@ -38,7 +36,6 @@ class OpenAIModeration(Moderation):
             flagged=flagged, action=ModerationAction.DIRECT_OUTPUT, preset_response=preset_response
         )
 
-    @override
     def moderation_for_outputs(self, text: str) -> ModerationOutputsResult:
         flagged = False
         preset_response = ""

api/core/plugin/impl/exc.py

@@ -1,5 +1,4 @@
 from collections.abc import Mapping
-from typing import override
 
 from pydantic import TypeAdapter
 
@@ -12,7 +11,6 @@ class PluginDaemonError(Exception):
     def __init__(self, description: str):
         self.description = description
 
-    @override
     def __str__(self) -> str:
         # returns the class name and description
         return f"req_id: {get_request_id()} {self.__class__.__name__}: {self.description}"

api/core/plugin/impl/model_runtime.py

@@ -3,7 +3,8 @@ from __future__ import annotations
 import hashlib
 import logging
 from collections.abc import Generator, Iterable, Sequence
-from typing import IO, Any, Literal, cast, overload, override
+from threading import Lock
+from typing import IO, Any, Literal, cast, overload
 
 from pydantic import ValidationError
 from redis import RedisError
@@ -12,9 +13,9 @@ from configs import dify_config
 from core.llm_generator.output_parser.structured_output import (
     invoke_llm_with_structured_output as invoke_llm_with_structured_output_helper,
 )
+from core.plugin.entities.plugin_daemon import PluginModelProviderEntity
 from core.plugin.impl.asset import PluginAssetManager
 from core.plugin.impl.model import PluginModelClient
-from core.plugin.plugin_service import PluginService
 from extensions.ext_redis import redis_client
 from graphon.model_runtime.entities.llm_entities import (
     LLMResult,
@@ -100,38 +101,35 @@ class _PluginStructuredOutputModelInstance:
 
 
 class PluginModelRuntime(ModelRuntime):
-    """Plugin-backed runtime adapter bound to tenant context and optional caller scope.
-
-    Provider discovery goes through ``PluginService`` so the plugin lifecycle
-    methods and provider reads share one tenant-scoped cache owner.
-    """
+    """Plugin-backed runtime adapter bound to tenant context and optional caller scope."""
 
     tenant_id: str
     user_id: str | None
     client: PluginModelClient
-    _plugin_service: type[PluginService]
+    _provider_entities: tuple[ProviderEntity, ...] | None
+    _provider_entities_lock: Lock
 
-    def __init__(
-        self,
-        tenant_id: str,
-        user_id: str | None,
-        client: PluginModelClient,
-        plugin_service: type[PluginService],
-    ) -> None:
+    def __init__(self, tenant_id: str, user_id: str | None, client: PluginModelClient) -> None:
         if client is None:
             raise ValueError("client is required.")
-        if plugin_service is None:
-            raise ValueError("plugin_service is required.")
         self.tenant_id = tenant_id
         self.user_id = user_id
         self.client = client
-        self._plugin_service = plugin_service
+        self._provider_entities = None
+        self._provider_entities_lock = Lock()
 
-    @override
     def fetch_model_providers(self) -> Sequence[ProviderEntity]:
-        return self._plugin_service.fetch_plugin_model_providers(tenant_id=self.tenant_id, client=self.client)
+        if self._provider_entities is not None:
+            return self._provider_entities
+
+        with self._provider_entities_lock:
+            if self._provider_entities is None:
+                self._provider_entities = tuple(
+                    self._to_provider_entity(provider) for provider in self.client.fetch_model_providers(self.tenant_id)
+                )
+
+        return self._provider_entities
 
-    @override
     def get_provider_icon(self, *, provider: str, icon_type: str, lang: str) -> tuple[bytes, str]:
         provider_schema = self._get_provider_schema(provider)
 
@@ -174,7 +172,6 @@ class PluginModelRuntime(ModelRuntime):
         mime_type = image_mime_types.get(extension, "image/png")
         return PluginAssetManager().fetch_asset(tenant_id=self.tenant_id, id=file_name), mime_type
 
-    @override
     def validate_provider_credentials(self, *, provider: str, credentials: dict[str, Any]) -> None:
         plugin_id, provider_name = self._split_provider(provider)
         self.client.validate_provider_credentials(
@@ -185,7 +182,6 @@ class PluginModelRuntime(ModelRuntime):
             credentials=credentials,
         )
 
-    @override
     def validate_model_credentials(
         self,
         *,
@@ -205,7 +201,6 @@ class PluginModelRuntime(ModelRuntime):
             credentials=credentials,
         )
 
-    @override
     def get_model_schema(
         self,
         *,
@@ -299,7 +294,6 @@ class PluginModelRuntime(ModelRuntime):
         stream: Literal[True],
     ) -> Generator[LLMResultChunk, None, None]: ...
 
-    @override
     def invoke_llm(
         self,
         *,
@@ -363,7 +357,6 @@ class PluginModelRuntime(ModelRuntime):
         stream: Literal[True],
     ) -> Generator[LLMResultChunkWithStructuredOutput, None, None]: ...
 
-    @override
     def invoke_llm_with_structured_output(
         self,
         *,
@@ -403,7 +396,6 @@ class PluginModelRuntime(ModelRuntime):
             stream=stream,
         )
 
-    @override
     def get_llm_num_tokens(
         self,
         *,
@@ -430,7 +422,6 @@ class PluginModelRuntime(ModelRuntime):
             tools=list(tools) if tools else None,
         )
 
-    @override
     def invoke_text_embedding(
         self,
         *,
@@ -452,7 +443,6 @@ class PluginModelRuntime(ModelRuntime):
             input_type=input_type,
         )
 
-    @override
     def invoke_multimodal_embedding(
         self,
         *,
@@ -474,7 +464,6 @@ class PluginModelRuntime(ModelRuntime):
             input_type=input_type,
         )
 
-    @override
     def get_text_embedding_num_tokens(
         self,
         *,
@@ -494,7 +483,6 @@ class PluginModelRuntime(ModelRuntime):
             texts=texts,
         )
 
-    @override
     def invoke_rerank(
         self,
         *,
@@ -520,7 +508,6 @@ class PluginModelRuntime(ModelRuntime):
             top_n=top_n,
         )
 
-    @override
     def invoke_multimodal_rerank(
         self,
         *,
@@ -546,7 +533,6 @@ class PluginModelRuntime(ModelRuntime):
             top_n=top_n,
         )
 
-    @override
     def invoke_tts(
         self,
         *,
@@ -568,7 +554,6 @@ class PluginModelRuntime(ModelRuntime):
             voice=voice,
         )
 
-    @override
     def get_tts_model_voices(
         self,
         *,
@@ -588,7 +573,6 @@ class PluginModelRuntime(ModelRuntime):
             language=language,
         )
 
-    @override
     def invoke_speech_to_text(
         self,
         *,
@@ -608,7 +592,6 @@ class PluginModelRuntime(ModelRuntime):
             file=file,
         )
 
-    @override
     def invoke_moderation(
         self,
         *,
@@ -628,6 +611,34 @@ class PluginModelRuntime(ModelRuntime):
             text=text,
         )
 
+    def _get_provider_short_name_alias(self, provider: PluginModelProviderEntity) -> str:
+        """
+        Expose a bare provider alias only for the canonical provider mapping.
+
+        Multiple plugins can publish the same short provider slug. If every
+        provider entity keeps that slug in ``provider_name``, callers that still
+        resolve by short name become order-dependent. Restrict the alias to the
+        provider selected by ``ModelProviderID`` so legacy short-name lookups
+        remain deterministic while the runtime surface stays canonical.
+        """
+        try:
+            canonical_provider_id = ModelProviderID(provider.provider)
+        except ValueError:
+            return ""
+
+        if canonical_provider_id.plugin_id != provider.plugin_id:
+            return ""
+        if canonical_provider_id.provider_name != provider.provider:
+            return ""
+
+        return provider.provider
+
+    def _to_provider_entity(self, provider: PluginModelProviderEntity) -> ProviderEntity:
+        declaration = provider.declaration.model_copy(deep=True)
+        declaration.provider = f"{provider.plugin_id}/{provider.provider}"
+        declaration.provider_name = self._get_provider_short_name_alias(provider)
+        return declaration
+
     def _get_provider_schema(self, provider: str) -> ProviderEntity:
         providers = self.fetch_model_providers()
         provider_entity = next((item for item in providers if item.provider == provider), None)

api/core/plugin/impl/model_runtime_factory.py

@@ -3,7 +3,6 @@ from __future__ import annotations
 from typing import TYPE_CHECKING
 
 from core.plugin.impl.model import PluginModelClient
-from core.plugin.plugin_service import PluginService
 from graphon.model_runtime.entities.model_entities import ModelType
 from graphon.model_runtime.entities.provider_entities import ProviderEntity
 from graphon.model_runtime.model_providers.base.ai_model import AIModel
@@ -118,7 +117,6 @@ def create_plugin_model_runtime(*, tenant_id: str, user_id: str | None = None) -
         tenant_id=tenant_id,
         user_id=user_id,
         client=PluginModelClient(),
-        plugin_service=PluginService,
     )