refactor(web): migrate shared icon commands

Replace ordinary ActionButton usage across shared, app, and chat surfaces with the Dify UI IconButton. Preserve legacy sizes and interaction colors, move popup styling to Base UI state attributes, and require accessible names.

.agents/skills/how-to-write-component/references/runtime.md

@@ -4,6 +4,7 @@ Read this document when a change introduces Effects, navigation side effects, me
 
 ## Effects
 
+- Keep render pure: do not read or write `ref.current` during render except for predictable null-guarded lazy initialization. Update interaction-owned refs in event handlers, synchronize external-system refs after commit, and use state or derivation for rendered values.
 - Use Effects only to synchronize with a named external system such as a browser API, subscription, timer, analytics integration, non-React widget, or imperative DOM API.
 - Do not use Effects to transform render state, handle user actions, copy query data, reset state from props, or fetch data owned by framework APIs or TanStack Query.
 - Initialize query-backed forms with keyed remounts or surface-entry hydration instead of copying data through Effects.

.github/CODEOWNERS

@@ -245,7 +245,7 @@
 
 # Frontend - Billing and Education
 /web/app/components/billing/ @iamjoel @zxhlyh
-/web/app/education-apply/ @iamjoel @zxhlyh
+/web/app/education/ @iamjoel @zxhlyh
 
 # Frontend - Workspace
 /web/app/components/header/account-dropdown/workplace-selector/ @iamjoel @zxhlyh

.github/workflows/db-migration-test.yml

@@ -40,7 +40,7 @@ jobs:
           cp envs/middleware.env.example middleware.env
 
       - name: Set up Middlewares
-        uses: hoverkraft-tech/compose-action@11beaa1c2dae4e8ed7b1665aa074723b6cecb0e4 # v3.0.0
+        uses: hoverkraft-tech/compose-action@ee6af68587292d72db67743171809c19787df4c9 # v3.1.0
         with:
           compose-file: |
             docker/docker-compose.middleware.yaml
@@ -94,7 +94,7 @@ jobs:
           sed -i 's/DB_USERNAME=postgres/DB_USERNAME=mysql/' middleware.env
 
       - name: Set up Middlewares
-        uses: hoverkraft-tech/compose-action@11beaa1c2dae4e8ed7b1665aa074723b6cecb0e4 # v3.0.0
+        uses: hoverkraft-tech/compose-action@ee6af68587292d72db67743171809c19787df4c9 # v3.1.0
         with:
           compose-file: |
             docker/docker-compose.middleware.yaml

.github/workflows/translate-i18n-claude.yml

@@ -158,7 +158,7 @@ jobs:
 
       - name: Run Claude Code for Translation Sync
         if: steps.context.outputs.CHANGED_FILES != ''
-        uses: anthropics/claude-code-action@c038e4dcdedfbbca18dfb17df35a17e40ded4ddc # v1.0.186
+        uses: anthropics/claude-code-action@1623c36729ac1cd5895198cded705a287de7db79 # v1.0.187
         with:
           anthropic_api_key: ${{ secrets.ANTHROPIC_API_KEY }}
           github_token: ${{ secrets.GITHUB_TOKEN }}

RESOURCE_BOUNDARY_CHANGE_GUIDE.md

@@ -0,0 +1,12 @@
+# Resource Boundary Change Guide
+
+- Resolve the tenant-scoped parent at the request boundary, then pass the validated model, owner reference, and actor downstream.
+- Put the complete owner tuple in the database query; do not load by a bare ID and check ownership afterward.
+- Treat missing and foreign-owned resources alike as `404` before locks, rate limits, tasks, plugin calls, network calls, or writes.
+- Reuse existing owner resolvers and trusted objects instead of adding parallel helpers or refetching the same resource.
+- Pass tenant, actor, and session explicitly; authenticated code must not depend on ambient account or tenant fallbacks.
+- Raise typed domain errors in services and translate them to HTTP errors in controllers; reserve `ValueError` for invalid values or state.
+- Let RBAC own authorization when enabled, and run legacy dataset permission checks only when RBAC is disabled.
+- Preserve successful HTTP responses and shared runtime contracts, especially Celery task names and argument shapes during rolling upgrades.
+- Keep runtime validation and OpenAPI schemas aligned, then regenerate Markdown and TypeScript contracts after schema changes.
+- Prove the boundary with a foreign-owner decoy and assert that rejected requests trigger no downstream side effects.

api/.env.example

@@ -488,6 +488,11 @@ SENDGRID_API_KEY=
 # Sentry configuration
 SENTRY_DSN=
 
+# Cloudflare Turnstile server-side verification for Dify Cloud sign-in
+TURNSTILE_SECRET_KEY=
+# Comma-separated parent or exact hostnames, for example: dify.ai,staging.dify.dev
+TURNSTILE_ALLOWED_HOSTNAMES=
+
 # DEBUG
 DEBUG=false
 ENABLE_REQUEST_LOGGING=False

api/.importlinter

@@ -65,6 +65,34 @@ forbidden_modules =
     sqlalchemy
     werkzeug
 
+[importlinter:contract:init-validation-service-boundary]
+name = Initialization validation application service is framework and persistence neutral
+type = forbidden
+source_modules =
+    services.init_validation_service
+forbidden_modules =
+    controllers
+    extensions
+    flask
+    models
+    repositories
+    sqlalchemy
+    werkzeug
+
+[importlinter:contract:explore-banner-query-service-boundary]
+name = Explore banner query application service is framework and persistence neutral
+type = forbidden
+source_modules =
+    services.explore_banner_query_service
+forbidden_modules =
+    controllers
+    extensions
+    flask
+    models
+    repositories
+    sqlalchemy
+    werkzeug
+
 [importlinter:contract:feature-query-service-boundary]
 name = Feature query application service is framework and persistence neutral
 type = forbidden

api/Dockerfile

@@ -33,7 +33,6 @@ RUN uv sync --frozen --no-dev --no-editable
 FROM base AS production
 
 ENV FLASK_APP=app.py
-ENV EDITION=SELF_HOSTED
 ENV DEPLOY_ENV=PRODUCTION
 ENV CONSOLE_API_URL=http://127.0.0.1:5001
 ENV CONSOLE_WEB_URL=http://127.0.0.1:3000

api/app_factory.py

@@ -14,6 +14,7 @@ from contexts.wrapper import RecyclableContextVar
 from controllers.console.error import UnauthorizedAndForceLogout
 from core.logging.context import init_request_context
 from dify_app import DifyApp
+from enums import DeploymentEdition
 from extensions.ext_socketio import sio
 from services.enterprise.enterprise_service import EnterpriseService
 from services.entities.feature_entities import LicenseStatus
@@ -112,7 +113,7 @@ def create_flask_app_with_configs() -> DifyApp:
         init_request_context()
         RecyclableContextVar.increment_thread_recycles()
 
-        if dify_config.ENTERPRISE_ENABLED:
+        if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.ENTERPRISE:
             surface = _match_license_gated_surface(request.path)
             if surface is not None:
                 try:

api/commands/account.py

@@ -33,7 +33,7 @@ def reset_password(email, new_password, password_confirm):
 
     try:
         valid_password(new_password)
-    except:
+    except ValueError:
         click.echo(click.style(f"Invalid password. Must match {password_pattern}", fg="red"))
         return
 
@@ -75,7 +75,7 @@ def reset_email(email, new_email, email_confirm):
 
     try:
         email_validate(normalized_new_email)
-    except:
+    except ValueError:
         click.echo(click.style(f"Invalid email: {new_email}", fg="red"))
         return
 

api/commands/retention.py

@@ -10,6 +10,8 @@ import click
 import sqlalchemy as sa
 from sqlalchemy.orm import Session, sessionmaker
 
+from configs import dify_config
+from enums import CloudPlan, DeploymentEdition
 from extensions.ext_database import db
 from libs.datetime_utils import naive_utc_now
 from services.clear_free_plan_tenant_expired_logs import ClearFreePlanTenantExpiredLogs
@@ -126,14 +128,12 @@ def _get_archive_candidate_tenant_ids_by_prefix(
 
 
 def _filter_paid_workflow_archive_tenant_ids(tenant_ids: list[str]) -> tuple[list[str], list[str]]:
-    from configs import dify_config
-    from enums.cloud_plan import CloudPlan
     from services.billing_service import BillingService
 
     tenant_ids = sorted(set(tenant_ids))
     if not tenant_ids:
         return [], []
-    if not dify_config.BILLING_ENABLED:
+    if dify_config.DEPLOYMENT_EDITION != DeploymentEdition.CLOUD:
         return tenant_ids, []
 
     plans = BillingService.get_plan_bulk_with_cache(tenant_ids)
@@ -1462,7 +1462,7 @@ def cleanup_orphaned_draft_variables(
     "--graceful-period",
     default=21,
     show_default=True,
-    help="Graceful period in days after subscription expiration, will be ignored when billing is disabled.",
+    help="Graceful period in days after subscription expiration; ignored outside the Cloud edition.",
 )
 @click.option("--dry-run", is_flag=True, default=False, help="Show messages logs would be cleaned without deleting")
 def clean_expired_messages(
@@ -1514,8 +1514,8 @@ def clean_expired_messages(
                 if from_days_ago <= before_days:
                     raise click.UsageError("--from-days-ago must be greater than --before-days.")
 
-        # Create policy based on billing configuration
-        # NOTE: graceful_period will be ignored when billing is disabled.
+        # Create the policy for the configured deployment edition.
+        # NOTE: graceful_period is ignored outside the Cloud edition.
         policy = create_message_clean_policy(graceful_period_days=graceful_period)
 
         if from_days_ago is not None and before_days is not None:

api/commands/system.py

@@ -6,7 +6,7 @@ from sqlalchemy import delete, select, update
 from sqlalchemy.orm import sessionmaker
 
 from configs import dify_config
-from enums.deployment_edition import DeploymentEdition
+from enums import DeploymentEdition
 from events.app_event import app_was_created
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client

api/configs/app_config.py

@@ -5,7 +5,6 @@ from typing import Any, override
 from pydantic.fields import FieldInfo
 from pydantic_settings import BaseSettings, PydanticBaseSettingsSource, SettingsConfigDict, TomlConfigSettingsSource
 
-from enums.deployment_edition import DeploymentEdition
 from libs.file_utils import search_file_upwards
 
 from .deploy import DeploymentConfig
@@ -117,11 +116,3 @@ class DifyConfig(
                 ),
             ),
         )
-
-    @property
-    def DEPLOYMENT_EDITION(self) -> DeploymentEdition:
-        if self.EDITION == "CLOUD":
-            return DeploymentEdition.CLOUD
-        if self.ENTERPRISE_ENABLED:
-            return DeploymentEdition.ENTERPRISE
-        return DeploymentEdition.COMMUNITY

api/configs/deploy/__init__.py

@@ -1,8 +1,8 @@
-from typing import Literal
-
 from pydantic import Field
 from pydantic_settings import BaseSettings
 
+from enums import DeploymentEdition
+
 
 class DeploymentConfig(BaseSettings):
     """
@@ -25,9 +25,14 @@ class DeploymentConfig(BaseSettings):
         default=False,
     )
 
-    EDITION: Literal["SELF_HOSTED", "CLOUD"] = Field(
-        description="Deployment edition of the application (e.g., 'SELF_HOSTED', 'CLOUD')",
-        default="SELF_HOSTED",
+    DEPLOYMENT_EDITION: DeploymentEdition = Field(
+        description="Product edition of the application.",
+        default=DeploymentEdition.COMMUNITY,
+    )
+
+    INIT_PASSWORD: str = Field(
+        description="Password required before initializing a self-hosted deployment",
+        default="",
     )
 
     DEPLOY_ENV: str = Field(

api/configs/enterprise/__init__.py

@@ -8,12 +8,6 @@ class EnterpriseFeatureConfig(BaseSettings):
     **Before using, please contact business@dify.ai by email to inquire about licensing matters.**
     """
 
-    ENTERPRISE_ENABLED: bool = Field(
-        description="Enable or disable enterprise-level features."
-        "Before using, please contact business@dify.ai by email to inquire about licensing matters.",
-        default=False,
-    )
-
     WEBAPP_PUBLIC_ACCESS_ENABLED: bool = Field(
         description="Whether admins are allowed to set a webapp's access mode to public (anyone with the link, "
         "no auth). Disable in security-sensitive on-prem deployments.",
@@ -59,7 +53,7 @@ class EnterpriseTelemetryConfig(BaseSettings):
     """
 
     ENTERPRISE_TELEMETRY_ENABLED: bool = Field(
-        description="Enable enterprise telemetry collection (also requires ENTERPRISE_ENABLED=true).",
+        description="Enable enterprise telemetry collection for enterprise deployments.",
         default=False,
     )
 

api/configs/extra/__init__.py

@@ -3,6 +3,7 @@ from configs.extra.archive_config import ArchiveStorageConfig
 from configs.extra.knowledge_fs_config import KnowledgeFSConfig
 from configs.extra.notion_config import NotionConfig
 from configs.extra.sentry_config import SentryConfig
+from configs.extra.turnstile_config import TurnstileConfig
 
 
 class ExtraServiceConfig(
@@ -12,5 +13,6 @@ class ExtraServiceConfig(
     KnowledgeFSConfig,
     NotionConfig,
     SentryConfig,
+    TurnstileConfig,
 ):
     pass

api/configs/extra/turnstile_config.py

@@ -0,0 +1,34 @@
+from pydantic import Field, SecretStr, field_validator
+from pydantic_settings import BaseSettings
+
+
+class TurnstileConfig(BaseSettings):
+    """Server-side Cloudflare Turnstile settings for Cloud sign-in."""
+
+    TURNSTILE_SECRET_KEY: SecretStr | None = Field(
+        default=None,
+        description="Secret key used to validate Cloudflare Turnstile tokens.",
+    )
+    TURNSTILE_ALLOWED_HOSTNAMES: str = Field(
+        default="",
+        description="Comma-separated parent or exact hostnames accepted from Turnstile.",
+    )
+
+    @field_validator("TURNSTILE_SECRET_KEY", mode="before")
+    @classmethod
+    def normalize_secret_key(cls, value: object) -> object:
+        if isinstance(value, SecretStr):
+            normalized = value.get_secret_value().strip()
+            return SecretStr(normalized) if normalized else None
+        if isinstance(value, str):
+            normalized = value.strip()
+            return normalized or None
+        return value
+
+    @property
+    def TURNSTILE_ALLOWED_HOSTNAME_SET(self) -> frozenset[str]:
+        return frozenset(
+            hostname.strip().lower().strip(".")
+            for hostname in self.TURNSTILE_ALLOWED_HOSTNAMES.split(",")
+            if hostname.strip().strip(".")
+        )

api/configs/feature/__init__.py

@@ -810,17 +810,6 @@ class ModelLoadBalanceConfig(BaseSettings):
     )
 
 
-class BillingConfig(BaseSettings):
-    """
-    Configuration for platform billing features
-    """
-
-    BILLING_ENABLED: bool = Field(
-        description="Enable or disable billing functionality",
-        default=False,
-    )
-
-
 class UpdateConfig(BaseSettings):
     """
     Configuration for application update checks
@@ -1621,7 +1610,6 @@ class FeatureConfig(
     # place the configs in alphabet order
     AppExecutionConfig,
     AuthConfig,  # Changed from OAuthConfig to AuthConfig
-    BillingConfig,
     CodeExecutionSandboxConfig,
     CreatorsPlatformConfig,
     TriggerConfig,

api/configs/feature/hosted_service/__init__.py

@@ -451,6 +451,12 @@ class HostedFetchAppTemplateConfig(BaseSettings):
         default="https://tmpl.dify.ai",
     )
 
+    HOSTED_FETCH_APP_TEMPLATES_CACHE_TTL: int = Field(
+        description="TTL in seconds for caching remote app template HTTP responses. 0 disables caching.",
+        default=600,
+        ge=0,
+    )
+
 
 class HostedFetchPipelineTemplateConfig(BaseSettings):
     """

api/controllers/common/fields.py

@@ -126,12 +126,6 @@ class UsageCountResponse(ResponseModel):
     count: int
 
 
-class IndexInfoResponse(ResponseModel):
-    welcome: str
-    api_version: str
-    server_version: str
-
-
 class AvatarUrlResponse(ResponseModel):
     avatar_url: str
 

api/controllers/common/wraps.py

@@ -152,7 +152,7 @@ def _extract_resource_id(
     if resource_type == RBACResourceScope.APP:
         app_id = matched_args.get("app_id")
         if app_id:
-            return str(app_id)
+            return str(app_id)  # pyrefly: ignore[unnecessary-type-conversion]
 
         agent_id = matched_args.get("agent_id")
         if agent_id:
@@ -161,7 +161,7 @@ def _extract_resource_id(
 
         resource_id = matched_args.get("resource_id")
         if resource_id:
-            return str(resource_id)
+            return str(resource_id)  # pyrefly: ignore[unnecessary-type-conversion]
         raise ValueError("Missing app_id in request path")
 
     if resource_type == RBACResourceScope.DATASET:
@@ -171,9 +171,11 @@ def _extract_resource_id(
 
         pipeline_id = matched_args.get("pipeline_id")
         if pipeline_id:
-            dataset = db.session.scalar(select(Dataset).where(Dataset.pipeline_id == str(pipeline_id)))
+            dataset = db.session.scalar(
+                select(Dataset).where(Dataset.pipeline_id == str(pipeline_id), Dataset.tenant_id == tenant_id)
+            )
             if not dataset:
                 raise NotFound("Dataset not found for pipeline")
-            return str(dataset.id)
+            return str(dataset.id)  # pyrefly: ignore[unnecessary-type-conversion]
         raise ValueError("Missing dataset_id or pipeline_id in request path")
     raise ValueError(f"Unknown resource_type: {resource_type}")

api/controllers/console/__init__.py

@@ -41,10 +41,9 @@ from . import (
     knowledge_fs_proxy,
     notification,
     onboarding,
-    ping,
     setup,
     spec,
-    version,
+    system,
     workflow_run_archive,
 )
 from .agent import composer as agent_composer
@@ -213,7 +212,6 @@ __all__ = [
     "onboarding",
     "ops_trace",
     "parameter",
-    "ping",
     "plugin",
     "rag_pipeline",
     "rag_pipeline_datasets",
@@ -231,11 +229,11 @@ __all__ = [
     "socketio_workflow",
     "spec",
     "statistic",
+    "system",
     "tags",
     "tool_providers",
     "trial",
     "trigger_providers",
-    "version",
     "website",
     "workflow",
     "workflow_app_log",

api/controllers/console/agent/composer.py

@@ -284,7 +284,7 @@ class WorkflowAgentComposerSaveToRosterApi(Resource):
 
 def _require_snippet_app_id(*, session: Session, tenant_id: str, snippet_id: UUID) -> str:
     snippet = SnippetService(session=session).get_snippet_by_id(
-        snippet_id=str(snippet_id),
+        snippet_id=str(snippet_id),  # pyrefly: ignore[unnecessary-type-conversion]
         tenant_id=tenant_id,
     )
     if snippet is None:

api/controllers/console/agent/roster.py

@@ -3,7 +3,7 @@ from uuid import UUID
 from flask import abort, request
 from flask_restx import Resource
 from pydantic import AliasChoices, BaseModel, Field, field_validator
-from sqlalchemy import func, select
+from sqlalchemy import func, or_, select
 from sqlalchemy.orm import Session
 
 from controllers.common.schema import (
@@ -484,12 +484,13 @@ def _resolve_agent_runtime_app_model(session: Session, *, tenant_id: str, agent_
     return _agent_roster_service(session).get_agent_runtime_app_model(tenant_id=tenant_id, agent_id=str(agent_id))
 
 
-def _agent_api_key_count(session: Session, app_id: str) -> int:
+def _agent_api_key_count(session: Session, app_model: App) -> int:
     return (
         session.scalar(
             select(func.count(ApiToken.id)).where(
+                or_(ApiToken.tenant_id == app_model.tenant_id, ApiToken.tenant_id.is_(None)),
                 ApiToken.type == ApiTokenType.APP,
-                ApiToken.app_id == app_id,
+                ApiToken.app_id == app_model.id,
             )
         )
         or 0
@@ -521,7 +522,7 @@ def _serialize_agent_api_access(session: Session, app_model: App) -> dict:
         meta_endpoint=f"{base_url}/meta",
         api_rpm=app_model.api_rpm or 0,
         api_rph=app_model.api_rph or 0,
-        api_key_count=_agent_api_key_count(session, str(app_model.id)),
+        api_key_count=_agent_api_key_count(session, app_model),
     )
     return response.model_dump(mode="json")
 
@@ -940,6 +941,8 @@ class AgentApiKeyListApi(BaseApiKeyListResource):
     @console_ns.response(200, "Agent service API keys", console_ns.models[ApiKeyList.__name__])
     @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_tenant_id
+    @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
     @with_session(write=False)
     def get(self, session: Session, tenant_id: str, agent_id: UUID) -> dict[str, object]:
         app_model = _resolve_agent_app_model(session, tenant_id=tenant_id, agent_id=agent_id)
@@ -1069,7 +1072,7 @@ class AgentLogMessagesApi(Resource):
             payload = _agent_observability_service(session).list_log_messages(
                 app=app_model,
                 agent_id=str(agent_id),
-                conversation_id=str(conversation_id),
+                conversation_id=str(conversation_id),  # pyrefly: ignore[unnecessary-type-conversion]
                 params=AgentLogQueryParams(
                     page=query.page,
                     limit=query.limit,

api/controllers/console/apikey.py

@@ -5,7 +5,7 @@ import flask_restx
 from flask_restx import Resource
 from flask_restx._http import HTTPStatus
 from pydantic import field_validator
-from sqlalchemy import delete, func, select
+from sqlalchemy import func, or_, select
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden
 
@@ -89,7 +89,9 @@ class BaseApiKeyListResource(Resource):
         _get_resource(resource_id, current_tenant_id, self.resource_model, session=session)
         keys = session.scalars(
             select(ApiToken).where(
-                ApiToken.type == self.resource_type, getattr(ApiToken, self.resource_id_field) == resource_id
+                or_(ApiToken.tenant_id == current_tenant_id, ApiToken.tenant_id.is_(None)),
+                ApiToken.type == self.resource_type,
+                getattr(ApiToken, self.resource_id_field) == resource_id,
             )
         ).all()
         return ApiKeyList.model_validate({"data": keys}, from_attributes=True)
@@ -110,7 +112,9 @@ class BaseApiKeyListResource(Resource):
         current_key_count: int = (
             session.scalar(
                 select(func.count(ApiToken.id)).where(
-                    ApiToken.type == self.resource_type, getattr(ApiToken, self.resource_id_field) == resource_id
+                    or_(ApiToken.tenant_id == current_tenant_id, ApiToken.tenant_id.is_(None)),
+                    ApiToken.type == self.resource_type,
+                    getattr(ApiToken, self.resource_id_field) == resource_id,
                 )
             )
             or 0
@@ -172,6 +176,7 @@ class BaseApiKeyResource(Resource):
         key = session.scalar(
             select(ApiToken)
             .where(
+                or_(ApiToken.tenant_id == current_tenant_id, ApiToken.tenant_id.is_(None)),
                 getattr(ApiToken, self.resource_id_field) == resource_id,
                 ApiToken.type == self.resource_type,
                 ApiToken.id == api_key_id,
@@ -187,7 +192,7 @@ class BaseApiKeyResource(Resource):
         assert key is not None  # nosec - for type checker only
         ApiTokenCache.delete(key.token, key.type)
 
-        session.execute(delete(ApiToken).where(ApiToken.id == api_key_id))
+        session.delete(key)
         session.commit()
 
 
@@ -198,13 +203,15 @@ class AppApiKeyListResource(BaseApiKeyListResource):
     @console_ns.doc(params={"resource_id": "App ID"})
     @console_ns.response(200, "API keys retrieved successfully", console_ns.models[ApiKeyList.__name__])
     @with_current_tenant_id
+    @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
     @agent_manage_required_for_agent_app
     @with_session(write=False)
     def get(self, session: Session, current_tenant_id: str, resource_id: UUID) -> dict[str, object]:
         """Get all API keys for an app"""
         return dump_response(
             ApiKeyList,
-            self._get_api_key_list(str(resource_id), current_tenant_id, session=session),
+            self._get_api_key_list(str(resource_id), current_tenant_id, session=session),  # pyrefly: ignore[unnecessary-type-conversion]
         )
 
     @console_ns.doc("create_app_api_key")
@@ -221,7 +228,7 @@ class AppApiKeyListResource(BaseApiKeyListResource):
         """Create a new API key for an app"""
         return dump_response(
             ApiKeyItem,
-            self._create_api_key(str(resource_id), current_tenant_id, session=session),
+            self._create_api_key(str(resource_id), current_tenant_id, session=session),  # pyrefly: ignore[unnecessary-type-conversion]
         ), 201
 
     resource_type = ApiTokenType.APP
@@ -251,7 +258,7 @@ class AppApiKeyResource(BaseApiKeyResource):
     ) -> tuple[str, int]:
         """Delete an API key for an app"""
         self._delete_api_key(
-            str(resource_id),
+            str(resource_id),  # pyrefly: ignore[unnecessary-type-conversion]
             str(api_key_id),
             current_tenant_id,
             current_user,
@@ -276,7 +283,7 @@ class DatasetApiKeyListResource(BaseApiKeyListResource):
         """Get all API keys for a dataset"""
         return dump_response(
             ApiKeyList,
-            self._get_api_key_list(str(resource_id), current_tenant_id, session=session),
+            self._get_api_key_list(str(resource_id), current_tenant_id, session=session),  # pyrefly: ignore[unnecessary-type-conversion]
         )
 
     @console_ns.doc("create_dataset_api_key")
@@ -292,7 +299,7 @@ class DatasetApiKeyListResource(BaseApiKeyListResource):
         """Create a new API key for a dataset"""
         return dump_response(
             ApiKeyItem,
-            self._create_api_key(str(resource_id), current_tenant_id, session=session),
+            self._create_api_key(str(resource_id), current_tenant_id, session=session),  # pyrefly: ignore[unnecessary-type-conversion]
         ), 201
 
     resource_type = ApiTokenType.DATASET
@@ -321,7 +328,7 @@ class DatasetApiKeyResource(BaseApiKeyResource):
     ) -> tuple[str, int]:
         """Delete an API key for a dataset"""
         self._delete_api_key(
-            str(resource_id),
+            str(resource_id),  # pyrefly: ignore[unnecessary-type-conversion]
             str(api_key_id),
             current_tenant_id,
             current_user,

api/controllers/console/app/workflow.py

@@ -315,6 +315,9 @@ class WorkflowResponse(ResponseModel):
     )
     hash: str = Field(validation_alias=AliasChoices("unique_hash", "hash"))
     version: str
+    # NULL for drafts and for versions published before numbering was introduced; those
+    # render as "Untitled Version" instead of `#N`. Never 0, so clients must test for null.
+    version_number: int | None = None
     marked_name: str
     marked_comment: str
     created_by: SimpleAccount | None = Field(

api/controllers/console/auth/activate.py

@@ -10,6 +10,7 @@ from controllers.console import console_ns
 from controllers.console.auth.error import InvitationAccountMismatchError
 from controllers.console.error import AccountInFreezeError, AlreadyActivateError
 from controllers.console.wraps import model_validate
+from enums import DeploymentEdition
 from extensions.ext_database import db
 from libs.datetime_utils import naive_utc_now
 from libs.helper import EmailStr, timezone
@@ -161,7 +162,9 @@ class ActivateApi(Resource):
             if current_account.id != account.id:
                 raise InvitationAccountMismatchError()
 
-        if dify_config.BILLING_ENABLED and BillingService.is_email_in_freeze(account.email):
+        if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD and BillingService.is_email_in_freeze(
+            account.email
+        ):
             raise AccountInFreezeError()
 
         tenant = invitation["tenant"]

api/controllers/console/auth/email_register.py

@@ -15,6 +15,7 @@ from controllers.console.auth.error import (
     InvalidTokenError,
     PasswordMismatchError,
 )
+from enums import DeploymentEdition
 from extensions.ext_database import db
 from fields.base import ResponseModel
 from libs.helper import EmailStr, extract_remote_ip
@@ -98,7 +99,9 @@ class EmailRegisterSendEmailApi(Resource):
         if req_data.language is not None and req_data.language in languages:
             language = req_data.language
 
-        if dify_config.BILLING_ENABLED and BillingService.is_email_in_freeze(normalized_email):
+        if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD and BillingService.is_email_in_freeze(
+            normalized_email
+        ):
             raise AccountInFreezeError()
 
         account = AccountService.get_account_by_email_with_case_fallback(req_data.email, session=db.session())

api/controllers/console/auth/error.py

@@ -95,6 +95,18 @@ class EmailPasswordLoginLimitError(BaseHTTPException):
     code = 429
 
 
+class TurnstileVerificationFailedError(BaseHTTPException):
+    error_code = "turnstile_verification_failed"
+    description = "Turnstile verification failed. Please try again."
+    code = 400
+
+
+class TurnstileServiceUnavailableError(BaseHTTPException):
+    error_code = "turnstile_service_unavailable"
+    description = "Turnstile verification is temporarily unavailable. Please try again later."
+    code = 503
+
+
 class EmailCodeLoginRateLimitExceededError(BaseHTTPException):
     error_code = "email_code_login_rate_limit_exceeded"
     description = "Too many login emails have been sent. Please try again in {minutes} minutes."

api/controllers/console/auth/login.py

@@ -25,6 +25,8 @@ from controllers.console.auth.error import (
     EmailPasswordLoginLimitError,
     InvalidEmailError,
     InvalidTokenError,
+    TurnstileServiceUnavailableError,
+    TurnstileVerificationFailedError,
 )
 from controllers.console.error import (
     AccountBannedError,
@@ -43,6 +45,7 @@ from controllers.console.wraps import (
     setup_required,
     with_current_user,
 )
+from enums import DeploymentEdition
 from extensions.ext_database import db
 from libs.helper import EmailStr, extract_remote_ip
 from libs.helper import timezone as validate_timezone_string
@@ -67,6 +70,11 @@ from services.errors.account import (
 )
 from services.errors.workspace import WorkSpaceNotAllowedCreateError, WorkspacesLimitExceededError
 from services.feature_service import FeatureService
+from services.turnstile_service import (
+    TurnstileChallengeRejectedError,
+    TurnstileService,
+    TurnstileUpstreamError,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -81,6 +89,13 @@ class EmailPayload(BaseModel):
     language: str | None = Field(default=None)
 
 
+class EmailCodeSendPayload(EmailPayload):
+    turnstile_token: str | None = Field(
+        default=None,
+        description="Cloudflare Turnstile token. Required at runtime for Dify Cloud.",
+    )
+
+
 class EmailCodeLoginPayload(BaseModel):
     email: EmailStr = Field(...)
     code: str = Field(...)
@@ -96,7 +111,7 @@ class EmailCodeLoginPayload(BaseModel):
         return validate_timezone_string(value)
 
 
-register_schema_models(console_ns, LoginPayload, EmailPayload, EmailCodeLoginPayload)
+register_schema_models(console_ns, LoginPayload, EmailPayload, EmailCodeSendPayload, EmailCodeLoginPayload)
 register_response_schema_models(
     console_ns,
     SimpleResultDataResponse,
@@ -121,7 +136,9 @@ class LoginApi(Resource):
         request_email = req_data.email
         normalized_email = request_email.lower()
 
-        if dify_config.BILLING_ENABLED and BillingService.is_email_in_freeze(normalized_email):
+        if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD and BillingService.is_email_in_freeze(
+            normalized_email
+        ):
             _log_console_login_failure(email=normalized_email, reason=LoginFailureReason.ACCOUNT_IN_FREEZE)
             raise AccountInFreezeError()
 
@@ -160,7 +177,6 @@ class LoginApi(Resource):
             AccountService.add_login_error_rate_limit(normalized_email)
             _log_console_login_failure(email=normalized_email, reason=LoginFailureReason.INVALID_CREDENTIALS)
             raise AuthenticationFailedError() from exc
-        # SELF_HOSTED only have one workspace
         tenants = TenantService.get_join_tenants(account, session=db.session())
         if len(tenants) == 0:
             if (
@@ -242,16 +258,26 @@ class ResetPasswordSendEmailApi(Resource):
 @console_ns.route("/email-code-login")
 class EmailCodeLoginSendEmailApi(Resource):
     @setup_required
-    @console_ns.expect(console_ns.models[EmailPayload.__name__])
+    @console_ns.expect(console_ns.models[EmailCodeSendPayload.__name__])
     @console_ns.response(200, "Success", console_ns.models[SimpleResultDataResponse.__name__])
-    @model_validate(EmailPayload)
-    def post(self, req_data: EmailPayload):
+    @model_validate(EmailCodeSendPayload)
+    def post(self, req_data: EmailCodeSendPayload):
         normalized_email = req_data.email.lower()
 
         ip_address = extract_remote_ip(request)
         if AccountService.is_email_send_ip_limit(ip_address):
             raise EmailSendIpLimitError()
 
+        if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD:
+            try:
+                TurnstileService.verify(token=req_data.turnstile_token, remote_ip=ip_address)
+            except TurnstileChallengeRejectedError as exc:
+                logger.info("Turnstile rejected an email-code login challenge")
+                raise TurnstileVerificationFailedError() from exc
+            except TurnstileUpstreamError as exc:
+                logger.warning("Turnstile verification is unavailable", exc_info=True)
+                raise TurnstileServiceUnavailableError() from exc
+
         if req_data.language is not None and req_data.language == "zh-Hans":
             language = "zh-Hans"
         else:

api/controllers/console/auth/oauth.py

@@ -12,6 +12,7 @@ from configs import dify_config
 from constants.languages import languages
 from controllers.common.fields import RedirectResponse
 from controllers.common.schema import query_params_from_model, register_response_schema_model, register_schema_models
+from enums import DeploymentEdition
 from extensions.ext_database import db
 from libs.datetime_utils import naive_utc_now
 from libs.helper import extract_remote_ip
@@ -301,7 +302,9 @@ def _generate_account(
         normalized_email = user_info.email.lower()
         oauth_new_user = True
         if not FeatureService.get_system_features().is_allow_register:
-            if dify_config.BILLING_ENABLED and BillingService.is_email_in_freeze(normalized_email):
+            if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD and BillingService.is_email_in_freeze(
+                normalized_email
+            ):
                 raise AccountRegisterError(
                     description=(
                         "This email account has been deleted within the past "

api/controllers/console/billing/billing.py

@@ -15,7 +15,7 @@ from controllers.console.wraps import (
     with_current_tenant_id,
     with_current_user,
 )
-from enums.cloud_plan import CloudPlan
+from enums import CloudPlan
 from extensions.ext_database import db
 from fields.base import ResponseModel
 from libs.login import login_required
@@ -40,14 +40,18 @@ class BillingInvoiceResponse(ResponseModel):
     url: str
 
 
+class BillingSubscriptionResponse(ResponseModel):
+    url: str
+
+
 register_schema_models(console_ns, SubscriptionQuery, PartnerTenantsPayload)
-register_response_schema_models(console_ns, BillingResponse, BillingInvoiceResponse)
+register_response_schema_models(console_ns, BillingResponse, BillingInvoiceResponse, BillingSubscriptionResponse)
 
 
 @console_ns.route("/billing/subscription")
 class Subscription(Resource):
     @console_ns.doc(params=query_params_from_model(SubscriptionQuery))
-    @console_ns.response(200, "Success", console_ns.models[BillingResponse.__name__])
+    @console_ns.response(200, "Success", console_ns.models[BillingSubscriptionResponse.__name__])
     @setup_required
     @login_required
     @account_initialization_required

api/controllers/console/datasets/datasets.py

@@ -53,6 +53,7 @@ from models.enums import ApiTokenType, SegmentStatus
 from models.provider_ids import ModelProviderID
 from services.api_token_service import ApiTokenCache
 from services.app_service import AppService
+from services.dataset_ref_service import DatasetRefService
 from services.dataset_service import DatasetPermissionService, DatasetService, DocumentService
 from services.enterprise import rbac_service as enterprise_rbac_service
 from services.enterprise.rbac_service import RBACResourceWhitelistScope, ReplaceMemberBindings
@@ -67,6 +68,18 @@ def _has_dataset_list_permission(permission_keys: list[str]) -> bool:
     return any(permission_key in DATASET_LIST_PERMISSION_KEYS for permission_key in permission_keys)
 
 
+def _get_accessible_dataset(dataset_id: UUID, tenant_id: str, current_user: Account, session: Session) -> Dataset:
+    dataset = DatasetService.get_dataset_for_tenant(str(dataset_id), tenant_id, session=session)
+    if dataset is None:
+        raise NotFound("Dataset not found.")
+    if not dify_config.RBAC_ENABLED:
+        try:
+            DatasetService.check_dataset_permission(dataset, current_user, session)
+        except services.errors.account.NoPermissionError as e:
+            raise Forbidden(str(e))
+    return dataset
+
+
 def _validate_indexing_technique(value: str | None) -> str | None:
     if value is None:
         return value
@@ -811,12 +824,13 @@ class DatasetUseCheckApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @with_current_user
+    @with_current_tenant_id
     @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_READONLY)
     @with_session(write=False)
-    def get(self, session: Session, dataset_id: UUID):
-        dataset_id_str = str(dataset_id)
-
-        dataset_is_using = DatasetService.dataset_use_check(dataset_id_str, session)
+    def get(self, session: Session, current_tenant_id: str, current_user: Account, dataset_id: UUID):
+        dataset = _get_accessible_dataset(dataset_id, current_tenant_id, current_user, session)
+        dataset_is_using = DatasetService.dataset_use_check(DatasetRefService.create_dataset_ref(dataset), session)
         return UsageCheckResponse(is_using=dataset_is_using).model_dump(mode="json"), 200
 
 
@@ -1027,13 +1041,15 @@ class DatasetIndexingStatusApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @with_current_user
     @with_current_tenant_id
     @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_READONLY)
     @with_session(write=False)
-    def get(self, session: Session, current_tenant_id: str, dataset_id: UUID):
-        dataset_id_str = str(dataset_id)
+    def get(self, session: Session, current_tenant_id: str, current_user: Account, dataset_id: UUID):
+        dataset = _get_accessible_dataset(dataset_id, current_tenant_id, current_user, session)
+
         documents = session.scalars(
-            select(Document).where(Document.dataset_id == dataset_id_str, Document.tenant_id == current_tenant_id)
+            select(Document).where(Document.dataset_id == dataset.id, Document.tenant_id == dataset.tenant_id)
         ).all()
         documents_status = []
         for document in documents:
@@ -1041,6 +1057,8 @@ class DatasetIndexingStatusApi(Resource):
                 session.scalar(
                     select(func.count(DocumentSegment.id)).where(
                         DocumentSegment.completed_at.isnot(None),
+                        DocumentSegment.tenant_id == dataset.tenant_id,
+                        DocumentSegment.dataset_id == dataset.id,
                         DocumentSegment.document_id == str(document.id),
                         DocumentSegment.status != SegmentStatus.RE_SEGMENT,
                     )
@@ -1050,6 +1068,8 @@ class DatasetIndexingStatusApi(Resource):
             total_segments = (
                 session.scalar(
                     select(func.count(DocumentSegment.id)).where(
+                        DocumentSegment.tenant_id == dataset.tenant_id,
+                        DocumentSegment.dataset_id == dataset.id,
                         DocumentSegment.document_id == str(document.id),
                         DocumentSegment.status != SegmentStatus.RE_SEGMENT,
                     )
@@ -1086,6 +1106,8 @@ class DatasetApiKeyApi(Resource):
     @console_ns.response(200, "API keys retrieved successfully", console_ns.models[ApiKeyList.__name__])
     @setup_required
     @login_required
+    @is_admin_or_owner_required
+    @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_API_KEY_MANAGE, resource_required=False)
     @account_initialization_required
     @with_current_tenant_id
     @with_session(write=False)
@@ -1177,12 +1199,16 @@ class DatasetEnableApiApi(Resource):
     @login_required
     @account_initialization_required
     @console_ns.response(200, "Success", console_ns.models[SimpleResultResponse.__name__])
+    @with_current_user
+    @with_current_tenant_id
     @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_EDIT)
     @with_session
-    def post(self, session: Session, dataset_id: UUID, status: str):
-        dataset_id_str = str(dataset_id)
+    def post(self, session: Session, current_tenant_id: str, current_user: Account, dataset_id: UUID, status: str):
+        dataset = _get_accessible_dataset(dataset_id, current_tenant_id, current_user, session)
+        if not current_user.is_dataset_editor:
+            raise Forbidden()
 
-        DatasetService.update_dataset_api_status(dataset_id_str, status == "enable", session)
+        DatasetService.update_dataset_api_status(dataset, status == "enable", current_user, session)
 
         return SimpleResultResponse(result="success").model_dump(mode="json"), 200
 
@@ -1248,14 +1274,15 @@ class DatasetErrorDocs(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @with_current_user
+    @with_current_tenant_id
     @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_READONLY)
     @with_session(write=False)
-    def get(self, session: Session, dataset_id: UUID):
-        dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, session)
-        if dataset is None:
-            raise NotFound("Dataset not found.")
-        results = DocumentService.get_error_documents_by_dataset_id(dataset_id_str, session)
+    def get(self, session: Session, current_tenant_id: str, current_user: Account, dataset_id: UUID):
+        dataset = _get_accessible_dataset(dataset_id, current_tenant_id, current_user, session)
+        results = DocumentService.get_error_documents_by_dataset_ref(
+            DatasetRefService.create_dataset_ref(dataset), session
+        )
 
         return dump_response(ErrorDocsResponse, {"data": results, "total": len(results)}), 200
 
@@ -1307,12 +1334,13 @@ class DatasetAutoDisableLogApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @with_current_user
+    @with_current_tenant_id
     @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_READONLY)
     @with_session(write=False)
-    def get(self, session: Session, dataset_id: UUID):
-        dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, session)
-        if dataset is None:
-            raise NotFound("Dataset not found.")
-        auto_disable_logs = DatasetService.get_dataset_auto_disable_logs(dataset_id_str, session)
+    def get(self, session: Session, current_tenant_id: str, current_user: Account, dataset_id: UUID):
+        dataset = _get_accessible_dataset(dataset_id, current_tenant_id, current_user, session)
+        auto_disable_logs = DatasetService.get_dataset_auto_disable_logs(
+            DatasetRefService.create_dataset_ref(dataset), session
+        )
         return dump_response(AutoDisableLogsResponse, auto_disable_logs), 200

api/controllers/console/datasets/datasets_document.py

@@ -16,6 +16,7 @@ from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden, NotFound
 
 import services
+from configs import dify_config
 from controllers.common.controller_schemas import DocumentBatchDownloadZipPayload
 from controllers.common.fields import SimpleResultMessageResponse, SimpleResultResponse, UrlResponse
 from controllers.common.schema import register_response_schema_models, register_schema_models
@@ -54,14 +55,17 @@ from libs.helper import dump_response, to_timestamp
 from libs.login import login_required
 from libs.pagination import paginate_query
 from models import Account, Document, DocumentSegment, UploadFile
-from models.dataset import DocumentPipelineExecutionLog
+from models.dataset import DatasetPermissionEnum, DocumentPipelineExecutionLog
 from models.enums import IndexingStatus, ProcessRuleMode, SegmentStatus
 from services.dataset_ref_service import DatasetRefService
 from services.dataset_service import DatasetService, DocumentService
+from services.enterprise import rbac_service as enterprise_rbac_service
+from services.enterprise.rbac_service import RBACResourceWhitelistScope, ReplaceMemberBindings
 from services.entities.knowledge_entities.knowledge_entities import KnowledgeConfig, ProcessRule, RetrievalModel
 from services.file_service import FileService
 from services.vector_space_admission_service import get_vector_space_admission_error_fields
 from tasks.generate_summary_index_task import generate_summary_index_task
+from tasks.initialize_created_app_rbac_access_task import initialize_created_app_rbac_access_task
 
 from ..app.error import (
     ProviderModelCurrentlyNotSupportError,
@@ -78,6 +82,7 @@ from ..datasets.error import (
 )
 from ..wraps import (
     account_initialization_required,
+    check_knowledge_rate_limit,
     cloud_edition_billing_rate_limit_check,
     cloud_edition_billing_resource_check,
     setup_required,
@@ -294,23 +299,23 @@ class DocumentResource(Resource):
     def get_document(
         self, session: Session, dataset_id: str, document_id: str, current_user: Account, current_tenant_id: str
     ) -> Document:
-        dataset = DatasetService.get_dataset(dataset_id, session)
+        dataset = DatasetService.get_dataset_for_tenant(dataset_id, current_tenant_id, session=session)
         if not dataset:
             raise NotFound("Dataset not found.")
 
-        try:
-            DatasetService.check_dataset_permission(dataset, current_user, session)
-        except services.errors.account.NoPermissionError as e:
-            raise Forbidden(str(e))
+        if not dify_config.RBAC_ENABLED:
+            try:
+                DatasetService.check_dataset_permission(dataset, current_user, session)
+            except services.errors.account.NoPermissionError as e:
+                raise Forbidden(str(e))
 
-        document = DocumentService.get_document(dataset_id, document_id, session=session)
+        dataset_ref = DatasetRefService.create_dataset_ref(dataset)
+        document_ref = DatasetRefService.create_document_ref_from_id(dataset_ref, document_id)
+        document = DatasetRefService.get_document_by_ref(document_ref, session=session)
 
         if not document:
             raise NotFound("Document not found.")
 
-        if document.tenant_id != current_tenant_id:
-            raise Forbidden("No permission.")
-
         return document
 
     def get_batch_documents(
@@ -578,18 +583,33 @@ class DatasetDocumentListApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @cloud_edition_billing_rate_limit_check("knowledge")
     @console_ns.response(204, "Documents deleted successfully")
+    @with_current_user
+    @with_current_tenant_id
     @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_EDIT)
     @with_session
-    def delete(self, session: Session, dataset_id: UUID):
+    def delete(
+        self,
+        session: Session,
+        current_tenant_id: str,
+        current_user: Account,
+        dataset_id: UUID,
+    ):
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, session)
+        dataset = DatasetService.get_dataset_for_tenant(dataset_id_str, current_tenant_id, session=session)
         if dataset is None:
             raise NotFound("Dataset not found.")
-        # check user's model setting
-        DatasetService.check_dataset_model_setting(dataset)
 
+        if not current_user.is_dataset_editor:
+            raise Forbidden()
+
+        if not dify_config.RBAC_ENABLED:
+            try:
+                DatasetService.check_dataset_permission(dataset, current_user, session)
+            except services.errors.account.NoPermissionError as e:
+                raise Forbidden(str(e))
+
+        check_knowledge_rate_limit()
         try:
             document_ids = request.args.getlist("document_id")
             dataset_ref = DatasetRefService.create_dataset_ref(dataset)
@@ -662,6 +682,21 @@ class DatasetInitApi(Resource):
         except ModelCurrentlyNotSupportError:
             raise ProviderModelCurrentlyNotSupportError()
 
+        if dify_config.RBAC_ENABLED:
+            dataset.permission = DatasetPermissionEnum.ALL_TEAM
+        else:
+            dataset.permission = DatasetPermissionEnum.ONLY_ME
+        session.flush()
+
+        if dify_config.RBAC_ENABLED:
+            enterprise_rbac_service.RBACService.DatasetAccess.replace_whitelist(
+                current_tenant_id,
+                current_user.id,
+                dataset.id,
+                ReplaceMemberBindings(scope=RBACResourceWhitelistScope.ALL),
+            )
+            initialize_created_app_rbac_access_task.delay(current_tenant_id, current_user.id, dataset_id=dataset.id)
+
         return dump_response(
             DatasetAndDocumentResponse,
             {"dataset": dataset, "documents": document_responses(documents, session=session), "batch": batch},
@@ -1367,29 +1402,32 @@ class DocumentPauseApi(DocumentResource):
     @setup_required
     @login_required
     @account_initialization_required
-    @cloud_edition_billing_rate_limit_check("knowledge")
     @console_ns.response(204, "Document paused successfully")
+    @with_current_user
+    @with_current_tenant_id
     @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_EDIT)
     @with_session
-    def patch(self, session: Session, dataset_id: UUID, document_id: UUID):
+    def patch(
+        self,
+        session: Session,
+        current_tenant_id: str,
+        current_user: Account,
+        dataset_id: UUID,
+        document_id: UUID,
+    ):
         """pause document."""
         dataset_id_str = str(dataset_id)
         document_id_str = str(document_id)
 
-        dataset = DatasetService.get_dataset(dataset_id_str, session)
-        if not dataset:
-            raise NotFound("Dataset not found.")
-
-        document = DocumentService.get_document(dataset.id, document_id_str, session=session)
-
-        # 404 if document not found
-        if document is None:
-            raise NotFound("Document Not Exists.")
+        document = self.get_document(session, dataset_id_str, document_id_str, current_user, current_tenant_id)
+        if not current_user.is_dataset_editor:
+            raise Forbidden()
 
         # 403 if document is archived
         if DocumentService.check_archived(document):
             raise ArchivedDocumentImmutableError()
 
+        check_knowledge_rate_limit()
         try:
             # pause document
             DocumentService.pause_document(document, session)
@@ -1404,26 +1442,31 @@ class DocumentRecoverApi(DocumentResource):
     @setup_required
     @login_required
     @account_initialization_required
-    @cloud_edition_billing_rate_limit_check("knowledge")
     @console_ns.response(204, "Document resumed successfully")
+    @with_current_user
+    @with_current_tenant_id
     @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_EDIT)
     @with_session
-    def patch(self, session: Session, dataset_id: UUID, document_id: UUID):
+    def patch(
+        self,
+        session: Session,
+        current_tenant_id: str,
+        current_user: Account,
+        dataset_id: UUID,
+        document_id: UUID,
+    ):
         """recover document."""
         dataset_id_str = str(dataset_id)
         document_id_str = str(document_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, session)
-        if not dataset:
-            raise NotFound("Dataset not found.")
-        document = DocumentService.get_document(dataset.id, document_id_str, session=session)
 
-        # 404 if document not found
-        if document is None:
-            raise NotFound("Document Not Exists.")
+        document = self.get_document(session, dataset_id_str, document_id_str, current_user, current_tenant_id)
+        if not current_user.is_dataset_editor:
+            raise Forbidden()
 
         # 403 if document is archived
         if DocumentService.check_archived(document):
             raise ArchivedDocumentImmutableError()
+        check_knowledge_rate_limit()
         try:
             # pause document
             DocumentService.recover_document(document, session)
@@ -1438,21 +1481,40 @@ class DocumentRetryApi(DocumentResource):
     @setup_required
     @login_required
     @account_initialization_required
-    @cloud_edition_billing_rate_limit_check("knowledge")
     @console_ns.expect(console_ns.models[DocumentRetryPayload.__name__])
     @console_ns.response(204, "Documents retry started successfully")
+    @with_current_user
+    @with_current_tenant_id
     @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_EDIT)
     @with_session
-    def post(self, session: Session, dataset_id: UUID):
+    def post(
+        self,
+        session: Session,
+        current_tenant_id: str,
+        current_user: Account,
+        dataset_id: UUID,
+    ):
         """retry document."""
-        payload = DocumentRetryPayload.model_validate(console_ns.payload or {})
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, session)
-        retry_documents = []
+        dataset = DatasetService.get_dataset_for_tenant(dataset_id_str, current_tenant_id, session=session)
         if not dataset:
             raise NotFound("Dataset not found.")
-        documents = DocumentService.get_documents_by_ids(dataset.id, payload.document_ids, session)
+
+        if not current_user.is_dataset_editor:
+            raise Forbidden()
+
+        if not dify_config.RBAC_ENABLED:
+            try:
+                DatasetService.check_dataset_permission(dataset, current_user, session)
+            except services.errors.account.NoPermissionError as e:
+                raise Forbidden(str(e))
+
+        payload = DocumentRetryPayload.model_validate(console_ns.payload or {})
+        documents = DocumentService.get_documents_by_ids(
+            DatasetRefService.create_dataset_ref(dataset), payload.document_ids, session
+        )
         documents_by_id = {document.id: document for document in documents}
+        retry_documents = []
         for document_id in payload.document_ids:
             try:
                 document = documents_by_id.get(document_id)
@@ -1473,6 +1535,7 @@ class DocumentRetryApi(DocumentResource):
                 logger.exception("Failed to retry document, document id: %s", document_id)
                 continue
         # retry document
+        check_knowledge_rate_limit()
         DocumentService.retry_document(dataset_id_str, retry_documents, session)
 
         return "", 204
@@ -1512,28 +1575,34 @@ class WebsiteDocumentSyncApi(DocumentResource):
     @login_required
     @account_initialization_required
     @console_ns.response(200, "Success", console_ns.models[SimpleResultResponse.__name__])
+    @with_current_user
     @with_current_tenant_id
-    @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_CREATE_AND_MANAGEMENT)
+    @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_EDIT)
     @with_session
-    def get(self, session: Session, current_tenant_id: str, dataset_id: UUID, document_id: UUID):
+    def get(
+        self,
+        session: Session,
+        current_tenant_id: str,
+        current_user: Account,
+        dataset_id: UUID,
+        document_id: UUID,
+    ):
         """sync website document."""
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, session)
+        dataset = DatasetService.get_dataset_for_tenant(dataset_id_str, current_tenant_id, session=session)
         if not dataset:
             raise NotFound("Dataset not found.")
+        if not current_user.is_dataset_editor:
+            raise Forbidden()
         document_id_str = str(document_id)
-        document = DocumentService.get_document(dataset.id, document_id_str, session=session)
-        if not document:
-            raise NotFound("Document not found.")
-        if document.tenant_id != current_tenant_id:
-            raise Forbidden("No permission.")
+        document = self.get_document(session, dataset.id, document_id_str, current_user, current_tenant_id)
         if document.data_source_type != "website_crawl":
             raise ValueError("Document is not a website document.")
         # 403 if document is archived
         if DocumentService.check_archived(document):
             raise ArchivedDocumentImmutableError()
         # sync document
-        DocumentService.sync_website_document(dataset_id_str, document, session)
+        DocumentService.sync_website_document(dataset, document, session)
 
         return SimpleResultResponse(result="success").model_dump(mode="json"), 200
 
@@ -1548,21 +1617,25 @@ class DocumentPipelineExecutionLogApi(DocumentResource):
     @setup_required
     @login_required
     @account_initialization_required
+    @with_current_user
+    @with_current_tenant_id
     @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_CREATE_AND_MANAGEMENT)
     @with_session(write=False)
-    def get(self, session: Session, dataset_id: UUID, document_id: UUID):
+    def get(
+        self,
+        session: Session,
+        current_tenant_id: str,
+        current_user: Account,
+        dataset_id: UUID,
+        document_id: UUID,
+    ):
         dataset_id_str = str(dataset_id)
         document_id_str = str(document_id)
 
-        dataset = DatasetService.get_dataset(dataset_id_str, session)
-        if not dataset:
-            raise NotFound("Dataset not found.")
-        document = DocumentService.get_document(dataset.id, document_id_str, session=session)
-        if not document:
-            raise NotFound("Document not found.")
+        document = self.get_document(session, dataset_id_str, document_id_str, current_user, current_tenant_id)
         log = session.scalar(
             select(DocumentPipelineExecutionLog)
-            .where(DocumentPipelineExecutionLog.document_id == document_id_str)
+            .where(DocumentPipelineExecutionLog.document_id == document.id)
             .order_by(DocumentPipelineExecutionLog.created_at.desc())
             .limit(1)
         )
@@ -1645,7 +1718,9 @@ class DocumentGenerateSummaryApi(Resource):
             raise ValueError("Summary index is not enabled for this dataset. Please enable it in the dataset settings.")
 
         # Verify all documents exist and belong to the dataset
-        documents = DocumentService.get_documents_by_ids(dataset_id_str, document_list, session)
+        documents = DocumentService.get_documents_by_ids(
+            DatasetRefService.create_dataset_ref(dataset), document_list, session
+        )
 
         if len(documents) != len(document_list):
             found_ids = {doc.id for doc in documents}

api/controllers/console/datasets/metadata.py

@@ -3,8 +3,10 @@ from uuid import UUID
 
 from flask_restx import Resource
 from sqlalchemy.orm import Session
-from werkzeug.exceptions import NotFound
+from werkzeug.exceptions import Forbidden, NotFound
 
+import services
+from configs import dify_config
 from controllers.common.controller_schemas import MetadataUpdatePayload
 from controllers.common.schema import register_response_schema_models, register_schema_models
 from controllers.common.session import with_session
@@ -35,6 +37,7 @@ from services.entities.knowledge_entities.knowledge_entities import (
     MetadataDetail,
     MetadataOperationData,
 )
+from services.errors.metadata import MetadataResourceNotFoundError
 from services.metadata_service import MetadataService
 
 register_schema_models(
@@ -87,13 +90,20 @@ class DatasetMetadataCreateApi(Resource):
     @console_ns.response(
         200, "Metadata retrieved successfully", console_ns.models[DatasetMetadataListResponse.__name__]
     )
+    @with_current_user
+    @with_current_tenant_id
     @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_CREATE_AND_MANAGEMENT)
     @with_session(write=False)
-    def get(self, session: Session, dataset_id: UUID):
+    def get(self, session: Session, current_tenant_id: str, current_user: Account, dataset_id: UUID):
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, session)
+        dataset = DatasetService.get_dataset_for_tenant(dataset_id_str, current_tenant_id, session=session)
         if dataset is None:
             raise NotFound("Dataset not found.")
+        if not dify_config.RBAC_ENABLED:
+            try:
+                DatasetService.check_dataset_permission(dataset, current_user, session)
+            except services.errors.account.NoPermissionError as e:
+                raise Forbidden(str(e))
         metadata = MetadataService.get_dataset_metadatas(dataset, session)
         return dump_response(DatasetMetadataListResponse, metadata), 200
 
@@ -124,14 +134,12 @@ class DatasetMetadataApi(Resource):
 
         dataset_id_str = str(dataset_id)
         metadata_id_str = str(metadata_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, session)
+        dataset = DatasetService.get_dataset_for_tenant(dataset_id_str, current_tenant_id, session=session)
         if dataset is None:
             raise NotFound("Dataset not found.")
         DatasetService.check_dataset_permission(dataset, current_user, session)
 
-        metadata = MetadataService.update_metadata_name(
-            dataset_id_str, metadata_id_str, name, current_user, current_tenant_id, session=session
-        )
+        metadata = MetadataService.update_metadata_name(dataset, metadata_id_str, name, current_user, session=session)
         return dump_response(DatasetMetadataResponse, metadata), 200
 
     @setup_required
@@ -140,17 +148,25 @@ class DatasetMetadataApi(Resource):
     @enterprise_license_required
     @console_ns.response(204, "Metadata deleted successfully")
     @with_current_user
+    @with_current_tenant_id
     @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_EDIT)
     @with_session
-    def delete(self, session: Session, current_user: Account, dataset_id: UUID, metadata_id: UUID):
+    def delete(
+        self,
+        session: Session,
+        current_tenant_id: str,
+        current_user: Account,
+        dataset_id: UUID,
+        metadata_id: UUID,
+    ):
         dataset_id_str = str(dataset_id)
         metadata_id_str = str(metadata_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, session)
+        dataset = DatasetService.get_dataset_for_tenant(dataset_id_str, current_tenant_id, session=session)
         if dataset is None:
             raise NotFound("Dataset not found.")
         DatasetService.check_dataset_permission(dataset, current_user, session)
 
-        MetadataService.delete_metadata(dataset_id_str, metadata_id_str, session)
+        MetadataService.delete_metadata(dataset, metadata_id_str, session)
         # Frontend callers only await success and invalidate metadata caches; no response body is consumed.
         return "", 204
 
@@ -208,18 +224,29 @@ class DocumentMetadataEditApi(Resource):
         204,
         "Documents metadata updated successfully",
     )
+    @console_ns.response(404, "Dataset, document, or metadata not found")
     @with_current_user
+    @with_current_tenant_id
     @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_EDIT)
     @with_session
     @model_validate(MetadataOperationData)
-    def post(self, req_data: MetadataOperationData, session: Session, current_user: Account, dataset_id: UUID):
-        dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, session)
+    def post(
+        self,
+        req_data: MetadataOperationData,
+        session: Session,
+        current_tenant_id: str,
+        current_user: Account,
+        dataset_id: UUID,
+    ):
+        dataset = DatasetService.get_dataset_for_tenant(str(dataset_id), current_tenant_id, session=session)
         if dataset is None:
             raise NotFound("Dataset not found.")
         DatasetService.check_dataset_permission(dataset, current_user, session)
 
-        MetadataService.update_documents_metadata(dataset, req_data, current_user, session=session)
+        try:
+            MetadataService.update_documents_metadata(dataset, req_data, current_user, session=session)
+        except MetadataResourceNotFoundError as exc:
+            raise NotFound(str(exc)) from exc
 
         # Frontend callers only await success and invalidate caches; no response body is consumed.
         return "", 204

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

@@ -3,10 +3,10 @@ from typing import Any
 
 from flask_restx import Resource
 from pydantic import BaseModel, Field
-from sqlalchemy import select
-from sqlalchemy.orm import Session, sessionmaker
-from werkzeug.exceptions import NotFound
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import Forbidden, NotFound
 
+from configs import dify_config
 from controllers.common.fields import SimpleDataResponse
 from controllers.common.schema import (
     JsonResponseWithStatus,
@@ -16,11 +16,15 @@ from controllers.common.schema import (
 )
 from controllers.console import console_ns
 from controllers.console.app.wraps import with_session
+from controllers.console.datasets.wraps import get_rag_pipeline
 from controllers.console.wraps import (
+    RBACPermission,
+    RBACResourceScope,
     account_initialization_required,
     enterprise_license_required,
     knowledge_pipeline_publish_enabled,
     model_validate,
+    rbac_permission_required,
     setup_required,
     with_current_tenant_id,
     with_current_user,
@@ -30,8 +34,11 @@ from fields.base import ResponseModel
 from libs.helper import dump_response
 from libs.login import login_required
 from models.account import Account
-from models.dataset import PipelineCustomizedTemplate
+from models.dataset import Pipeline
+from services.dataset_service import DatasetService
 from services.entities.knowledge_entities.rag_pipeline_entities import IconInfo, PipelineTemplateInfoEntity
+from services.errors.account import NoPermissionError
+from services.errors.rag_pipeline import RagPipelineResourceNotFoundError
 from services.rag_pipeline.rag_pipeline import RagPipelineService
 
 logger: logging.Logger = logging.getLogger(__name__)
@@ -125,15 +132,24 @@ class PipelineTemplateListApi(Resource):
 class PipelineTemplateDetailApi(Resource):
     @console_ns.doc(params=query_params_from_model(PipelineTemplateDetailQuery))
     @console_ns.response(200, "Pipeline template", console_ns.models[PipelineTemplateDetailResponse.__name__])
+    @console_ns.response(404, "Pipeline template not found")
     @setup_required
     @login_required
     @account_initialization_required
     @enterprise_license_required
-    @with_session
+    @with_current_tenant_id
+    @with_session(write=False)
     @model_validate(PipelineTemplateDetailQuery)
-    def get(self, req_data: PipelineTemplateDetailQuery, session: Session, template_id: str) -> JsonResponseWithStatus:
+    def get(
+        self,
+        req_data: PipelineTemplateDetailQuery,
+        session: Session,
+        current_tenant_id: str,
+        template_id: str,
+    ) -> JsonResponseWithStatus:
         pipeline_template = RagPipelineService.get_pipeline_template_detail(
             template_id,
+            current_tenant_id,
             type=req_data.type,
             session=session,
         )
@@ -181,38 +197,57 @@ class CustomizedPipelineTemplateApi(Resource):
     @account_initialization_required
     @enterprise_license_required
     @console_ns.response(200, "Success", console_ns.models[SimpleDataResponse.__name__])
-    def post(self, template_id: str) -> JsonResponseWithStatus:
-        with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
-            template = session.scalar(
-                select(PipelineCustomizedTemplate).where(PipelineCustomizedTemplate.id == template_id).limit(1)
+    @console_ns.response(404, "Customized pipeline template not found")
+    @with_current_tenant_id
+    @with_session(write=False)
+    def post(self, session: Session, current_tenant_id: str, template_id: str) -> JsonResponseWithStatus:
+        try:
+            yaml_content = RagPipelineService.get_customized_pipeline_template_yaml(
+                template_id, current_tenant_id, session=session
             )
-            if not template:
-                raise ValueError("Customized pipeline template not found.")
+        except RagPipelineResourceNotFoundError as exc:
+            raise NotFound(str(exc)) from exc
 
-        return dump_response(SimpleDataResponse, {"data": template.yaml_content}), 200
+        return dump_response(SimpleDataResponse, {"data": yaml_content}), 200
 
 
 @console_ns.route("/rag/pipelines/<string:pipeline_id>/customized/publish")
 class PublishCustomizedPipelineTemplateApi(Resource):
     @console_ns.expect(console_ns.models[CustomizedPipelineTemplatePayload.__name__])
     @console_ns.response(204, "Pipeline template published")
+    @console_ns.response(404, "Pipeline, workflow, or dataset not found")
     @setup_required
     @login_required
     @account_initialization_required
     @enterprise_license_required
     @knowledge_pipeline_publish_enabled
     @with_current_user
-    @with_current_tenant_id
+    @get_rag_pipeline
+    @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_PIPELINE_RELEASE)
     @model_validate(CustomizedPipelineTemplatePayload)
     def post(
         self,
         req_data: CustomizedPipelineTemplatePayload,
-        current_tenant_id: str,
         current_user: Account,
-        pipeline_id: str,
+        pipeline: Pipeline,
     ) -> tuple[str, int]:
-        rag_pipeline_service = RagPipelineService(db.session())
-        rag_pipeline_service.publish_customized_pipeline_template(
-            pipeline_id, req_data.model_dump(), current_user, current_tenant_id, session=db.session()
-        )
+        session = db.session()
+        dataset = pipeline.retrieve_dataset(session=session)
+        if dataset is None:
+            raise NotFound("Dataset not found")
+
+        if not dify_config.RBAC_ENABLED:
+            if not current_user.is_dataset_editor:
+                raise Forbidden()
+            try:
+                DatasetService.check_dataset_permission(dataset, current_user, session)
+            except NoPermissionError as exc:
+                raise Forbidden(str(exc)) from exc
+
+        try:
+            RagPipelineService.publish_customized_pipeline_template(
+                pipeline, dataset, req_data.model_dump(), current_user, session=session
+            )
+        except RagPipelineResourceNotFoundError as exc:
+            raise NotFound(str(exc)) from exc
         return "", 204

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

@@ -10,6 +10,7 @@ from sqlalchemy.orm import Session, sessionmaker
 from werkzeug.exceptions import BadRequest, Forbidden, InternalServerError, NotFound
 
 import services
+from configs import dify_config
 from controllers.common.controller_schemas import DefaultBlockConfigQuery, WorkflowListQuery, WorkflowUpdatePayload
 from controllers.common.fields import SimpleResultResponse
 from controllers.common.schema import query_params_from_model, register_response_schema_models, register_schema_models
@@ -60,8 +61,10 @@ from models import Account
 from models.dataset import Pipeline
 from models.model import EndUser
 from models.workflow import Workflow
+from services.dataset_service import DatasetService
 from services.errors.app import IsDraftWorkflowError, WorkflowHashNotEqualError, WorkflowNotFoundError
 from services.errors.llm import InvokeRateLimitError
+from services.errors.rag_pipeline import RagPipelineResourceNotFoundError
 from services.rag_pipeline.pipeline_generate_service import PipelineGenerateService
 from services.rag_pipeline.rag_pipeline import RagPipelineService
 from services.rag_pipeline.rag_pipeline_manage_service import RagPipelineManageService
@@ -1016,19 +1019,31 @@ class RagPipelineWorkflowLastRunApi(Resource):
 @console_ns.route("/rag/pipelines/transform/datasets/<uuid:dataset_id>")
 class RagPipelineTransformApi(Resource):
     @console_ns.response(200, "Success", console_ns.models[RagPipelineOpaqueResponse.__name__])
+    @console_ns.response(404, "Dataset or pipeline not found")
     @setup_required
     @login_required
     @account_initialization_required
     @with_current_user
+    @with_current_tenant_id
+    @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_EDIT)
     @with_session
-    def post(self, session: Session, current_user: Account, dataset_id: UUID):
-        if not (current_user.has_edit_permission or current_user.is_dataset_operator):
-            raise Forbidden()
+    def post(self, session: Session, current_tenant_id: str, current_user: Account, dataset_id: UUID):
+        dataset = DatasetService.get_dataset_for_tenant(str(dataset_id), current_tenant_id, session=session)
+        if dataset is None:
+            raise NotFound("Dataset not found.")
 
-        dataset_id_str = str(dataset_id)
-        rag_pipeline_transform_service = RagPipelineTransformService()
-        result = rag_pipeline_transform_service.transform_dataset(dataset_id_str, session)
-        return result
+        if not dify_config.RBAC_ENABLED:
+            if not (current_user.has_edit_permission or current_user.is_dataset_operator):
+                raise Forbidden()
+            try:
+                DatasetService.check_dataset_permission(dataset, current_user, session)
+            except services.errors.account.NoPermissionError as exc:
+                raise Forbidden(str(exc)) from exc
+
+        try:
+            return RagPipelineTransformService().transform_dataset(dataset, current_user.id, session)
+        except RagPipelineResourceNotFoundError as exc:
+            raise NotFound(str(exc)) from exc
 
 
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/datasource/variables-inspect")

api/controllers/console/error.py

@@ -109,6 +109,12 @@ class EducationActivateLimitError(BaseHTTPException):
     code = 429
 
 
+class EducationDiscountTemporarilyPausedError(BaseHTTPException):
+    error_code = "education_discount_temporarily_paused"
+    description = "Education discount temporarily paused, while we upgrade our security measures."
+    code = 503
+
+
 class ComplianceRateLimitError(BaseHTTPException):
     error_code = "compliance_rate_limit"
     description = "Rate limit exceeded for downloading compliance report."

api/controllers/console/explore/banner.py

@@ -3,17 +3,14 @@ from typing import cast
 
 from flask_restx import Namespace, Resource
 from pydantic import BaseModel, Field, RootModel, field_validator
-from sqlalchemy import select
 
 from controllers.common.schema import query_params_from_model, register_response_schema_models
 from controllers.console import api
 from controllers.console.wraps import model_validate
-from extensions.ext_database import db
+from extensions.ext_application_services import application_services
 from fields.base import ResponseModel
 from libs.helper import dump_response
 from models.enums import BannerStatus
-from models.model import ExporleBanner
-from services.feature_service import FeatureService
 
 
 class BannerListQuery(BaseModel):
@@ -67,23 +64,7 @@ class BannerApi(Resource):
     @model_validate(BannerListQuery)
     def get(self, req_data: BannerListQuery):
         """Get banner list."""
-        if not FeatureService.is_explore_banner_enabled():
-            return dump_response(BannerListResponse, [])
-
-        # Build base query for enabled banners
-        base_query = select(ExporleBanner).where(ExporleBanner.status == BannerStatus.ENABLED)
-
-        # Try to get banners in the requested language
-        banners = db.session.scalars(
-            base_query.where(ExporleBanner.language == req_data.language).order_by(ExporleBanner.sort)
-        ).all()
-
-        # Fallback to en-US if no banners found and language is not en-US
-        if not banners and req_data.language != "en-US":
-            banners = db.session.scalars(
-                base_query.where(ExporleBanner.language == "en-US").order_by(ExporleBanner.sort)
-            ).all()
-
+        banners = application_services().explore_banner_queries.list_for_language(req_data.language)
         return dump_response(BannerListResponse, banners)
 
 

api/controllers/console/flask_admission.py

@@ -9,7 +9,7 @@ from flask import Response, abort, request
 from configs import dify_config
 from controllers.console.wraps import account_initialization_required, enterprise_license_required, setup_required
 from core.logging.context import get_request_id, get_trace_id
-from enums.deployment_edition import DeploymentEdition
+from enums import DeploymentEdition
 from libs.login import current_account_with_tenant, login_required
 from machinery.context import RequestContext
 

api/controllers/console/init_validate.py

@@ -1,17 +1,11 @@
-import os
 from typing import Literal
 
 from flask import session
 from pydantic import BaseModel, Field
-from sqlalchemy import select
-from sqlalchemy.orm import Session
 
-from configs import dify_config
 from controllers.fastopenapi import console_router
-from enums.deployment_edition import DeploymentEdition
-from extensions.ext_database import db
-from models.model import DifySetup
-from services.account_service import TenantService
+from extensions.ext_application_services import application_services
+from services.init_validation_service import AlreadyInitializedError, InvalidInitializationPasswordError
 
 from .error import AlreadySetupError, InitValidateFailedError
 from .wraps import only_edition_self_hosted
@@ -36,7 +30,7 @@ class InitValidateResponse(BaseModel):
 )
 def get_init_status() -> InitStatusResponse:
     """Get initialization validation status."""
-    init_status = get_init_validate_status()
+    init_status = is_init_validated()
     if init_status:
         return InitStatusResponse(status="finished")
     return InitStatusResponse(status="not_started")
@@ -51,25 +45,19 @@ def get_init_status() -> InitStatusResponse:
 @only_edition_self_hosted
 def validate_init_password(payload: InitValidatePayload) -> InitValidateResponse:
     """Validate initialization password."""
-    tenant_count = TenantService.get_tenant_count(session=db.session())
-    if tenant_count > 0:
-        raise AlreadySetupError()
-
-    if payload.password != os.environ.get("INIT_PASSWORD"):
+    try:
+        application_services().init_validation.validate_password(payload.password)
+    except AlreadyInitializedError:
+        raise AlreadySetupError() from None
+    except InvalidInitializationPasswordError:
         session["is_init_validated"] = False
-        raise InitValidateFailedError()
+        raise InitValidateFailedError() from None
 
     session["is_init_validated"] = True
     return InitValidateResponse(result="success")
 
 
-def get_init_validate_status() -> bool:
-    if dify_config.DEPLOYMENT_EDITION != DeploymentEdition.CLOUD:
-        if os.environ.get("INIT_PASSWORD"):
-            if session.get("is_init_validated"):
-                return True
-
-            with Session(db.engine) as db_session:
-                return db_session.execute(select(DifySetup)).scalar_one_or_none() is not None
-
-    return True
+def is_init_validated() -> bool:
+    return application_services().init_validation.is_validated(
+        session_validated=bool(session.get("is_init_validated")),
+    )

api/controllers/console/ping.py

@@ -1,17 +0,0 @@
-from pydantic import BaseModel, Field
-
-from controllers.fastopenapi import console_router
-
-
-class PingResponse(BaseModel):
-    result: str = Field(description="Health check result", examples=["pong"])
-
-
-@console_router.get(
-    "/ping",
-    response_model=PingResponse,
-    tags=["console"],
-)
-def ping() -> PingResponse:
-    """Health check endpoint for connection testing."""
-    return PingResponse(result="pong")

api/controllers/console/setup.py

@@ -14,7 +14,7 @@ from services.setup_service import (
 )
 
 from .error import AlreadySetupError, NotInitValidateError
-from .init_validate import get_init_validate_status
+from .init_validate import is_init_validated
 from .wraps import mark_setup_completed, only_edition_self_hosted
 
 
@@ -73,7 +73,7 @@ def setup_system(payload: SetupRequestPayload) -> SetupResponse:
     """Initialize system setup with admin account.
 
     NOTE: This endpoint is unauthenticated by design for first-time bootstrap.
-    Access is restricted by deployment mode (`SELF_HOSTED`), one-time setup guards,
+    Access is restricted to self-hosted editions (`COMMUNITY` and `ENTERPRISE`), one-time setup guards,
     and init-password validation rather than user session authentication.
     """
     try:
@@ -85,7 +85,7 @@ def setup_system(payload: SetupRequestPayload) -> SetupResponse:
                 ip_address=extract_remote_ip(request),
                 language=payload.language,
             ),
-            initialization_validated=get_init_validate_status(),
+            initialization_validated=is_init_validated(),
         )
     except SetupAlreadyCompletedError:
         raise AlreadySetupError() from None

api/controllers/console/system.py

@@ -10,21 +10,27 @@ from controllers.fastopenapi import console_router
 logger = logging.getLogger(__name__)
 
 
+class PingResponse(BaseModel):
+    result: str = Field(description="Health check result", examples=["pong"])
+
+
 class VersionQuery(BaseModel):
     current_version: str = Field(..., description="Current application version")
 
 
-class VersionFeatures(BaseModel):
-    can_replace_logo: bool = Field(description="Whether logo replacement is supported")
-    model_load_balancing_enabled: bool = Field(description="Whether model load balancing is enabled")
-
-
 class VersionResponse(BaseModel):
     version: str = Field(description="Latest version number")
-    release_date: str = Field(description="Release date of latest version")
     release_notes: str = Field(description="Release notes for latest version")
-    can_auto_update: bool = Field(description="Whether auto-update is supported")
-    features: VersionFeatures = Field(description="Feature flags and capabilities")
+
+
+@console_router.get(
+    "/ping",
+    response_model=PingResponse,
+    tags=["console"],
+)
+def ping() -> PingResponse:
+    """Health check endpoint for connection testing."""
+    return PingResponse(result="pong")
 
 
 @console_router.get(
@@ -38,13 +44,7 @@ def check_version_update(query: VersionQuery) -> VersionResponse:
 
     result = VersionResponse(
         version=dify_config.project.version,
-        release_date="",
         release_notes="",
-        can_auto_update=False,
-        features=VersionFeatures(
-            can_replace_logo=dify_config.CAN_REPLACE_LOGO,
-            model_load_balancing_enabled=dify_config.MODEL_LB_ENABLED,
-        ),
     )
 
     if not check_update_url:
@@ -62,11 +62,9 @@ def check_version_update(query: VersionQuery) -> VersionResponse:
         result.version = query.current_version
         return result
     latest_version = content.get("version", result.version)
-    if _has_new_version(latest_version=latest_version, current_version=f"{query.current_version}"):
+    if _has_new_version(latest_version=latest_version, current_version=query.current_version):
         result.version = latest_version
-        result.release_date = content.get("releaseDate", "")
         result.release_notes = content.get("releaseNotes", "")
-        result.can_auto_update = content.get("canAutoUpdate", False)
     return result
 
 
@@ -75,7 +73,6 @@ def _has_new_version(*, latest_version: str, current_version: str) -> bool:
         latest = version.parse(latest_version)
         current = version.parse(current_version)
 
-        # Compare versions
         return latest > current
     except version.InvalidVersion:
         logger.warning("Invalid version format: latest=%s, current=%s", latest_version, current_version)

api/controllers/console/workflow_run_archive.py

@@ -11,7 +11,6 @@ from controllers.common.schema import register_response_schema_models, register_
 from controllers.console import console_ns
 from controllers.console.wraps import (
     account_initialization_required,
-    cloud_edition_billing_enabled,
     cloud_edition_billing_paid_plan_required,
     model_validate,
     only_edition_cloud,
@@ -122,7 +121,6 @@ class WorkflowRunArchivesApi(Resource):
     @login_required
     @account_initialization_required
     @only_edition_cloud
-    @cloud_edition_billing_enabled
     @cloud_edition_billing_paid_plan_required
     def get(self):
         tenant_id, _ = _current_owner_or_admin_ids()
@@ -143,7 +141,6 @@ class WorkflowRunArchiveDownloadsApi(Resource):
     @login_required
     @account_initialization_required
     @only_edition_cloud
-    @cloud_edition_billing_enabled
     @cloud_edition_billing_paid_plan_required
     @model_validate(WorkflowRunArchiveDownloadPayload)
     def post(self, req_data: WorkflowRunArchiveDownloadPayload):
@@ -170,7 +167,6 @@ class WorkflowRunArchiveDownloadApi(Resource):
     @login_required
     @account_initialization_required
     @only_edition_cloud
-    @cloud_edition_billing_enabled
     @cloud_edition_billing_paid_plan_required
     def get(self, download_id: str):
         tenant_id, _ = _current_owner_or_admin_ids()
@@ -195,7 +191,6 @@ class WorkflowRunArchiveDownloadFileApi(Resource):
     @login_required
     @account_initialization_required
     @only_edition_cloud
-    @cloud_edition_billing_enabled
     @cloud_edition_billing_paid_plan_required
     def get(self, download_id: str):
         tenant_id, _ = _current_owner_or_admin_ids()

api/controllers/console/workspace/account.py

@@ -28,7 +28,12 @@ from controllers.console.auth.error import (
     InvalidEmailError,
     InvalidTokenError,
 )
-from controllers.console.error import AccountInFreezeError, AccountNotFound, EmailSendIpLimitError
+from controllers.console.error import (
+    AccountInFreezeError,
+    AccountNotFound,
+    EducationDiscountTemporarilyPausedError,
+    EmailSendIpLimitError,
+)
 from controllers.console.workspace.error import (
     AccountAlreadyInitedError,
     CurrentPasswordIncorrectError,
@@ -38,7 +43,6 @@ from controllers.console.workspace.error import (
 )
 from controllers.console.wraps import (
     account_initialization_required,
-    cloud_edition_billing_enabled,
     enable_change_email,
     enterprise_license_required,
     model_validate,
@@ -46,7 +50,7 @@ from controllers.console.wraps import (
     setup_required,
     with_current_user,
 )
-from enums.deployment_edition import DeploymentEdition
+from enums import DeploymentEdition
 from extensions.ext_database import db
 from fields.base import ResponseModel
 from fields.member_fields import AccountResponse
@@ -536,7 +540,6 @@ class EducationVerifyApi(Resource):
     @login_required
     @account_initialization_required
     @only_edition_cloud
-    @cloud_edition_billing_enabled
     @console_ns.response(HTTPStatus.OK, "Success", console_ns.models[EducationVerifyResponse.__name__])
     @with_current_user
     def get(self, account: Account):
@@ -554,20 +557,14 @@ class EducationApi(Resource):
     @login_required
     @account_initialization_required
     @only_edition_cloud
-    @cloud_edition_billing_enabled
     @with_current_user
     def post(self, account: Account):
-        payload = console_ns.payload or {}
-        args = EducationActivatePayload.model_validate(payload)
-
-        result = BillingService.EducationIdentity.activate(account, args.token, args.institution, args.role)
-        return result
+        raise EducationDiscountTemporarilyPausedError()
 
     @setup_required
     @login_required
     @account_initialization_required
     @only_edition_cloud
-    @cloud_edition_billing_enabled
     @console_ns.response(HTTPStatus.OK, "Success", console_ns.models[EducationStatusResponse.__name__])
     @with_current_user
     def get(self, account: Account):
@@ -585,7 +582,6 @@ class EducationAutoCompleteApi(Resource):
     @login_required
     @account_initialization_required
     @only_edition_cloud
-    @cloud_edition_billing_enabled
     @console_ns.response(HTTPStatus.OK, "Success", console_ns.models[EducationAutocompleteResponse.__name__])
     def get(self):
         payload = request.args.to_dict(flat=True)

api/controllers/console/workspace/members.py

@@ -32,6 +32,7 @@ from controllers.console.wraps import (
     setup_required,
     with_current_user,
 )
+from enums import DeploymentEdition
 from extensions.ext_application_services import application_services
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
@@ -179,7 +180,7 @@ def _check_member_invite_limits(tenant_id: str, new_member_count: int, new_accou
 
     features = FeatureService.get_features(tenant_id=tenant_id, exclude_vector_space=True)
 
-    if dify_config.ENTERPRISE_ENABLED:
+    if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.ENTERPRISE:
         workspace_members = features.workspace_members
         if workspace_members.enabled is True and not workspace_members.is_available(new_member_count):
             raise WorkspaceMembersLimitExceeded()
@@ -189,7 +190,7 @@ def _check_member_invite_limits(tenant_id: str, new_member_count: int, new_accou
                 raise SeatsLimitExceeded()
         return
 
-    if dify_config.BILLING_ENABLED and features.billing.enabled is True:
+    if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD:
         members = features.members
         current_member_count = _count_current_members(tenant_id)
         if 0 < members.limit < current_member_count + new_member_count:
@@ -265,7 +266,10 @@ class MemberInviteEmailApi(Resource):
 
         tenant_id = inviter.current_tenant.id
         with redis_client.lock(f"workspace_member_invite:{tenant_id}", timeout=60):
-            if dify_config.ENTERPRISE_ENABLED is True or dify_config.BILLING_ENABLED is True:
+            if dify_config.DEPLOYMENT_EDITION in {
+                DeploymentEdition.CLOUD,
+                DeploymentEdition.ENTERPRISE,
+            }:
                 new_member_count, new_account_count = _count_new_member_invites(tenant_id, invitee_emails)
                 _check_member_invite_limits(tenant_id, new_member_count, new_account_count)
 

api/controllers/console/workspace/rbac.py

@@ -14,6 +14,7 @@ from controllers.console import console_ns
 from controllers.console.wraps import RBACPermission, RBACResourceScope, model_validate, rbac_permission_required
 from core.db.session_factory import session_factory
 from core.rbac import RBACResourceWhitelistScope
+from enums import DeploymentEdition
 from extensions.ext_database import db
 from libs.login import current_account_with_tenant, login_required
 from models import Account
@@ -316,14 +317,16 @@ class RBACRolesApi(Resource):
         options = req_data.to_inner_options()
         if not dify_config.RBAC_ENABLED:
             result = _legacy_workspace_roles(
-                options, include_owner=req_data.include_owner, billing_enabled=dify_config.BILLING_ENABLED
+                options,
+                include_owner=req_data.include_owner,
+                billing_enabled=dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD,
             )
         else:
             result = svc.RBACService.Roles.list(
                 tenant_id,
                 account_id,
                 include_owner=req_data.include_owner,
-                biiling_enabled=dify_config.BILLING_ENABLED,
+                biiling_enabled=dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD,
                 options=options,
             )
 
@@ -351,7 +354,12 @@ class RBACRoleItemApi(Resource):
     def get(self, role_id):
         tenant_id, account_id = _current_ids()
         return _dump(
-            svc.RBACService.Roles.get(tenant_id, account_id, role_id, billing_enabled=dify_config.BILLING_ENABLED)
+            svc.RBACService.Roles.get(
+                tenant_id,
+                account_id,
+                role_id,
+                billing_enabled=dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD,
+            )
         )
 
     @login_required
@@ -719,14 +727,17 @@ class RBACDatasetWhitelistApi(Resource):
     def put(self, dataset_id):
         tenant_id, account_id = _current_ids()
         request = _payload(_ResourceAccessScopeRequest)
-        return _dump(
-            svc.RBACService.DatasetAccess.replace_whitelist(
-                tenant_id,
-                account_id,
-                str(dataset_id),
-                svc.ReplaceMemberBindings(scope=request.scope.value),
-            )
+        result = svc.RBACService.DatasetAccess.replace_whitelist(
+            tenant_id,
+            account_id,
+            str(dataset_id),
+            svc.ReplaceMemberBindings(scope=request.scope.value),
         )
+        # Widening the scope only records it: the members still need the default access policy
+        # before they can reach the dataset, same as the app whitelist route above.
+        if dify_config.RBAC_ENABLED and request.scope is RBACResourceWhitelistScope.ALL:
+            initialize_created_app_rbac_access_task.delay(tenant_id, account_id, dataset_id=str(dataset_id))
+        return _dump(result)
 
 
 @console_ns.route("/workspaces/current/rbac/datasets/<uuid:dataset_id>/user-access-policies")

api/controllers/console/workspace/tool_providers.py

@@ -61,6 +61,7 @@ from core.tools.entities.tool_entities import (
     ToolProviderType,
     WorkflowToolParameterConfiguration,
 )
+from enums import DeploymentEdition
 from extensions.ext_database import db
 from fields.base import ResponseModel
 from libs.helper import alphanumeric, dump_response, uuid_value
@@ -282,10 +283,10 @@ def _resolve_identity_mode(requested: IdentityMode | None, *, current: IdentityM
       can never imply forwarding that the runtime won't perform. This gates the
       API surface to match the backend gate in
       ``MCPTool._forwarding_requested`` — both the API and the backend
-      invocation must be gated on ``dify_config.ENTERPRISE_ENABLED``.
+      invocation must be gated on the Enterprise deployment edition.
     """
     mode = current if requested is None else requested
-    if mode != IdentityMode.OFF and not dify_config.ENTERPRISE_ENABLED:
+    if mode != IdentityMode.OFF and dify_config.DEPLOYMENT_EDITION != DeploymentEdition.ENTERPRISE:
         return IdentityMode.OFF
     return mode
 
@@ -1417,7 +1418,7 @@ class ToolProviderMCPApi(Resource):
         with sessionmaker(db.engine).begin() as session:
             service = MCPToolManageService(session=session)
             # Resolve "leave unchanged" (None) against the stored value, and gate
-            # the result on ENTERPRISE_ENABLED — both are API-layer concerns, so
+            # the result on the Enterprise edition — both are API-layer concerns, so
             # the service receives a concrete IdentityMode.
             existing = service.get_provider(provider_id=req_data.provider_id, tenant_id=current_tenant_id)
             identity_mode = _resolve_identity_mode(req_data.identity_mode, current=IdentityMode(existing.identity_mode))

api/controllers/console/workspace/workspace.py

@@ -38,6 +38,7 @@ from controllers.console.wraps import (
     with_current_tenant_id,
     with_current_user,
 )
+from enums import CloudPlan
 from extensions.ext_application_services import application_services
 from extensions.ext_database import db
 from fields.base import ResponseModel
@@ -80,7 +81,7 @@ class WorkspaceInfoPayload(BaseModel):
 class TenantInfoResponse(ResponseModel):
     id: str
     name: str | None = None
-    plan: str | None = None
+    plan: CloudPlan | None = None
     status: str | None = None
     created_at: int | None = None
     role: str | None = None
@@ -92,7 +93,7 @@ class TenantInfoResponse(ResponseModel):
     trial_credits_exhausted_at: int | None = None
     next_credit_reset_date: int | None = None
 
-    @field_validator("plan", "status", "trial_end_reason", mode="before")
+    @field_validator("status", "trial_end_reason", mode="before")
     @classmethod
     def _normalize_enum_like(cls, value):
         if value is None:
@@ -111,20 +112,20 @@ class CurrentWorkspaceSummaryResponse(ResponseModel):
     id: str
     name: str
     role: TenantAccountRole
-    plan: str | None
+    plan: CloudPlan | None
     credits: int | None = Field(description="Remaining credits in the effective pool; -1 means unlimited.")
 
 
 class TenantListItemResponse(ResponseModel):
     id: str
     name: str | None = None
-    plan: str | None = None
+    plan: CloudPlan | None = None
     status: str | None = None
     created_at: int | None = None
     last_opened_at: int | None = None
     current: bool
 
-    @field_validator("plan", "status", mode="before")
+    @field_validator("status", mode="before")
     @classmethod
     def _normalize_enum_like(cls, value):
         if value is None:

api/controllers/console/wraps.py

@@ -1,6 +1,5 @@
 import contextlib
 import json
-import os
 import time
 from collections.abc import Callable
 from functools import wraps
@@ -19,8 +18,7 @@ from controllers.common.wraps import (
 )
 from controllers.console.auth.error import AuthenticationFailedError, EmailCodeError
 from controllers.console.workspace.error import AccountNotInitializedError
-from enums.cloud_plan import CloudPlan
-from enums.deployment_edition import DeploymentEdition
+from enums import CloudPlan, DeploymentEdition
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from libs.encryption import FieldEncryption
@@ -142,7 +140,7 @@ def only_edition_cloud[**P, R](view: Callable[P, R]) -> Callable[P, R]:
 def only_edition_enterprise[**P, R](view: Callable[P, R]) -> Callable[P, R]:
     @wraps(view)
     def decorated(*args: P.args, **kwargs: P.kwargs):
-        if not dify_config.ENTERPRISE_ENABLED:
+        if dify_config.DEPLOYMENT_EDITION != DeploymentEdition.ENTERPRISE:
             abort(404)
 
         return view(*args, **kwargs)
@@ -161,16 +159,6 @@ def only_edition_self_hosted[**P, R](view: Callable[P, R]) -> Callable[P, R]:
     return decorated
 
 
-def cloud_edition_billing_enabled[**P, R](view: Callable[P, R]) -> Callable[P, R]:
-    @wraps(view)
-    def decorated(*args: P.args, **kwargs: P.kwargs):
-        if not dify_config.BILLING_ENABLED:
-            abort(403, "Billing feature is not enabled.")
-        return view(*args, **kwargs)
-
-    return decorated
-
-
 def cloud_edition_billing_paid_plan_required[**P, R](view: Callable[P, R]) -> Callable[P, R]:
     @wraps(view)
     def decorated(*args: P.args, **kwargs: P.kwargs):
@@ -192,7 +180,7 @@ def cloud_edition_billing_resource_check[**P, R](resource: str) -> Callable[[Cal
         def decorated(*args: P.args, **kwargs: P.kwargs):
             _, current_tenant_id = current_account_with_tenant()
             if resource == "vector_space":
-                if not dify_config.BILLING_ENABLED:
+                if dify_config.DEPLOYMENT_EDITION != DeploymentEdition.CLOUD:
                     return view(*args, **kwargs)
 
                 vector_space = FeatureService.get_vector_space(current_tenant_id)
@@ -260,35 +248,35 @@ def cloud_edition_billing_knowledge_limit_check[**P, R](
     return interceptor
 
 
+def check_knowledge_rate_limit() -> None:
+    _, current_tenant_id = current_account_with_tenant()
+    knowledge_rate_limit = FeatureService.get_knowledge_rate_limit(current_tenant_id)
+    if not knowledge_rate_limit.enabled:
+        return
+
+    current_time = int(time.time() * 1000)
+    key = f"rate_limit_{current_tenant_id}"
+    redis_client.zadd(key, {current_time: current_time})
+    redis_client.zremrangebyscore(key, 0, current_time - 60000)
+
+    if redis_client.zcard(key) > knowledge_rate_limit.limit:
+        db.session.add(  # guard-ignore: no-new-controller-sqlalchemy -- existing decorator audit write
+            RateLimitLog(
+                tenant_id=current_tenant_id,
+                subscription_plan=knowledge_rate_limit.subscription_plan,
+                operation="knowledge",
+            )
+        )
+        db.session.commit()
+        abort(403, "Sorry, you have reached the knowledge base request rate limit of your subscription.")
+
+
 def cloud_edition_billing_rate_limit_check[**P, R](resource: str) -> Callable[[Callable[P, R]], Callable[P, R]]:
     def interceptor(view: Callable[P, R]):
         @wraps(view)
         def decorated(*args: P.args, **kwargs: P.kwargs):
             if resource == "knowledge":
-                _, current_tenant_id = current_account_with_tenant()
-                knowledge_rate_limit = FeatureService.get_knowledge_rate_limit(current_tenant_id)
-                if knowledge_rate_limit.enabled:
-                    current_time = int(time.time() * 1000)
-                    key = f"rate_limit_{current_tenant_id}"
-
-                    redis_client.zadd(key, {current_time: current_time})
-
-                    redis_client.zremrangebyscore(key, 0, current_time - 60000)
-
-                    request_count = redis_client.zcard(key)
-
-                    if request_count > knowledge_rate_limit.limit:
-                        # add ratelimit record
-                        rate_limit_log = RateLimitLog(
-                            tenant_id=current_tenant_id,
-                            subscription_plan=knowledge_rate_limit.subscription_plan,
-                            operation="knowledge",
-                        )
-                        db.session.add(rate_limit_log)
-                        db.session.commit()
-                        abort(
-                            403, "Sorry, you have reached the knowledge base request rate limit of your subscription."
-                        )
+                check_knowledge_rate_limit()
             return view(*args, **kwargs)
 
         return decorated
@@ -301,7 +289,7 @@ def cloud_utm_record[**P, R](view: Callable[P, R]) -> Callable[P, R]:
     def decorated(*args: P.args, **kwargs: P.kwargs):
         with contextlib.suppress(Exception):
             utm_info = request.cookies.get("utm_info")
-            if dify_config.BILLING_ENABLED and utm_info:
+            if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD and utm_info:
                 _, current_tenant_id = current_account_with_tenant()
                 utm_info_dict: UtmInfo = json.loads(utm_info)
                 OperationService.record_utm(current_tenant_id, utm_info_dict)
@@ -330,7 +318,7 @@ def setup_required[R](view: Callable[..., R]) -> Callable[..., R]:
         # preserving support for plain functions used in tests and utilities.
         # check setup
         if dify_config.DEPLOYMENT_EDITION != DeploymentEdition.CLOUD and not _is_setup_completed():
-            if os.environ.get("INIT_PASSWORD"):
+            if dify_config.INIT_PASSWORD:
                 raise NotInitValidateError()
             raise NotSetupError()
 

api/controllers/inner_api/app/dsl.py

@@ -5,13 +5,15 @@ to attribute the created app; workspace/membership validation is done by the
 Go admin-api caller.
 """
 
+from uuid import UUID
+
 from flask import request
 from flask_restx import Resource
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, ValidationError, field_validator
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
-from controllers.common.schema import register_schema_model
+from controllers.common.schema import query_params_from_model, register_schema_model
 from controllers.console.wraps import setup_required
 from controllers.inner_api import inner_api_ns
 from controllers.inner_api.wraps import enterprise_inner_api_only
@@ -20,6 +22,7 @@ from models import Account, App
 from models.account import AccountStatus
 from services.app_dsl_service import AppDslService
 from services.entities.dsl_entities import ImportMode, ImportStatus
+from services.errors.app import IsDraftWorkflowError, WorkflowNotFoundError
 
 
 class InnerAppDSLImportPayload(BaseModel):
@@ -29,6 +32,18 @@ class InnerAppDSLImportPayload(BaseModel):
     description: str | None = Field(default=None, description="Override app description from DSL")
 
 
+class EnterpriseAppDSLExportQuery(BaseModel):
+    include_secret: bool = Field(default=False, description="Whether to include secret values in the exported DSL")
+    workflow_id: UUID | None = Field(default=None, description="Published workflow version ID to export")
+
+    @field_validator("include_secret", mode="before")
+    @classmethod
+    def parse_include_secret(cls, value: object) -> bool:
+        if isinstance(value, str):
+            return value.lower() == "true"
+        return bool(value)
+
+
 register_schema_model(inner_api_ns, InnerAppDSLImportPayload)
 
 
@@ -82,24 +97,48 @@ class EnterpriseAppDSLExport(Resource):
     @enterprise_inner_api_only
     @inner_api_ns.doc(
         "enterprise_app_dsl_export",
+        params=query_params_from_model(EnterpriseAppDSLExportQuery),
         responses={
             200: "Export successful",
-            404: "App not found",
+            400: "Invalid workflow ID or unpublished workflow version",
+            404: "App or workflow version not found",
         },
     )
     def get(self, app_id: str):
         """Export an app's DSL as YAML."""
-        include_secret = request.args.get("include_secret", "false").lower() == "true"
+        try:
+            query = EnterpriseAppDSLExportQuery.model_validate(request.args.to_dict(flat=True))
+        except ValidationError:
+            return {
+                "code": "invalid_workflow_id",
+                "message": "workflow_id must be a valid UUID",
+                "status": 400,
+            }, 400
+
+        workflow_id = str(query.workflow_id) if query.workflow_id else None
 
         app_model = db.session.get(App, app_id)
         if not app_model:
             return {"message": "app not found"}, 404
 
-        data = AppDslService.export_dsl(
-            app_model=app_model,
-            session=db.session(),
-            include_secret=include_secret,
-        )
+        if not workflow_id:
+            data = AppDslService.export_dsl(
+                app_model=app_model,
+                session=db.session(),
+                include_secret=query.include_secret,
+            )
+        else:
+            try:
+                data = AppDslService.export_dsl(
+                    app_model=app_model,
+                    session=db.session(),
+                    include_secret=query.include_secret,
+                    workflow_id=workflow_id,
+                )
+            except WorkflowNotFoundError as exc:
+                return {"code": "workflow_version_not_found", "message": str(exc), "status": 404}, 404
+            except IsDraftWorkflowError as exc:
+                return {"code": "workflow_version_not_published", "message": str(exc), "status": 400}, 400
 
         return {"data": data}, 200
 

api/controllers/openapi/__init__.py

@@ -139,7 +139,6 @@ register_response_schema_models(
 register_enum_models(openapi_ns, OpenApiErrorCode)
 
 from . import (
-    _meta,
     account,
     app_dsl,
     app_run,
@@ -157,7 +156,6 @@ from . import (
 # Request models are imported from _models.py and registered above.
 
 __all__ = [
-    "_meta",
     "account",
     "app_dsl",
     "app_run",

api/controllers/openapi/_meta.py

@@ -1,24 +0,0 @@
-"""Meta endpoint: `GET /openapi/v1/_version` — no auth.
-
-Returns the server's project version and edition so the difyctl CLI can probe
-compatibility without needing to be logged in. Mirrors the `_health` endpoint
-in `index.py`.
-"""
-
-from flask_restx import Resource
-
-from configs import dify_config
-from controllers.openapi import openapi_ns
-from controllers.openapi._contract import returns
-from controllers.openapi._models import ServerVersionResponse
-
-
-@openapi_ns.route("/_version")
-class VersionApi(Resource):
-    @returns(200, ServerVersionResponse, description="Server version")
-    def get(self):
-        edition = dify_config.EDITION if dify_config.EDITION in ("SELF_HOSTED", "CLOUD") else "SELF_HOSTED"
-        return ServerVersionResponse(
-            version=dify_config.project.version,
-            edition=edition,
-        )

api/controllers/openapi/_models.py

@@ -7,6 +7,7 @@ from typing import Any, Final, Literal
 
 from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
+from enums import DeploymentEdition
 from libs.helper import EmailStr, UUIDStr, UUIDStrOrEmpty, uuid_value
 from models.model import AppMode
 
@@ -258,7 +259,7 @@ class ServerVersionResponse(BaseModel):
     """Meta endpoint payload for `GET /openapi/v1/_version` — no auth required."""
 
     version: str
-    edition: Literal["SELF_HOSTED", "CLOUD"]
+    edition: DeploymentEdition
 
 
 class HealthResponse(BaseModel):

api/controllers/openapi/apps_permitted_external.py

@@ -23,7 +23,8 @@ from controllers.openapi._models import (
 )
 from controllers.openapi.apps import build_app_describe_response
 from controllers.openapi.auth.composition import auth_router
-from controllers.openapi.auth.data import AuthData, Edition
+from controllers.openapi.auth.data import AuthData
+from enums import DeploymentEdition
 from libs.oauth_bearer import Scope, TokenType
 from models import App
 from models.enums import AppStatus
@@ -37,7 +38,7 @@ class PermittedExternalAppsListApi(Resource):
     @auth_router.guard(
         scope=Scope.APPS_READ_PERMITTED_EXTERNAL,
         allowed_token_types=frozenset({TokenType.OAUTH_EXTERNAL_SSO}),
-        edition=frozenset({Edition.EE}),
+        edition=frozenset({DeploymentEdition.ENTERPRISE}),
     )
     @returns(200, PermittedExternalAppsListResponse, description="Permitted external apps list")
     @accepts(query=PermittedExternalAppsListQuery)
@@ -94,7 +95,7 @@ class PermittedExternalAppDescribeApi(Resource):
     @auth_router.guard(
         scope=Scope.APPS_READ_PERMITTED_EXTERNAL,
         allowed_token_types=frozenset({TokenType.OAUTH_EXTERNAL_SSO}),
-        edition=frozenset({Edition.EE}),
+        edition=frozenset({DeploymentEdition.ENTERPRISE}),
     )
     @returns(200, AppDescribeResponse, description="Permitted external app description")
     @accepts(query=AppDescribeQuery)

api/controllers/openapi/auth/composition.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from controllers.openapi.auth.conditions import (
-    EDITION_EE,
+    EDITION_ENTERPRISE,
     HAS_ALLOWED_ROLES,
     HAS_RBAC,
     LOADED_APP_IS_PRIVATE,
@@ -11,7 +11,6 @@ from controllers.openapi.auth.conditions import (
     WORKSPACE_MEMBERSHIP_REQUIRED,
     WORKSPACE_SCOPED,
 )
-from controllers.openapi.auth.data import Edition
 from controllers.openapi.auth.flow import When
 from controllers.openapi.auth.pipeline import AuthPipeline, PipelineRoute, PipelineRouter
 from controllers.openapi.auth.prepare import (
@@ -33,6 +32,7 @@ from controllers.openapi.auth.verify import (
     check_workspace_mismatch,
     check_workspace_role,
 )
+from enums import DeploymentEdition
 from libs.oauth_bearer import TokenType
 
 account_pipeline = AuthPipeline(
@@ -42,7 +42,7 @@ account_pipeline = AuthPipeline(
         When(WORKSPACE_MEMBERSHIP_REQUIRED, then=load_tenant_from_request),
         load_account,
         When(WORKSPACE_SCOPED, then=load_workspace_role),
-        When(PATH_HAS_APP_ID & EDITION_EE, then=load_app_access_mode),
+        When(PATH_HAS_APP_ID & EDITION_ENTERPRISE, then=load_app_access_mode),
     ],
     auth=[
         When(PATH_HAS_APP_ID, then=check_app_api_enabled),
@@ -51,8 +51,8 @@ account_pipeline = AuthPipeline(
         When(PATH_HAS_APP_ID, then=check_workspace_mismatch),
         When(HAS_ALLOWED_ROLES, then=check_workspace_role),
         When(HAS_RBAC, then=check_rbac_permission),
-        When(PATH_HAS_APP_ID & EDITION_EE & WEBAPP_AUTH_ENABLED & WEBAPP_RUN_SCOPED, then=check_acl),
-        When(EDITION_EE & LOADED_APP_IS_PRIVATE & WEBAPP_RUN_SCOPED, then=check_private_app_permission),
+        When(PATH_HAS_APP_ID & EDITION_ENTERPRISE & WEBAPP_AUTH_ENABLED & WEBAPP_RUN_SCOPED, then=check_acl),
+        When(EDITION_ENTERPRISE & LOADED_APP_IS_PRIVATE & WEBAPP_RUN_SCOPED, then=check_private_app_permission),
     ],
 )
 
@@ -74,6 +74,9 @@ external_sso_pipeline = AuthPipeline(
 auth_router = PipelineRouter(
     {
         TokenType.OAUTH_ACCOUNT: PipelineRoute(account_pipeline),
-        TokenType.OAUTH_EXTERNAL_SSO: PipelineRoute(external_sso_pipeline, required_edition=frozenset({Edition.EE})),
+        TokenType.OAUTH_EXTERNAL_SSO: PipelineRoute(
+            external_sso_pipeline,
+            required_edition=frozenset({DeploymentEdition.ENTERPRISE}),
+        ),
     }
 )

api/controllers/openapi/auth/conditions.py

@@ -2,7 +2,9 @@ from __future__ import annotations
 
 from collections.abc import Callable
 
-from controllers.openapi.auth.data import AuthData, Edition, RequestContext, current_edition
+from configs import dify_config
+from controllers.openapi.auth.data import AuthData, RequestContext
+from enums import DeploymentEdition
 from libs.oauth_bearer import Scope, TokenType
 from services.enterprise.enterprise_service import WebAppAccessMode
 from services.feature_service import FeatureService
@@ -44,9 +46,9 @@ TOKEN_IS_OAUTH_EXTERNAL_SSO = request_cond(lambda ctx: ctx.token_type == TokenTy
 
 PATH_HAS_APP_ID = request_cond(lambda ctx: "app_id" in ctx.path_params)
 
-EDITION_CE = config_cond(lambda: current_edition() == Edition.CE)
-EDITION_EE = config_cond(lambda: current_edition() == Edition.EE)
-EDITION_SAAS = config_cond(lambda: current_edition() == Edition.SAAS)
+EDITION_COMMUNITY = config_cond(lambda: dify_config.DEPLOYMENT_EDITION == DeploymentEdition.COMMUNITY)
+EDITION_ENTERPRISE = config_cond(lambda: dify_config.DEPLOYMENT_EDITION == DeploymentEdition.ENTERPRISE)
+EDITION_CLOUD = config_cond(lambda: dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD)
 
 WEBAPP_AUTH_ENABLED = config_cond(lambda: FeatureService.get_system_features().webapp_auth.enabled)
 

api/controllers/openapi/auth/data.py

@@ -6,34 +6,18 @@ from enum import StrEnum
 from pydantic import BaseModel, ConfigDict, Field
 from werkzeug.exceptions import InternalServerError
 
-from configs import dify_config
 from core.rbac import RBACPermission, RBACResourceScope
-from enums.deployment_edition import DeploymentEdition
 from libs.oauth_bearer import Scope, TokenType
 from models.account import Account, Tenant, TenantAccountRole
 from models.model import App, EndUser
 from services.enterprise.enterprise_service import WebAppAccessMode
 
 
-class Edition(StrEnum):
-    CE = "ce"
-    EE = "ee"
-    SAAS = "saas"
-
-
 class CallerKind(StrEnum):
     ACCOUNT = "account"
     END_USER = "end_user"
 
 
-def current_edition() -> Edition:
-    if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD:
-        return Edition.SAAS
-    if dify_config.ENTERPRISE_ENABLED:
-        return Edition.EE
-    return Edition.CE
-
-
 class ExternalIdentity(BaseModel):
     model_config = ConfigDict(frozen=True)
 

api/controllers/openapi/auth/pipeline.py

@@ -16,16 +16,16 @@ from flask import current_app, request
 from flask_login import user_logged_in
 from werkzeug.exceptions import Forbidden, NotFound, Unauthorized
 
+from configs import dify_config
 from controllers.openapi._audit import emit_wrong_surface
 from controllers.openapi.auth.data import (
     AuthData,
-    Edition,
     ExternalIdentity,
     RBACRequirement,
     RequestContext,
-    current_edition,
 )
 from controllers.openapi.auth.flow import When
+from enums import DeploymentEdition
 from libs.oauth_bearer import (
     AuthContext,
     Scope,
@@ -112,7 +112,7 @@ class AuthPipeline:
 @dataclass(frozen=True)
 class PipelineRoute:
     pipeline: AuthPipeline
-    required_edition: frozenset[Edition] | None = None
+    required_edition: frozenset[DeploymentEdition] | None = None
 
 
 class PipelineRouter:
@@ -131,7 +131,7 @@ class PipelineRouter:
         *,
         scope: Scope | None = None,
         allowed_token_types: frozenset[TokenType] | None = None,
-        edition: frozenset[Edition] | None = None,
+        edition: frozenset[DeploymentEdition] | None = None,
         workspace_membership: bool = False,
         allowed_roles: frozenset[TenantAccountRole] | None = None,
         rbac: RBACRequirement | None = None,
@@ -150,7 +150,7 @@ class PipelineRouter:
         *,
         scope: Scope | None = None,
         allowed_token_types: frozenset[TokenType] | None = None,
-        edition: frozenset[Edition] | None = None,
+        edition: frozenset[DeploymentEdition] | None = None,
         allowed_roles: frozenset[TenantAccountRole] | None = None,
         rbac: RBACRequirement | None = None,
     ) -> Callable:
@@ -168,7 +168,7 @@ class PipelineRouter:
         *,
         scope: Scope | None,
         allowed_token_types: frozenset[TokenType] | None,
-        edition: frozenset[Edition] | None,
+        edition: frozenset[DeploymentEdition] | None,
         workspace_membership: bool,
         allowed_roles: frozenset[TenantAccountRole] | None,
         rbac: RBACRequirement | None,
@@ -200,17 +200,17 @@ class PipelineRouter:
         *,
         scope: Scope | None,
         allowed_token_types: frozenset[TokenType] | None,
-        edition: frozenset[Edition] | None,
+        edition: frozenset[DeploymentEdition] | None,
         workspace_membership: bool = False,
         allowed_roles: frozenset[TenantAccountRole] | None = None,
         rbac: RBACRequirement | None = None,
     ) -> Any:
         # 404 not 403 — this edition doesn't expose the feature at all
-        if edition is not None and current_edition() not in edition:
+        if edition is not None and dify_config.DEPLOYMENT_EDITION not in edition:
             raise NotFound()
 
         license_checked = False
-        if edition is not None and Edition.EE in edition:
+        if edition is not None and DeploymentEdition.ENTERPRISE in edition:
             _check_license()
             license_checked = True
 
@@ -234,9 +234,9 @@ class PipelineRouter:
             raise Forbidden("unsupported_token_type")
 
         if route.required_edition is not None:
-            if current_edition() not in route.required_edition:
+            if dify_config.DEPLOYMENT_EDITION not in route.required_edition:
                 raise Forbidden("external_sso_requires_ee")
-            if not license_checked and Edition.EE in route.required_edition:
+            if not license_checked and DeploymentEdition.ENTERPRISE in route.required_edition:
                 _check_license()
 
         return route.pipeline._run(

api/controllers/openapi/index.py

@@ -1,8 +1,11 @@
+"""Unauthenticated health and version probes for the OpenAPI surface."""
+
 from flask_restx import Resource
 
+from configs import dify_config
 from controllers.openapi import openapi_ns
 from controllers.openapi._contract import returns
-from controllers.openapi._models import HealthResponse
+from controllers.openapi._models import HealthResponse, ServerVersionResponse
 
 
 @openapi_ns.route("/_health")
@@ -10,3 +13,13 @@ class HealthApi(Resource):
     @returns(200, HealthResponse, description="Health check")
     def get(self):
         return HealthResponse(ok=True)
+
+
+@openapi_ns.route("/_version")
+class VersionApi(Resource):
+    @returns(200, ServerVersionResponse, description="Server version")
+    def get(self):
+        return ServerVersionResponse(
+            version=dify_config.project.version,
+            edition=dify_config.DEPLOYMENT_EDITION,
+        )

api/controllers/service_api/app/completion.py

@@ -42,7 +42,7 @@ from core.errors.error import (
     QuotaExceededError,
 )
 from core.helper.trace_id_helper import get_external_trace_id, get_trace_session_id, omit_trace_session_id_from_payload
-from enums.cloud_plan import CloudPlan
+from enums import CloudPlan, DeploymentEdition
 from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from libs.helper import UUIDStrOrEmpty
@@ -376,7 +376,11 @@ class ChatApi(Resource):
 
         payload = ChatRequestPayload.model_validate(omit_trace_session_id_from_payload(service_api_ns.payload) or {})
 
-        if app_mode == AppMode.ADVANCED_CHAT and payload.workflow_id and dify_config.BILLING_ENABLED:
+        if (
+            app_mode == AppMode.ADVANCED_CHAT
+            and payload.workflow_id
+            and dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD
+        ):
             billing_info = BillingService.get_info(app_model.tenant_id, exclude_vector_space=True)
             if billing_info["enabled"] and billing_info["subscription"]["plan"] == CloudPlan.SANDBOX:
                 raise WorkflowVersionExecutionNotAllowedError()

api/controllers/service_api/app/workflow.py

@@ -45,7 +45,7 @@ from core.errors.error import (
     QuotaExceededError,
 )
 from core.helper.trace_id_helper import get_external_trace_id, get_trace_session_id, omit_trace_session_id_from_payload
-from enums.cloud_plan import CloudPlan
+from enums import CloudPlan, DeploymentEdition
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from fields.base import ResponseModel
@@ -451,7 +451,7 @@ class WorkflowRunByIdApi(Resource):
         if app_mode != AppMode.WORKFLOW:
             raise NotWorkflowAppError()
 
-        if dify_config.BILLING_ENABLED:
+        if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD:
             billing_info = BillingService.get_info(app_model.tenant_id, exclude_vector_space=True)
             if billing_info["enabled"] and billing_info["subscription"]["plan"] == CloudPlan.SANDBOX:
                 raise WorkflowVersionExecutionNotAllowedError()

api/controllers/service_api/dataset/metadata.py

@@ -1,4 +1,4 @@
-from typing import Literal
+from typing import Literal, cast
 from uuid import UUID
 
 from flask_login import current_user
@@ -17,6 +17,7 @@ from fields.dataset_fields import (
     DatasetMetadataResponse,
 )
 from libs.helper import dump_response
+from models import Account
 from services.dataset_service import DatasetService
 from services.entities.knowledge_entities.knowledge_entities import (
     DocumentMetadataOperation,
@@ -24,6 +25,7 @@ from services.entities.knowledge_entities.knowledge_entities import (
     MetadataDetail,
     MetadataOperationData,
 )
+from services.errors.metadata import MetadataResourceNotFoundError
 from services.metadata_service import MetadataService
 
 BUILT_IN_METADATA_ACTION_PARAM = {
@@ -158,12 +160,14 @@ class DatasetMetadataServiceApi(DatasetApiResource):
 
         dataset_id_str = str(dataset_id)
         metadata_id_str = str(metadata_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, session)
+        dataset = DatasetService.get_dataset_for_tenant(dataset_id_str, tenant_id, session=session)
         if dataset is None:
             raise NotFound("Dataset not found.")
         DatasetService.check_dataset_permission(dataset, current_user, session)
 
-        metadata = MetadataService.update_metadata_name(dataset_id_str, metadata_id_str, payload.name, session=session)
+        metadata = MetadataService.update_metadata_name(
+            dataset, metadata_id_str, payload.name, cast(Account, current_user), session=session
+        )
         return dump_response(DatasetMetadataResponse, metadata), 200
 
     @service_api_ns.doc(
@@ -194,12 +198,12 @@ class DatasetMetadataServiceApi(DatasetApiResource):
         """Delete metadata."""
         dataset_id_str = str(dataset_id)
         metadata_id_str = str(metadata_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, session)
+        dataset = DatasetService.get_dataset_for_tenant(dataset_id_str, tenant_id, session=session)
         if dataset is None:
             raise NotFound("Dataset not found.")
         DatasetService.check_dataset_permission(dataset, current_user, session)
 
-        MetadataService.delete_metadata(dataset_id_str, metadata_id_str, session)
+        MetadataService.delete_metadata(dataset, metadata_id_str, session)
         return "", 204
 
 
@@ -297,7 +301,7 @@ class DocumentMetadataEditServiceApi(DatasetApiResource):
         responses={
             200: "Documents metadata updated successfully",
             401: "Unauthorized - invalid API token",
-            404: "Dataset not found",
+            404: "Dataset, document, or metadata not found",
         }
     )
     @service_api_ns.response(
@@ -309,14 +313,18 @@ class DocumentMetadataEditServiceApi(DatasetApiResource):
     @with_session
     def post(self, session: Session, tenant_id, dataset_id: UUID):
         """Update metadata for multiple documents."""
-        dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, session)
+        dataset = DatasetService.get_dataset_for_tenant(str(dataset_id), str(tenant_id), session=session)
         if dataset is None:
             raise NotFound("Dataset not found.")
         DatasetService.check_dataset_permission(dataset, current_user, session)
 
         metadata_args = MetadataOperationData.model_validate(service_api_ns.payload or {})
 
-        MetadataService.update_documents_metadata(dataset, metadata_args, session=session)
+        try:
+            MetadataService.update_documents_metadata(
+                dataset, metadata_args, cast(Account, current_user), session=session
+            )
+        except MetadataResourceNotFoundError as exc:
+            raise NotFound(str(exc)) from exc
 
         return dump_response(DatasetMetadataActionResponse, {"result": "success"}), 200

api/controllers/service_api/index.py

@@ -1,9 +1,16 @@
 from flask_restx import Resource
 
 from configs import dify_config
-from controllers.common.fields import IndexInfoResponse
 from controllers.common.schema import register_response_schema_models
 from controllers.service_api import service_api_ns
+from fields.base import ResponseModel
+
+
+class IndexInfoResponse(ResponseModel):
+    welcome: str
+    api_version: str
+    server_version: str
+
 
 register_response_schema_models(service_api_ns, IndexInfoResponse)
 
@@ -12,8 +19,8 @@ register_response_schema_models(service_api_ns, IndexInfoResponse)
 class IndexApi(Resource):
     @service_api_ns.response(200, "Success", service_api_ns.models[IndexInfoResponse.__name__])
     def get(self) -> dict[str, str]:
-        return {
-            "welcome": "Dify OpenAPI",
-            "api_version": "v1",
-            "server_version": dify_config.project.version,
-        }
+        return IndexInfoResponse(
+            welcome="Dify OpenAPI",
+            api_version="v1",
+            server_version=dify_config.project.version,
+        ).model_dump(mode="json")

api/controllers/service_api/wraps.py

@@ -22,7 +22,7 @@ from controllers.service_api.schema import (
     USER_QUERY_PARAM,
     USER_REQUIRED_ATTR,
 )
-from enums.cloud_plan import CloudPlan
+from enums import CloudPlan, DeploymentEdition
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from libs.login import current_user
@@ -186,7 +186,7 @@ def cloud_edition_billing_resource_check[**P, R](
         def decorated(*args: P.args, **kwargs: P.kwargs):
             api_token = validate_and_get_api_token(api_token_type)
             if resource == "vector_space":
-                if not dify_config.BILLING_ENABLED:
+                if dify_config.DEPLOYMENT_EDITION != DeploymentEdition.CLOUD:
                     return view(*args, **kwargs)
 
                 vector_space = FeatureService.get_vector_space(api_token.tenant_id)

api/controllers/trigger/webhook.py

@@ -7,7 +7,7 @@ from werkzeug.exceptions import NotFound, RequestEntityTooLarge
 from controllers.trigger import bp
 from core.trigger.debug.event_bus import TriggerDebugEventBus
 from core.trigger.debug.events import WebhookDebugEvent, build_webhook_pool_key
-from enums.quota_type import QuotaType
+from enums import QuotaType
 from services.errors.app import QuotaExceededError
 from services.trigger.webhook_service import RawWebhookDataDict, WebhookService
 

api/controllers/web/feature.py

@@ -2,9 +2,9 @@ from flask_restx import Resource
 
 from controllers.common.schema import register_response_schema_models
 from controllers.web import web_ns
+from extensions.ext_application_services import application_services
 from libs.helper import dump_response
 from services.entities.feature_entities import SystemFeatureModel
-from services.feature_service import FeatureService
 
 register_response_schema_models(web_ns, SystemFeatureModel)
 
@@ -30,4 +30,7 @@ class SystemFeatureApi(Resource):
 
         Authentication configuration must be available before the authentication flow can be selected.
         """
-        return dump_response(SystemFeatureModel, FeatureService.get_system_features())
+        return dump_response(
+            SystemFeatureModel,
+            application_services().feature_queries.get_system_features(),
+        )

api/controllers/web/login.py

@@ -30,6 +30,7 @@ from controllers.console.wraps import (
 )
 from controllers.web import web_ns
 from controllers.web.wraps import decode_jwt_token
+from enums import DeploymentEdition
 from extensions.ext_database import db
 from libs.helper import EmailStr, extract_remote_ip
 from libs.passport import PassportService
@@ -146,8 +147,9 @@ class LoginStatusApi(Resource):
         if not app_code:
             return LoginStatusResponse(logged_in=bool(token), app_logged_in=False).model_dump(mode="json")
         app_id = AppService.get_app_id_by_code(app_code, session=db.session())
-        is_public = not dify_config.ENTERPRISE_ENABLED or not WebAppAuthService.is_app_require_permission_check(
-            app_id=app_id, session=db.session()
+        is_public = (
+            dify_config.DEPLOYMENT_EDITION != DeploymentEdition.ENTERPRISE
+            or not WebAppAuthService.is_app_require_permission_check(app_id=app_id, session=db.session())
         )
         user_logged_in = False
 

api/controllers/web/site.py

@@ -8,7 +8,7 @@ from configs import dify_config
 from controllers.common.schema import register_response_schema_models
 from controllers.web import web_ns
 from controllers.web.wraps import WebApiResource
-from enums.deployment_edition import DeploymentEdition
+from enums import DeploymentEdition
 from extensions.ext_database import db
 from extensions.storage.storage_type import StorageType
 from fields.base import ResponseModel

api/core/app/llm/__init__.py

@@ -5,6 +5,7 @@ from .quota import (
     deduct_llm_quota_for_model,
     ensure_llm_quota_available,
     ensure_llm_quota_available_for_model,
+    reserve_llm_quota_for_model,
 )
 
 __all__ = [
@@ -12,4 +13,5 @@ __all__ = [
     "deduct_llm_quota_for_model",
     "ensure_llm_quota_available",
     "ensure_llm_quota_available_for_model",
+    "reserve_llm_quota_for_model",
 ]

api/core/app/llm/quota.py

@@ -7,6 +7,10 @@ with a non-LLM model.
 """
 
 import warnings
+from dataclasses import dataclass, field
+from enum import StrEnum, auto
+from typing import Any
+from uuid import uuid4
 
 from sqlalchemy import select
 from sqlalchemy.orm import sessionmaker
@@ -23,6 +27,68 @@ from graphon.model_runtime.entities.model_entities import ModelType
 from libs.datetime_utils import naive_utc_now
 from models.provider import Provider, ProviderType
 from models.provider_ids import ModelProviderID
+from services.credit_pool_service import CreditPoolReservation, CreditPoolService
+
+
+class LLMQuotaReservationState(StrEnum):
+    RESERVED = auto()
+    COMMITTED = auto()
+    RELEASED = auto()
+
+
+@dataclass
+class LLMQuotaReservation:
+    """Quota reserved for one system-hosted LLM invocation."""
+
+    tenant_id: str
+    provider: str
+    model: str
+    provider_configuration: Any
+    quota_unit: QuotaUnit | None = None
+    credit_pool_reservation: CreditPoolReservation | None = None
+    requires_usage: bool = False
+    _state: LLMQuotaReservationState = field(default=LLMQuotaReservationState.RESERVED, init=False, repr=False)
+
+    @property
+    def state(self) -> LLMQuotaReservationState:
+        return self._state
+
+    @property
+    def commit_before_delivery(self) -> bool:
+        return self.credit_pool_reservation is not None
+
+    def commit(self, usage: LLMUsage | None = None) -> None:
+        if self._state == LLMQuotaReservationState.COMMITTED:
+            return
+        if self._state == LLMQuotaReservationState.RELEASED:
+            raise RuntimeError("Cannot commit a released LLM quota reservation.")
+
+        if self.credit_pool_reservation is not None:
+            self.credit_pool_reservation.commit()
+        elif self.requires_usage:
+            if usage is None:
+                raise ValueError("Accurate terminal usage is required for token-based LLM quota settlement.")
+            used_quota = _resolve_llm_used_quota(
+                system_configuration=self.provider_configuration.system_configuration,
+                model=self.model,
+                usage=usage,
+            )
+            _deduct_used_llm_quota(
+                tenant_id=self.tenant_id,
+                provider=self.provider,
+                provider_configuration=self.provider_configuration,
+                used_quota=used_quota,
+            )
+
+        self._state = LLMQuotaReservationState.COMMITTED
+
+    def release(self) -> None:
+        if self._state in {LLMQuotaReservationState.COMMITTED, LLMQuotaReservationState.RELEASED}:
+            return
+
+        if self.credit_pool_reservation is not None:
+            self.credit_pool_reservation.release()
+        self._state = LLMQuotaReservationState.RELEASED
 
 
 def _get_provider_configuration(*, tenant_id: str, provider: str):
@@ -34,6 +100,67 @@ def _get_provider_configuration(*, tenant_id: str, provider: str):
     return provider_configuration
 
 
+def _get_current_quota_configuration(system_configuration):
+    return next(
+        (
+            quota_configuration
+            for quota_configuration in system_configuration.quota_configurations
+            if quota_configuration.quota_type == system_configuration.current_quota_type
+        ),
+        None,
+    )
+
+
+def reserve_llm_quota_for_model(*, tenant_id: str, provider: str, model: str) -> LLMQuotaReservation:
+    """Reserve system-hosted LLM quota before invoking the provider."""
+    provider_configuration = _get_provider_configuration(tenant_id=tenant_id, provider=provider)
+    reservation = LLMQuotaReservation(
+        tenant_id=tenant_id,
+        provider=provider,
+        model=model,
+        provider_configuration=provider_configuration,
+    )
+    if provider_configuration.using_provider_type != ProviderType.SYSTEM:
+        return reservation
+
+    provider_model = provider_configuration.get_provider_model(model_type=ModelType.LLM, model=model)
+    if provider_model and provider_model.status == ModelStatus.QUOTA_EXCEEDED:
+        raise QuotaExceededError(f"Model provider {provider} quota exceeded.")
+
+    system_configuration = provider_configuration.system_configuration
+    quota_configuration = _get_current_quota_configuration(system_configuration)
+    if quota_configuration is None or quota_configuration.quota_limit == -1:
+        return reservation
+
+    reservation.quota_unit = quota_configuration.quota_unit
+    quota_type = system_configuration.current_quota_type
+    if quota_type in {ProviderQuotaType.TRIAL, ProviderQuotaType.PAID}:
+        match quota_configuration.quota_unit:
+            case QuotaUnit.CREDITS:
+                amount = dify_config.get_model_credits(model)
+            case QuotaUnit.TIMES:
+                amount = 1
+            case QuotaUnit.TOKENS:
+                # Token usage is unknown before invocation. Enabling TOKENS for a hosted
+                # credit pool requires accurate terminal usage and an upper-bound reservation strategy.
+                raise ValueError("Token-based hosted credit pools do not support pre-invocation reservation.")
+            case _:
+                raise ValueError(f"Unsupported hosted credit pool quota unit: {quota_configuration.quota_unit}")
+
+        reservation.credit_pool_reservation = CreditPoolService.reserve_credits(
+            tenant_id=tenant_id,
+            credits_required=amount,
+            pool_type="paid" if quota_type == ProviderQuotaType.PAID else "trial",
+            request_id=str(uuid4()),
+            session_factory=db.session,
+            meta={"source": "llm.invoke", "provider": provider, "model": model},
+        )
+    elif quota_type == ProviderQuotaType.FREE:
+        reservation.requires_usage = True
+
+    return reservation
+
+
 def ensure_llm_quota_available_for_model(*, tenant_id: str, provider: str, model: str) -> None:
     """Raise when a tenant-bound LLM model is already out of quota."""
     provider_configuration = _get_provider_configuration(tenant_id=tenant_id, provider=provider)

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

@@ -1,11 +1,9 @@
 """Workflow-level GraphEngine layers that depend on outer infrastructure."""
 
-from .llm_quota import LLMQuotaLayer
 from .observability import ObservabilityLayer
 from .persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
 
 __all__ = [
-    "LLMQuotaLayer",
     "ObservabilityLayer",
     "PersistenceWorkflowInfo",
     "WorkflowPersistenceLayer",

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

@@ -1,194 +0,0 @@
-"""
-LLM quota deduction layer for GraphEngine.
-
-This layer centralizes model-quota handling outside node implementations.
-
-Graphon LLM-backed nodes expose provider/model identity through public node
-configuration and, after execution, through ``node_run_result.inputs``. Resolve
-quota billing from that public identity instead of depending on
-``ModelInstance`` reconstruction inside the workflow layer. Missing identity on
-quota-tracked nodes is treated as a workflow bug and aborts execution so quota
-handling is never silently skipped.
-"""
-
-import logging
-from typing import final, override
-
-from core.app.llm import deduct_llm_quota_for_model, ensure_llm_quota_available_for_model
-from core.errors.error import QuotaExceededError
-from graphon.enums import BuiltinNodeTypes, WorkflowNodeExecutionStatus
-from graphon.graph_engine.entities.commands import AbortCommand, CommandType
-from graphon.graph_engine.layers import GraphEngineLayer
-from graphon.graph_events import GraphEngineEvent, GraphNodeEventBase, NodeRunSucceededEvent
-from graphon.node_events import NodeRunResult
-from graphon.nodes.base.node import Node
-
-logger = logging.getLogger(__name__)
-_QUOTA_NODE_TYPES = frozenset(
-    [
-        BuiltinNodeTypes.LLM,
-        BuiltinNodeTypes.PARAMETER_EXTRACTOR,
-        BuiltinNodeTypes.QUESTION_CLASSIFIER,
-    ]
-)
-
-
-@final
-class LLMQuotaLayer(GraphEngineLayer):
-    """Graph layer that applies tenant-scoped quota checks to LLM-backed nodes."""
-
-    tenant_id: str
-    _abort_sent: bool
-
-    def __init__(self, tenant_id: str) -> None:
-        super().__init__()
-        self.tenant_id = tenant_id
-        self._abort_sent = False
-
-    @override
-    def on_graph_start(self) -> None:
-        self._abort_sent = False
-
-    @override
-    def on_event(self, event: GraphEngineEvent) -> None:
-        _ = event
-
-    @override
-    def on_graph_end(self, error: Exception | None) -> None:
-        _ = error
-
-    @override
-    def on_node_run_start(self, node: Node) -> None:
-        if self._abort_sent:
-            return
-
-        if not self._supports_quota(node):
-            return
-
-        model_identity = self._extract_model_identity_from_node(node)
-        if model_identity is None:
-            reason = "LLM quota check requires public node model identity before execution."
-            self._abort_before_node_run(node=node, reason=reason, error_type="LLMQuotaIdentityError")
-            logger.error("LLM quota handling aborted, node_id=%s, reason=%s", node.id, reason)
-            return
-
-        provider, model_name = model_identity
-        try:
-            ensure_llm_quota_available_for_model(
-                tenant_id=self.tenant_id,
-                provider=provider,
-                model=model_name,
-            )
-        except QuotaExceededError as exc:
-            self._abort_before_node_run(node=node, reason=str(exc), error_type=QuotaExceededError.__name__)
-            logger.warning("LLM quota check failed, node_id=%s, error=%s", node.id, exc)
-
-    @override
-    def on_node_run_end(
-        self, node: Node, error: Exception | None, result_event: GraphNodeEventBase | None = None
-    ) -> None:
-        if error is not None or not isinstance(result_event, NodeRunSucceededEvent) or not self._supports_quota(node):
-            return
-
-        model_identity = self._extract_model_identity_from_result_event(result_event)
-        if model_identity is None:
-            self._abort_for_missing_model_identity(
-                node=node,
-                reason="LLM quota deduction requires model identity in the node result event.",
-            )
-            return
-
-        provider, model_name = model_identity
-
-        try:
-            deduct_llm_quota_for_model(
-                tenant_id=self.tenant_id,
-                provider=provider,
-                model=model_name,
-                usage=result_event.node_run_result.llm_usage,
-            )
-        except QuotaExceededError as exc:
-            self._set_stop_event(node)
-            self._send_abort_command(reason=str(exc))
-            logger.warning("LLM quota deduction exceeded, node_id=%s, error=%s", node.id, exc)
-        except Exception:
-            logger.exception("LLM quota deduction failed, node_id=%s", node.id)
-
-    @staticmethod
-    def _set_stop_event(node: Node) -> None:
-        stop_event = getattr(node.graph_runtime_state, "stop_event", None)
-        if stop_event is not None:
-            stop_event.set()
-
-    def _abort_before_node_run(self, *, node: Node, reason: str, error_type: str) -> None:
-        self._set_stop_event(node)
-        node.node_data.error_strategy = None
-        node.node_data.retry_config.retry_enabled = False
-
-        def quota_aborted_run() -> NodeRunResult:
-            return NodeRunResult(
-                status=WorkflowNodeExecutionStatus.FAILED,
-                error=reason,
-                error_type=error_type,
-            )
-
-        # TODO: Push Graphon to expose a public pre-run failure/skip hook, then replace this private _run override.
-        node._run = quota_aborted_run  # type: ignore[method-assign]
-        self._send_abort_command(reason=reason)
-
-    def _abort_for_missing_model_identity(self, *, node: Node, reason: str) -> None:
-        self._set_stop_event(node)
-        self._send_abort_command(reason=reason)
-        logger.error("LLM quota handling aborted, node_id=%s, reason=%s", node.id, reason)
-
-    def _send_abort_command(self, *, reason: str) -> None:
-        if not self.command_channel or self._abort_sent:
-            return
-
-        try:
-            self.command_channel.send_command(
-                AbortCommand(
-                    command_type=CommandType.ABORT,
-                    reason=reason,
-                )
-            )
-            self._abort_sent = True
-        except Exception:
-            logger.exception("Failed to send quota abort command")
-
-    @staticmethod
-    def _supports_quota(node: Node) -> bool:
-        return node.node_type in _QUOTA_NODE_TYPES
-
-    @staticmethod
-    def _extract_model_identity_from_result_event(result_event: NodeRunSucceededEvent) -> tuple[str, str] | None:
-        provider = result_event.node_run_result.inputs.get("model_provider")
-        model_name = result_event.node_run_result.inputs.get("model_name")
-        if isinstance(provider, str) and provider and isinstance(model_name, str) and model_name:
-            return provider, model_name
-        return None
-
-    @staticmethod
-    def _extract_model_identity_from_node(node: Node) -> tuple[str, str] | None:
-        node_data = getattr(node, "node_data", None)
-        if node_data is None:
-            node_data = getattr(node, "data", None)
-
-        model_config = getattr(node_data, "model", None)
-        if model_config is None:
-            logger.warning(
-                "LLMQuotaLayer skipped quota handling because node model config is missing, node_id=%s",
-                node.id,
-            )
-            return None
-
-        provider = getattr(model_config, "provider", None)
-        model_name = getattr(model_config, "name", None)
-        if isinstance(provider, str) and provider and isinstance(model_name, str) and model_name:
-            return provider, model_name
-
-        logger.warning(
-            "LLMQuotaLayer skipped quota handling because node model identity is invalid, node_id=%s",
-            node.id,
-        )
-        return None

api/core/entities/parameter_entities.py

@@ -16,6 +16,8 @@ class CommonParameterType(StrEnum):
     TOOLS_SELECTOR = "array[tools]"
     CHECKBOX = "checkbox"
     ANY = auto()
+    DATE = "date"
+    DATE_RANGE = "date-range"
 
     # Dynamic select parameter
     # Once you are not sure about the available options until authorization is done

api/core/entities/provider_configuration.py

@@ -1857,10 +1857,10 @@ class ProviderConfiguration(BaseModel):
                             )
                         )
 
-            # if llm name not in restricted llm list, remove it
+            # Hosted allowlists currently use exact model names across model types.
             restrict_model_names = [rm.model for rm in restrict_models]
             for provider_model in provider_models:
-                if provider_model.model_type == ModelType.LLM and provider_model.model not in restrict_model_names:
+                if provider_model.model not in restrict_model_names:
                     provider_model.status = ModelStatus.NO_PERMISSION
                 elif not quota_configuration.is_valid:
                     provider_model.status = ModelStatus.QUOTA_EXCEEDED

api/core/hosting_configuration.py

@@ -6,7 +6,7 @@ from pydantic import BaseModel
 from configs import dify_config
 from core.entities import DEFAULT_PLUGIN_ID
 from core.entities.provider_entities import ProviderQuotaType, QuotaUnit, RestrictModel
-from enums.deployment_edition import DeploymentEdition
+from enums import DeploymentEdition
 from graphon.model_runtime.entities.model_entities import ModelType
 
 

api/core/indexing_runner.py

@@ -34,6 +34,7 @@ from core.rag.splitter.fixed_text_splitter import (
 )
 from core.rag.splitter.text_splitter import TextSplitter
 from core.tools.utils.web_reader_tool import get_image_upload_file_ids
+from enums import DeploymentEdition
 from extensions.ext_redis import redis_client
 from extensions.ext_storage import storage
 from graphon.model_runtime.entities.model_entities import ModelType
@@ -330,7 +331,7 @@ class IndexingRunner:
         Estimate the indexing for the document.
         """
         # check document limit
-        if dify_config.BILLING_ENABLED:
+        if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD:
             count = len(extract_settings)
             batch_upload_limit = dify_config.BATCH_UPLOAD_LIMIT
             if count > batch_upload_limit:

api/core/model_manager.py

@@ -1,19 +1,19 @@
 import logging
 from collections.abc import Callable, Generator, Iterable, Mapping, Sequence
 from copy import deepcopy
-from typing import IO, Any, Literal, Optional, ParamSpec, TypeVar, Union, cast, overload
+from typing import IO, Any, Literal, Optional, ParamSpec, TypeVar, Union, cast, overload, override
 
 from configs import dify_config
 from core.entities import PluginCredentialType
 from core.entities.embedding_type import EmbeddingInputType
 from core.entities.provider_configuration import ProviderConfiguration, ProviderModelBundle
 from core.entities.provider_entities import ModelLoadBalancingConfiguration
-from core.errors.error import ProviderTokenNotInitError
+from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError
 from core.plugin.impl.model_runtime_factory import create_plugin_provider_manager
 from core.provider_manager import ProviderManager
 from extensions.ext_redis import redis_client
 from graphon.model_runtime.callbacks.base_callback import Callback
-from graphon.model_runtime.entities.llm_entities import LLMResult
+from graphon.model_runtime.entities.llm_entities import LLMResult, LLMUsage
 from graphon.model_runtime.entities.message_entities import PromptMessage, PromptMessageTool
 from graphon.model_runtime.entities.model_entities import AIModelEntity, ModelFeature, ModelType
 from graphon.model_runtime.entities.rerank_entities import MultimodalRerankInput, RerankResult
@@ -442,6 +442,149 @@ class ModelInstance:
         )
 
 
+class QuotaManagedModelInstance(ModelInstance):
+    """A system-hosted LLM instance that owns quota settlement per invocation."""
+
+    def reserve_quota(self):
+        from core.app.llm.quota import reserve_llm_quota_for_model
+
+        return reserve_llm_quota_for_model(
+            tenant_id=self.provider_model_bundle.configuration.tenant_id,
+            provider=self.provider,
+            model=self.model_name,
+        )
+
+    @staticmethod
+    def release_quota_safely(reservation) -> None:
+        try:
+            reservation.release()
+        except Exception:
+            logger.exception("Failed to release LLM quota reservation")
+
+    @overload
+    def invoke_llm(
+        self,
+        prompt_messages: Sequence[PromptMessage],
+        model_parameters: dict[str, Any] | None = None,
+        tools: Sequence[PromptMessageTool] | None = None,
+        stop: list[str] | None = None,
+        stream: Literal[True] = True,
+        callbacks: list[Callback] | None = None,
+        request_metadata: Mapping[str, object] | None = None,
+    ) -> Generator: ...
+
+    @overload
+    def invoke_llm(
+        self,
+        prompt_messages: list[PromptMessage],
+        model_parameters: dict[str, Any] | None = None,
+        tools: Sequence[PromptMessageTool] | None = None,
+        stop: list[str] | None = None,
+        stream: Literal[False] = False,
+        callbacks: list[Callback] | None = None,
+        request_metadata: Mapping[str, object] | None = None,
+    ) -> LLMResult: ...
+
+    @overload
+    def invoke_llm(
+        self,
+        prompt_messages: list[PromptMessage],
+        model_parameters: dict[str, Any] | None = None,
+        tools: Sequence[PromptMessageTool] | None = None,
+        stop: list[str] | None = None,
+        stream: bool = True,
+        callbacks: list[Callback] | None = None,
+        request_metadata: Mapping[str, object] | None = None,
+    ) -> Union[LLMResult, Generator]: ...
+
+    @override
+    def invoke_llm(
+        self,
+        prompt_messages: Sequence[PromptMessage],
+        model_parameters: dict[str, Any] | None = None,
+        tools: Sequence[PromptMessageTool] | None = None,
+        stop: Sequence[str] | None = None,
+        stream: bool = True,
+        callbacks: list[Callback] | None = None,
+        request_metadata: Mapping[str, object] | None = None,
+    ) -> Union[LLMResult, Generator]:
+        normalized_prompt_messages = list(prompt_messages)
+        normalized_stop = list(stop) if stop else None
+        if stream:
+            return self._invoke_llm_stream(
+                prompt_messages=normalized_prompt_messages,
+                model_parameters=model_parameters,
+                tools=tools,
+                stop=normalized_stop,
+                callbacks=callbacks,
+                request_metadata=request_metadata,
+            )
+
+        reservation = self.reserve_quota()
+        try:
+            response = super().invoke_llm(
+                prompt_messages=normalized_prompt_messages,
+                model_parameters=model_parameters,
+                tools=tools,
+                stop=normalized_stop,
+                stream=False,
+                callbacks=callbacks,
+                request_metadata=request_metadata,
+            )
+            if isinstance(response, Generator):
+                raise TypeError("Non-streaming LLM invocation returned a generator.")
+            reservation.commit(response.usage)
+            return response
+        finally:
+            self.release_quota_safely(reservation)
+
+    def _invoke_llm_stream(
+        self,
+        *,
+        prompt_messages: list[PromptMessage],
+        model_parameters: dict[str, Any] | None,
+        tools: Sequence[PromptMessageTool] | None,
+        stop: list[str] | None,
+        callbacks: list[Callback] | None,
+        request_metadata: Mapping[str, object] | None,
+    ) -> Generator:
+        reservation = self.reserve_quota()
+        usage: LLMUsage | None = None
+        try:
+            response = super().invoke_llm(
+                prompt_messages=prompt_messages,
+                model_parameters=model_parameters,
+                tools=tools,
+                stop=stop,
+                stream=True,
+                callbacks=callbacks,
+                request_metadata=request_metadata,
+            )
+            if not isinstance(response, Generator):
+                raise TypeError("Streaming LLM invocation did not return a generator.")
+
+            if reservation.commit_before_delivery:
+                for chunk in response:
+                    chunk_usage = chunk.delta.usage
+                    if chunk_usage is not None:
+                        usage = chunk_usage
+                    reservation.commit(usage)
+                    yield chunk
+                return
+
+            buffered_chunks = []
+            for chunk in response:
+                chunk_usage = chunk.delta.usage
+                if chunk_usage is not None:
+                    usage = chunk_usage
+                buffered_chunks.append(chunk)
+
+            reservation.commit(usage)
+            yield from buffered_chunks
+        finally:
+            self.release_quota_safely(reservation)
+
+
 class ModelManager:
     """Resolves :class:`ModelInstance` objects for a tenant and provider.
 
@@ -472,6 +615,43 @@ class ModelManager:
     def for_tenant(cls, tenant_id: str, user_id: str | None = None) -> "ModelManager":
         return cls(provider_manager=create_plugin_provider_manager(tenant_id=tenant_id, user_id=user_id))
 
+    @staticmethod
+    def _validate_system_model_access(
+        provider_model_bundle: ProviderModelBundle,
+        *,
+        model_type: ModelType,
+        model: str,
+    ) -> None:
+        configuration = provider_model_bundle.configuration
+        if configuration.using_provider_type != ProviderType.SYSTEM:
+            return
+
+        # Hosted allowlists retain the existing comma-separated format. Model names
+        # are matched exactly; model-type-specific entries will be introduced later.
+        quota_configuration = next(
+            (
+                quota
+                for quota in configuration.system_configuration.quota_configurations
+                if quota.quota_type == configuration.system_configuration.current_quota_type
+            ),
+            None,
+        )
+        if quota_configuration is None or not quota_configuration.restrict_models:
+            return
+        if any(restricted_model.model == model for restricted_model in quota_configuration.restrict_models):
+            return
+
+        raise ModelCurrentlyNotSupportError(f"System model {model_type.value}/{model} is not allowed.")
+
+    @staticmethod
+    def _model_instance_class(provider_model_bundle: ProviderModelBundle, model_type: ModelType) -> type[ModelInstance]:
+        if (
+            model_type == ModelType.LLM
+            and provider_model_bundle.configuration.using_provider_type == ProviderType.SYSTEM
+        ):
+            return QuotaManagedModelInstance
+        return ModelInstance
+
     def get_model_instance(
         self,
         tenant_id: str,
@@ -493,17 +673,19 @@ class ModelManager:
         provider_model_bundle = self._provider_manager.get_provider_model_bundle(
             tenant_id=tenant_id, provider=provider, model_type=model_type
         )
+        self._validate_system_model_access(provider_model_bundle, model_type=model_type, model=model)
+        model_instance_class = self._model_instance_class(provider_model_bundle, model_type)
 
         cred_cache_key = (tenant_id, provider, model_type.value, model)
 
         if cred_cache_key in self._credentials_cache:
-            return ModelInstance(
+            return model_instance_class(
                 provider_model_bundle,
                 model,
                 deepcopy(self._credentials_cache[cred_cache_key]),
             )
 
-        ret = ModelInstance(provider_model_bundle, model)
+        ret = model_instance_class(provider_model_bundle, model)
         if self._enable_credentials_cache:
             self._credentials_cache[cred_cache_key] = deepcopy(ret.credentials)
         return ret

api/core/plugin/backwards_invocation/model.py

@@ -3,7 +3,6 @@ from binascii import hexlify, unhexlify
 from collections.abc import Generator
 from typing import Any
 
-from core.app.llm import deduct_llm_quota
 from core.llm_generator.output_parser.structured_output import invoke_llm_with_structured_output
 from core.model_manager import ModelManager
 from core.plugin.backwards_invocation.base import BaseBackwardsInvocation
@@ -80,15 +79,11 @@ class PluginModelBackwardsInvocation(BaseBackwardsInvocation):
 
             def handle() -> Generator[LLMResultChunk, None, None]:
                 for chunk in response:
-                    if chunk.delta.usage:
-                        deduct_llm_quota(tenant_id=tenant.id, model_instance=model_instance, usage=chunk.delta.usage)
                     chunk.prompt_messages = []
                     yield chunk
 
             return handle()
         else:
-            if response.usage:
-                deduct_llm_quota(tenant_id=tenant.id, model_instance=model_instance, usage=response.usage)
 
             def handle_non_streaming(response: LLMResult) -> Generator[LLMResultChunk, None, None]:
                 yield LLMResultChunk(
@@ -141,15 +136,11 @@ class PluginModelBackwardsInvocation(BaseBackwardsInvocation):
 
             def handle() -> Generator[LLMResultChunkWithStructuredOutput, None, None]:
                 for chunk in response:
-                    if chunk.delta.usage:
-                        deduct_llm_quota(tenant_id=tenant.id, model_instance=model_instance, usage=chunk.delta.usage)
                     chunk.prompt_messages = []
                     yield chunk
 
             return handle()
         else:
-            if response.usage:
-                deduct_llm_quota(tenant_id=tenant.id, model_instance=model_instance, usage=response.usage)
 
             def handle_non_streaming(
                 response: LLMResultWithStructuredOutput,

api/core/plugin/entities/parameters.py

@@ -1,4 +1,5 @@
 import json
+from datetime import date
 from enum import StrEnum, auto
 from typing import Any, Union
 
@@ -46,6 +47,8 @@ class PluginParameterType(StrEnum):
     # MCP object and array type parameters
     ARRAY = CommonParameterType.ARRAY
     OBJECT = CommonParameterType.OBJECT
+    DATE = CommonParameterType.DATE
+    DATE_RANGE = CommonParameterType.DATE_RANGE
 
 
 class MCPServerParameterType(StrEnum):
@@ -92,15 +95,32 @@ class PluginParameter(BaseModel):
 
 
 def as_normal_type(typ: StrEnum):
+    if typ.value == PluginParameterType.DATE_RANGE:
+        return "object"
     if typ.value in {
         PluginParameterType.SECRET_INPUT,
         PluginParameterType.SELECT,
         PluginParameterType.CHECKBOX,
+        PluginParameterType.DATE,
     }:
         return "string"
     return typ.value
 
 
+def _validate_date(value: Any, name: str = "date") -> str:
+    if not isinstance(value, str):
+        raise ValueError(f"The {name} parameter must be a string in YYYY-MM-DD format.")
+
+    try:
+        parsed = date.fromisoformat(value)
+    except ValueError as exc:
+        raise ValueError(f"The {name} parameter must be a valid date in YYYY-MM-DD format.") from exc
+
+    if parsed.isoformat() != value:
+        raise ValueError(f"The {name} parameter must use YYYY-MM-DD format.")
+    return value
+
+
 def cast_parameter_value(typ: StrEnum, value: Any, /):
     try:
         match typ.value:
@@ -115,6 +135,34 @@ def cast_parameter_value(typ: StrEnum, value: Any, /):
                     return ""
                 else:
                     return value if isinstance(value, str) else str(value)
+            case PluginParameterType.DATE:
+                if value is None or value == "":
+                    return ""
+                return _validate_date(value)
+            case PluginParameterType.DATE_RANGE:
+                if value is None or value == "":
+                    return {}
+                if isinstance(value, dict):
+                    out: dict[str, str] = {}
+                    for key in ("start", "end"):
+                        if key not in value or value[key] is None or value[key] == "":
+                            continue
+                        out[key] = _validate_date(value[key], f"date-range {key}")
+                    if out.get("start") and out.get("end") and out["start"] > out["end"]:
+                        raise ValueError("The date-range start date must not be after the end date.")
+                    return out
+                if isinstance(value, str):
+                    try:
+                        parsed_value = json.loads(value)
+                        if isinstance(parsed_value, dict):
+                            return cast_parameter_value(typ, parsed_value)
+                    except json.JSONDecodeError:
+                        pass
+                    stripped = value.strip()
+                    if not stripped:
+                        return {}
+                    return {"start": _validate_date(stripped, "date-range start")}
+                raise ValueError("The date-range parameter must be a JSON object, JSON string, or empty.")
 
             case PluginParameterType.BOOLEAN:
                 match value:

api/core/plugin/plugin_service.py

@@ -58,6 +58,7 @@ from core.plugin.impl.debugging import PluginDebuggingClient
 from core.plugin.impl.endpoint import PluginEndpointClient
 from core.plugin.impl.model import PluginModelClient
 from core.plugin.impl.plugin import PluginInstaller
+from enums import DeploymentEdition
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from models.provider import Provider, ProviderCredential, TenantPreferredModelProvider
@@ -1256,7 +1257,7 @@ class PluginService:
                 PluginService.invalidate_plugin_model_providers_cache(tenant_id)
             return result
 
-        if dify_config.ENTERPRISE_ENABLED:
+        if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.ENTERPRISE:
             PluginManagerService.try_pre_uninstall_plugin(
                 PreUninstallPluginRequest(
                     tenant_id=tenant_id,

api/core/provider_manager.py

@@ -36,7 +36,7 @@ from core.helper.model_provider_cache import ProviderCredentialsCache, ProviderC
 from core.helper.position_helper import is_filtered
 from core.plugin.entities.plugin import PluginInstallationSource
 from core.plugin.entities.plugin_daemon import PluginModelProviderDeclaration
-from enums.deployment_edition import DeploymentEdition
+from enums import DeploymentEdition
 from extensions import ext_hosting_provider
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client

api/core/rag/retrieval/dataset_retrieval.py

@@ -9,6 +9,7 @@ from collections.abc import Generator, Mapping
 from typing import Any, Union, cast
 
 from flask import Flask, current_app
+from opentelemetry.trace import get_current_span
 from sqlalchemy import and_, func, literal, or_, select, update
 from sqlalchemy.orm import Session, sessionmaker
 
@@ -650,16 +651,19 @@ class DatasetRetrieval:
         self._record_usage(router_usage)
         timer = None
         if dataset_id:
-            # get retrieval model config
-            dataset_stmt = select(Dataset).where(Dataset.id == dataset_id)
-            selected_dataset = session.scalar(dataset_stmt)
+            allowed_dataset = next((dataset for dataset in available_datasets if dataset.id == dataset_id), None)
+            selected_dataset = (
+                session.scalar(select(Dataset).where(Dataset.id == allowed_dataset.id, Dataset.tenant_id == tenant_id))
+                if allowed_dataset
+                else None
+            )
             if selected_dataset:
                 results = []
                 if selected_dataset.provider == "external":
                     external_documents = ExternalDatasetService.fetch_external_knowledge_retrieval(
                         session=session,
                         tenant_id=selected_dataset.tenant_id,
-                        dataset_id=dataset_id,
+                        dataset_id=selected_dataset.id,
                         query=query,
                         external_retrieval_parameters=selected_dataset.retrieval_model,
                         metadata_condition=metadata_condition,
@@ -673,7 +677,7 @@ class DatasetRetrieval:
                         if document.metadata is not None:
                             document.metadata["score"] = external_document.get("score")
                             document.metadata["title"] = external_document.get("title")
-                            document.metadata["dataset_id"] = dataset_id
+                            document.metadata["dataset_id"] = selected_dataset.id
                             document.metadata["dataset_name"] = selected_dataset.name
                         results.append(document)
                 else:
@@ -723,7 +727,7 @@ class DatasetRetrieval:
                             weights=retrieval_model_config.get("weights", None),
                             document_ids_filter=document_ids_filter,
                         )
-                self._on_query(query, None, [dataset_id], app_id, user_from, user_id)
+                self._on_query(query, None, [selected_dataset.id], app_id, user_from, user_id)
 
                 if results:
                     thread = threading.Thread(
@@ -1204,8 +1208,9 @@ class DatasetRetrieval:
         attachment_ids: list[str] | None,
         cancel_event: threading.Event | None,
         thread_exceptions: list[Exception] | None,
+        skip_on_error: bool = False,
     ) -> None:
-        """Collect errors only after they pass through the traced retrieval method."""
+        """Collect errors after tracing, or skip dataset-level failures when requested."""
         try:
             self._run_retriever_thread(
                 flask_app=flask_app,
@@ -1218,6 +1223,25 @@ class DatasetRetrieval:
                 attachment_ids=attachment_ids,
             )
         except Exception as exc:
+            if skip_on_error:
+                logger.warning(
+                    "Skipping dataset retrieval because retriever failed, dataset_id=%s, error_type=%s, error=%s",
+                    dataset_id,
+                    type(exc).__name__,
+                    str(exc),
+                )
+                span = get_current_span()
+                if span and span.is_recording():
+                    span.add_event(
+                        "dataset_retrieval.skipped",
+                        attributes={
+                            "dataset_id": dataset_id,
+                            "error.type": type(exc).__name__,
+                            "error.message": str(exc),
+                        },
+                    )
+                return
+
             if cancel_event:
                 cancel_event.set()
             if thread_exceptions is not None:
@@ -1878,6 +1902,7 @@ class DatasetRetrieval:
                             "attachment_ids": [attachment_id] if attachment_id else None,
                             "cancel_event": cancel_event,
                             "thread_exceptions": retrieval_thread_exceptions,
+                            "skip_on_error": True,
                         },
                     )
                     threads.append(retrieval_thread)

api/core/rag/retrieval/router/multi_dataset_react_route.py

@@ -2,7 +2,6 @@ from collections.abc import Generator, Sequence
 from typing import Any, Union
 
 from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
-from core.app.llm import deduct_llm_quota
 from core.model_manager import ModelInstance, ModelManager
 from core.prompt.advanced_prompt_transform import AdvancedPromptTransform
 from core.prompt.entities.advanced_prompt_entities import ChatModelMessage, CompletionModelPromptTemplate
@@ -168,9 +167,6 @@ class ReactMultiDatasetRouter:
         # handle invoke result
         text, usage = self._handle_invoke_result(invoke_result=invoke_result)
 
-        # deduct quota
-        deduct_llm_quota(tenant_id=tenant_id, model_instance=bound_model_instance, usage=usage)
-
         return text, usage
 
     def _handle_invoke_result(self, invoke_result: Generator) -> tuple[str, LLMUsage]:

api/core/tools/builtin_tool/providers/audio/audio.yaml

@@ -3,6 +3,7 @@ identity:
   name: audio
   label:
     en_US: Audio
+    zh_Hans: 音频
   description:
     en_US: A tool for tts and asr.
     zh_Hans: 一个用于文本转语音和语音转文本的工具。

api/core/tools/builtin_tool/providers/audio/tools/asr.yaml

@@ -3,6 +3,7 @@ identity:
   author: hjlarry
   label:
     en_US: Speech To Text
+    zh_Hans: 语音转文本
 description:
   human:
     en_US: Convert audio file to text.

api/core/tools/builtin_tool/providers/audio/tools/tts.yaml

@@ -3,6 +3,7 @@ identity:
   author: hjlarry
   label:
     en_US: Text To Speech
+    zh_Hans: 文本转语音
 description:
   human:
     en_US: Convert text to audio file.

api/core/tools/entities/tool_entities.py

@@ -311,6 +311,8 @@ class ToolParameter(PluginParameter):
         MODEL_SELECTOR = PluginParameterType.MODEL_SELECTOR
         ANY = PluginParameterType.ANY
         DYNAMIC_SELECT = PluginParameterType.DYNAMIC_SELECT
+        DATE = PluginParameterType.DATE
+        DATE_RANGE = PluginParameterType.DATE_RANGE
 
         # MCP object and array type parameters
         ARRAY = MCPServerParameterType.ARRAY

api/core/tools/mcp_tool/tool.py

@@ -25,6 +25,7 @@ from core.tools.__base.tool import Tool
 from core.tools.__base.tool_runtime import ToolRuntime
 from core.tools.entities.tool_entities import ToolEntity, ToolInvokeMessage, ToolProviderType
 from core.tools.errors import ToolInvokeError
+from enums import DeploymentEdition
 from graphon.model_runtime.entities.llm_entities import LLMUsage, LLMUsageMetadata
 
 logger = logging.getLogger(__name__)
@@ -272,7 +273,7 @@ class MCPTool(Tool):
         the deployment actually has the enterprise side that can mint tokens.
         Non-enterprise installs treat the DB value as a no-op — a stale row
         won't trigger a 5xx against a missing inner-API endpoint."""
-        return self.identity_mode != IdentityMode.OFF and dify_config.ENTERPRISE_ENABLED
+        return self.identity_mode != IdentityMode.OFF and dify_config.DEPLOYMENT_EDITION == DeploymentEdition.ENTERPRISE
 
     def invoke_remote_mcp_tool(
         self,

api/core/workflow/llm_node.py

@@ -0,0 +1,44 @@
+from collections.abc import Callable, Generator, Sequence
+from typing import Any, override
+
+from graphon.model_runtime.entities.llm_entities import LLMStructuredOutput
+from graphon.model_runtime.entities.message_entities import PromptMessage
+from graphon.node_events.base import NodeEventBase
+from graphon.nodes.llm.node import LLMNode
+from graphon.nodes.llm.runtime_protocols import LLMPollingCapableProtocol
+
+
+# TODO: Remove this Dify-specific node once graphon exposes a polling finalization hook.
+class DifyLLMNode(LLMNode):
+    """Dify-owned LLM node lifecycle extensions."""
+
+    @classmethod
+    @override
+    def version(cls) -> str:
+        return "1"
+
+    def __init__(
+        self,
+        *args: Any,
+        polling_finalizer: Callable[[], None],
+        **kwargs: Any,
+    ) -> None:
+        super().__init__(*args, **kwargs)
+        self._polling_finalizer = polling_finalizer
+
+    @override
+    def _invoke_llm_with_polling(
+        self,
+        *,
+        polling_model: LLMPollingCapableProtocol,
+        prompt_messages: Sequence[PromptMessage],
+        stop: Sequence[str] | None,
+    ) -> Generator[NodeEventBase | LLMStructuredOutput, None, None]:
+        try:
+            yield from super()._invoke_llm_with_polling(
+                polling_model=polling_model,
+                prompt_messages=prompt_messages,
+                stop=stop,
+            )
+        finally:
+            self._polling_finalizer()

api/core/workflow/node_factory.py

@@ -27,6 +27,7 @@ from core.workflow.llm_environment_variable import (
     resolve_llm_model_config,
     should_resolve_llm_model_selector,
 )
+from core.workflow.llm_node import DifyLLMNode
 from core.workflow.node_runtime import (
     DifyFileReferenceFactory,
     DifyHumanInputNodeRuntime,
@@ -493,6 +494,8 @@ class DifyNodeFactory(NodeFactory):
 
     @staticmethod
     def _resolve_node_class(*, node_type: NodeType, node_version: str) -> type[Node]:
+        if node_type == BuiltinNodeTypes.LLM:
+            return DifyLLMNode
         return resolve_workflow_node_class(node_type=node_type, node_version=node_version)
 
     def _resolve_llm_model_reference(self, node_data: LLMNodeData) -> LLMNodeData:
@@ -586,18 +589,19 @@ class DifyNodeFactory(NodeFactory):
     ) -> dict[str, object]:
         validated_node_data = cast(LLMCompatibleNodeData, node_data)
         model_instance = self._build_model_instance_for_llm_node(validated_node_data)
+        node_model_instance = (
+            self._wrap_model_instance_for_node(
+                node_data=validated_node_data,
+                model_instance=model_instance,
+                request_metadata={"app_id": self._dify_context.app_id},
+            )
+            if wrap_model_instance
+            else model_instance
+        )
         node_init_kwargs: dict[str, object] = {
             "credentials_provider": self._llm_credentials_provider,
             "model_factory": self._llm_model_factory,
-            "model_instance": (
-                self._wrap_model_instance_for_node(
-                    node_data=validated_node_data,
-                    model_instance=model_instance,
-                    request_metadata={"app_id": self._dify_context.app_id},
-                )
-                if wrap_model_instance
-                else model_instance
-            ),
+            "model_instance": node_model_instance,
             "memory": self._build_memory_for_llm_node(
                 node_data=validated_node_data,
                 model_instance=model_instance,
@@ -619,6 +623,7 @@ class DifyNodeFactory(NodeFactory):
             node_init_kwargs["jinja2_template_renderer"] = self._jinja2_template_renderer
         if validated_node_data.type == BuiltinNodeTypes.LLM:
             node_init_kwargs["default_query_selector"] = system_variable_selector(SystemVariableKey.QUERY)
+            node_init_kwargs["polling_finalizer"] = cast(DifyPreparedLLM, node_model_instance).finalize_llm_polling
         return node_init_kwargs
 
     @staticmethod

api/core/workflow/node_runtime.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 from collections.abc import Callable, Generator, Mapping, Sequence
 from dataclasses import dataclass
 from enum import Enum
-from typing import TYPE_CHECKING, Any, Literal, cast, overload, override
+from typing import TYPE_CHECKING, Any, Literal, Protocol, cast, overload, override
 
 from pydantic import JsonValue
 from sqlalchemy import select
@@ -20,7 +20,7 @@ from core.db.session_factory import session_factory
 from core.helper.trace_id_helper import ParentTraceContext
 from core.llm_generator.output_parser.errors import OutputParserError
 from core.llm_generator.output_parser.structured_output import invoke_llm_with_structured_output
-from core.model_manager import ModelInstance
+from core.model_manager import ModelInstance, QuotaManagedModelInstance
 from core.plugin.impl.exc import PluginDaemonClientSideError, PluginInvokeError
 from core.plugin.impl.plugin import PluginInstaller
 from core.prompt.utils.prompt_message_util import PromptMessageUtil
@@ -49,6 +49,7 @@ from graphon.file import File, FileTransferMethod, FileType
 from graphon.model_runtime.entities import LLMMode
 from graphon.model_runtime.entities.llm_entities import (
     LLMPollingResult,
+    LLMPollingStatus,
     LLMResult,
     LLMResultChunk,
     LLMResultChunkWithStructuredOutput,
@@ -87,6 +88,33 @@ from .human_input_adapter import (
 )
 from .system_variables import SystemVariableKey, get_system_text
 
+
+class PollingLLMRuntimeProtocol(Protocol):
+    """Runtime capability required by the workflow polling adapter."""
+
+    def start_llm_polling(
+        self,
+        *,
+        provider: str,
+        model: str,
+        credentials: dict[str, Any],
+        model_parameters: dict[str, Any],
+        prompt_messages: Sequence[PromptMessage],
+        tools: Sequence[PromptMessageTool] | None,
+        stop: Sequence[str] | None,
+        json_schema: dict[str, Any] | None,
+    ) -> LLMPollingResult: ...
+
+    def check_llm_polling(
+        self,
+        *,
+        provider: str,
+        model: str,
+        credentials: dict[str, Any],
+        plugin_state: dict[str, JsonValue],
+    ) -> LLMPollingResult: ...
+
+
 if TYPE_CHECKING:
     from core.tools.__base.tool import Tool
     from core.tools.entities.tool_entities import ToolInvokeMessage as CoreToolInvokeMessage
@@ -281,23 +309,45 @@ class DifyPreparedLLM(LLMProtocol):
     def is_structured_output_parse_error(self, error: Exception) -> bool:
         return isinstance(error, OutputParserError)
 
+    def finalize_llm_polling(self) -> None:
+        """Finalize resources held by a polling invocation, if any."""
+
 
 class DifyPreparedPollingLLM(DifyPreparedLLM, LLMPollingCapableProtocol):
     """Prepared workflow LLM adapter that exposes Graphon's polling protocol."""
 
     def __init__(self, model_instance: ModelInstance, request_metadata: Mapping[str, object] | None = None) -> None:
-        from core.plugin.impl.model_runtime import PluginModelRuntime
-
         super().__init__(model_instance, request_metadata=request_metadata)
-        model_type_instance = model_instance.model_type_instance
-        if not isinstance(model_type_instance, LargeLanguageModel):
-            raise TypeError("Polling wrapper requires a large-language-model instance.")
+        model_type_instance = cast(LargeLanguageModel, model_instance.model_type_instance)
+        self._polling_runtime = cast(PollingLLMRuntimeProtocol, model_type_instance.model_runtime)
+        self._polling_quota_reservation = None
 
-        plugin_model_runtime = model_type_instance.model_runtime
-        if not isinstance(plugin_model_runtime, PluginModelRuntime):
-            raise TypeError("Polling wrapper requires a plugin-backed model runtime.")
+    @override
+    def finalize_llm_polling(self) -> None:
+        reservation = self._polling_quota_reservation
+        self._polling_quota_reservation = None
+        if reservation is not None:
+            QuotaManagedModelInstance.release_quota_safely(reservation)
 
-        self._plugin_model_runtime = plugin_model_runtime
+    def _settle_polling_quota(self, polling_result: LLMPollingResult) -> LLMPollingResult:
+        reservation = self._polling_quota_reservation
+        if reservation is None or polling_result.status == LLMPollingStatus.RUNNING:
+            return polling_result
+
+        try:
+            if polling_result.status == LLMPollingStatus.SUCCEEDED:
+                if polling_result.result is None:
+                    raise ValueError("A successful LLM polling result must include a model result.")
+                reservation.commit(polling_result.result.usage)
+            else:
+                reservation.release()
+        except Exception:
+            QuotaManagedModelInstance.release_quota_safely(reservation)
+            raise
+        finally:
+            self._polling_quota_reservation = None
+
+        return polling_result
 
     @override
     def start_llm_polling(
@@ -309,16 +359,26 @@ class DifyPreparedPollingLLM(DifyPreparedLLM, LLMPollingCapableProtocol):
         stop: Sequence[str] | None,
         json_schema: Mapping[str, Any] | None,
     ) -> LLMPollingResult:
-        return self._plugin_model_runtime.start_llm_polling(
-            provider=self.provider,
-            model=self.model_name,
-            credentials=self._model_instance.credentials,
-            prompt_messages=prompt_messages,
-            model_parameters=dict(model_parameters),
-            tools=tools,
-            stop=stop,
-            json_schema=dict(json_schema) if json_schema is not None else None,
-        )
+        self.finalize_llm_polling()
+
+        if isinstance(self._model_instance, QuotaManagedModelInstance):
+            self._polling_quota_reservation = self._model_instance.reserve_quota()
+
+        try:
+            polling_result = self._polling_runtime.start_llm_polling(
+                provider=self.provider,
+                model=self.model_name,
+                credentials=self._model_instance.credentials,
+                prompt_messages=prompt_messages,
+                model_parameters=dict(model_parameters),
+                tools=tools,
+                stop=stop,
+                json_schema=dict(json_schema) if json_schema is not None else None,
+            )
+            return self._settle_polling_quota(polling_result)
+        except Exception:
+            self.finalize_llm_polling()
+            raise
 
     @override
     def check_llm_polling(
@@ -326,12 +386,17 @@ class DifyPreparedPollingLLM(DifyPreparedLLM, LLMPollingCapableProtocol):
         *,
         plugin_state: Mapping[str, JsonValue],
     ) -> LLMPollingResult:
-        return self._plugin_model_runtime.check_llm_polling(
-            provider=self.provider,
-            model=self.model_name,
-            credentials=self._model_instance.credentials,
-            plugin_state=dict(plugin_state),
-        )
+        try:
+            polling_result = self._polling_runtime.check_llm_polling(
+                provider=self.provider,
+                model=self.model_name,
+                credentials=self._model_instance.credentials,
+                plugin_state=dict(plugin_state),
+            )
+            return self._settle_polling_quota(polling_result)
+        except Exception:
+            self.finalize_llm_polling()
+            raise
 
 
 class DifyPromptMessageSerializer(PromptMessageSerializerProtocol):

api/core/workflow/workflow_entry.py

@@ -9,7 +9,6 @@ from context import capture_current_context
 from core.app.apps.exc import GenerateTaskStoppedError
 from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom, build_dify_run_context
 from core.app.file_access import DatabaseFileAccessController
-from core.app.workflow.layers.llm_quota import LLMQuotaLayer
 from core.app.workflow.layers.observability import ObservabilityLayer
 from core.workflow.node_factory import (
     DifyGraphInitContext,
@@ -34,12 +33,12 @@ from graphon.filters import GraphEventFilterContext, ResponseStreamFilter, filte
 from graphon.graph import Graph
 from graphon.graph_engine import GraphEngine, GraphEngineConfig
 from graphon.graph_engine.command_channels import CommandChannel, InMemoryChannel
-from graphon.graph_engine.layers import DebugLoggingLayer, ExecutionLimitsLayer
-from graphon.graph_events import GraphEngineEvent, GraphNodeEventBase, GraphRunFailedEvent
+from graphon.graph_engine.layers import DebugLoggingLayer, ExecutionLimitsLayer, GraphEngineLayer
+from graphon.graph_events import GraphEngineEvent, GraphNodeEventBase, GraphRunFailedEvent, is_node_result_event
 from graphon.nodes import BuiltinNodeTypes
 from graphon.nodes.base.node import Node
 from graphon.nodes.container_effects import ContainerAwaitRequest
-from graphon.runtime import GraphRuntimeState, VariablePool
+from graphon.runtime import GraphRuntimeState, ReadOnlyGraphRuntimeStateWrapper, VariablePool
 from graphon.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader, load_into_variable_pool
 from models.workflow import Workflow
 
@@ -170,7 +169,6 @@ class WorkflowEntry:
             max_steps=dify_config.WORKFLOW_MAX_EXECUTION_STEPS, max_time=dify_config.WORKFLOW_MAX_EXECUTION_TIME
         )
         self.graph_engine.layer(limits_layer)
-        self.graph_engine.layer(LLMQuotaLayer(tenant_id=tenant_id))
 
         # Add observability layer when OTel is enabled
         if dify_config.ENABLE_OTEL or is_instrument_flag_enabled():
@@ -289,7 +287,7 @@ class WorkflowEntry:
         node = node_factory.create_node(node_config)
 
         try:
-            generator = cls._traced_node_run(node)
+            generator = cls._run_node_with_layers(node, tenant_id=workflow.tenant_id)
         except Exception as e:
             logger.exception(
                 "error while running node, workflow_id=%s, node_id=%s, node_type=%s, node_version=%s",
@@ -428,7 +426,7 @@ class WorkflowEntry:
                 tenant_id=tenant_id,
             )
 
-            generator = cls._traced_node_run(node)
+            generator = cls._run_node_with_layers(node, tenant_id=tenant_id)
 
             return node, generator
         except Exception as e:
@@ -544,24 +542,48 @@ class WorkflowEntry:
                 variable_pool.add([variable_node_id] + variable_key_list, input_value)
 
     @staticmethod
-    def _traced_node_run(node: Node) -> Generator[GraphNodeEventBase | ContainerAwaitRequest, None, None]:
+    def _run_node_with_layers(
+        node: Node, *, tenant_id: str
+    ) -> Generator[GraphNodeEventBase | ContainerAwaitRequest, None, None]:
         """
-        Wraps a node's run method with OpenTelemetry tracing and returns a generator.
+        Run a standalone node with the same quota and observability hooks as GraphEngine.
         """
-        # Wrap node.run() with ObservabilityLayer hooks to produce node-level spans
-        layer = ObservabilityLayer()
-        layer.on_graph_start()
+        layers: Sequence[GraphEngineLayer] = (ObservabilityLayer(),)
+        command_channel = InMemoryChannel()
+        runtime_state = ReadOnlyGraphRuntimeStateWrapper(node.graph_runtime_state)
+        for layer in layers:
+            layer.initialize(runtime_state, command_channel)
+            layer.on_graph_start()
+
         node.bind_execution_id(str(uuid4()))
 
         def _gen():
             error: Exception | None = None
-            layer.on_node_run_start(node)
+            result_event: GraphNodeEventBase | None = None
+            layers_finished = False
+
+            def finish_layers() -> None:
+                nonlocal layers_finished
+                if layers_finished:
+                    return
+                layers_finished = True
+                for layer in layers:
+                    layer.on_node_run_end(node, error, result_event)
+                for layer in layers:
+                    layer.on_graph_end(error)
+
             try:
-                yield from node.run()
+                for layer in layers:
+                    layer.on_node_run_start(node)
+                for event in node.run():
+                    if isinstance(event, GraphNodeEventBase) and is_node_result_event(event):
+                        result_event = event
+                        finish_layers()
+                    yield event
             except Exception as exc:
                 error = exc
                 raise
             finally:
-                layer.on_node_run_end(node, error)
+                finish_layers()
 
         return _gen()

api/docker/entrypoint.sh

@@ -31,13 +31,13 @@ if [[ "${MODE}" == "worker" ]]; then
     CONCURRENCY_OPTION="-c ${CELERY_WORKER_AMOUNT:-1}"
   fi
 
-  # Configure queues based on edition if not explicitly set
+  # Configure queues based on product edition if not explicitly set
   if [[ -z "${CELERY_QUEUES}" ]]; then
-    if [[ "${EDITION}" == "CLOUD" ]]; then
+    if [[ "${DEPLOYMENT_EDITION:-COMMUNITY}" == "CLOUD" ]]; then
       # Cloud edition: separate queues for dataset and trigger tasks
       DEFAULT_QUEUES="api_token,dataset,dataset_summary,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,app_rbac,plugin,workflow_storage,conversation,workflow_professional,workflow_team,workflow_sandbox,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_publisher,trigger_refresh_executor,retention,workflow_based_app_execution"
     else
-      # Community edition (SELF_HOSTED): dataset, pipeline and workflow have separate queues
+      # Self-hosted editions: dataset, pipeline and workflow have separate queues
       DEFAULT_QUEUES="api_token,dataset,dataset_summary,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,app_rbac,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_publisher,trigger_refresh_executor,retention,workflow_based_app_execution"
     fi
   else

api/enterprise/telemetry/README.md

@@ -32,7 +32,7 @@ The Enterprise OTEL exporter is configured via environment variables.
 
 | Variable | Description | Default |
 |----------|-------------|---------|
-| `ENTERPRISE_ENABLED` | Master switch for all enterprise features. | `false` |
+| `DEPLOYMENT_EDITION` | Product edition; enterprise telemetry is only available in `ENTERPRISE`. | `COMMUNITY` |
 | `ENTERPRISE_TELEMETRY_ENABLED` | Master switch for enterprise telemetry. | `false` |
 | `ENTERPRISE_OTLP_ENDPOINT` | OTLP collector endpoint (e.g., `http://otel-collector:4318`). | - |
 | `ENTERPRISE_OTLP_HEADERS` | Custom headers for OTLP requests (e.g., `x-scope-orgid=tenant1`). | - |

api/enterprise/telemetry/exporter.py

@@ -42,12 +42,15 @@ from enterprise.telemetry.id_generator import (
     set_correlation_id,
     set_span_id_source,
 )
+from enums import DeploymentEdition
 
 logger = logging.getLogger(__name__)
 
 
 def is_enterprise_telemetry_enabled() -> bool:
-    return bool(dify_config.ENTERPRISE_ENABLED and dify_config.ENTERPRISE_TELEMETRY_ENABLED)
+    return bool(
+        dify_config.DEPLOYMENT_EDITION == DeploymentEdition.ENTERPRISE and dify_config.ENTERPRISE_TELEMETRY_ENABLED
+    )
 
 
 def _parse_otlp_headers(raw: str) -> dict[str, str]:

api/enums/__init__.py

@@ -0,0 +1,59 @@
+from enum import StrEnum, auto
+
+
+class CloudPlan(StrEnum):
+    """
+    Enum representing user plan types in the cloud platform.
+
+    SANDBOX: Free/default plan with limited features
+    PROFESSIONAL: Professional paid plan
+    TEAM: Team collaboration paid plan
+    """
+
+    SANDBOX = auto()
+    PROFESSIONAL = auto()
+    TEAM = auto()
+
+
+class DeploymentEdition(StrEnum):
+    """Enum representing the deployment edition of the platform."""
+
+    COMMUNITY = "COMMUNITY"
+    ENTERPRISE = "ENTERPRISE"
+    CLOUD = "CLOUD"
+
+
+class HostedTrialProvider(StrEnum):
+    """Enum representing hosted model provider names for trial access."""
+
+    OPENAI = "langgenius/openai/openai"
+    ANTHROPIC = "langgenius/anthropic/anthropic"
+    GEMINI = "langgenius/gemini/google"
+    X = "langgenius/x/x"
+    DEEPSEEK = "langgenius/deepseek/deepseek"
+    TONGYI = "langgenius/tongyi/tongyi"
+
+    @property
+    def config_key(self) -> str:
+        """Return the config key used in dify_config (e.g., HOSTED_{config_key}_PAID_ENABLED)."""
+        if self == HostedTrialProvider.X:
+            return "XAI"
+        return self.name
+
+
+class QuotaType(StrEnum):
+    """Supported quota types for tenant feature usage."""
+
+    TRIGGER = auto()
+    WORKFLOW = auto()
+    UNLIMITED = auto()
+
+    @property
+    def billing_key(self) -> str:
+        match self:
+            case QuotaType.TRIGGER:
+                return "trigger_event"
+            case QuotaType.WORKFLOW:
+                return "api_rate_limit"
+            case _:
+                raise ValueError(f"Invalid quota type: {self}")

api/enums/cloud_plan.py

@@ -1,15 +0,0 @@
-from enum import StrEnum, auto
-
-
-class CloudPlan(StrEnum):
-    """
-    Enum representing user plan types in the cloud platform.
-
-    SANDBOX: Free/default plan with limited features
-    PROFESSIONAL: Professional paid plan
-    TEAM: Team collaboration paid plan
-    """
-
-    SANDBOX = auto()
-    PROFESSIONAL = auto()
-    TEAM = auto()

api/enums/deployment_edition.py

@@ -1,11 +0,0 @@
-from enum import StrEnum
-
-
-class DeploymentEdition(StrEnum):
-    """
-    Enum representing the deployment edition of the platform.
-    """
-
-    COMMUNITY = "COMMUNITY"
-    ENTERPRISE = "ENTERPRISE"
-    CLOUD = "CLOUD"