perf(web): load integration tools by category

fix(web): defer model status plugin lookup
fix(web): optimize data source plugin loading
2026-07-24 17:40:26 -07:00 · 2026-07-24 16:39:05 -07:00 · 2026-07-24 16:37:53 -07:00 · 2026-07-24 15:08:27 -07:00 · 2026-07-24 15:30:52 +00:00 · 2026-07-24 15:27:35 +00:00
1039 changed files with 69360 additions and 12740 deletions
@@ -32,12 +32,11 @@ Keep this skill focused on Cucumber, Playwright, and package-level E2E guidance.
 - `e2e/` uses Cucumber for scenarios and Playwright as the browser layer.
 - `DifyWorld` is the per-scenario context object. Type `this` as `DifyWorld` and use `async function`, not arrow functions.
 - Keep glue organized by capability under `e2e/features/step-definitions/`; use `common/` only for broadly reusable steps.
- Browser session behavior comes from `features/support/hooks.ts`:
-  - default: authenticated session with shared storage state
-  - `@unauthenticated`: clean browser context
-  - `@authenticated`: readability/selective-run tag only unless implementation changes
-  - `@fresh`: only for `e2e:full*` flows
+- Treat `e2e/AGENTS.md`, `features/support/hooks.ts`, and the Cucumber configuration as the owners of current session and tag semantics. Verify them when behavior depends on session state instead of copying a tag inventory into this skill.
 - Do not import Playwright Test runner patterns that bypass the current Cucumber + `DifyWorld` architecture unless the task is explicitly about changing that architecture.
+- Perform the behavior under test through Playwright. APIs are allowed for setup, seed preparation, persistence polling, and cleanup, but ordinary Console JSON and representable multipart operations must use the scenario- or process-owned generated oRPC client with request and response validation enabled. Keep the setup/cleanup API identity independent from an unauthenticated or logged-out behavior browser.
+- Consume generated operations directly. Do not add one-to-one API wrappers, handwritten endpoint URLs, response DTO casts, duplicate schemas, global mutable clients, or TanStack Query caching in Cucumber. Keep helpers only for real fixture construction, multi-operation orchestration, invariants, polling, derived test views, or protocol adapters.
+- Keep SSE, binary, redirect-only, external-service, and readiness exceptions centralized under their protocol owner. A contract mismatch must fail and be fixed at the backend schema owner followed by regeneration; never weaken validation to make E2E pass.

 ## Workflow

@@ -66,7 +65,7 @@ Keep this skill focused on Cucumber, Playwright, and package-level E2E guidance.
   - If a product element has real user-facing semantics but no accessible name, prefer fixing that accessible contract over adding a test id.
 5. Validate narrowly.
   - Run the narrowest tagged scenario or flow that exercises the change.
-   - Run `vpr lint --fix --quiet` from the repository root and `pnpm -C e2e type-check`.
+   - Run the package-required static checks documented in `e2e/AGENTS.md`.
   - Broaden verification only when the change affects hooks, tags, setup, or shared step semantics.

 ## Review Checklist
@@ -77,6 +76,8 @@ Keep this skill focused on Cucumber, Playwright, and package-level E2E guidance.
 - Are locators user-facing and assertions web-first?
 - Does the change introduce hidden coupling across scenarios, tags, or instance state?
 - Does it document or implement behavior that differs from the real hooks or configuration?
+- Does setup/cleanup use the generated client directly, with any remaining helper owning more than a one-to-one endpoint forward?
+- Is every raw HTTP call a documented protocol or infrastructure exception rather than an ordinary Console operation?

 Lead findings with correctness, flake risk, and architecture drift.

@@ -8,7 +8,6 @@

 # Lint bulk suppression baselines.
 /oxlint-suppressions.json
-/eslint-suppressions.json

 # CODEOWNERS file
 /.github/CODEOWNERS @laipz8200 @crazywoola
@@ -33,31 +32,9 @@
 # Backend (default owner, more specific rules below will override)
 /api/ @QuantumGhost

-# Backend - MCP
-/api/core/mcp/ @Nov1c444
-/api/core/entities/mcp_provider.py @Nov1c444
-/api/services/tools/mcp_tools_manage_service.py @Nov1c444
-/api/controllers/mcp/ @Nov1c444
-/api/controllers/console/app/mcp_server.py @Nov1c444
-
 # Backend - Tests
 /api/tests/ @laipz8200 @QuantumGhost

-/api/tests/**/*mcp* @Nov1c444
-
-# Backend - Workflow - Engine (Core graph execution engine)
-/api/core/workflow/graph_engine/ @laipz8200 @QuantumGhost
-/api/core/workflow/runtime/ @laipz8200 @QuantumGhost
-/api/core/workflow/graph/ @laipz8200 @QuantumGhost
-/api/core/workflow/graph_events/ @laipz8200 @QuantumGhost
-/api/core/workflow/node_events/ @laipz8200 @QuantumGhost
-
-# Backend - Workflow - Nodes (Agent, Iteration, Loop, LLM)
-/api/core/workflow/nodes/agent/ @Nov1c444
-/api/core/workflow/nodes/iteration/ @Nov1c444
-/api/core/workflow/nodes/loop/ @Nov1c444
-/api/core/workflow/nodes/llm/ @Nov1c444
-
 # Backend - RAG (Retrieval Augmented Generation)
 /api/core/rag/ @JohnJyong
 /api/services/rag_pipeline/ @JohnJyong
@@ -111,7 +88,6 @@
 /api/core/app/layers/trigger_post_layer.py @CourTeous33
 /api/services/trigger/ @CourTeous33
 /api/models/trigger.py @CourTeous33
-/api/fields/workflow_trigger_fields.py @CourTeous33
 /api/repositories/workflow_trigger_log_repository.py @CourTeous33
 /api/repositories/sqlalchemy_workflow_trigger_log_repository.py @CourTeous33
 /api/libs/schedule_utils.py @CourTeous33
@@ -136,11 +112,11 @@
 /api/controllers/console/billing/ @hj24 @zyssyz123

 # Backend - Enterprise
-/api/configs/enterprise/ @GarfieldDai @GareArc
-/api/services/enterprise/ @GarfieldDai @GareArc
-/api/services/feature_service.py @GarfieldDai @GareArc
-/api/controllers/console/feature.py @GarfieldDai @GareArc
-/api/controllers/web/feature.py @GarfieldDai @GareArc
+/api/configs/enterprise/ @GareArc
+/api/services/enterprise/ @GareArc
+/api/services/feature_service.py @GareArc
+/api/controllers/console/feature.py @GareArc
+/api/controllers/web/feature.py @GareArc

 # Backend - Database Migrations
 /api/migrations/ @snakevash @laipz8200 @MRZHUH
@@ -153,7 +129,6 @@

 # Frontend - Platform and Features
 /web/config/ @lyzno1
-/web/contract/ @lyzno1
 /web/env.ts @lyzno1
 /web/features/ @lyzno1
 /web/hooks/ @lyzno1
@@ -212,7 +187,6 @@
 /web/app/components/rag-pipeline/store/ @iamjoel @zxhlyh

 # Frontend - RAG - Documents List
-/web/app/components/datasets/documents/list.tsx @iamjoel @WTW0313
 /web/app/components/datasets/documents/create-from-pipeline/ @iamjoel @WTW0313

 # Frontend - RAG - Segments List
@@ -231,22 +205,22 @@
 /web/app/components/plugins/marketplace/ @iamjoel @Yessenia-d

 # Frontend - Login and Registration
-/web/app/signin/ @douxc @iamjoel
-/web/app/signup/ @douxc @iamjoel
-/web/app/reset-password/ @douxc @iamjoel
-/web/app/install/ @douxc @iamjoel
-/web/app/init/ @douxc @iamjoel
-/web/app/forgot-password/ @douxc @iamjoel
-/web/app/account/ @douxc @iamjoel
+/web/app/signin/ @iamjoel
+/web/app/signup/ @iamjoel
+/web/app/reset-password/ @iamjoel
+/web/app/install/ @iamjoel
+/web/app/init/ @iamjoel
+/web/app/forgot-password/ @iamjoel
+/web/app/account/ @iamjoel

 # Frontend - Service Authentication
-/web/service/base.ts @douxc @iamjoel
+/web/service/base.ts @iamjoel

 # Frontend - WebApp Authentication and Access Control
-/web/app/(shareLayout)/components/ @douxc @iamjoel
-/web/app/(shareLayout)/webapp-signin/ @douxc @iamjoel
-/web/app/(shareLayout)/webapp-reset-password/ @douxc @iamjoel
-/web/app/components/app/app-access-control/ @douxc @iamjoel
+/web/app/(shareLayout)/components/ @iamjoel
+/web/app/(shareLayout)/webapp-signin/ @iamjoel
+/web/app/(shareLayout)/webapp-reset-password/ @iamjoel
+/web/app/components/app/app-access-control/ @iamjoel

 # Frontend - Explore Page
 /web/app/components/explore/ @CodingOnStar @iamjoel
@@ -265,7 +239,6 @@
 /web/app/components/base/**/*.spec.tsx @hyoban @CodingOnStar

 # Frontend - Utils and Hooks
-/web/utils/classnames.ts @iamjoel @zxhlyh
 /web/utils/time.ts @iamjoel @zxhlyh
 /web/utils/format.ts @iamjoel @zxhlyh
 /web/utils/clipboard.ts @iamjoel @zxhlyh
@@ -277,5 +250,22 @@
 # Frontend - Workspace
 /web/app/components/header/account-dropdown/workplace-selector/ @iamjoel @zxhlyh

+# Frontend - App Shell and Console Bootstrap
+/web/app/layout.tsx @iamjoel @lyzno1
+/web/app/error.tsx @iamjoel @lyzno1
+/web/app/(commonLayout)/layout.tsx @iamjoel @lyzno1
+/web/app/(commonLayout)/providers.tsx @iamjoel @lyzno1
+/web/app/(commonLayout)/hydration-boundary.tsx @iamjoel @lyzno1
+/web/app/(commonLayout)/profile-bootstrap-gate.tsx @iamjoel @lyzno1
+/web/app/(commonLayout)/error.tsx @iamjoel @lyzno1
+/web/app/account/(commonLayout)/layout.tsx @iamjoel @lyzno1
+/web/app/components/main-nav/* @iamjoel @lyzno1
+/web/app/components/main-nav/components/* @iamjoel @lyzno1
+/web/context/query-client.tsx @iamjoel @lyzno1
+/web/context/query-client-server.ts @iamjoel @lyzno1
+/web/proxy.ts @iamjoel @lyzno1
+/web/app/auth/refresh/route.ts @iamjoel @lyzno1
+/web/service/server.ts @iamjoel @lyzno1
+
 # Docker
 /docker/* @laipz8200
@@ -0,0 +1,28 @@
+name: Deploy Knowledge
+
+permissions:
+  contents: read
+
+on:
+  workflow_run:
+    workflows: ["Build and Push API & Web"]
+    branches:
+      - "deploy/konwledge"
+    types:
+      - completed
+
+jobs:
+  deploy:
+    runs-on: depot-ubuntu-24.04
+    if: |
+      github.event.workflow_run.conclusion == 'success' &&
+      github.event.workflow_run.head_branch == 'deploy/konwledge'
+    steps:
+      - name: Deploy to server
+        uses: appleboy/ssh-action@0ff4204d59e8e51228ff73bce53f80d53301dee2 # v1.2.5
+        with:
+          host: ${{ secrets.SSH_NEW_RAG_HOST }}
+          username: ${{ secrets.SSH_USER }}
+          key: ${{ secrets.SSH_PRIVATE_KEY }}
+          script: |
+            ${{ vars.SSH_SCRIPT || secrets.SSH_SCRIPT }}
@@ -335,6 +335,8 @@ jobs:
      - check-changes
    if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.e2e-changed == 'true'
    uses: ./.github/workflows/web-e2e.yml
+    with:
+      run-external-runtime: false
    secrets: inherit

  web-e2e-skip:
@@ -26,19 +26,30 @@ jobs:
            external_e2e:
              - 'e2e/features/agent-v2/**'
              - 'e2e/features/step-definitions/agent-v2/**'
+              - 'e2e/features/step-definitions/common/**'
              - 'e2e/features/support/**'
+              - 'e2e/fixtures/auth.ts'
              - 'e2e/fixtures/test-materials/**'
              - 'e2e/scripts/**'
              - 'e2e/support/**'
              - 'e2e/cucumber.config.ts'
              - 'e2e/package.json'
              - 'e2e/test-env.ts'
+              - 'e2e/tsconfig.json'
              - 'e2e/tsx-register.js'
+              - 'package.json'
+              - 'pnpm-lock.yaml'
+              - '.nvmrc'
              - '.github/workflows/post-merge.yml'
              - '.github/workflows/web-e2e.yml'
              - '.github/actions/setup-web/**'
+              - 'docker/docker-compose.middleware.yaml'
+              - 'docker/envs/middleware.env.example'
              - 'dify-agent/**'
              - 'dify-agent-runtime/**'
+              - 'api/pyproject.toml'
+              - 'api/uv.lock'
+              - 'api/tests/integration_tests/.env.example'
              - 'api/clients/agent_backend/**'
              - 'api/core/app/apps/agent_app/**'
              - 'api/core/workflow/nodes/agent_v2/**'
@@ -48,8 +59,13 @@ jobs:
              - 'api/services/plugin/**'
              - 'api/core/tools/**'
              - 'api/services/tools/**'
+              - 'packages/contracts/package.json'
              - 'packages/contracts/generated/api/console/agent/**'
+              - 'packages/contracts/generated/api/console/apps/**'
+              - 'packages/contracts/generated/api/console/datasets/**'
              - 'packages/contracts/generated/api/console/orpc.gen.ts'
+              - 'packages/contracts/generated/api/console/workspaces/**'
+              - 'packages/contracts/generated/api/service/**'
              - 'web/features/agent-v2/**'
              - 'web/app/(commonLayout)/agents/**'
              - 'web/app/(commonLayout)/@detailSidebar/agents/**'
@@ -4,9 +4,9 @@ on:
  workflow_call:
    inputs:
      run-external-runtime:
-        required: false
+        description: Run only the prepared and external runtime suite instead of the core suites.
+        required: true
        type: boolean
-        default: false

 permissions:
  contents: read
@@ -46,6 +46,7 @@ jobs:
        run: uv sync --project api --dev

      - name: Run E2E support unit tests
+        if: ${{ !inputs.run-external-runtime }}
        working-directory: ./e2e
        run: vp run test:unit

@@ -54,6 +55,7 @@ jobs:
        run: vp run e2e:install

      - name: Run isolated source-api and built-web Cucumber E2E tests
+        if: ${{ !inputs.run-external-runtime }}
        working-directory: ./e2e
        env:
          E2E_ADMIN_EMAIL: e2e-admin@example.com
@@ -64,7 +66,7 @@ jobs:
        run: vp run e2e:full

      - name: Preserve Chromium E2E report and logs
-        if: ${{ !cancelled() }}
+        if: ${{ !cancelled() && !inputs.run-external-runtime }}
        run: |
          if [[ -d e2e/cucumber-report ]]; then
            mv e2e/cucumber-report e2e/cucumber-report-non-external
@@ -74,6 +76,7 @@ jobs:
          fi

      - name: Run WebKit keyboard and browser smoke tests
+        if: ${{ !inputs.run-external-runtime }}
        working-directory: ./e2e
        env:
          E2E_ADMIN_EMAIL: e2e-admin@example.com
@@ -99,7 +102,7 @@ jobs:
          vp run e2e -- --tags '@browser-smoke'

      - name: Preserve WebKit E2E report and logs
-        if: ${{ !cancelled() }}
+        if: ${{ !cancelled() && !inputs.run-external-runtime }}
        run: |
          if [[ -d e2e/cucumber-report ]]; then
            mv e2e/cucumber-report e2e/cucumber-report-webkit
@@ -71,7 +71,7 @@ Dify is an open-source LLM app development platform. Its intuitive interface com

 <br/>

-The easiest way to start the Dify server is through [Docker Compose](docker/docker-compose.yaml). Before running Dify with the following commands, make sure that [Docker](https://docs.docker.com/get-docker/) and [Docker Compose](https://docs.docker.com/compose/install/) are installed on your machine:
+The easiest way to start the Dify server is through [Docker Compose](docker/docker-compose.yaml). Before running Dify with the following commands, make sure that [Docker](https://docs.docker.com/get-docker/) and Docker Compose v2.24.0 or later are installed on your machine:

 ```bash
 cd dify
@@ -6,6 +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 events.app_event import app_was_created
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
@@ -42,7 +43,7 @@ def reset_encrypt_key_pair():
    After the reset, all LLM credentials will become invalid, requiring re-entry.
    Only support SELF_HOSTED mode.
    """
-    if dify_config.EDITION != "SELF_HOSTED":
+    if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD:
        click.echo(click.style("This command is only for SELF_HOSTED installations.", fg="red"))
        return
    with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
@@ -5,6 +5,7 @@ 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
@@ -116,3 +117,11 @@ 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
@@ -1497,6 +1497,11 @@ class LoginConfig(BaseSettings):


 class AccountConfig(BaseSettings):
+    ENABLE_CHANGE_EMAIL: bool = Field(
+        description="whether users can change their email address",
+        default=True,
+    )
+
    ACCOUNT_DELETION_TOKEN_EXPIRY_MINUTES: PositiveInt = Field(
        description="Duration in minutes for which a account deletion token remains valid",
        default=5,
@@ -10,6 +10,7 @@ from extensions.ext_database import db
 from libs.login import current_account_with_tenant
 from models.dataset import Dataset
 from models.model import App
+from services.agent.roster_service import AgentRosterService
 from services.enterprise.rbac_service import RBACService

 __all__ = ["RBACPermission", "RBACResourceScope", "enforce_rbac_access", "rbac_permission_required"]
@@ -51,7 +52,7 @@ def enforce_rbac_access(
    check_resource_type = None if resource_type == RBACResourceScope.WORKSPACE else resource_type
    resource_id = None
    if resource_required and check_resource_type:
-        resource_id = _extract_resource_id(resource_type, path_args)
+        resource_id = _extract_resource_id(resource_type, tenant_id, path_args)
        if _is_resource_owned_by_current_user(tenant_id, account_id, resource_type, resource_id):
            return
    allowed = RBACService.CheckAccess.check(
@@ -131,11 +132,14 @@ def _is_resource_owned_by_current_user(
    return False


-def _extract_resource_id(resource_type: RBACResourceScope, path_args: dict[str, object] | None = None) -> str:
+def _extract_resource_id(
+    resource_type: RBACResourceScope, tenant_id: str, path_args: dict[str, object] | None = None
+) -> str:
    """Extract the resource ID from matched path arguments.

    Some legacy route classes use neutral names such as ``resource_id`` for
-    app/dataset resources, and Agent App routes use ``agent_id`` as the app id.
+    app/dataset resources, and Agent routes carry ``agent_id``, which is
+    resolved to the App backing that Agent.
    Dataset endpoints behind a rag-pipeline route contain ``pipeline_id``
    instead of ``dataset_id``. In that case we look up the associated
    ``Dataset`` row via ``Dataset.pipeline_id``.
@@ -146,10 +150,19 @@ def _extract_resource_id(resource_type: RBACResourceScope, path_args: dict[str,
    matched_args = {**view_args, **(path_args or {})}

    if resource_type == RBACResourceScope.APP:
-        app_id = matched_args.get("app_id") or matched_args.get("agent_id") or matched_args.get("resource_id")
-        if not app_id:
-            raise ValueError("Missing app_id in request path")
-        return str(app_id)
+        app_id = matched_args.get("app_id")
+        if app_id:
+            return str(app_id)
+
+        agent_id = matched_args.get("agent_id")
+        if agent_id:
+            authz_app_id = AgentRosterService(db.session).peek_authz_app_id(tenant_id=tenant_id, agent_id=str(agent_id))
+            return authz_app_id or str(agent_id)
+
+        resource_id = matched_args.get("resource_id")
+        if resource_id:
+            return str(resource_id)
+        raise ValueError("Missing app_id in request path")

    if resource_type == RBACResourceScope.DATASET:
        dataset_id = matched_args.get("dataset_id") or matched_args.get("resource_id")
@@ -230,6 +230,7 @@ class WorkflowAgentComposerSaveToRosterApi(Resource):
    @account_initialization_required
    @edit_permission_required
    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
    @with_current_user_id
    @with_current_tenant_id
    @with_session
@@ -439,6 +440,7 @@ class SnippetAgentComposerSaveToRosterApi(Resource):
    @rbac_permission_required(
        RBACResourceScope.WORKSPACE, RBACPermission.SNIPPETS_CREATE_AND_MODIFY, resource_required=False
    )
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
    @with_current_user_id
    @with_current_tenant_id
    @with_session
@@ -478,6 +480,7 @@ class AgentComposerApi(Resource):
    @account_initialization_required
    @edit_permission_required
    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
    @with_current_user_id
    @with_current_tenant_id
    @with_session
@@ -62,7 +62,7 @@ from libs.datetime_utils import parse_time_range
 from libs.helper import dump_response
 from libs.login import login_required
 from models import Account
-from models.agent import Agent, AgentStatus
+from models.agent import Agent, AgentConfigDraftType, AgentStatus
 from models.agent_config_entities import AgentSoulConfig
 from models.enums import ApiTokenType
 from models.model import ApiToken, App, IconType
@@ -266,6 +266,13 @@ class AgentDebugConversationRefreshResponse(BaseModel):
    debug_conversation_message_count: int = 0


+class AgentDebugConversationRefreshPayload(BaseModel):
+    draft_type: AgentConfigDraftType = Field(
+        default=AgentConfigDraftType.DEBUG_BUILD,
+        description="Agent draft surface whose conversation should be refreshed",
+    )
+
+
 class AgentPublishPayload(BaseModel):
    version_note: str | None = Field(default=None, description="Optional note for this published Agent version")

@@ -309,6 +316,7 @@ register_schema_models(
    AgentAppCopyPayload,
    AgentPublishPayload,
    AgentBuildDraftCheckoutPayload,
+    AgentDebugConversationRefreshPayload,
    ComposerSavePayload,
    AgentApiStatusPayload,
    AgentInviteOptionsQuery,
@@ -392,6 +400,7 @@ def _serialize_agent_app_detail(
        tenant_id=app_model.tenant_id,
        agent_id=agent.id,
        account_id=current_user.id,
+        draft_type=AgentConfigDraftType.DEBUG_BUILD,
        commit=False,
    )
    message_count = roster_service.count_agent_app_debug_conversation_messages(
@@ -439,6 +448,7 @@ def _serialize_agent_app_pagination(session: Session, app_pagination, *, tenant_
        tenant_id=tenant_id,
        agents=list(agents_by_app_id.values()),
        account_id=current_user.id,
+        draft_type=AgentConfigDraftType.DEBUG_BUILD,
    )
    payload = AgentAppPagination.model_validate(
        app_pagination,
@@ -529,9 +539,23 @@ def _parse_observability_time_range(start: str | None, end: str | None, account:


 def _query_values(name: str, alias_name: str | None = None) -> list[str]:
-    values = request.args.getlist(name)
+    def _get_values(field_name: str) -> list[str]:
+        values = request.args.getlist(field_name)
+        indexed_values: list[tuple[int, list[str]]] = []
+        prefix = f"{field_name}["
+        for key in request.args:
+            if not key.startswith(prefix) or not key.endswith("]"):
+                continue
+            index = key[len(prefix) : -1]
+            if index.isdigit():
+                indexed_values.append((int(index), request.args.getlist(key)))
+        for _, items in sorted(indexed_values):
+            values.extend(items)
+        return values
+
+    values = _get_values(name)
    if alias_name:
-        values.extend(request.args.getlist(alias_name))
+        values.extend(_get_values(alias_name))
    return [value.strip() for value in values if value.strip()]


@@ -542,6 +566,7 @@ class AgentAppListApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
    @with_current_user
    @with_current_tenant_id
    @with_session
@@ -579,6 +604,7 @@ class AgentAppListApi(Resource):
    @login_required
    @account_initialization_required
    @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
    @with_current_user
    @with_current_tenant_id
    @with_session
@@ -620,6 +646,7 @@ class AgentAppApi(Resource):
    @login_required
    @account_initialization_required
    @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
    @with_current_user
    @with_current_tenant_id
    @with_session
@@ -645,6 +672,7 @@ class AgentAppApi(Resource):
    @login_required
    @account_initialization_required
    @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
    @with_current_tenant_id
    @with_session
    def delete(self, session: Session, tenant_id: str, agent_id: UUID):
@@ -655,6 +683,16 @@ class AgentAppApi(Resource):

@console_ns.route("/agent/<uuid:agent_id>/debug-conversation/refresh")
 class AgentDebugConversationRefreshApi(Resource):
+    @console_ns.expect(console_ns.models[AgentDebugConversationRefreshPayload.__name__])
+    @console_ns.doc(
+        params={
+            "payload": {
+                "in": "body",
+                "required": False,
+                "schema": {"$ref": f"#/components/schemas/{AgentDebugConversationRefreshPayload.__name__}"},
+            }
+        }
+    )
    @console_ns.response(
        200,
        "Agent debug conversation refreshed",
@@ -669,10 +707,12 @@ class AgentDebugConversationRefreshApi(Resource):
    @with_current_tenant_id
    @with_session
    def post(self, session: Session, tenant_id: str, current_user: Account, agent_id: UUID):
+        args = AgentDebugConversationRefreshPayload.model_validate(request.get_json(silent=True) or {})
        debug_conversation_id = _agent_roster_service(session).refresh_agent_app_debug_conversation_id(
            tenant_id=tenant_id,
            agent_id=str(agent_id),
            account_id=current_user.id,
+            draft_type=args.draft_type,
        )
        return AgentDebugConversationRefreshResponse(
            debug_conversation_id=debug_conversation_id,
@@ -690,6 +730,7 @@ class AgentPublishApi(Resource):
    @login_required
    @account_initialization_required
    @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
    @with_current_user
    @with_current_tenant_id
    @with_session
@@ -712,6 +753,7 @@ class AgentBuildDraftCheckoutApi(Resource):
    @login_required
    @account_initialization_required
    @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
    @with_current_user
    @with_current_tenant_id
    @with_session
@@ -729,6 +771,7 @@ class AgentBuildDraftCheckoutApi(Resource):
@console_ns.route("/agent/<uuid:agent_id>/build-draft")
 class AgentBuildDraftApi(Resource):
    @console_ns.response(200, "Agent build draft", console_ns.models[AgentBuildDraftResponse.__name__])
+    @console_ns.response(404, "Agent build draft not found")
    @setup_required
    @login_required
    @account_initialization_required
@@ -787,6 +830,7 @@ class AgentBuildDraftApplyApi(Resource):
    @login_required
    @account_initialization_required
    @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
    @with_current_user
    @with_current_tenant_id
    @with_session
@@ -809,6 +853,7 @@ class AgentAppCopyApi(Resource):
    @login_required
    @account_initialization_required
    @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
    @with_current_user
    @with_current_tenant_id
    @with_session
@@ -834,6 +879,7 @@ class AgentApiAccessApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
    @with_current_tenant_id
    @with_session(write=False)
    def get(self, session: Session, tenant_id: str, agent_id: UUID):
@@ -850,6 +896,7 @@ class AgentApiStatusApi(Resource):
    @login_required
    @is_admin_or_owner_required
    @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
    @with_current_tenant_id
    @with_session
@@ -868,6 +915,7 @@ class AgentApiKeyListApi(BaseApiKeyListResource):
    token_prefix = "app-"

    @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
    @with_session(write=False)
    def get(self, session: Session, tenant_id: str, agent_id: UUID) -> dict[str, object]:
@@ -878,6 +926,7 @@ class AgentApiKeyListApi(BaseApiKeyListResource):
    @console_ns.response(400, "Maximum keys exceeded")
    @with_current_tenant_id
    @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
    @with_session
    def post(self, session: Session, tenant_id: str, agent_id: UUID) -> tuple[dict[str, object], int]:
@@ -897,6 +946,7 @@ class AgentApiKeyApi(BaseApiKeyResource):
    @console_ns.response(204, "Agent service API key deleted")
    @with_current_user
    @with_current_tenant_id
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
    @with_session
    def delete(
@@ -942,6 +992,7 @@ class AgentLogsApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
    @with_current_user
    @with_current_tenant_id
    @with_session(write=False)
@@ -980,6 +1031,7 @@ class AgentLogMessagesApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
    @with_current_user
    @with_current_tenant_id
    @with_session(write=False)
@@ -1018,6 +1070,7 @@ class AgentLogSourcesApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
    @with_current_user
    @with_current_tenant_id
    @with_session(write=False)
@@ -1038,6 +1091,7 @@ class AgentStatisticsSummaryApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
    @with_current_user
    @with_current_tenant_id
    @with_session(write=False)
@@ -1063,6 +1117,7 @@ class AgentRosterVersionsApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
    @with_current_tenant_id
    @with_session(write=False)
    def get(self, session: Session, tenant_id: str, agent_id: UUID):
@@ -1078,6 +1133,7 @@ class AgentRosterVersionDetailApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
    @with_current_tenant_id
    @with_session(write=False)
    def get(self, session: Session, tenant_id: str, agent_id: UUID, version_id: UUID):
@@ -1098,6 +1154,7 @@ class AgentRosterVersionRestoreApi(Resource):
    @login_required
    @account_initialization_required
    @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
    @with_current_user
    @with_current_tenant_id
    @with_session
@@ -12,6 +12,7 @@ from werkzeug.exceptions import Forbidden
 from configs import dify_config
 from controllers.common.schema import register_response_schema_models
 from controllers.common.session import with_session
+from controllers.console.app.wraps import agent_manage_required_for_agent_app
 from fields.base import ResponseModel
 from libs.helper import dump_response, to_timestamp
 from libs.login import login_required
@@ -194,6 +195,7 @@ 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
+    @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"""
@@ -210,6 +212,7 @@ class AppApiKeyListResource(BaseApiKeyListResource):
    @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
    def post(self, session: Session, current_tenant_id: str, resource_id: UUID) -> tuple[dict[str, object], int]:
        """Create a new API key for an app"""
@@ -233,6 +236,7 @@ class AppApiKeyResource(BaseApiKeyResource):
    @with_current_user
    @with_current_tenant_id
    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
+    @agent_manage_required_for_agent_app
    @with_session
    def delete(
        self,
@@ -9,7 +9,7 @@ from flask_restx import Resource
 from pydantic import AliasChoices, BaseModel, Field, ValidationInfo, computed_field, field_validator, model_validator
 from sqlalchemy import select
 from sqlalchemy.orm import Session
-from werkzeug.exceptions import BadRequest, NotFound
+from werkzeug.exceptions import BadRequest, Forbidden, NotFound

 from configs import dify_config
 from controllers.common.app_access import resolve_app_access_filter
@@ -23,7 +23,7 @@ from controllers.common.schema import (
    register_schema_models,
 )
 from controllers.console import console_ns
-from controllers.console.app.wraps import get_app_model, with_session
+from controllers.console.app.wraps import agent_manage_required_for_agent_app, get_app_model, with_session
 from controllers.console.workspace.models import LoadBalancingPayload
 from controllers.console.wraps import (
    RBACPermission,
@@ -75,6 +75,7 @@ from services.entities.knowledge_entities.knowledge_entities import (
    WeightModel,
    WeightVectorSetting,
 )
+from services.errors.account import NoPermissionError
 from services.feature_service import FeatureService
 from tasks.initialize_created_app_rbac_access_task import initialize_created_app_rbac_access_task

@@ -246,7 +247,7 @@ class ModelConfigPartial(ResponseModel):
        return to_timestamp(value)


-class ModelConfig(ResponseModel):
+class AppModelConfigResponse(ResponseModel):
    opening_statement: str | None = None
    suggested_questions: Any | None = Field(
        default=None, validation_alias=AliasChoices("suggested_questions_list", "suggested_questions")
@@ -419,7 +420,7 @@ class AppDetail(AppResponseModel):
    icon_background: str | None = None
    enable_site: bool
    enable_api: bool
-    model_config_: ModelConfig | None = Field(
+    model_config_: AppModelConfigResponse | None = Field(
        default=None,
        validation_alias=AliasChoices("app_model_config", "model_config"),
        alias="model_config",
@@ -525,7 +526,13 @@ def _enrich_app_list_items(session: Session, *, apps: Sequence[App], tenant_id:

 register_enum_models(console_ns, RetrievalMethod, WorkflowExecutionStatus, DatasetPermissionEnum)
 register_response_schema_models(
-    console_ns, RedirectUrlResponse, SimpleResultResponse, AppImportResponse, AppTraceResponse
+    console_ns,
+    RedirectUrlResponse,
+    SimpleResultResponse,
+    AppImportResponse,
+    AppTraceResponse,
+    AppModelConfigResponse,
+    AppDetail,
 )

 register_schema_models(
@@ -544,10 +551,8 @@ register_schema_models(
    Tag,
    WorkflowPartial,
    ModelConfigPartial,
-    ModelConfig,
    AppDetailSiteResponse,
    DeletedTool,
-    AppDetail,
    AppExportResponse,
    Segmentation,
    PreProcessingRule,
@@ -823,6 +828,7 @@ class AppApi(Resource):
    @account_initialization_required
    @edit_permission_required
    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
+    @agent_manage_required_for_agent_app
    @with_session
    @get_app_model(mode=None)
    def put(self, session: Session, app_model: App):
@@ -857,6 +863,7 @@ class AppApi(Resource):
    @account_initialization_required
    @edit_permission_required
    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_DELETE)
+    @agent_manage_required_for_agent_app
    @with_session
    @get_app_model
    def delete(self, session: Session, app_model: App):
@@ -881,6 +888,7 @@ class AppCopyApi(Resource):
    @account_initialization_required
    @edit_permission_required
    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_CREATE_AND_MANAGEMENT)
+    @agent_manage_required_for_agent_app
    @with_current_user
    @with_current_tenant_id
    @get_app_model(mode=None)
@@ -892,16 +900,19 @@ class AppCopyApi(Resource):
        with Session(db.engine, expire_on_commit=False) as session:
            import_service = AppDslService(session)
            yaml_content = import_service.export_dsl(app_model=app_model, session=session, include_secret=True)
-            result = import_service.import_app(
-                account=current_user,
-                import_mode=ImportMode.YAML_CONTENT,
-                yaml_content=yaml_content,
-                name=args.name,
-                description=args.description,
-                icon_type=args.icon_type,
-                icon=args.icon,
-                icon_background=args.icon_background,
-            )
+            try:
+                result = import_service.import_app(
+                    account=current_user,
+                    import_mode=ImportMode.YAML_CONTENT,
+                    yaml_content=yaml_content,
+                    name=args.name,
+                    description=args.description,
+                    icon_type=args.icon_type,
+                    icon=args.icon,
+                    icon_background=args.icon_background,
+                )
+            except NoPermissionError as e:
+                raise Forbidden(str(e))
            if result.status == ImportStatus.FAILED:
                session.rollback()
                return dump_response(AppImportResponse, result), 400
@@ -955,6 +966,7 @@ class AppExportApi(Resource):
    @account_initialization_required
    @edit_permission_required
    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_IMPORT_EXPORT_DSL)
+    @agent_manage_required_for_agent_app
    @get_app_model
    def get(self, app_model: App):
        """Export app"""
@@ -979,6 +991,7 @@ class AppPublishToCreatorsPlatformApi(Resource):
    @account_initialization_required
    @edit_permission_required
    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_IMPORT_EXPORT_DSL)
+    @agent_manage_required_for_agent_app
    @with_current_user_id
    @get_app_model(mode=None)
    def post(self, current_user_id: str, app_model: App):
@@ -1009,6 +1022,7 @@ class AppNameApi(Resource):
    @account_initialization_required
    @edit_permission_required
    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
+    @agent_manage_required_for_agent_app
    @with_session
    @get_app_model(mode=None)
    def post(self, session: Session, app_model: App):
@@ -1036,6 +1050,7 @@ class AppIconApi(Resource):
    @account_initialization_required
    @edit_permission_required
    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
+    @agent_manage_required_for_agent_app
    @with_session
    @get_app_model(mode=None)
    def post(self, session: Session, app_model: App):
@@ -1069,6 +1084,7 @@ class AppSiteStatus(Resource):
    @account_initialization_required
    @edit_permission_required
    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
+    @agent_manage_required_for_agent_app
    @with_session
    @get_app_model(mode=None)
    def post(self, session: Session, app_model: App):
@@ -1096,6 +1112,7 @@ class AppApiStatus(Resource):
    @is_admin_or_owner_required
    @account_initialization_required
    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
+    @agent_manage_required_for_agent_app
    @with_session
    @get_app_model(mode=None)
    def post(self, session: Session, app_model: App):
@@ -1,6 +1,7 @@
 from flask_restx import Resource
 from pydantic import BaseModel, Field
 from sqlalchemy.orm import Session
+from werkzeug.exceptions import Forbidden

 from configs import dify_config
 from controllers.common.schema import register_enum_models, register_schema_models
@@ -28,6 +29,7 @@ from services.app_dsl_service import (
 )
 from services.enterprise.enterprise_service import EnterpriseService
 from services.entities.dsl_entities import CheckDependenciesResult, ImportStatus
+from services.errors.account import NoPermissionError
 from services.feature_service import FeatureService

 from .. import console_ns
@@ -91,18 +93,21 @@ class AppImportApi(Resource):
            import_service = AppDslService(session)
            # Import app
            account = current_user
-            result = import_service.import_app(
-                account=account,
-                import_mode=args.mode,
-                yaml_content=args.yaml_content,
-                yaml_url=args.yaml_url,
-                name=args.name,
-                description=args.description,
-                icon_type=args.icon_type,
-                icon=args.icon,
-                icon_background=args.icon_background,
-                app_id=args.app_id,
-            )
+            try:
+                result = import_service.import_app(
+                    account=account,
+                    import_mode=args.mode,
+                    yaml_content=args.yaml_content,
+                    yaml_url=args.yaml_url,
+                    name=args.name,
+                    description=args.description,
+                    icon_type=args.icon_type,
+                    icon=args.icon,
+                    icon_background=args.icon_background,
+                    app_id=args.app_id,
+                )
+            except NoPermissionError as e:
+                raise Forbidden(str(e))
            if result.status == ImportStatus.FAILED:
                session.rollback()
            else:
@@ -157,7 +162,10 @@ class AppImportConfirmApi(Resource):
            import_service = AppDslService(session)
            # Confirm import
            account = current_user
-            result = import_service.confirm_import(import_id=import_id, account=account)
+            try:
+                result = import_service.confirm_import(import_id=import_id, account=account)
+            except NoPermissionError as e:
+                raise Forbidden(str(e))
            if result.status == ImportStatus.FAILED:
                session.rollback()
            else:
@@ -49,6 +49,7 @@ from libs import helper
 from libs.helper import uuid_value
 from libs.login import login_required
 from models import Account
+from models.agent import AgentConfigDraftType
 from models.model import App, AppMode
 from services.agent.errors import AgentNotFoundError
 from services.agent.roster_service import AgentRosterService
@@ -343,23 +344,40 @@ class AgentChatMessageStopApi(Resource):


 def _resolve_current_user_agent_debug_conversation_id(
-    *, session: Session, current_tenant_id: str, current_user: Account, app_model: App, agent_id: str | None
+    *,
+    session: Session,
+    current_tenant_id: str,
+    current_user: Account,
+    app_model: App,
+    agent_id: str | None,
+    draft_type: AgentConfigDraftType,
+    start_new: bool = False,
 ) -> str:
-    roster_service = AgentRosterService(session)
-    if agent_id:
-        return roster_service.get_or_create_agent_app_debug_conversation_id(
-            tenant_id=current_tenant_id,
-            agent_id=agent_id,
-            account_id=current_user.id,
-        )
+    """Resolve or rotate the current editor's conversation within one draft surface.

-    agent = roster_service.get_app_backing_agent(tenant_id=current_tenant_id, app_id=str(app_model.id))
-    if agent is None:
-        raise AgentNotFoundError()
-    return roster_service.get_or_create_agent_app_debug_conversation_id(
+    ``start_new`` rotates the scoped mapping through ``AgentRosterService`` so
+    the old runtime session is retired before the new conversation is used.
+    Continuations and Build chat keep resolving the existing mapping.
+    """
+
+    roster_service = AgentRosterService(session)
+    resolved_agent_id = agent_id
+    if not resolved_agent_id:
+        agent = roster_service.get_app_backing_agent(tenant_id=current_tenant_id, app_id=str(app_model.id))
+        if agent is None:
+            raise AgentNotFoundError()
+        resolved_agent_id = agent.id
+
+    resolve_conversation = (
+        roster_service.refresh_agent_app_debug_conversation_id
+        if start_new
+        else roster_service.get_or_create_agent_app_debug_conversation_id
+    )
+    return resolve_conversation(
        tenant_id=current_tenant_id,
-        agent_id=agent.id,
+        agent_id=resolved_agent_id,
        account_id=current_user.id,
+        draft_type=draft_type,
    )


@@ -376,12 +394,17 @@ def _create_chat_message(
    args = args_model.model_dump(exclude_none=True, by_alias=True)

    if AppMode.value_of(app_model.mode) == AppMode.AGENT:
+        draft_type = AgentConfigDraftType(args_model.draft_type)
+        # Preview follows the normal chat contract: an omitted/empty conversation ID starts a new
+        # conversation. Build chat keeps its stable mapping so build drafts and finalization stay continuous.
        debug_conversation_id = _resolve_current_user_agent_debug_conversation_id(
            session=session,
            current_tenant_id=current_tenant_id or app_model.tenant_id,
            current_user=current_user,
            app_model=app_model,
            agent_id=agent_id,
+            draft_type=draft_type,
+            start_new=draft_type == AgentConfigDraftType.DRAFT and not args_model.conversation_id,
        )
        if args_model.conversation_id and args_model.conversation_id != debug_conversation_id:
            raise NotFound("Conversation Not Exists.")
@@ -418,6 +441,7 @@ def _create_build_chat_finalization_message(
        current_user=current_user,
        app_model=app_model,
        agent_id=agent_id,
+        draft_type=AgentConfigDraftType.DEBUG_BUILD,
    )
    args: dict[str, Any] = {
        "query": _BUILD_CHAT_FINALIZATION_QUERY,
@@ -10,7 +10,7 @@ from constants.languages import supported_language
 from controllers.common.schema import register_schema_models
 from controllers.common.session import with_session
 from controllers.console import console_ns
-from controllers.console.app.wraps import get_app_model
+from controllers.console.app.wraps import agent_manage_required_for_agent_app, get_app_model
 from controllers.console.wraps import (
    RBACPermission,
    RBACResourceScope,
@@ -93,6 +93,7 @@ class AppSite(Resource):
    @login_required
    @edit_permission_required
    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
+    @agent_manage_required_for_agent_app
    @account_initialization_required
    @with_current_user
    @with_session
@@ -145,6 +146,7 @@ class AppSiteAccessTokenReset(Resource):
    @login_required
    @is_admin_or_owner_required
    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
+    @agent_manage_required_for_agent_app
    @account_initialization_required
    @with_current_user
    @with_session
@@ -359,6 +359,12 @@ class WorkflowPublishResponse(ResponseModel):
    created_at: int


+class SyncDraftWorkflowResponse(ResponseModel):
+    result: str
+    hash: str
+    updated_at: int
+
+
 class WorkflowRestoreResponse(ResponseModel):
    result: str
    hash: str
@@ -441,6 +447,7 @@ register_response_schema_models(
    WorkflowOnlineUsersByApp,
    WorkflowOnlineUsersResponse,
    WorkflowPublishResponse,
+    SyncDraftWorkflowResponse,
    WorkflowRestoreResponse,
    DefaultBlockConfigsResponse,
    DefaultBlockConfigResponse,
@@ -556,14 +563,7 @@ class DraftWorkflowApi(Resource):
    @console_ns.response(
        200,
        "Draft workflow synced successfully",
-        console_ns.model(
-            "SyncDraftWorkflowResponse",
-            {
-                "result": fields.String,
-                "hash": fields.String,
-                "updated_at": fields.String,
-            },
-        ),
+        console_ns.models[SyncDraftWorkflowResponse.__name__],
    )
    @console_ns.response(400, "Invalid workflow configuration")
    @console_ns.response(403, "Permission denied")
@@ -618,11 +618,14 @@ class DraftWorkflowApi(Resource):
        except VariableError as e:
            raise InvalidArgumentError(description=str(e))

-        return {
-            "result": "success",
-            "hash": workflow.unique_hash,
-            "updated_at": TimestampField().format(workflow.updated_at or workflow.created_at),
-        }
+        return dump_response(
+            SyncDraftWorkflowResponse,
+            {
+                "result": "success",
+                "hash": workflow.unique_hash,
+                "updated_at": TimestampField().format(workflow.updated_at or workflow.created_at),
+            },
+        )


@console_ns.route("/apps/<uuid:app_id>/advanced-chat/workflows/draft/run")
@@ -12,14 +12,22 @@ from typing import cast, overload
 from sqlalchemy import select
 from sqlalchemy.orm import Session

+from configs import dify_config
 from controllers.common.session import with_session
+from controllers.common.wraps import RBACPermission, RBACResourceScope, enforce_rbac_access
 from controllers.console.app.error import AppNotFoundError
 from extensions.ext_database import db
 from libs.login import current_account_with_tenant
 from models import App, AppMode, TrialApp
+from models.agent import AgentScope
 from services.recommended_app_service import RecommendedAppService

-__all__ = ["get_app_model", "get_app_model_with_trial", "with_session"]
+__all__ = [
+    "agent_manage_required_for_agent_app",
+    "get_app_model",
+    "get_app_model_with_trial",
+    "with_session",
+]


 def _load_app_model(session: Session, app_id: str) -> App | None:
@@ -48,6 +56,45 @@ def _load_app_model_with_trial(session: Session, app_id: str) -> App | None:
    return app_model


+def agent_manage_required_for_agent_app[**P, R](view: Callable[P, R]) -> Callable[P, R]:
+    """Gate generic app management routes that target an Agent App.
+
+    A hidden workflow-only backing App only reuses the App runtime and is not
+    part of the general app management plane, so generic routes reject it
+    outright. Managing a roster Agent App mutates the roster Agent behind it
+    (rename/icon sync, archive, API enablement), so it additionally requires
+    workspace ``agent.manage`` on top of the route's existing App permission
+    checks when RBAC is enabled. A no-op for non-agent Apps. Must be placed
+    above ``get_app_model`` so the ``app_id`` path parameter is still present.
+    """
+
+    @wraps(view)
+    def decorated(*args: P.args, **kwargs: P.kwargs) -> R:
+        raw_app_id = kwargs.get("app_id") or kwargs.get("resource_id")
+        if raw_app_id is not None:
+            app_model = _load_app_model_from_scoped_session(str(raw_app_id))
+            binding = (
+                app_model.agent_app_binding_with_session(session=db.session(), include_archived=True)
+                if app_model is not None
+                else None
+            )
+            if binding is not None:
+                if binding.scope == AgentScope.WORKFLOW_ONLY:
+                    raise AppNotFoundError()
+                if dify_config.RBAC_ENABLED:
+                    current_user, current_tenant_id = current_account_with_tenant()
+                    enforce_rbac_access(
+                        tenant_id=current_tenant_id,
+                        account_id=current_user.id,
+                        resource_type=RBACResourceScope.WORKSPACE,
+                        scene=RBACPermission.AGENT_MANAGE,
+                        resource_required=False,
+                    )
+        return view(*args, **kwargs)
+
+    return decorated
+
+
 def _get_injected_session(args: tuple[object, ...]) -> Session | None:
    """Return the request session inserted by `with_session`, if this handler has been migrated."""
    if len(args) < 2:
@@ -7,10 +7,13 @@ from configs import dify_config
 from constants.languages import supported_language
 from controllers.common.schema import query_params_from_model, register_schema_models
 from controllers.console import console_ns
+from controllers.console.auth.error import InvitationAccountMismatchError
 from controllers.console.error import AccountInFreezeError, AlreadyActivateError
 from extensions.ext_database import db
 from libs.datetime_utils import naive_utc_now
 from libs.helper import EmailStr, timezone
+from libs.login import current_account_with_tenant
+from libs.token import extract_access_token
 from models import AccountStatus
 from models.account import TenantAccountJoin, TenantAccountRole
 from services.account_service import RegisterService, TenantService
@@ -136,6 +139,12 @@ class ActivateApi(Resource):
    )
    @console_ns.response(400, "Already activated or invalid token")
    def post(self):
+        """Accept an invitation without letting an existing session act for another account.
+
+        Token-only activation remains available for legacy clients. When the request already
+        carries a console session, that session must belong to the account encoded in the
+        invitation before the token is consumed or tenant membership is changed.
+        """
        args = ActivatePayload.model_validate(console_ns.payload)

        normalized_request_email = args.email.lower() if args.email else None
@@ -146,6 +155,11 @@ class ActivateApi(Resource):
            raise AlreadyActivateError()

        account = invitation["account"]
+        if extract_access_token(request):
+            current_account, _ = current_account_with_tenant()
+            if current_account.id != account.id:
+                raise InvitationAccountMismatchError()
+
        if dify_config.BILLING_ENABLED and BillingService.is_email_in_freeze(account.email):
            raise AccountInFreezeError()

@@ -13,6 +13,12 @@ class InvalidEmailError(BaseHTTPException):
    code = 400


+class InvitationAccountMismatchError(BaseHTTPException):
+    error_code = "invitation_account_mismatch"
+    description = "This invitation was sent to another account. Please sign in with the invited account."
+    code = 403
+
+
 class PasswordMismatchError(BaseHTTPException):
    error_code = "password_mismatch"
    description = "The passwords do not match."
@@ -162,9 +162,10 @@ class LoginApi(Resource):
        # SELF_HOSTED only have one workspace
        tenants = TenantService.get_join_tenants(account, session=db.session())
        if len(tenants) == 0:
-            system_features = FeatureService.get_system_features()
-
-            if system_features.is_allow_create_workspace and not system_features.license.workspaces.is_available():
+            if (
+                FeatureService.get_system_features().is_allow_create_workspace
+                and not FeatureService.get_license().workspaces.is_available()
+            ):
                raise WorkspacesLimitExceeded()
            else:
                return SimpleResultOptionalDataResponse(
@@ -310,7 +311,7 @@ class EmailCodeLoginApi(Resource):
        if account:
            tenants = TenantService.get_join_tenants(account, session=db.session())
            if not tenants:
-                workspaces = FeatureService.get_system_features().license.workspaces
+                workspaces = FeatureService.get_license().workspaces
                if not workspaces.is_available():
                    raise WorkspacesLimitExceeded()
                if not FeatureService.get_system_features().is_allow_create_workspace:
@@ -6,6 +6,7 @@ from flask import current_app, redirect, request
 from flask_restx import Resource
 from pydantic import BaseModel, Field
 from werkzeug.exceptions import Unauthorized
+from werkzeug.wrappers import Response

 from configs import dify_config
 from constants.languages import languages
@@ -127,6 +128,20 @@ def _preferred_interface_language(language: str | None = None) -> str:
    return languages[0]


+def _redirect_with_console_session(account: Account, target_url: str) -> Response:
+    """Create a console session and attach its cookies to a redirect response."""
+    token_pair = AccountService.login(
+        account=account,
+        session=db.session(),
+        ip_address=extract_remote_ip(request),
+    )
+    response = redirect(target_url)
+    set_access_token_to_cookie(request, response, token_pair.access_token)
+    set_refresh_token_to_cookie(request, response, token_pair.refresh_token)
+    set_csrf_token_to_cookie(request, response, token_pair.csrf_token)
+    return response
+
+
@console_ns.route("/oauth/login/<provider>")
 class OAuthLogin(Resource):
    @console_ns.doc("oauth_login")
@@ -195,16 +210,26 @@ class OAuthCallback(Resource):
            return redirect(f"{dify_config.CONSOLE_WEB_URL}/signin?message={urllib.parse.quote(str(e))}")

        if invite_token and RegisterService.is_valid_invite_token(invite_token):
-            invitation = RegisterService.get_invitation_by_token(token=invite_token)
-            if invitation:
-                invitation_email = invitation.get("email", None)
-                invitation_email_normalized = (
-                    invitation_email.lower() if isinstance(invitation_email, str) else invitation_email
-                )
-                if invitation_email_normalized != user_info.email.lower():
-                    return redirect(f"{dify_config.CONSOLE_WEB_URL}/signin?message=Invalid invitation token.")
+            invitation = RegisterService.get_invitation_if_token_valid(
+                None,
+                None,
+                invite_token,
+                session=db.session(),
+            )
+            if not invitation:
+                return redirect(f"{dify_config.CONSOLE_WEB_URL}/signin?message=Invalid invitation token.")
+            if invitation["data"]["email"].lower() != user_info.email.lower():
+                message = "This invitation was sent to another account. Please sign in with the invited account."
+                query = urllib.parse.urlencode({"message": message, "invite_token": invite_token})
+                return redirect(f"{dify_config.CONSOLE_WEB_URL}/signin?{query}")

-            return redirect(f"{dify_config.CONSOLE_WEB_URL}/signin/invite-settings?invite_token={invite_token}")
+            account = invitation["account"]
+            if account.status == AccountStatus.BANNED:
+                return redirect(f"{dify_config.CONSOLE_WEB_URL}/signin?message=Account is banned.")
+
+            AccountService.link_account_integrate(provider, user_info.id, account, session=db.session())
+            target_url = f"{dify_config.CONSOLE_WEB_URL}/signin/invite-settings?invite_token={invite_token}"
+            return _redirect_with_console_session(account, target_url)

        try:
            account, oauth_new_user = _generate_account(provider, user_info, timezone=timezone, language=language)
@@ -239,21 +264,10 @@ class OAuthCallback(Resource):
                "?message=Workspace not found, please contact system admin to invite you to join in a workspace."
            )

-        token_pair = AccountService.login(
-            account=account,
-            session=db.session(),
-            ip_address=extract_remote_ip(request),
-        )
-
        target_url = _get_redirect_target(redirect_url)
        query_char = "&" if "?" in target_url else "?"
        target_url = f"{target_url}{query_char}oauth_new_user={str(oauth_new_user).lower()}"
-        response = redirect(target_url)
-
-        set_access_token_to_cookie(request, response, token_pair.access_token)
-        set_refresh_token_to_cookie(request, response, token_pair.refresh_token)
-        set_csrf_token_to_cookie(request, response, token_pair.csrf_token)
-        return response
+        return _redirect_with_console_session(account, target_url)


 def _get_account_by_openid_or_email(provider: str, user_info: OAuthUserInfo) -> Account | None:
@@ -56,6 +56,7 @@ class OAuthProviderTokenResponse(BaseModel):


 class OAuthProviderAccountResponse(BaseModel):
+    id: str
    name: str
    email: str
    avatar: str | None = None
@@ -251,6 +252,7 @@ class OAuthServerUserAccountApi(Resource):
    def post(self, oauth_provider_app: OAuthProviderApp, account: Account):
        return jsonable_encoder(
            {
+                "id": account.id,
                "name": account.name,
                "email": account.email,
                "avatar": account.avatar,
@@ -3,10 +3,11 @@ from flask_restx import Resource
 from controllers.common.schema import register_response_schema_models
 from fields.base import ResponseModel
 from libs.helper import dump_response
-from libs.login import current_account_with_tenant_optional, login_required
+from libs.login import login_required
 from services.feature_service import (
    FeatureModel,
    FeatureService,
+    LicenseModel,
    LimitationModel,
    SystemFeatureModel,
 )
@@ -32,6 +33,7 @@ register_response_schema_models(
    console_ns,
    AppDslVersionResponse,
    FeatureModel,
+    LicenseModel,
    LimitationModel,
    SystemFeatureModel,
    TrialModelsResponse,
@@ -135,8 +137,28 @@ class SystemFeatureApi(Resource):

        Authentication would create circular dependency (can't login without dashboard loading).

-        Only non-sensitive configuration data should be returned by this endpoint.
+        Only non-sensitive configuration data should be returned by this endpoint. Authenticated
+        license detail is served separately by SystemFeatureLicenseApi.
        """
-        current_user, _ = current_account_with_tenant_optional()
-        is_authenticated = current_user is not None
-        return FeatureService.get_system_features(is_authenticated=is_authenticated).model_dump()
+        return FeatureService.get_system_features().model_dump()
+
+
+@console_ns.route("/system-features/license")
+class SystemFeatureLicenseApi(Resource):
+    @console_ns.doc("get_system_license")
+    @console_ns.doc(description="Get license status and usage detail")
+    @console_ns.response(
+        200,
+        "Success",
+        console_ns.models[LicenseModel.__name__],
+    )
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self):
+        """Get full license detail (status, expiry, workspace/seat usage).
+
+        Authenticated counterpart to the license *status* exposed on the public
+        system-features endpoint.
+        """
+        return FeatureService.get_license().model_dump()
@@ -8,6 +8,7 @@ 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
@@ -63,7 +64,7 @@ def validate_init_password(payload: InitValidatePayload) -> InitValidateResponse


 def get_init_validate_status() -> bool:
-    if dify_config.EDITION == "SELF_HOSTED":
+    if dify_config.DEPLOYMENT_EDITION != DeploymentEdition.CLOUD:
        if os.environ.get("INIT_PASSWORD"):
            if session.get("is_init_validated"):
                return True
@@ -8,8 +8,9 @@ can be validated explicitly against the pinned KnowledgeFS contract during devel
 Console auth and contract-specific dataset RBAC run before forwarding. Request
 bodies are capped at 64 MiB, JSON and binary responses have separate bounds,
 SSE responses remain streaming with a bounded idle read timeout, and only safe
-response headers are exposed. Upstream 401 responses become 502 so they cannot
-trigger Dify browser-session recovery; resource-level 403 responses remain 403.
+response headers are exposed. Operation-specific upstream error mappings are
+applied before Console JSON error handling; the default maps 401 to 502 so it
+cannot trigger browser-session recovery and preserves resource-level 403.
 """

 from __future__ import annotations
@@ -27,9 +28,11 @@ from werkzeug.exceptions import (
    BadGateway,
    Forbidden,
    GatewayTimeout,
+    HTTPException,
    NotFound,
    RequestEntityTooLarge,
    ServiceUnavailable,
+    default_exceptions,
 )

 from configs import dify_config
@@ -41,21 +44,28 @@ from controllers.console.wraps import (
 )
 from core.helper import ssrf_proxy
 from libs.login import current_account_with_tenant, login_required
+from services.knowledge_fs_operations import KnowledgeFSMethod
 from services.knowledge_fs_proxy import (
    KnowledgeFSAccessDeniedError,
+    KnowledgeFSAuthorization,
    KnowledgeFSConfigurationError,
-    KnowledgeFSMethod,
    KnowledgeFSRouteNotAllowedError,
    KnowledgeFSTimeoutError,
    KnowledgeFSTransportError,
    KnowledgeFSUpstreamResponse,
    authorize_knowledge_fs_request,
    get_knowledge_fs_operation,
+    proxy_authorized_knowledge_fs_request,
    proxy_knowledge_fs_request,
 )

 logger = logging.getLogger(__name__)

+type _KnowledgeFSRequestForwarder = Callable[
+    [str | None, str | None, bytes | None, bytes | None],
+    KnowledgeFSUpstreamResponse,
+]
+
 _MAX_PROXY_BODY_BYTES = 64 * 1024 * 1024
 _RESPONSE_HEADER_ALLOWLIST = (
    "Cache-Control",
@@ -128,27 +138,25 @@ def _translate_proxy_error(exc: Exception, *, tenant_id: str) -> NoReturn:


 def _knowledge_fs_operation_access_required(
-    view: Callable[[KnowledgeFSMethod, str], ResponseReturnValue],
+    view: Callable[[KnowledgeFSAuthorization], ResponseReturnValue],
 ) -> Callable[[KnowledgeFSMethod, str], ResponseReturnValue]:
    """Authorize one declared operation before billing and request-body work."""

    @wraps(view)
    def decorated(method: KnowledgeFSMethod, upstream_path: str) -> ResponseReturnValue:
-        try:
-            operation = get_knowledge_fs_operation(method, upstream_path)
-        except KnowledgeFSRouteNotAllowedError as exc:
-            raise NotFound() from exc
-
        current_user, tenant_id = current_account_with_tenant()
        try:
-            authorize_knowledge_fs_request(
+            authorization = authorize_knowledge_fs_request(
                account=current_user,
                tenant_id=tenant_id,
-                operation=operation,
+                method=method,
+                path=upstream_path,
            )
+        except KnowledgeFSRouteNotAllowedError as exc:
+            raise NotFound() from exc
        except KnowledgeFSAccessDeniedError as exc:
            _translate_proxy_error(exc, tenant_id=tenant_id)
-        return view(method, upstream_path)
+        return view(authorization)

    return decorated

@@ -190,21 +198,26 @@ def _proxy_response(
    """Expose raw content, status, and allowlisted headers from KnowledgeFS.

    Raises:
-        BadGateway: KnowledgeFS rejects the configured server credential.
-        Forbidden: KnowledgeFS denies the account access to the requested resource.
+        HTTPException: KnowledgeFS returns a status normalized by the operation contract.
    """
    upstream = upstream_result.response
-    if upstream.status_code == HTTPStatus.UNAUTHORIZED:
+    mapped_status = dict(upstream_result.operation.error_status_map).get(upstream.status_code)
+    if mapped_status is not None:
        upstream.close()
-        logger.error(
-            "KnowledgeFS rejected the Dify server credential with HTTP %s for tenant_id=%s",
-            upstream.status_code,
-            tenant_id,
-        )
-        raise BadGateway("KnowledgeFS authentication failed")
-    if upstream.status_code == HTTPStatus.FORBIDDEN:
-        upstream.close()
-        raise Forbidden()
+        description = "KnowledgeFS upstream request failed"
+        if upstream.status_code == HTTPStatus.UNAUTHORIZED:
+            description = "KnowledgeFS authentication failed"
+            logger.error(
+                "KnowledgeFS rejected the Dify server credential with HTTP %s for tenant_id=%s",
+                upstream.status_code,
+                tenant_id,
+            )
+        exception_type = default_exceptions.get(mapped_status)
+        if exception_type is None:
+            exception = HTTPException(description)
+            exception.code = mapped_status
+            raise exception
+        raise exception_type(description)

    allowed_header_names = dict.fromkeys(
        name.lower() for name in (*_RESPONSE_HEADER_ALLOWLIST, *contract_response_headers)
@@ -237,26 +250,21 @@ def _proxy_response(
    return Response(content, status=upstream.status_code, headers=headers)


-def _proxy_request(method: KnowledgeFSMethod, upstream_path: str) -> Response:
-    """Forward the current raw request and return its filtered upstream response.
-
-    The call performs one outbound KnowledgeFS request. Integration failures are
-    converted to Console HTTP exceptions for the outer JSON error adapter.
-    """
+def _proxy_current_request(
+    *,
+    method: KnowledgeFSMethod,
+    tenant_id: str,
+    forward: _KnowledgeFSRequestForwarder,
+) -> Response:
+    """Forward the current raw request through one preconfigured service entry."""
    if not dify_config.KNOWLEDGE_FS_ENABLED:
        raise NotFound()
-    current_user, tenant_id = current_account_with_tenant()
    try:
-        proxy_result = proxy_knowledge_fs_request(
-            account=current_user,
-            method=method,
-            path=upstream_path,
-            tenant_id=tenant_id,
-            accept=request.headers.get("Accept"),
-            content_type=request.content_type,
-            query=request.query_string or None,
-            body=_request_body() if method != "GET" else None,
-            request_headers=request.headers,
+        proxy_result = forward(
+            request.headers.get("Accept"),
+            request.content_type,
+            request.query_string or None,
+            _request_body() if method != "GET" else None,
        )
    except (
        KnowledgeFSConfigurationError,
@@ -274,20 +282,99 @@ def _proxy_request(method: KnowledgeFSMethod, upstream_path: str) -> Response:
    )


+def _proxy_request(
+    method: KnowledgeFSMethod,
+    upstream_path: str,
+) -> Response:
+    """Authorize and forward the current request through the combined service use case."""
+    if not dify_config.KNOWLEDGE_FS_ENABLED:
+        raise NotFound()
+    current_user, tenant_id = current_account_with_tenant()
+
+    def forward(
+        accept: str | None,
+        content_type: str | None,
+        query: bytes | None,
+        body: bytes | None,
+    ) -> KnowledgeFSUpstreamResponse:
+        return proxy_knowledge_fs_request(
+            account=current_user,
+            method=method,
+            path=upstream_path,
+            tenant_id=tenant_id,
+            accept=accept,
+            content_type=content_type,
+            query=query,
+            body=body,
+            request_headers=request.headers,
+        )
+
+    return _proxy_current_request(method=method, tenant_id=tenant_id, forward=forward)
+
+
+def _proxy_authorized_request(authorization: KnowledgeFSAuthorization) -> Response:
+    """Forward the current request using one previously authorized operation capability.
+
+    Args:
+        authorization: Request-scoped capability produced before billing and body parsing.
+
+    Returns:
+        The filtered response returned by KnowledgeFS.
+
+    Raises:
+        HTTPException: The integration is disabled or forwarding fails.
+    """
+    operation = authorization.operation
+    tenant_id = authorization.tenant_id
+
+    def forward(
+        accept: str | None,
+        content_type: str | None,
+        query: bytes | None,
+        body: bytes | None,
+    ) -> KnowledgeFSUpstreamResponse:
+        return proxy_authorized_knowledge_fs_request(
+            authorization=authorization,
+            accept=accept,
+            content_type=content_type,
+            query=query,
+            body=body,
+            request_headers=request.headers,
+        )
+
+    return _proxy_current_request(method=operation.method, tenant_id=tenant_id, forward=forward)
+
+
@_knowledge_fs_enabled
@_knowledge_fs_operation_access_required
@cloud_edition_billing_rate_limit_check("knowledge")
 def _proxy_knowledge_fs_non_get(
-    method: KnowledgeFSMethod,
-    upstream_path: str,
+    authorization: KnowledgeFSAuthorization,
 ) -> ResponseReturnValue:
    """Apply knowledge billing checks to one allowlisted non-GET operation."""
-    return _proxy_request(method, upstream_path)
+    return _proxy_authorized_request(authorization)


@bp.route(
    "/knowledge-fs/<path:upstream_path>",
-    methods=["GET", "OPTIONS"],
+    methods=["OPTIONS"],
+    provide_automatic_options=False,
+)
+@_console_api_errors
+@_knowledge_fs_enabled
+def proxy_knowledge_fs_options(upstream_path: str) -> ResponseReturnValue:
+    """Complete a CORS preflight only for an enabled Console operation."""
+    requested_method = cast(KnowledgeFSMethod, request.headers.get("Access-Control-Request-Method", "").upper())
+    try:
+        get_knowledge_fs_operation(requested_method, upstream_path)
+    except KnowledgeFSRouteNotAllowedError as exc:
+        raise NotFound() from exc
+    return Response(status=HTTPStatus.NO_CONTENT)
+
+
+@bp.route(
+    "/knowledge-fs/<path:upstream_path>",
+    methods=["GET"],
    provide_automatic_options=False,
 )
@_console_api_errors
@@ -6,6 +6,7 @@ from sqlalchemy import select

 from configs import dify_config
 from controllers.fastopenapi import console_router
+from enums.deployment_edition import DeploymentEdition
 from libs.helper import EmailStr, extract_remote_ip
 from libs.password import valid_password
 from models.model import DifySetup, db
@@ -52,7 +53,7 @@ def get_setup_status_api() -> SetupStatusResponse:

    Only bootstrap-safe status information should be returned by this endpoint.
    """
-    if dify_config.EDITION == "SELF_HOSTED":
+    if dify_config.DEPLOYMENT_EDITION != DeploymentEdition.CLOUD:
        setup_status = get_setup_status()
        if setup_status and not isinstance(setup_status, bool):
            return SetupStatusResponse(step="finished", setup_at=setup_status.setup_at.isoformat())
@@ -102,7 +103,7 @@ def setup_system(payload: SetupRequestPayload) -> SetupResponse:


 def get_setup_status() -> DifySetup | bool | None:
-    if dify_config.EDITION == "SELF_HOSTED":
+    if dify_config.DEPLOYMENT_EDITION != DeploymentEdition.CLOUD:
        return db.session.scalar(select(DifySetup).limit(1))

    return True
@@ -46,6 +46,7 @@ from controllers.console.wraps import (
    with_current_tenant_id,
    with_current_user,
 )
+from enums.deployment_edition import DeploymentEdition
 from extensions.ext_database import db
 from fields.base import ResponseModel
 from fields.member_fields import AccountResponse
@@ -262,7 +263,7 @@ class AccountInitApi(Resource):
        payload = console_ns.payload or {}
        args = AccountInitPayload.model_validate(payload)

-        if dify_config.EDITION == "CLOUD":
+        if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD:
            if not args.invitation_code:
                raise ValueError("invitation_code is required")

@@ -198,7 +198,7 @@ def _check_member_invite_limits(tenant_id: str, new_member_count: int, new_accou
        if workspace_members.enabled is True and not workspace_members.is_available(new_member_count):
            raise WorkspaceMembersLimitExceeded()
        if new_account_count > 0:
-            seats = FeatureService.get_system_features(is_authenticated=True).license.seats
+            seats = FeatureService.get_license().seats
            if not seats.is_available(new_account_count):
                raise SeatsLimitExceeded()
        return
@@ -1,5 +1,5 @@
 import io
-from collections.abc import Mapping
+from collections.abc import Mapping, Sequence
 from datetime import datetime
 from typing import Any, Literal, TypedDict

@@ -43,6 +43,7 @@ from core.plugin.entities.plugin_daemon import PluginDecodeResponse, PluginInsta
 from core.plugin.impl.exc import PluginDaemonClientSideError
 from core.plugin.plugin_service import PluginService
 from core.tools.builtin_tool.providers._positions import BuiltinToolProviderSort
+from core.tools.entities.api_entities import ToolProviderApiEntity
 from core.tools.entities.common_entities import I18nObject
 from core.tools.entities.tool_entities import ToolProviderType
 from core.tools.tool_manager import ToolManager
@@ -67,6 +68,15 @@ from services.plugin.plugin_parameter_service import PluginParameterService
 from services.plugin.plugin_permission_service import PluginPermissionService
 from services.tools.tools_transform_service import ToolTransformService

+_PLUGIN_PACKAGE_UPLOAD_PARAMS = {
+    "pkg": {
+        "description": "Plugin package to upload",
+        "in": "formData",
+        "type": "file",
+        "required": True,
+    }
+}
+

 class AutoUpgradeSettingsResponse(TypedDict):
    strategy_setting: TenantPluginAutoUpgradeStrategySetting
@@ -84,6 +94,15 @@ class ParserList(BaseModel):
 class PluginCategoryListQuery(BaseModel):
    page: int = Field(default=1, ge=1, description="Page number")
    page_size: int = Field(default=256, ge=1, le=256, description="Page size (1-256)")
+    query: str = Field(default="", max_length=256, description="Case-insensitive search query")
+    tags: list[str] = Field(default_factory=list, max_length=128, description="Match any plugin tag")
+    language: Literal["en_US", "zh_Hans", "ja_JP", "pt_BR"] = Field(
+        default="en_US", description="Language used for localized label and description search"
+    )
+
+
+class PluginInstalledIdsQuery(BaseModel):
+    category: PluginCategory = Field(description="Plugin category to include")


 class ParserLatest(BaseModel):
@@ -316,6 +335,10 @@ class PluginListResponse(ResponseModel):
    total: int


+class PluginInstalledIdsResponse(ResponseModel):
+    plugin_ids: list[str]
+
+
 class PluginVersionsResponse(ResponseModel):
    versions: Mapping[str, PluginService.LatestPluginCache | None]

@@ -375,6 +398,7 @@ register_schema_models(
    console_ns,
    ParserList,
    PluginCategoryListQuery,
+    PluginInstalledIdsQuery,
    PluginAutoUpgradeSettingsPayload,
    PluginPermissionSettingsPayload,
    ParserLatest,
@@ -411,6 +435,7 @@ register_response_schema_models(
    PluginDebuggingKeyResponse,
    PluginDynamicOptionsResponse,
    PluginInstallationsResponse,
+    PluginInstalledIdsResponse,
    PluginInstallTaskStartResponse,
    PluginListResponse,
    PluginManifestResponse,
@@ -433,12 +458,10 @@ register_enum_models(
 )


-def _default_auto_upgrade_settings(
-    tenant_id: str,
-    category: TenantPluginAutoUpgradeCategory,
-) -> AutoUpgradeSettingsResponse:
+def _missing_auto_upgrade_settings(tenant_id: str) -> AutoUpgradeSettingsResponse:
+    """Represent a missing persisted strategy as effectively disabled."""
    return {
-        "strategy_setting": PluginAutoUpgradeService.default_strategy_setting_for_category(category),
+        "strategy_setting": TenantPluginAutoUpgradeStrategySetting.DISABLED,
        "upgrade_time_of_day": PluginAutoUpgradeService.default_upgrade_time_of_day(tenant_id),
        "upgrade_mode": TenantPluginAutoUpgradeMode.EXCLUDE,
        "exclude_plugins": [],
@@ -470,7 +493,39 @@ def _read_upload_content(file: FileStorage, max_size: int) -> bytes:
    return content


-def _list_hardcoded_builtin_tool_providers(tenant_id: str) -> list[dict[str, Any]]:
+def _localized_builtin_tool_text(value: I18nObject, language: str) -> str:
+    return getattr(value, language, None) or value.en_US
+
+
+def _builtin_tool_provider_matches_filters(
+    provider: ToolProviderApiEntity,
+    *,
+    query: str,
+    tags: Sequence[str],
+    language: str,
+) -> bool:
+    if tags and not any(tag in provider.labels for tag in tags):
+        return False
+    if not query:
+        return True
+
+    lower_query = query.lower()
+    candidates = (
+        provider.name,
+        _localized_builtin_tool_text(provider.label, language),
+        _localized_builtin_tool_text(provider.description, language),
+    )
+    return any(lower_query in candidate.lower() for candidate in candidates)
+
+
+def _list_hardcoded_builtin_tool_providers(
+    tenant_id: str,
+    *,
+    query: str = "",
+    tags: Sequence[str] = (),
+    language: str = "en_US",
+) -> list[dict[str, Any]]:
+    """List builtin providers using the same search and tag semantics as category plugins."""
    db_builtin_providers = {
        str(ToolProviderID(provider.provider)): provider
        for provider in ToolManager.list_default_builtin_providers(tenant_id)
@@ -491,6 +546,13 @@ def _list_hardcoded_builtin_tool_providers(tenant_id: str) -> list[dict[str, Any
            db_provider=db_builtin_providers.get(provider.entity.identity.name),
            decrypt_credentials=False,
        )
+        if not _builtin_tool_provider_matches_filters(
+            user_provider,
+            query=query,
+            tags=tags,
+            language=language,
+        ):
+            continue
        ToolTransformService.repack_provider(tenant_id=tenant_id, provider=user_provider)
        builtin_providers.append(user_provider)

@@ -545,7 +607,9 @@ class PluginCategoryListApi(Resource):
    @account_initialization_required
    @with_current_tenant_id
    def get(self, tenant_id: str, category: str):
-        args = PluginCategoryListQuery.model_validate(request.args.to_dict(flat=True))
+        args = PluginCategoryListQuery.model_validate(
+            {**request.args.to_dict(flat=True), "tags": request.args.getlist("tags")}
+        )

        try:
            plugin_category = PluginCategory(category)
@@ -553,13 +617,26 @@ class PluginCategoryListApi(Resource):
            return {"code": "invalid_param", "message": "invalid plugin category"}, 400

        try:
-            plugins = PluginService.list_by_category(tenant_id, plugin_category, args.page, args.page_size)
+            plugins = PluginService.list_by_category(
+                tenant_id,
+                plugin_category,
+                args.page,
+                args.page_size,
+                query=args.query,
+                tags=args.tags,
+                language=args.language,
+            )
        except PluginDaemonClientSideError as e:
            return {"code": "plugin_error", "message": e.description}, 400

        builtin_tools = []
        if plugin_category == PluginCategory.Tool:
-            builtin_tools = _list_hardcoded_builtin_tool_providers(tenant_id)
+            builtin_tools = _list_hardcoded_builtin_tool_providers(
+                tenant_id,
+                query=args.query,
+                tags=args.tags,
+                language=args.language,
+            )

        return dump_response(
            PluginCategoryListResponse,
@@ -571,6 +648,24 @@ class PluginCategoryListApi(Resource):
        )


+@console_ns.route("/workspaces/current/plugin/installed-ids")
+class PluginInstalledIdsApi(Resource):
+    @console_ns.doc(params=query_params_from_model(PluginInstalledIdsQuery))
+    @console_ns.response(200, "Success", console_ns.models[PluginInstalledIdsResponse.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @with_current_tenant_id
+    def get(self, tenant_id: str):
+        args = PluginInstalledIdsQuery.model_validate(request.args.to_dict(flat=True))
+        try:
+            plugin_ids = PluginService.list_installed_plugin_ids(tenant_id, args.category)
+        except PluginDaemonClientSideError as e:
+            return {"code": "plugin_error", "message": e.description}, 400
+
+        return dump_response(PluginInstalledIdsResponse, {"plugin_ids": plugin_ids})
+
+
@console_ns.route("/workspaces/current/plugin/list/latest-versions")
 class PluginListLatestVersionsApi(Resource):
    @console_ns.expect(console_ns.models[ParserLatest.__name__])
@@ -645,6 +740,7 @@ class PluginAssetApi(Resource):

@console_ns.route("/workspaces/current/plugin/upload/pkg")
 class PluginUploadFromPkgApi(Resource):
+    @console_ns.doc(consumes=["multipart/form-data"], params=_PLUGIN_PACKAGE_UPLOAD_PARAMS)
    @console_ns.response(200, "Success", console_ns.models[PluginDecodeResponse.__name__])
    @setup_required
    @login_required
@@ -1125,9 +1221,7 @@ class PluginFetchAutoUpgradeApi(Resource):
        args = ParserAutoUpgradeFetch.model_validate(request.args.to_dict(flat=True))
        auto_upgrade = PluginAutoUpgradeService.get_strategy(tenant_id, args.category, session=db.session())
        auto_upgrade_dict = (
-            _auto_upgrade_settings_to_dict(auto_upgrade)
-            if auto_upgrade
-            else _default_auto_upgrade_settings(tenant_id, args.category)
+            _auto_upgrade_settings_to_dict(auto_upgrade) if auto_upgrade else _missing_auto_upgrade_settings(tenant_id)
        )

        return jsonable_encoder(
@@ -37,6 +37,7 @@ from controllers.console.wraps import (
    with_current_user,
 )
 from enums.cloud_plan import CloudPlan
+from enums.deployment_edition import DeploymentEdition
 from extensions.ext_database import db
 from fields.base import ResponseModel
 from libs.helper import dump_response, to_timestamp
@@ -233,7 +234,7 @@ class TenantListApi(Resource):
        tenants = [tenant for tenant, _ in tenant_rows]
        tenant_dicts = []
        is_enterprise_only = dify_config.ENTERPRISE_ENABLED and not dify_config.BILLING_ENABLED
-        is_saas = dify_config.EDITION == "CLOUD" and dify_config.BILLING_ENABLED
+        is_saas = dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD and dify_config.BILLING_ENABLED
        tenant_plans: dict[str, SubscriptionPlan] = {}

        if is_saas:
@@ -20,6 +20,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 extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from libs.encryption import FieldEncryption
@@ -129,7 +130,7 @@ def account_initialization_required[R](view: Callable[..., R]) -> Callable[...,
 def only_edition_cloud[**P, R](view: Callable[P, R]) -> Callable[P, R]:
    @wraps(view)
    def decorated(*args: P.args, **kwargs: P.kwargs):
-        if dify_config.EDITION != "CLOUD":
+        if dify_config.DEPLOYMENT_EDITION != DeploymentEdition.CLOUD:
            abort(404)

        return view(*args, **kwargs)
@@ -151,7 +152,7 @@ def only_edition_enterprise[**P, R](view: Callable[P, R]) -> Callable[P, R]:
 def only_edition_self_hosted[**P, R](view: Callable[P, R]) -> Callable[P, R]:
    @wraps(view)
    def decorated(*args: P.args, **kwargs: P.kwargs):
-        if dify_config.EDITION != "SELF_HOSTED":
+        if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD:
            abort(404)

        return view(*args, **kwargs)
@@ -327,7 +328,7 @@ def setup_required[R](view: Callable[..., R]) -> Callable[..., R]:
        # The overloads keep Resource methods method-aware for pyrefly while
        # preserving support for plain functions used in tests and utilities.
        # check setup
-        if dify_config.EDITION == "SELF_HOSTED" and not _is_setup_completed():
+        if dify_config.DEPLOYMENT_EDITION != DeploymentEdition.CLOUD and not _is_setup_completed():
            if os.environ.get("INIT_PASSWORD"):
                raise NotInitValidateError()
            raise NotSetupError()
@@ -8,6 +8,7 @@ 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
@@ -26,7 +27,7 @@ class CallerKind(StrEnum):


 def current_edition() -> Edition:
-    if dify_config.EDITION == "CLOUD":
+    if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD:
        return Edition.SAAS
    if dify_config.ENTERPRISE_ENABLED:
        return Edition.EE
@@ -1,6 +1,6 @@
 from typing import Any, Self

-from pydantic import AliasChoices, Field, computed_field
+from pydantic import AliasChoices, Field
 from sqlalchemy import select
 from werkzeug.exceptions import Forbidden

@@ -8,12 +8,15 @@ 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 extensions.ext_database import db
+from extensions.storage.storage_type import StorageType
 from fields.base import ResponseModel
 from libs.helper import build_icon_url
 from models.account import Tenant, TenantStatus
-from models.model import App, EndUser, Site
+from models.model import App, EndUser, IconType, Site
 from services.feature_service import FeatureModel, FeatureService
+from services.file_service import FileService


 class WebSiteResponse(ResponseModel):
@@ -32,11 +35,7 @@ class WebSiteResponse(ResponseModel):
    prompt_public: bool | None = None
    show_workflow_steps: bool | None = None
    use_icon_as_answer_icon: bool | None = None
-
-    @computed_field(return_type=str | None)  # type: ignore[prop-decorator]
-    @property
-    def icon_url(self) -> str | None:
-        return build_icon_url(self.icon_type, self.icon)
+    icon_url: str | None = None


 class WebModelConfigResponse(ResponseModel):
@@ -88,6 +87,7 @@ class WebAppSiteResponse(ResponseModel):
        end_user_id: str | None,
        features: FeatureModel,
        can_replace_logo: bool,
+        icon_url: str | None = None,
    ) -> Self:
        custom_config = None
        if can_replace_logo:
@@ -102,6 +102,7 @@ class WebAppSiteResponse(ResponseModel):
            )

        site_response = WebSiteResponse.model_validate(site, from_attributes=True)
+        site_response.icon_url = icon_url if icon_url is not None else build_icon_url(site.icon_type, site.icon)
        if features.billing.enabled and not features.webapp_copyright_enabled:
            site_response.copyright = None
            site_response.input_placeholder = None
@@ -123,6 +124,17 @@ register_response_schema_models(
 )


+def _build_site_icon_url(*, site: Site, tenant_id: str) -> str | None:
+    """Use direct S3 URLs only in Cloud Mode and preserve preview URLs elsewhere."""
+    if site.icon_type != IconType.IMAGE or not site.icon:
+        return None
+    if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD and (
+        StorageType(dify_config.STORAGE_TYPE) == StorageType.S3
+    ):
+        return FileService(db.engine).get_file_presigned_url(file_id=site.icon, tenant_id=tenant_id)
+    return build_icon_url(site.icon_type, site.icon)
+
+
@web_ns.route("/site")
 class AppSiteApi(WebApiResource):
    @web_ns.doc("Get App Site Info")
@@ -159,4 +171,5 @@ class AppSiteApi(WebApiResource):
            end_user_id=end_user.id,
            features=features,
            can_replace_logo=features.can_replace_logo,
+            icon_url=_build_site_icon_url(site=site, tenant_id=tenant.id),
        ).model_dump(mode="json")
@@ -11,6 +11,7 @@ from werkzeug.exceptions import BadRequest, NotFound, Unauthorized

 from constants import HEADER_NAME_APP_CODE
 from controllers.web.error import WebAppAuthAccessDeniedError, WebAppAuthRequiredError
+from core.logging.context import set_identity_context
 from extensions.ext_database import db
 from libs.passport import PassportService
 from libs.token import extract_webapp_passport
@@ -28,6 +29,11 @@ def validate_jwt_token[**P, R](
        @wraps(view)
        def decorated(*args: P.args, **kwargs: P.kwargs) -> R:
            app_model, end_user = decode_jwt_token()
+            set_identity_context(
+                tenant_id=end_user.tenant_id,
+                user_id=end_user.id,
+                user_type=end_user.type or "end_user",
+            )
            return view(app_model, end_user, *args, **kwargs)

        return decorated
@@ -0,0 +1,93 @@
+"""Publication visibility rules for calling roster Agents from Workflows.
+
+``Agent.active_config_is_published`` describes whether the editable shared
+draft still matches the active snapshot. It is false both before the first
+publish and after a published Agent receives new draft edits, so it must not be
+used as a runtime availability flag. App-backed Agents are callable from a
+Workflow only when the active snapshot has a revision created by a
+publish-visible operation. Direct roster Agents are publish-visible by
+construction and only need an active snapshot.
+"""
+
+from sqlalchemy import and_, or_, select
+from sqlalchemy.orm import Session
+from sqlalchemy.sql.elements import ColumnElement
+
+from models.agent import Agent, AgentConfigRevision, AgentConfigRevisionOperation, AgentScope, AgentSource
+
+PUBLISH_VISIBLE_APP_BACKED_REVISION_OPERATIONS = frozenset(
+    {
+        AgentConfigRevisionOperation.PUBLISH_DRAFT,
+        AgentConfigRevisionOperation.SAVE_CURRENT_VERSION,
+        AgentConfigRevisionOperation.SAVE_NEW_VERSION,
+        AgentConfigRevisionOperation.SAVE_NEW_AGENT,
+        AgentConfigRevisionOperation.SAVE_TO_ROSTER,
+        AgentConfigRevisionOperation.RESTORE_VERSION,
+    }
+)
+
+
+def workflow_callable_active_snapshot_filter() -> ColumnElement[bool]:
+    """Return the SQL predicate for an Agent with a Workflow-callable active snapshot.
+
+    The caller remains responsible for tenant, roster scope, lifecycle status,
+    and model configuration filters. The correlated revision lookup makes the
+    predicate safe to compose into roster pagination queries.
+    """
+
+    app_backed_agent = or_(
+        Agent.source == AgentSource.AGENT_APP,
+        and_(
+            Agent.source == AgentSource.IMPORTED,
+            Agent.scope == AgentScope.ROSTER,
+            Agent.app_id.is_not(None),
+        ),
+    )
+    publish_visible_revision_exists = (
+        select(AgentConfigRevision.id)
+        .where(
+            AgentConfigRevision.tenant_id == Agent.tenant_id,
+            AgentConfigRevision.agent_id == Agent.id,
+            AgentConfigRevision.current_snapshot_id == Agent.active_config_snapshot_id,
+            AgentConfigRevision.operation.in_(PUBLISH_VISIBLE_APP_BACKED_REVISION_OPERATIONS),
+        )
+        .correlate(Agent)
+        .exists()
+    )
+    return and_(
+        Agent.active_config_snapshot_id.is_not(None),
+        or_(
+            ~app_backed_agent,
+            publish_visible_revision_exists,
+        ),
+    )
+
+
+def agent_has_workflow_callable_active_snapshot(*, session: Session, agent: Agent) -> bool:
+    """Return whether ``agent`` has an active snapshot visible to Workflow.
+
+    This object-level form is useful after ownership and lifecycle checks have
+    already loaded an Agent. It intentionally ignores dirty draft state so a
+    previously published snapshot keeps serving while later edits remain
+    unpublished.
+    """
+
+    if not agent.active_config_snapshot_id:
+        return False
+    is_app_backed = agent.source == AgentSource.AGENT_APP or (
+        agent.source == AgentSource.IMPORTED and agent.scope == AgentScope.ROSTER and agent.app_id is not None
+    )
+    if not is_app_backed:
+        return True
+    return bool(
+        session.scalar(
+            select(AgentConfigRevision.id)
+            .where(
+                AgentConfigRevision.tenant_id == agent.tenant_id,
+                AgentConfigRevision.agent_id == agent.id,
+                AgentConfigRevision.current_snapshot_id == agent.active_config_snapshot_id,
+                AgentConfigRevision.operation.in_(PUBLISH_VISIBLE_APP_BACKED_REVISION_OPERATIONS),
+            )
+            .limit(1)
+        )
+    )
@@ -682,42 +682,27 @@ class AgentAppGenerator(MessageBasedAppGenerator):
            if draft_type == AgentConfigDraftType.DEBUG_BUILD.value
            else AgentConfigDraftType.DRAFT
        )
+        if effective_draft_type == AgentConfigDraftType.DRAFT:
+            from services.agent.composer_service import AgentComposerService
+
+            return AgentComposerService.get_or_create_normal_agent_draft(
+                session=session,
+                tenant_id=tenant_id,
+                agent=agent,
+                created_by=agent.updated_by or agent.created_by,
+            )
+        if not account_id:
+            raise AgentAppGeneratorError("Build draft requires an account user")
        stmt = select(AgentConfigDraft).where(
            AgentConfigDraft.tenant_id == tenant_id,
            AgentConfigDraft.agent_id == agent.id,
-            AgentConfigDraft.draft_type == effective_draft_type,
+            AgentConfigDraft.draft_type == AgentConfigDraftType.DEBUG_BUILD,
+            AgentConfigDraft.account_id == account_id,
        )
-        if effective_draft_type == AgentConfigDraftType.DEBUG_BUILD:
-            if not account_id:
-                raise AgentAppGeneratorError("Build draft requires an account user")
-            stmt = stmt.where(AgentConfigDraft.account_id == account_id)
-        else:
-            stmt = stmt.where(AgentConfigDraft.account_id.is_(None))
        draft = session.scalar(stmt.order_by(AgentConfigDraft.updated_at.desc()).limit(1))
        if draft is not None:
            return draft
-        if effective_draft_type == AgentConfigDraftType.DEBUG_BUILD:
-            raise AgentAppGeneratorError("Agent build draft not found")
-        _, snapshot, agent_soul = AgentAppGenerator._resolve_agent_by_id(
-            tenant_id=tenant_id,
-            agent_id=agent.id,
-            snapshot_id=agent.active_config_snapshot_id,
-            session=session,
-        )
-        draft = AgentConfigDraft(
-            tenant_id=tenant_id,
-            agent_id=agent.id,
-            draft_type=AgentConfigDraftType.DRAFT,
-            account_id=None,
-            draft_owner_key="",
-            base_snapshot_id=snapshot.id,
-            config_snapshot=agent_soul,
-            created_by=agent.created_by,
-            updated_by=agent.updated_by,
-        )
-        session.add(draft)
-        session.flush()
-        return draft
+        raise AgentAppGeneratorError("Agent build draft not found")

    @staticmethod
    def _resolve_agent_by_id(
@@ -6,6 +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 graphon.model_runtime.entities.model_entities import ModelType


@@ -49,7 +50,7 @@ class HostingConfiguration:
        self.moderation_config = None

    def init_app(self, app: Flask):
-        if dify_config.EDITION != "CLOUD":
+        if dify_config.DEPLOYMENT_EDITION != DeploymentEdition.CLOUD:
            return

        self.provider_map[f"{DEFAULT_PLUGIN_ID}/azure_openai/azure_openai"] = self.init_azure_openai()
@@ -21,6 +21,7 @@ from core.model_manager import ModelInstance, ModelManager
 from core.rag.cleaner.clean_processor import CleanProcessor
 from core.rag.datasource.keyword.keyword_factory import Keyword
 from core.rag.docstore.dataset_docstore import DatasetDocumentStore
+from core.rag.embedding.token_counter import calculate_segment_token_counts
 from core.rag.extractor.entity.datasource_type import DatasourceType
 from core.rag.extractor.entity.extract_setting import ExtractSetting, NotionInfo, WebsiteInfo
 from core.rag.index_processor.constant.index_type import IndexStructureType, IndexTechniqueType
@@ -113,17 +114,25 @@ class IndexingRunner:
                    current_user=current_user,
                    session=session,
                )
+                token_counts = calculate_segment_token_counts(dataset=dataset, documents=documents)
+                total_tokens = sum(token_counts)
                # save segment
-                self._load_segments(dataset, requeried_document, documents, session)
+                self._load_segments(
+                    session=session,
+                    dataset=dataset,
+                    dataset_document=requeried_document,
+                    documents=documents,
+                    token_counts=token_counts,
+                )
                session.commit()

                # load
                self._load(
-                    index_processor=index_processor,
+                    session=session,
                    dataset=dataset,
                    dataset_document=requeried_document,
                    documents=documents,
-                    session=session,
+                    total_tokens=total_tokens,
                )
            except DocumentIsPausedError:
                raise DocumentIsPausedError(f"Document paused, document id: {document_id}")
@@ -190,17 +199,25 @@ class IndexingRunner:
                current_user=current_user,
                session=session,
            )
+            token_counts = calculate_segment_token_counts(dataset=dataset, documents=documents)
+            total_tokens = sum(token_counts)
            # save segment
-            self._load_segments(dataset, requeried_document, documents, session)
+            self._load_segments(
+                session=session,
+                dataset=dataset,
+                dataset_document=requeried_document,
+                documents=documents,
+                token_counts=token_counts,
+            )
            session.commit()

            # load
            self._load(
-                index_processor=index_processor,
+                session=session,
                dataset=dataset,
                dataset_document=requeried_document,
                documents=documents,
-                session=session,
+                total_tokens=total_tokens,
            )
        except DocumentIsPausedError:
            raise DocumentIsPausedError(f"Document paused, document id: {document_id}")
@@ -225,7 +242,7 @@ class IndexingRunner:
            if not dataset:
                raise ValueError("no dataset found")

-            # get exist document_segment list and delete
+            # get existing document segments
            document_segments = session.scalars(
                select(DocumentSegment).where(
                    DocumentSegment.dataset_id == dataset.id,
@@ -264,15 +281,15 @@ class IndexingRunner:
                                    child_documents.append(child_document)
                                document.children = child_documents
                        documents.append(document)
+            # Preserve the full document total even when only incomplete segments are re-indexed.
+            total_tokens = sum(document_segment.tokens for document_segment in document_segments)
            # build index
-            index_type = requeried_document.doc_form
-            index_processor = IndexProcessorFactory(index_type).init_index_processor()
            self._load(
-                index_processor=index_processor,
+                session=session,
                dataset=dataset,
                dataset_document=requeried_document,
                documents=documents,
-                session=session,
+                total_tokens=total_tokens,
            )
        except DocumentIsPausedError:
            raise DocumentIsPausedError(f"Document paused, document id: {document_id}")
@@ -601,28 +618,16 @@ class IndexingRunner:

    def _load(
        self,
-        index_processor: BaseIndexProcessor,
+        session: Session,
        dataset: Dataset,
        dataset_document: DatasetDocument,
        documents: list[Document],
-        session: Session,
-    ):
-        """
-        insert index and update document/segment status to completed
-        """
+        total_tokens: int,
+    ) -> None:
+        """Build indexes and mark the document complete using the token total computed before hash sharding."""

-        embedding_model_instance = None
-        if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
-            embedding_model_instance = self._get_model_manager(dataset.tenant_id).get_model_instance(
-                tenant_id=dataset.tenant_id,
-                provider=dataset.embedding_model_provider,
-                model_type=ModelType.TEXT_EMBEDDING,
-                model=dataset.embedding_model,
-            )
-
-        # chunk nodes by chunk size
+        # Build indexes using the existing hash-based worker groups.
        indexing_start_at = time.perf_counter()
-        tokens = 0
        create_keyword_thread = None
        if (
            dataset_document.doc_form != IndexStructureType.PARENT_CHILD_INDEX
@@ -659,12 +664,11 @@ class IndexingRunner:
                            chunk_documents,
                            dataset.id,
                            dataset_document.id,
-                            embedding_model_instance,
                        )
                    )

                for future in futures:
-                    tokens += future.result()
+                    future.result()
        if (
            dataset_document.doc_form != IndexStructureType.PARENT_CHILD_INDEX
            and dataset.indexing_technique == IndexTechniqueType.ECONOMY
@@ -679,7 +683,7 @@ class IndexingRunner:
            document_id=dataset_document.id,
            after_indexing_status=IndexingStatus.COMPLETED,
            extra_update_params={
-                DatasetDocument.tokens: tokens,
+                DatasetDocument.tokens: total_tokens,
                DatasetDocument.completed_at: naive_utc_now(),
                DatasetDocument.indexing_latency: indexing_end_at - indexing_start_at,
                DatasetDocument.error: None,
@@ -720,8 +724,7 @@ class IndexingRunner:
        chunk_documents: list[Document],
        dataset_id: str,
        dataset_document_id: str,
-        embedding_model_instance: ModelInstance | None,
-    ):
+    ) -> None:
        with flask_app.app_context():
            with session_factory.create_session() as session:
                dataset = session.get(Dataset, dataset_id)
@@ -735,11 +738,6 @@ class IndexingRunner:
                # check document is paused
                self._check_document_paused_status(dataset_document.id)

-                tokens = 0
-                if embedding_model_instance:
-                    page_content_list = [document.page_content for document in chunk_documents]
-                    tokens += sum(embedding_model_instance.get_text_embedding_num_tokens(page_content_list))
-
                multimodal_documents = []
                for document in chunk_documents:
                    if document.attachments and dataset.is_multimodal:
@@ -773,8 +771,6 @@ class IndexingRunner:

                session.commit()

-                return tokens
-
    @staticmethod
    def _check_document_paused_status(document_id: str):
        indexing_cache_key = f"document_{document_id}_is_paused"
@@ -864,8 +860,14 @@ class IndexingRunner:
        return documents

    def _load_segments(
-        self, dataset: Dataset, dataset_document: DatasetDocument, documents: list[Document], session: Session
-    ):
+        self,
+        session: Session,
+        dataset: Dataset,
+        dataset_document: DatasetDocument,
+        documents: list[Document],
+        token_counts: list[int],
+    ) -> None:
+        """Persist transformed documents and their precomputed token counts before indexing starts."""
        # save node to document segment
        doc_store = DatasetDocumentStore(
            dataset=dataset, user_id=dataset_document.created_by, document_id=dataset_document.id
@@ -873,9 +875,10 @@ class IndexingRunner:

        # add document segments
        doc_store.add_documents(
+            session=session,
            docs=documents,
            save_child=dataset_document.doc_form == IndexStructureType.PARENT_CHILD_INDEX,
-            session=session,
+            token_counts=token_counts,
        )

        # update document status to indexing
@@ -900,7 +903,6 @@ class IndexingRunner:
                DocumentSegment.indexing_at: naive_utc_now(),
            },
        )
-        pass


 class DocumentIsPausedError(Exception):
@@ -6,9 +6,21 @@ using Python's contextvars for thread-safe and async-safe storage.

 import uuid
 from contextvars import ContextVar
+from typing import NamedTuple
+
+
+class IdentityContext(NamedTuple):
+    """Immutable identity values captured for logging."""
+
+    tenant_id: str
+    user_id: str
+    user_type: str
+

 _request_id: ContextVar[str] = ContextVar("log_request_id", default="")
 _trace_id: ContextVar[str] = ContextVar("log_trace_id", default="")
+_EMPTY_IDENTITY_CONTEXT = IdentityContext(tenant_id="", user_id="", user_type="")
+_identity: ContextVar[IdentityContext] = ContextVar("log_identity", default=_EMPTY_IDENTITY_CONTEXT)


 def get_request_id() -> str:
@@ -21,15 +33,35 @@ def get_trace_id() -> str:
    return _trace_id.get()


+def get_identity_context() -> IdentityContext:
+    """Get the immutable tenant, user, and user-type snapshot for logging."""
+    return _identity.get()
+
+
+def set_identity_context(
+    *, tenant_id: str | None = None, user_id: str | None = None, user_type: str | None = None
+) -> None:
+    """Set primitive identity values already resolved by an authentication boundary."""
+    _identity.set(
+        IdentityContext(
+            tenant_id=tenant_id or "",
+            user_id=user_id or "",
+            user_type=user_type or "",
+        )
+    )
+
+
 def init_request_context() -> None:
-    """Initialize request context. Call at start of each request."""
+    """Initialize request context and discard identity left by earlier work."""
    req_id = uuid.uuid4().hex[:10]
    trace_id = uuid.uuid5(uuid.NAMESPACE_DNS, req_id).hex
    _request_id.set(req_id)
    _trace_id.set(trace_id)
+    _identity.set(_EMPTY_IDENTITY_CONTEXT)


 def clear_request_context() -> None:
-    """Clear request context. Call at end of request (optional)."""
+    """Clear request context at a request or task lifecycle boundary."""
    _request_id.set("")
    _trace_id.set("")
+    _identity.set(_EMPTY_IDENTITY_CONTEXT)
@@ -4,10 +4,7 @@ import contextlib
 import logging
 from typing import override

-import flask
-
-from core.logging.context import get_request_id, get_trace_id
-from core.logging.structured_formatter import IdentityDict
+from core.logging.context import get_identity_context, get_request_id, get_trace_id


 class TraceContextFilter(logging.Filter):
@@ -51,49 +48,16 @@ class TraceContextFilter(logging.Filter):


 class IdentityContextFilter(logging.Filter):
-    """
-    Filter that adds user identity context to log records.
-    Extracts tenant_id, user_id, and user_type from Flask-Login current_user.
+    """Add an identity snapshot without invoking authentication or database work.
+
+    Logging can run while other libraries hold internal locks, so this filter must
+    only read primitive ContextVar values populated by authentication boundaries.
    """

    @override
    def filter(self, record: logging.LogRecord) -> bool:
-        identity = self._extract_identity()
-        record.tenant_id = identity.get("tenant_id", "")
-        record.user_id = identity.get("user_id", "")
-        record.user_type = identity.get("user_type", "")
+        identity = get_identity_context()
+        record.tenant_id = identity.tenant_id
+        record.user_id = identity.user_id
+        record.user_type = identity.user_type
        return True
-
-    def _extract_identity(self) -> IdentityDict:
-        """Extract identity from current_user if in request context."""
-        try:
-            if not flask.has_request_context():
-                return {}
-            from flask_login import current_user
-
-            # Check if user is authenticated using the proxy
-            if not current_user.is_authenticated:
-                return {}
-
-            # Access the underlying user object
-            user = current_user
-
-            from models import Account
-            from models.model import EndUser
-
-            identity: IdentityDict = {}
-
-            match user:
-                case Account():
-                    if user.current_tenant_id:
-                        identity["tenant_id"] = user.current_tenant_id
-                    identity["user_id"] = user.id
-                    identity["user_type"] = "account"
-                case EndUser():
-                    identity["tenant_id"] = user.tenant_id
-                    identity["user_id"] = user.id
-                    identity["user_type"] = user.type or "end_user"
-
-            return identity
-        except Exception:
-            return {}
@@ -207,6 +207,10 @@ class PluginListResponse(BaseModel):
    total: int


+class PluginInstalledIdsDaemonResponse(BaseModel):
+    plugin_ids: list[str]
+
+
 class PluginListWithoutTotalResponse(BaseModel):
    list: list[PluginEntity]
    has_more: bool
@@ -1,7 +1,7 @@
 import inspect
 import json
 import logging
-from collections.abc import Callable, Generator
+from collections.abc import Callable, Generator, Mapping
 from typing import Any, cast
 from urllib.parse import unquote

@@ -23,6 +23,7 @@ from core.plugin.impl.exc import (
    PluginLLMPollingUnsupportedError,
    PluginNotFoundError,
    PluginPermissionDeniedError,
+    PluginRuntimeError,
    PluginUniqueIdentifierError,
 )
 from core.trigger.errors import (
@@ -375,6 +376,18 @@ class BasePluginClient:
                    # type `PluginLLMPollingUnsupportedError`.
                    case PluginLLMPollingUnsupportedError.__name__:
                        raise PluginLLMPollingUnsupportedError(description=error_object.get("message"))
+                    case PluginRuntimeError.__name__:
+                        args = error_object.get("args")
+                        lambda_request_id = args.get("request_id") if isinstance(args, Mapping) else None
+                        if not isinstance(lambda_request_id, str):
+                            lambda_request_id = None
+                        runtime_message = error_object.get("message")
+                        if not isinstance(runtime_message, str):
+                            runtime_message = "Plugin runtime request failed"
+                        raise PluginRuntimeError(
+                            description=runtime_message,
+                            lambda_request_id=lambda_request_id,
+                        )
                    case _:
                        raise PluginInvokeError(description=message)
            case PluginDaemonInternalServerError.__name__:
@@ -49,6 +49,18 @@ class PluginDaemonBadRequestError(PluginDaemonClientSideError):
    description: str = "Bad Request"


+class PluginRuntimeError(PluginDaemonInternalError):
+    """A plugin runtime failed before it could return a valid plugin response."""
+
+    lambda_request_id: str | None
+
+    def __init__(self, description: str, lambda_request_id: str | None = None) -> None:
+        self.lambda_request_id = lambda_request_id
+        if lambda_request_id:
+            description = description.replace(f"RequestId: {lambda_request_id} Error: ", "", 1)
+        super().__init__(description)
+
+
 class PluginInvokeError(PluginDaemonClientSideError, ValueError):
    description: str = "Invoke Error"

@@ -14,6 +14,7 @@ from core.plugin.entities.plugin import (
 )
 from core.plugin.entities.plugin_daemon import (
    PluginDecodeResponse,
+    PluginInstalledIdsDaemonResponse,
    PluginInstallTask,
    PluginInstallTaskStartResponse,
    PluginListResponse,
@@ -68,6 +69,16 @@ class PluginInstaller(BasePluginClient):
        )
        return result.list

+    def list_installed_plugin_ids(self, tenant_id: str, category: PluginCategory) -> list[str]:
+        """List all currently installed plugin IDs in one category."""
+        result = self._request_with_plugin_daemon_response(
+            "GET",
+            f"plugin/{tenant_id}/management/installation/ids",
+            PluginInstalledIdsDaemonResponse,
+            params={"category": category.value},
+        )
+        return result.plugin_ids
+
    def list_plugins_with_total(self, tenant_id: str, page: int, page_size: int) -> PluginListResponse:
        return self._request_with_plugin_daemon_response(
            "GET",
@@ -77,13 +88,28 @@ class PluginInstaller(BasePluginClient):
        )

    def list_plugins_by_category(
-        self, tenant_id: str, category: PluginCategory, page: int, page_size: int
+        self,
+        tenant_id: str,
+        category: PluginCategory,
+        page: int,
+        page_size: int,
+        *,
+        query: str = "",
+        tags: Sequence[str] = (),
+        language: str = "en_US",
    ) -> PluginListWithoutTotalResponse:
        return self._request_with_plugin_daemon_response(
            "GET",
            f"plugin/{tenant_id}/management/{category.value}/list",
            PluginListWithoutTotalResponse,
-            params={"page": page, "page_size": page_size, "response_type": "paged"},
+            params={
+                "page": page,
+                "page_size": page_size,
+                "response_type": "paged",
+                "query": query,
+                "tags": list(tags),
+                "language": language,
+            },
        )

    def upload_pkg(
@@ -656,6 +656,12 @@ class PluginService:
        plugins = manager.list_plugins(tenant_id)
        return plugins

+    @staticmethod
+    def list_installed_plugin_ids(tenant_id: str, category: PluginCategory) -> Sequence[str]:
+        """List all currently installed plugin IDs in one category through the daemon's lightweight query."""
+        manager = PluginInstaller()
+        return manager.list_installed_plugin_ids(tenant_id, category)
+
    @staticmethod
    def list_with_total(tenant_id: str, user_id: str, page: int, page_size: int) -> PluginListResponse:
        """List tenant plugins with endpoint counts reconciled from live records.
@@ -672,17 +678,32 @@ class PluginService:

    @staticmethod
    def list_by_category(
-        tenant_id: str, category: PluginCategory, page: int, page_size: int
+        tenant_id: str,
+        category: PluginCategory,
+        page: int,
+        page_size: int,
+        *,
+        query: str = "",
+        tags: Sequence[str] = (),
+        language: str = "en_US",
    ) -> PluginListWithoutTotalResponse:
        """
        List plugins in one category with a has-more cursor signal and without calculating total.

-        The daemon scans tenant installations in the existing list order and stops once it finds one extra match.
-        This keeps pagination usable before category is persisted on installation rows.
+        The daemon applies category, search, and tag filters before pagination, then stops once it finds one extra
+        match. Filtered model reads are partial views and therefore do not reconcile the full model-provider cache.
        """
        manager = PluginInstaller()
-        plugins = manager.list_plugins_by_category(tenant_id, category, page, page_size)
-        if category == PluginCategory.Model:
+        plugins = manager.list_plugins_by_category(
+            tenant_id,
+            category,
+            page,
+            page_size,
+            query=query,
+            tags=tags,
+            language=language,
+        )
+        if category == PluginCategory.Model and not query and not tags:
            should_invalidate_model_provider_cache = (
                PluginService._should_invalidate_model_provider_cache_for_remote_model_plugins(
                    tenant_id,
@@ -34,6 +34,7 @@ from core.entities.provider_entities import (
 from core.helper import encrypter
 from core.helper.model_provider_cache import ProviderCredentialsCache, ProviderCredentialsCacheType
 from core.helper.position_helper import is_filtered
+from enums.deployment_edition import DeploymentEdition
 from extensions import ext_hosting_provider
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
@@ -743,7 +744,7 @@ class ProviderManager:

            if preferred_provider_type_record:
                preferred_provider_type = preferred_provider_type_record.preferred_provider_type
-            elif dify_config.EDITION == "CLOUD" and system_configuration.enabled:
+            elif dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD and system_configuration.enabled:
                preferred_provider_type = ProviderType.SYSTEM
            elif custom_configuration.provider or custom_configuration.models:
                preferred_provider_type = ProviderType.CUSTOM
@@ -1538,7 +1539,7 @@ class ProviderManager:
                quota_type_to_provider_records_dict[provider_record.quota_type] = provider_record  # type: ignore[index]
        quota_configurations = []

-        if dify_config.EDITION == "CLOUD":
+        if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD:
            from services.credit_pool_service import CreditPoolService

            trail_pool = CreditPoolService.get_pool(
@@ -10,6 +10,7 @@ from core.rag.rerank.entity.weight import KeywordSetting, VectorSetting, Weights
 from core.rag.rerank.rerank_base import BaseRerankRunner
 from core.rag.rerank.rerank_factory import RerankRunnerFactory
 from core.rag.rerank.rerank_type import RerankMode
+from extensions.otel import trace_span
 from graphon.model_runtime.entities.model_entities import ModelType
 from graphon.model_runtime.errors.invoke import InvokeAuthorizationError

@@ -52,6 +53,7 @@ class DataPostProcessor:
        )
        self.reorder_runner = self._get_reorder_runner(reorder_enabled)

+    @trace_span()
    def invoke(
        self,
        query: str,
@@ -1,12 +1,10 @@
 import concurrent.futures
-import functools
 import logging
-from collections.abc import Callable, Sequence
+from collections.abc import Sequence
 from concurrent.futures import ThreadPoolExecutor
 from typing import Any, NotRequired, TypedDict

 from flask import Flask, current_app
-from opentelemetry import context as otel_context
 from sqlalchemy import select
 from sqlalchemy.orm import Session, load_only

@@ -26,7 +24,7 @@ from core.rag.rerank.rerank_type import RerankMode
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
 from core.tools.signature import sign_upload_file_preview_url
 from extensions.ext_database import db
-from extensions.otel import trace_span
+from extensions.otel import propagate_context, trace_span
 from graphon.model_runtime.entities.model_entities import ModelType
 from models.dataset import (
    ChildChunk,
@@ -92,20 +90,6 @@ default_retrieval_model: DefaultRetrievalModelDict = {
 logger = logging.getLogger(__name__)


-def _propagate_otel_context[**P, R](func: Callable[P, R]) -> Callable[P, R]:
-    captured_context = otel_context.get_current()
-
-    @functools.wraps(func)
-    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
-        token = otel_context.attach(captured_context)
-        try:
-            return func(*args, **kwargs)
-        finally:
-            otel_context.detach(token)
-
-    return wrapper
-
-
 class RetrievalService:
    # Cache precompiled regular expressions to avoid repeated compilation
    @classmethod
@@ -139,7 +123,7 @@ class RetrievalService:
            if query:
                futures.append(
                    executor.submit(
-                        _propagate_otel_context(retrieval_service._retrieve),
+                        propagate_context(retrieval_service._retrieve),
                        flask_app=current_app._get_current_object(),  # type: ignore
                        retrieval_method=retrieval_method,
                        dataset=dataset,
@@ -159,7 +143,7 @@ class RetrievalService:
                for attachment_id in attachment_ids:
                    futures.append(
                        executor.submit(
-                            _propagate_otel_context(retrieval_service._retrieve),
+                            propagate_context(retrieval_service._retrieve),
                            flask_app=current_app._get_current_object(),  # type: ignore
                            retrieval_method=retrieval_method,
                            dataset=dataset,
@@ -820,7 +804,7 @@ class RetrievalService:
                if retrieval_method == RetrievalMethod.KEYWORD_SEARCH and query:
                    futures.append(
                        executor.submit(
-                            _propagate_otel_context(self.keyword_search),
+                            propagate_context(self.keyword_search),
                            flask_app=current_app._get_current_object(),  # type: ignore
                            dataset_id=dataset.id,
                            query=query,
@@ -834,7 +818,7 @@ class RetrievalService:
                    if query:
                        futures.append(
                            executor.submit(
-                                _propagate_otel_context(self.embedding_search),
+                                propagate_context(self.embedding_search),
                                flask_app=current_app._get_current_object(),  # type: ignore
                                dataset_id=dataset.id,
                                query=query,
@@ -851,7 +835,7 @@ class RetrievalService:
                    if attachment_id:
                        futures.append(
                            executor.submit(
-                                _propagate_otel_context(self.embedding_search),
+                                propagate_context(self.embedding_search),
                                flask_app=current_app._get_current_object(),  # type: ignore
                                dataset_id=dataset.id,
                                query=attachment_id,
@@ -868,7 +852,7 @@ class RetrievalService:
                if RetrievalMethod.is_support_fulltext_search(retrieval_method) and query:
                    futures.append(
                        executor.submit(
-                            _propagate_otel_context(self.full_text_index_search),
+                            propagate_context(self.full_text_index_search),
                            flask_app=current_app._get_current_object(),  # type: ignore
                            dataset_id=dataset.id,
                            query=query,
@@ -6,10 +6,7 @@ from typing import Any
 from sqlalchemy import delete, func, select
 from sqlalchemy.orm import Session

-from core.model_manager import ModelManager
-from core.rag.index_processor.constant.index_type import IndexTechniqueType
 from core.rag.models.document import AttachmentDocument, Document
-from graphon.model_runtime.entities.model_entities import ModelType
 from models.dataset import ChildChunk, Dataset, DocumentSegment, SegmentAttachmentBinding
 from models.enums import SegmentType

@@ -69,34 +66,22 @@ class DatasetDocumentStore:

    def add_documents(
        self,
-        docs: Sequence[Document],
        session: Session,
+        docs: Sequence[Document],
+        token_counts: list[int],
        allow_update: bool = True,
        save_child: bool = False,
-    ):
+    ) -> None:
+        document_token_pairs = list(zip(docs, token_counts, strict=True))
+
        max_position = session.scalar(
            select(func.max(DocumentSegment.position)).where(DocumentSegment.document_id == self._document_id)
        )

        if max_position is None:
            max_position = 0
-        embedding_model = None
-        if self._dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
-            model_manager = ModelManager.for_tenant(tenant_id=self._dataset.tenant_id)
-            embedding_model = model_manager.get_model_instance(
-                tenant_id=self._dataset.tenant_id,
-                provider=self._dataset.embedding_model_provider,
-                model_type=ModelType.TEXT_EMBEDDING,
-                model=self._dataset.embedding_model,
-            )

-        if embedding_model:
-            page_content_list = [doc.page_content for doc in docs]
-            tokens_list = embedding_model.get_text_embedding_num_tokens(page_content_list)
-        else:
-            tokens_list = [0] * len(docs)
-
-        for doc, tokens in zip(docs, tokens_list):
+        for doc, tokens in document_token_pairs:
            if not isinstance(doc, Document):
                raise ValueError("doc must be a Document")

@@ -0,0 +1,25 @@
+"""Token counting for document segments."""
+
+from core.model_manager import ModelManager
+from core.rag.index_processor.constant.index_type import IndexTechniqueType
+from core.rag.models.document import Document
+from graphon.model_runtime.entities.model_entities import ModelType
+from models.dataset import Dataset
+
+
+def calculate_segment_token_counts(dataset: Dataset, documents: list[Document]) -> list[int]:
+    """Return one token count per document, invoking the embedding model only for high-quality indexes."""
+    if not documents:
+        return []
+
+    if dataset.indexing_technique != IndexTechniqueType.HIGH_QUALITY:
+        return [0] * len(documents)
+
+    model_manager = ModelManager.for_tenant(tenant_id=dataset.tenant_id)
+    embedding_model = model_manager.get_model_instance(
+        tenant_id=dataset.tenant_id,
+        provider=dataset.embedding_model_provider,
+        model_type=ModelType.TEXT_EMBEDDING,
+        model=dataset.embedding_model,
+    )
+    return embedding_model.get_text_embedding_num_tokens([document.page_content for document in documents])
@@ -19,6 +19,7 @@ from core.rag.cleaner.clean_processor import CleanProcessor
 from core.rag.datasource.keyword.keyword_factory import Keyword
 from core.rag.datasource.vdb.vector_factory import Vector
 from core.rag.docstore.dataset_docstore import DatasetDocumentStore
+from core.rag.embedding.token_counter import calculate_segment_token_counts
 from core.rag.entities import Rule
 from core.rag.extractor.entity.extract_setting import ExtractSetting
 from core.rag.extractor.extract_processor import ExtractProcessor
@@ -244,10 +245,16 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
                        all_multimodal_documents.extend(doc.attachments)
                documents.append(doc)
        if documents:
+            token_counts = calculate_segment_token_counts(dataset=dataset, documents=documents)
            # save node to document segment
            doc_store = DatasetDocumentStore(dataset=dataset, user_id=document.created_by, document_id=document.id)
            # add document segments
-            doc_store.add_documents(docs=documents, save_child=False, session=session)
+            doc_store.add_documents(
+                session=session,
+                docs=documents,
+                token_counts=token_counts,
+                save_child=False,
+            )
            session.commit()
            if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
                vector = Vector(dataset, session=session)
@@ -15,6 +15,7 @@ from core.model_manager import ModelInstance
 from core.rag.cleaner.clean_processor import CleanProcessor
 from core.rag.datasource.vdb.vector_factory import Vector
 from core.rag.docstore.dataset_docstore import DatasetDocumentStore
+from core.rag.embedding.token_counter import calculate_segment_token_counts
 from core.rag.entities import ParentMode, Rule
 from core.rag.extractor.entity.extract_setting import ExtractSetting
 from core.rag.extractor.extract_processor import ExtractProcessor
@@ -304,6 +305,7 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
                doc.attachments = self._get_content_files(doc, current_user=account, session=session)
            documents.append(doc)
        if documents:
+            token_counts = calculate_segment_token_counts(dataset=dataset, documents=documents)
            # update document parent mode
            dataset_process_rule = DatasetProcessRule(
                dataset_id=dataset.id,
@@ -321,7 +323,12 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
            # save node to document segment
            doc_store = DatasetDocumentStore(dataset=dataset, user_id=document.created_by, document_id=document.id)
            # add document segments
-            doc_store.add_documents(docs=documents, save_child=True, session=session)
+            doc_store.add_documents(
+                session=session,
+                docs=documents,
+                token_counts=token_counts,
+                save_child=True,
+            )
            session.commit()
            if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
                all_child_documents = []
@@ -17,6 +17,7 @@ from core.llm_generator.llm_generator import LLMGenerator
 from core.rag.cleaner.clean_processor import CleanProcessor
 from core.rag.datasource.vdb.vector_factory import Vector
 from core.rag.docstore.dataset_docstore import DatasetDocumentStore
+from core.rag.embedding.token_counter import calculate_segment_token_counts
 from core.rag.entities import Rule
 from core.rag.extractor.entity.extract_setting import ExtractSetting
 from core.rag.extractor.extract_processor import ExtractProcessor
@@ -205,9 +206,15 @@ class QAIndexProcessor(BaseIndexProcessor):
            doc = Document(page_content=qa_chunk.question, metadata=metadata)
            documents.append(doc)
        if documents:
+            token_counts = calculate_segment_token_counts(dataset=dataset, documents=documents)
            # save node to document segment
            doc_store = DatasetDocumentStore(dataset=dataset, user_id=document.created_by, document_id=document.id)
-            doc_store.add_documents(docs=documents, save_child=False, session=session)
+            doc_store.add_documents(
+                session=session,
+                docs=documents,
+                token_counts=token_counts,
+                save_child=False,
+            )
            session.commit()
            if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
                vector = Vector(dataset, session=session)
@@ -9,6 +9,7 @@ from core.rag.index_processor.constant.query_type import QueryType
 from core.rag.models.document import Document
 from core.rag.rerank.rerank_base import BaseRerankRunner
 from extensions.ext_storage import storage
+from extensions.otel import trace_span
 from graphon.model_runtime.entities.model_entities import ModelType
 from graphon.model_runtime.entities.rerank_entities import MultimodalRerankInput, RerankResult
 from models.model import UploadFile
@@ -22,6 +23,7 @@ class RerankModelRunner(BaseRerankRunner):
        self._session = session

    @override
+    @trace_span()
    def run(
        self,
        query: str,
@@ -65,6 +65,7 @@ from core.workflow.nodes.knowledge_retrieval.retrieval import (
 )
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
+from extensions.otel import propagate_context, trace_span
 from graphon.file import File, FileTransferMethod, FileType
 from graphon.model_runtime.entities.llm_entities import LLMMode, LLMResult, LLMUsage
 from graphon.model_runtime.entities.message_entities import PromptMessage, PromptMessageRole, PromptMessageTool
@@ -116,6 +117,7 @@ class DatasetRetrieval:
        else:
            self._llm_usage = self._llm_usage.plus(usage)

+    @trace_span()
    def knowledge_retrieval(self, session: Session, request: KnowledgeRetrievalRequest) -> list[Source]:
        self._check_knowledge_rate_limit(request.tenant_id)
        available_datasets = self._get_available_datasets(request.tenant_id, request.dataset_ids)
@@ -599,6 +601,7 @@ class DatasetRetrieval:
            return "\n".join([document_context.content for document_context in document_context_list]), context_files
        return "", context_files

+    @trace_span()
    def single_retrieve(
        self,
        session: Session,
@@ -724,7 +727,7 @@ class DatasetRetrieval:

                if results:
                    thread = threading.Thread(
-                        target=self._on_retrieval_end,
+                        target=propagate_context(self._on_retrieval_end),
                        kwargs={
                            "flask_app": current_app._get_current_object(),  # type: ignore
                            "documents": results,
@@ -737,6 +740,7 @@ class DatasetRetrieval:
                return results
        return []

+    @trace_span()
    def multiple_retrieve(
        self,
        app_id: str,
@@ -798,7 +802,7 @@ class DatasetRetrieval:

            if query:
                query_thread = threading.Thread(
-                    target=self._multiple_retrieve_thread,
+                    target=propagate_context(self._multiple_retrieve_thread_safely),
                    kwargs={
                        "flask_app": current_app._get_current_object(),  # type: ignore
                        "available_datasets": available_datasets,
@@ -824,7 +828,7 @@ class DatasetRetrieval:
            if attachment_ids:
                for attachment_id in attachment_ids:
                    attachment_thread = threading.Thread(
-                        target=self._multiple_retrieve_thread,
+                        target=propagate_context(self._multiple_retrieve_thread_safely),
                        kwargs={
                            "flask_app": current_app._get_current_object(),  # type: ignore
                            "available_datasets": available_datasets,
@@ -865,7 +869,7 @@ class DatasetRetrieval:
        if all_documents:
            # add thread to call _on_retrieval_end
            retrieval_end_thread = threading.Thread(
-                target=self._on_retrieval_end,
+                target=propagate_context(self._on_retrieval_end),
                kwargs={
                    "flask_app": current_app._get_current_object(),  # type: ignore
                    "documents": all_documents,
@@ -1161,6 +1165,7 @@ class DatasetRetrieval:

                        all_documents.extend(documents)

+    @trace_span()
    def _run_retriever_thread(
        self,
        *,
@@ -1172,27 +1177,51 @@ class DatasetRetrieval:
        document_ids_filter: list[str] | None,
        metadata_condition: MetadataFilteringCondition | None,
        attachment_ids: list[str] | None,
+    ) -> None:
+        with session_factory.create_session() as session:
+            self._retriever(
+                flask_app=flask_app,
+                session=session,
+                dataset_id=dataset_id,
+                query=query or "",
+                top_k=top_k,
+                all_documents=all_documents,
+                document_ids_filter=document_ids_filter,
+                metadata_condition=metadata_condition,
+                attachment_ids=attachment_ids,
+            )
+
+    def _run_retriever_thread_safely(
+        self,
+        *,
+        flask_app: Flask,
+        dataset_id: str,
+        query: str | None,
+        top_k: int,
+        all_documents: list[Document],
+        document_ids_filter: list[str] | None,
+        metadata_condition: MetadataFilteringCondition | None,
+        attachment_ids: list[str] | None,
        cancel_event: threading.Event | None,
        thread_exceptions: list[Exception] | None,
    ) -> None:
+        """Collect errors only after they pass through the traced retrieval method."""
        try:
-            with session_factory.create_session() as session:
-                self._retriever(
-                    flask_app=flask_app,
-                    session=session,
-                    dataset_id=dataset_id,
-                    query=query or "",
-                    top_k=top_k,
-                    all_documents=all_documents,
-                    document_ids_filter=document_ids_filter,
-                    metadata_condition=metadata_condition,
-                    attachment_ids=attachment_ids,
-                )
-        except Exception as e:
+            self._run_retriever_thread(
+                flask_app=flask_app,
+                dataset_id=dataset_id,
+                query=query,
+                top_k=top_k,
+                all_documents=all_documents,
+                document_ids_filter=document_ids_filter,
+                metadata_condition=metadata_condition,
+                attachment_ids=attachment_ids,
+            )
+        except Exception as exc:
            if cancel_event:
                cancel_event.set()
            if thread_exceptions is not None:
-                thread_exceptions.append(e)
+                thread_exceptions.append(exc)

    def to_dataset_retriever_tool(
        self,
@@ -1795,6 +1824,7 @@ class DatasetRetrieval:

        return full_text, usage

+    @trace_span()
    def _multiple_retrieve_thread(
        self,
        flask_app: Flask,
@@ -1813,11 +1843,11 @@ class DatasetRetrieval:
        attachment_id: str | None,
        dataset_count: int,
        cancel_event: threading.Event | None = None,
-        thread_exceptions: list[Exception] | None = None,
-    ):
+    ) -> None:
        try:
            with flask_app.app_context():
                threads = []
+                retrieval_thread_exceptions: list[Exception] = []
                all_documents_item: list[Document] = []
                index_type = None
                for dataset in available_datasets:
@@ -1836,7 +1866,7 @@ class DatasetRetrieval:
                            else:
                                continue
                    retrieval_thread = threading.Thread(
-                        target=self._run_retriever_thread,
+                        target=propagate_context(self._run_retriever_thread_safely),
                        kwargs={
                            "flask_app": flask_app,
                            "dataset_id": dataset.id,
@@ -1847,7 +1877,7 @@ class DatasetRetrieval:
                            "metadata_condition": metadata_condition,
                            "attachment_ids": [attachment_id] if attachment_id else None,
                            "cancel_event": cancel_event,
-                            "thread_exceptions": thread_exceptions,
+                            "thread_exceptions": retrieval_thread_exceptions,
                        },
                    )
                    threads.append(retrieval_thread)
@@ -1862,6 +1892,9 @@ class DatasetRetrieval:
                    if cancel_event and cancel_event.is_set():
                        break

+                if retrieval_thread_exceptions:
+                    raise retrieval_thread_exceptions[0]
+
                # Skip second reranking when there is only one dataset
                if reranking_enable and dataset_count > 1:
                    # do rerank for searched documents
@@ -1902,11 +1935,55 @@ class DatasetRetrieval:
                        all_documents_item = all_documents_item[:top_k] if top_k else all_documents_item
                if all_documents_item:
                    all_documents.extend(all_documents_item)
-        except Exception as e:
+        except Exception:
+            raise
+
+    def _multiple_retrieve_thread_safely(
+        self,
+        *,
+        flask_app: Flask,
+        available_datasets: list[Dataset],
+        metadata_condition: MetadataFilteringCondition | None,
+        metadata_filter_document_ids: dict[str, list[str]] | None,
+        all_documents: list[Document],
+        tenant_id: str,
+        reranking_enable: bool,
+        reranking_mode: str,
+        reranking_model: RerankingModelDict | None,
+        weights: WeightsDict | None,
+        top_k: int,
+        score_threshold: float,
+        query: str | None,
+        attachment_id: str | None,
+        dataset_count: int,
+        cancel_event: threading.Event | None = None,
+        thread_exceptions: list[Exception] | None = None,
+    ) -> None:
+        """Collect errors only after they pass through the traced multi-retrieval method."""
+        try:
+            self._multiple_retrieve_thread(
+                flask_app=flask_app,
+                available_datasets=available_datasets,
+                metadata_condition=metadata_condition,
+                metadata_filter_document_ids=metadata_filter_document_ids,
+                all_documents=all_documents,
+                tenant_id=tenant_id,
+                reranking_enable=reranking_enable,
+                reranking_mode=reranking_mode,
+                reranking_model=reranking_model,
+                weights=weights,
+                top_k=top_k,
+                score_threshold=score_threshold,
+                query=query,
+                attachment_id=attachment_id,
+                dataset_count=dataset_count,
+                cancel_event=cancel_event,
+            )
+        except Exception as exc:
            if cancel_event:
                cancel_event.set()
            if thread_exceptions is not None:
-                thread_exceptions.append(e)
+                thread_exceptions.append(exc)

    def _get_available_datasets(self, tenant_id: str, dataset_ids: list[str]) -> list[Dataset]:
        with session_factory.create_session() as session:
@@ -61,6 +61,7 @@ class RBACPermission(StrEnum):
    WORKSPACE_ROLE_MANAGE = "workspace_role_manage"
    API_EXTENSION_MANAGE = "api_extension_manage"
    CUSTOMIZATION_MANAGE = "customization_manage"
+    AGENT_MANAGE = "agent_manage"

    SNIPPETS_CREATE_AND_MODIFY = "snippets_create_and_modify"
    SNIPPETS_MANAGE = "snippets_management"
@@ -58,7 +58,6 @@ class Tool(ABC):
        if self.runtime and self.runtime.runtime_parameters:
            tool_parameters.update(self.runtime.runtime_parameters)

-        # try parse tool parameters into the correct type
        tool_parameters = self._transform_tool_parameters_type(tool_parameters)

        result = self._invoke(
@@ -87,14 +86,14 @@ class Tool(ABC):
                return result

    def _transform_tool_parameters_type(self, tool_parameters: dict[str, Any]) -> dict[str, Any]:
-        """
-        Transform tool parameters type
-        """
-        # Temp fix for the issue that the tool parameters will be converted to empty while validating the credentials
+        """Transform declared tool parameter values without resolving runtime schemas."""
        result = deepcopy(tool_parameters)
        for parameter in self.entity.parameters or []:
            if parameter.name in tool_parameters:
-                result[parameter.name] = parameter.type.cast_value(tool_parameters[parameter.name])
+                if parameter.multiple:
+                    result[parameter.name] = parameter.init_frontend_parameter(result.get(parameter.name))
+                else:
+                    result[parameter.name] = parameter.type.cast_value(tool_parameters[parameter.name])

        return result

@@ -196,17 +195,31 @@ class Tool(ABC):
            }:
                continue

-            parameter_schema: dict[str, Any] = (
-                {
-                    "type": parameter.type.as_normal_type(),
-                    "description": parameter.llm_description or "",
-                }
-                if parameter.input_schema is None
-                else deepcopy(parameter.input_schema)
-            )
+            is_multiple_select = parameter.multiple and parameter.type in {
+                ToolParameter.ToolParameterType.SELECT,
+                ToolParameter.ToolParameterType.DYNAMIC_SELECT,
+            }
+            if is_multiple_select:
+                item_schema: dict[str, Any] = {"type": "string"}
+                if parameter.type == ToolParameter.ToolParameterType.SELECT and parameter.options:
+                    item_schema["enum"] = [option.value for option in parameter.options]
+                parameter_schema: dict[str, Any] = {"type": "array", "items": item_schema}
+            else:
+                parameter_schema = (
+                    {
+                        "type": parameter.type.as_normal_type(),
+                        "description": parameter.llm_description or "",
+                    }
+                    if parameter.input_schema is None
+                    else deepcopy(parameter.input_schema)
+                )
            parameter_schema.setdefault("description", parameter.llm_description or "")

-            if parameter.type == ToolParameter.ToolParameterType.SELECT and parameter.options:
+            if (
+                not is_multiple_select
+                and parameter.type == ToolParameter.ToolParameterType.SELECT
+                and parameter.options
+            ):
                parameter_schema["enum"] = [option.value for option in parameter.options]

            schema["properties"][parameter.name] = parameter_schema
@@ -292,9 +292,7 @@ class ToolInvokeMessageBinary(BaseModel):


 class ToolParameter(PluginParameter):
-    """
-    Overrides type
-    """
+    """Tool-specific parameter declaration and invocation-value normalization."""

    class ToolParameterType(StrEnum):
        """
@@ -333,12 +331,28 @@ class ToolParameter(PluginParameter):
        LLM = auto()  # will be set by LLM

    type: ToolParameterType = Field(..., description="The type of the parameter")
+    multiple: bool = Field(
+        default=False,
+        description="Whether the parameter is multiple select, only valid for select or dynamic-select type",
+    )
    human_description: I18nObject | None = Field(default=None, description="The description presented to the user")
    form: ToolParameterForm = Field(..., description="The form of the parameter, schema/form/llm")
    llm_description: str | None = None
    # MCP object and array type parameters use this field to store the schema
    input_schema: dict[str, Any] | None = None

+    @model_validator(mode="after")
+    def validate_multiple(self) -> ToolParameter:
+        supports_multiple = self.type in {
+            self.ToolParameterType.SELECT,
+            self.ToolParameterType.DYNAMIC_SELECT,
+        }
+        if self.multiple and not supports_multiple:
+            raise ValueError("multiple is only valid for select and dynamic-select parameters")
+        if supports_multiple and self.default is not None and (isinstance(self.default, list) != self.multiple):
+            raise ValueError("default must be a list exactly when multiple is true")
+        return self
+
    @classmethod
    def get_simple_instance(
        cls,
@@ -378,8 +392,25 @@ class ToolParameter(PluginParameter):
            options=option_objs,
        )

-    def init_frontend_parameter(self, value: Any):
-        return init_frontend_parameter(self, self.type, value)
+    def init_frontend_parameter(self, value: Any) -> Any:
+        """Normalize a value against this tool parameter's full declaration."""
+        if not self.multiple:
+            return init_frontend_parameter(self, self.type, value)
+
+        parameter_value = self.default if value is None else value
+        if parameter_value is None:
+            parameter_value = []
+        if not isinstance(parameter_value, list):
+            raise ValueError(f"tool parameter {self.name} must be a list when multiple is true")
+        if not all(isinstance(item, str) for item in parameter_value):
+            raise ValueError(f"tool parameter {self.name} must contain only strings")
+        if self.required and not parameter_value:
+            raise ValueError(f"tool parameter {self.name} not found in tool config")
+        if self.type == self.ToolParameterType.SELECT:
+            options = [option.value for option in self.options]
+            if any(item not in options for item in parameter_value):
+                raise ValueError(f"tool parameter {self.name} value {parameter_value} not in options {options}")
+        return parameter_value


 class ToolProviderIdentity(BaseModel):
@@ -4,8 +4,16 @@ from dataclasses import dataclass

 from sqlalchemy import select

+from core.agent.publish_visibility import workflow_callable_active_snapshot_filter
 from core.db.session_factory import session_factory
-from models.agent import Agent, AgentConfigSnapshot, AgentStatus, WorkflowAgentBindingType, WorkflowAgentNodeBinding
+from models.agent import (
+    Agent,
+    AgentConfigSnapshot,
+    AgentScope,
+    AgentStatus,
+    WorkflowAgentBindingType,
+    WorkflowAgentNodeBinding,
+)


 class WorkflowAgentBindingError(Exception):
@@ -24,7 +32,7 @@ class WorkflowAgentBindingBundle:


 class WorkflowAgentBindingResolver:
-    """Resolve the Agent binding owned by the current workflow id and node id."""
+    """Resolve an owned binding without allowing unpublished roster snapshots to run."""

    def resolve(
        self,
@@ -53,18 +61,20 @@ class WorkflowAgentBindingResolver:
            if binding.agent_id is None:
                raise WorkflowAgentBindingError("agent_not_available", "Workflow Agent binding has no agent.")

-            agent = session.scalar(
-                select(Agent)
-                .where(
-                    Agent.tenant_id == tenant_id,
-                    Agent.id == binding.agent_id,
-                )
-                .limit(1)
+            agent_stmt = select(Agent).where(
+                Agent.tenant_id == tenant_id,
+                Agent.id == binding.agent_id,
            )
+            if binding.binding_type == WorkflowAgentBindingType.ROSTER_AGENT:
+                agent_stmt = agent_stmt.where(
+                    Agent.scope == AgentScope.ROSTER,
+                    workflow_callable_active_snapshot_filter(),
+                )
+            agent = session.scalar(agent_stmt.limit(1))
            if agent is None or agent.status == AgentStatus.ARCHIVED:
                raise WorkflowAgentBindingError(
                    "agent_not_available",
-                    f"Agent {binding.agent_id} is not available.",
+                    f"Agent {binding.agent_id} is not available or has not been published.",
                )

            snapshot_id = (
@@ -2,7 +2,7 @@ from __future__ import annotations

 from collections.abc import Mapping
 from dataclasses import dataclass
-from typing import Any, Literal, Protocol, cast
+from typing import Any, Final, Literal, Protocol

 from dify_agent.layers.dify_core_tools import DifyCoreToolConfig, DifyCoreToolProviderType, DifyCoreToolsLayerConfig
 from dify_agent.layers.dify_plugin import (
@@ -29,6 +29,13 @@ from models.provider_ids import ToolProviderID
 from models.tools import WorkflowToolProvider
 from services.tools.mcp_tools_manage_service import MCPToolManageService

+_CORE_TOOL_PROVIDER_TYPES: Final[dict[ToolProviderType, DifyCoreToolProviderType]] = {
+    ToolProviderType.BUILT_IN: "builtin",
+    ToolProviderType.API: "api",
+    ToolProviderType.WORKFLOW: "workflow",
+    ToolProviderType.MCP: "mcp",
+}
+

 class WorkflowAgentDifyToolsBuildError(ValueError):
    """Raised when Agent Soul tools cannot be prepared for Agent backend."""
@@ -235,7 +242,7 @@ class WorkflowAgentDifyToolsBuilder:
            if tool_config.tool_name is not None:
                expanded.append(tool_config)
                continue
-            provider_type = ToolProviderType.value_of(tool_config.provider_type)
+            provider_type = tool_config.provider_type
            provider_id = self._provider_id(tool_config)
            try:
                tool_names = self._provider_declared_tool_names(
@@ -279,7 +286,7 @@ class WorkflowAgentDifyToolsBuilder:
        tenant_id: str,
        tool_config: AgentSoulDifyToolConfig,
    ) -> AgentSoulDifyToolConfig:
-        if tool_config.provider_type != ToolProviderType.MCP.value:
+        if tool_config.provider_type is not ToolProviderType.MCP:
            return tool_config
        provider_id = self._mcp_provider_id_resolver(tenant_id=tenant_id, provider_id=self._provider_id(tool_config))
        return tool_config.model_copy(update={"provider_id": provider_id, "plugin_id": None, "provider": None})
@@ -326,7 +333,7 @@ class WorkflowAgentDifyToolsBuilder:
    def _to_agent_tool_entity(tool_config: AgentSoulDifyToolConfig) -> AgentToolEntity:
        assert tool_config.tool_name is not None
        return AgentToolEntity(
-            provider_type=ToolProviderType.value_of(tool_config.provider_type),
+            provider_type=tool_config.provider_type,
            provider_id=WorkflowAgentDifyToolsBuilder._provider_id(tool_config),
            tool_name=tool_config.tool_name,
            tool_parameters=dict(tool_config.runtime_parameters),
@@ -343,24 +350,16 @@ class WorkflowAgentDifyToolsBuilder:

    @staticmethod
    def _provider_key(tool_config: AgentSoulDifyToolConfig) -> tuple[ToolProviderType, str]:
-        return (
-            ToolProviderType.value_of(tool_config.provider_type),
-            WorkflowAgentDifyToolsBuilder._provider_id(tool_config),
-        )
+        return (tool_config.provider_type, WorkflowAgentDifyToolsBuilder._provider_id(tool_config))

    @staticmethod
    def _tool_layer_destination(tool_config: AgentSoulDifyToolConfig) -> Literal["plugin", "core"]:
-        provider_type = ToolProviderType.value_of(tool_config.provider_type)
+        provider_type = tool_config.provider_type
        if provider_type is ToolProviderType.PLUGIN or (
            provider_type is ToolProviderType.BUILT_IN and _is_plugin_provider_id(tool_config.provider_id)
        ):
            return "plugin"
-        if provider_type in {
-            ToolProviderType.BUILT_IN,
-            ToolProviderType.API,
-            ToolProviderType.WORKFLOW,
-            ToolProviderType.MCP,
-        }:
+        if provider_type in _CORE_TOOL_PROVIDER_TYPES:
            return "core"
        if provider_type is ToolProviderType.DATASET_RETRIEVAL:
            raise WorkflowAgentDifyToolsBuildError(
@@ -417,7 +416,7 @@ class WorkflowAgentDifyToolsBuilder:
    ) -> DifyCoreToolConfig:
        parameters = self._prepared_parameters(tool_runtime)
        return DifyCoreToolConfig(
-            provider_type=cast(DifyCoreToolProviderType, tool_config.provider_type),
+            provider_type=_CORE_TOOL_PROVIDER_TYPES[tool_config.provider_type],
            provider_id=self._provider_id(tool_config),
            tool_name=tool_config.tool_name or exposed_name,
            credential_id=tool_config.credential_ref.id if tool_config.credential_ref else None,
@@ -6,9 +6,17 @@ from typing import Any
 from sqlalchemy import select
 from sqlalchemy.orm import Session

+from core.agent.publish_visibility import workflow_callable_active_snapshot_filter
 from core.workflow.graph_topology import WorkflowGraphTopology
 from graphon.enums import BuiltinNodeTypes
-from models.agent import Agent, AgentConfigSnapshot, AgentStatus, WorkflowAgentBindingType, WorkflowAgentNodeBinding
+from models.agent import (
+    Agent,
+    AgentConfigSnapshot,
+    AgentScope,
+    AgentStatus,
+    WorkflowAgentBindingType,
+    WorkflowAgentNodeBinding,
+)
 from models.agent_config_entities import (
    AgentFileRefConfig,
    AgentHumanContactConfig,
@@ -117,21 +125,28 @@ class WorkflowAgentNodeValidator:
        binding: WorkflowAgentNodeBinding,
        topology: _WorkflowGraphTopology | None = None,
    ) -> None:
+        """Validate binding ownership, publication state, Agent Soul, and node-job references."""
+
        if binding.agent_id is None:
            raise WorkflowAgentNodeValidationError(f"Workflow Agent node {binding.node_id} is missing agent binding.")

-        agent = session.scalar(
-            select(Agent)
-            .where(
-                Agent.tenant_id == binding.tenant_id,
-                Agent.id == binding.agent_id,
-            )
-            .limit(1)
+        agent_stmt = select(Agent).where(
+            Agent.tenant_id == binding.tenant_id,
+            Agent.id == binding.agent_id,
        )
-        if agent is None or agent.status == AgentStatus.ARCHIVED:
-            raise WorkflowAgentNodeValidationError(
-                f"Workflow Agent node {binding.node_id} references an unavailable agent."
+        if binding.binding_type == WorkflowAgentBindingType.ROSTER_AGENT:
+            agent_stmt = agent_stmt.where(
+                Agent.scope == AgentScope.ROSTER,
+                workflow_callable_active_snapshot_filter(),
            )
+        agent = session.scalar(agent_stmt.limit(1))
+        if agent is None or agent.status == AgentStatus.ARCHIVED:
+            availability = (
+                "an unavailable or unpublished roster agent"
+                if binding.binding_type == WorkflowAgentBindingType.ROSTER_AGENT
+                else "an unavailable agent"
+            )
+            raise WorkflowAgentNodeValidationError(f"Workflow Agent node {binding.node_id} references {availability}.")

        snapshot_id = (
            agent.active_config_snapshot_id
@@ -12,6 +12,7 @@ import json
 import subprocess
 import sys
 import tempfile
+from copy import deepcopy
 from pathlib import Path
 from typing import Any, Literal, TypedDict

@@ -24,6 +25,16 @@ LOCK_PATH = API_ROOT / "knowledge-fs-contract.lock.json"
 DEFAULT_REPOSITORY = WORKSPACE_ROOT.parent / "knowledge-fs"
 OPENAPI_METHODS = ("delete", "get", "head", "options", "patch", "post", "put", "trace")
 PROXY_METHODS = frozenset({"delete", "get", "patch", "post", "put"})
+CONSOLE_PROXY_ERROR_SCHEMA_NAME = "ConsoleProxyError"
+CONSOLE_PROXY_ERROR_SCHEMA: dict[str, Any] = {
+    "type": "object",
+    "required": ["code", "message", "status"],
+    "properties": {
+        "code": {"type": "string"},
+        "message": {"type": "string"},
+        "status": {"type": "integer"},
+    },
+}


 class ContractDeclaration(TypedDict):
@@ -38,6 +49,7 @@ class ContractDeclaration(TypedDict):
    request_headers: tuple[str, ...]
    response_headers: tuple[str, ...]
    response_media_types: tuple[str, ...]
+    error_status_map: tuple[tuple[int, int], ...]


 type DeclarationField = Literal[
@@ -70,6 +82,7 @@ def main() -> None:
    mode.add_argument("--check", action="store_true")
    mode.add_argument("--update-lock", action="store_true")
    parser.add_argument("--repository", type=Path, default=DEFAULT_REPOSITORY)
+    parser.add_argument("--output-openapi", type=Path)
    args = parser.parse_args()

    repository = args.repository.resolve()
@@ -101,7 +114,15 @@ def main() -> None:
        )

    document: dict[str, Any] = json.loads(openapi_content)
-    validate_declarations(document, console_contract_declarations())
+    declarations = console_contract_declarations()
+    validate_declarations(document, declarations)
+
+    if args.output_openapi:
+        filtered_document = filter_openapi_document(document, declarations)
+        filtered_document["x-dify-source-openapi-sha256"] = openapi_sha256
+        filtered_document["x-dify-console-declarations-sha256"] = contract_declarations_sha256(declarations)
+        args.output_openapi.parent.mkdir(parents=True, exist_ok=True)
+        args.output_openapi.write_text(json.dumps(filtered_document, indent=2) + "\n")

    if args.update_lock:
        LOCK_PATH.write_text(
@@ -147,8 +168,7 @@ def validate_declarations(document: dict[str, Any], declarations: tuple[Contract
            raise ValueError(f"KnowledgeFS OpenAPI path must be absolute: {path}")
        if method not in PROXY_METHODS:
            raise ValueError(f"KnowledgeFS proxy does not support {method.upper()} {path}")
-        expected: ContractDeclaration = {
-            "operation_id": operation_id,
+        expected: dict[DeclarationField, object] = {
            "method": method.upper(),
            "path": path[1:],
            "required_scope": required_scope(operation),
@@ -166,11 +186,114 @@ def validate_declarations(document: dict[str, Any], declarations: tuple[Contract
                    f"KnowledgeFS operation {operation_id} field {field} drifted: "
                    f"expected {expected_value!r}, received {received_value!r}"
                )
+        validate_error_status_map(operation_id, declaration["error_status_map"])
+
+
+def filter_openapi_document(
+    document: dict[str, Any],
+    declarations: tuple[ContractDeclaration, ...],
+) -> dict[str, Any]:
+    """Return a code-generation document containing only Console-allowlisted operations."""
+    filtered_document: dict[str, Any] = {
+        key: value for key, value in document.items() if key not in {"components", "paths"}
+    }
+    source_paths = document.get("paths", {})
+    filtered_paths: dict[str, Any] = {}
+
+    for declaration in declarations:
+        path = f"/{declaration['path']}"
+        method = declaration["method"].lower()
+        source_path_item = source_paths[path]
+        path_metadata = {key: value for key, value in source_path_item.items() if key not in OPENAPI_METHODS}
+        filtered_path_item = filtered_paths.setdefault(path, path_metadata)
+        filtered_operation = deepcopy(source_path_item[method])
+        _rewrite_proxy_error_responses(filtered_operation, declaration["error_status_map"])
+        filtered_path_item[method] = filtered_operation
+
+    filtered_document["paths"] = filtered_paths
+
+    source_components = document.get("components", {})
+    filtered_components = {key: value for key, value in source_components.items() if key != "schemas"}
+    source_schemas = source_components.get("schemas", {})
+    available_schemas = {**source_schemas, CONSOLE_PROXY_ERROR_SCHEMA_NAME: CONSOLE_PROXY_ERROR_SCHEMA}
+    schema_names = _referenced_schema_names(filtered_paths, available_schemas)
+    filtered_components["schemas"] = {
+        name: schema for name, schema in available_schemas.items() if name in schema_names
+    }
+    filtered_document["components"] = filtered_components
+    return filtered_document
+
+
+def validate_error_status_map(operation_id: str, error_status_map: tuple[tuple[int, int], ...]) -> None:
+    """Validate the status normalization advertised by one Console operation."""
+    upstream_statuses: set[int] = set()
+    for upstream_status, console_status in error_status_map:
+        if upstream_status in upstream_statuses:
+            raise ValueError(f"KnowledgeFS operation {operation_id} has duplicate error status: {upstream_status}")
+        if not 400 <= upstream_status <= 599 or not 400 <= console_status <= 599:
+            raise ValueError(f"KnowledgeFS operation {operation_id} has invalid error status mapping")
+        upstream_statuses.add(upstream_status)
+
+
+def _rewrite_proxy_error_responses(
+    operation: dict[str, Any],
+    error_status_map: tuple[tuple[int, int], ...],
+) -> None:
+    responses = operation.setdefault("responses", {})
+    proxy_error_response = {
+        "description": "Error normalized by the Dify Console KnowledgeFS proxy.",
+        "content": {
+            "application/json": {"schema": {"$ref": f"#/components/schemas/{CONSOLE_PROXY_ERROR_SCHEMA_NAME}"}}
+        },
+    }
+    for upstream_status, console_status in error_status_map:
+        existing_target = responses.get(str(console_status)) if upstream_status != console_status else None
+        responses.pop(str(upstream_status), None)
+        normalized_response: dict[str, Any] = deepcopy(proxy_error_response)
+        existing_schema = (
+            existing_target.get("content", {}).get("application/json", {}).get("schema")
+            if isinstance(existing_target, dict)
+            else None
+        )
+        if existing_schema is not None:
+            normalized_response["content"]["application/json"]["schema"] = {
+                "oneOf": [
+                    deepcopy(existing_schema),
+                    {"$ref": f"#/components/schemas/{CONSOLE_PROXY_ERROR_SCHEMA_NAME}"},
+                ]
+            }
+        responses[str(console_status)] = normalized_response
+
+
+def _referenced_schema_names(value: Any, schemas: dict[str, Any]) -> set[str]:
+    reference_prefix = "#/components/schemas/"
+    selected: set[str] = set()
+    pending: list[Any] = [value]
+
+    while pending:
+        current = pending.pop()
+        if isinstance(current, list):
+            pending.extend(current)
+            continue
+        if not isinstance(current, dict):
+            continue
+
+        reference = current.get("$ref")
+        if isinstance(reference, str) and reference.startswith(reference_prefix):
+            name = reference.removeprefix(reference_prefix)
+            if name not in selected:
+                if name not in schemas:
+                    raise ValueError(f"KnowledgeFS OpenAPI references missing schema: {name}")
+                selected.add(name)
+                pending.append(schemas[name])
+        pending.extend(current.values())
+
+    return selected


 def console_contract_declarations() -> tuple[ContractDeclaration, ...]:
    """Return transport declarations from the runtime Console operation registry."""
-    from services.knowledge_fs_proxy import KNOWLEDGE_FS_CONSOLE_OPERATIONS
+    from services.knowledge_fs_operations import KNOWLEDGE_FS_CONSOLE_OPERATIONS

    return tuple(
        {
@@ -183,11 +306,18 @@ def console_contract_declarations() -> tuple[ContractDeclaration, ...]:
            "request_headers": operation.request_headers,
            "response_headers": operation.response_headers,
            "response_media_types": operation.response_media_types,
+            "error_status_map": operation.error_status_map,
        }
        for operation in KNOWLEDGE_FS_CONSOLE_OPERATIONS
    )


+def contract_declarations_sha256(declarations: tuple[ContractDeclaration, ...]) -> str:
+    """Return a stable digest for the runtime Console operation declarations."""
+    content = json.dumps(declarations, separators=(",", ":"), sort_keys=True).encode()
+    return sha256(content)
+
+
 def response_kind(operation: dict[str, Any]) -> str:
    media_types = response_media_types(operation)
    if "text/event-stream" in media_types:
@@ -0,0 +1,11 @@
+from enum import StrEnum
+
+
+class DeploymentEdition(StrEnum):
+    """
+    Enum representing the deployment edition of the platform.
+    """
+
+    COMMUNITY = "COMMUNITY"
+    ENTERPRISE = "ENTERPRISE"
+    CLOUD = "CLOUD"
@@ -157,6 +157,7 @@ def init_app(app: DifyApp) -> Celery:
        "tasks.regenerate_summary_index_task",  # summary index regeneration
        "tasks.initialize_created_app_rbac_access_task",  # app access initialization
        "tasks.install_default_plugins_task",  # tenant default plugin installation
+        "tasks.refresh_billing_vector_space_task",  # billing vector-space cache refresh
        "tasks.app_generate.resume_agent_app_task",  # ENG-635: Agent v2 chat ask_human resume
        "tasks.workflow_run_archive_download_tasks",  # workflow-run archive download preparation
    ]
@@ -1,5 +1,6 @@
 import json
-from typing import cast, override
+import logging
+from typing import assert_never, cast, override

 import flask_login
 from flask import Request, Response, request
@@ -11,6 +12,7 @@ from werkzeug.exceptions import NotFound, Unauthorized
 from configs import dify_config
 from constants import HEADER_NAME_APP_CODE
 from core.db.session_factory import session_factory
+from core.logging.context import set_identity_context
 from dify_app import DifyApp
 from libs.passport import PassportService
 from libs.token import extract_access_token, extract_console_cookie_token, extract_webapp_passport
@@ -19,6 +21,8 @@ from models.enums import EndUserType
 from models.model import AppMCPServer, EndUser
 from services.account_service import AccountService

+logger = logging.getLogger(__name__)
+
 type LoginUser = Account | EndUser


@@ -156,13 +160,24 @@ def _load_user_from_request(request_from_flask_login: Request, session: Session)
@user_logged_in.connect
@user_loaded_from_request.connect
 def on_user_logged_in(_sender: object, user: LoginUser) -> None:
-    """Called when a user logged in.
+    """Snapshot authenticated identity into the side-effect-free logging context.

    Note: AccountService.load_logged_in_account will populate user.current_tenant_id
    through the load_user method, which calls account.set_tenant_id_with_session().
    """
-    # tenant_id context variable removed - using current_user.current_tenant_id directly
-    pass
+    set_identity_context()
+    try:
+        match user:
+            case Account():
+                set_identity_context(tenant_id=user.current_tenant_id, user_id=user.id, user_type="account")
+            case EndUser():
+                set_identity_context(tenant_id=user.tenant_id, user_id=user.id, user_type=user.type or "end_user")
+            case _ as unreachable:
+                assert_never(unreachable)
+    except Exception:
+        # Logging enrichment must never make authentication fail.
+        logger.exception("Failed to set logging identity context")
+        return


@login_manager.unauthorized_handler
@@ -119,6 +119,19 @@ class Storage:
    def delete(self, filename: str):
        return self.storage_runner.delete(filename)

+    def generate_presigned_url(
+        self,
+        filename: str,
+        *,
+        expires_in: int,
+        content_type: str | None = None,
+    ) -> str:
+        return self.storage_runner.generate_presigned_url(
+            filename,
+            expires_in=expires_in,
+            content_type=content_type,
+        )
+
    def scan(self, path: str, files: bool = True, directories: bool = False) -> list[str]:
        return self.storage_runner.scan(path, files=files, directories=directories)

@@ -1,3 +1,4 @@
+from extensions.otel.context import propagate_context
 from extensions.otel.decorators.base import trace_span
 from extensions.otel.decorators.handler import SpanHandler
 from extensions.otel.decorators.handlers.generate_handler import AppGenerateHandler
@@ -7,5 +8,6 @@ __all__ = [
    "AppGenerateHandler",
    "SpanHandler",
    "WorkflowAppRunnerHandler",
+    "propagate_context",
    "trace_span",
 ]
@@ -0,0 +1,21 @@
+"""Utilities for propagating OpenTelemetry context across execution boundaries."""
+
+import functools
+from collections.abc import Callable
+
+from opentelemetry import context as otel_context
+
+
+def propagate_context[**P, R](func: Callable[P, R]) -> Callable[P, R]:
+    """Capture the current context and attach it whenever ``func`` executes."""
+    captured_context = otel_context.get_current()
+
+    @functools.wraps(func)
+    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
+        token = otel_context.attach(captured_context)
+        try:
+            return func(*args, **kwargs)
+        finally:
+            otel_context.detach(token)
+
+    return wrapper
@@ -69,12 +69,17 @@ def on_user_loaded(_sender, user: Union["Account", "EndUser"]):
        if user:
            try:
                current_span = get_current_span()
+                if not current_span.is_recording():
+                    return
                tenant_id = extract_tenant_id(user)
                if not tenant_id:
                    return
-                if current_span:
-                    current_span.set_attribute(DifySpanAttributes.TENANT_ID, tenant_id)
-                    current_span.set_attribute(GenAIAttributes.USER_ID, user.id)
+                current_span.set_attributes(
+                    {
+                        DifySpanAttributes.TENANT_ID: tenant_id,
+                        GenAIAttributes.USER_ID: user.id,
+                    }
+                )
            except Exception:
                logger.exception("Error setting tenant and user attributes")
                pass
@@ -92,3 +92,21 @@ class AwsS3Storage(BaseStorage):
    @override
    def delete(self, filename: str):
        self.client.delete_object(Bucket=self.bucket_name, Key=filename)
+
+    @override
+    def generate_presigned_url(
+        self,
+        filename: str,
+        *,
+        expires_in: int,
+        content_type: str | None = None,
+    ) -> str:
+        params = {"Bucket": self.bucket_name, "Key": filename}
+        if content_type:
+            params["ResponseContentType"] = content_type
+
+        return self.client.generate_presigned_url(
+            "get_object",
+            Params=params,
+            ExpiresIn=expires_in,
+        )
@@ -31,6 +31,16 @@ class BaseStorage(ABC):
    def delete(self, filename: str):
        raise NotImplementedError

+    def generate_presigned_url(
+        self,
+        filename: str,
+        *,
+        expires_in: int,
+        content_type: str | None = None,
+    ) -> str:
+        """Generate a temporary direct-download URL when the backend supports it."""
+        raise NotImplementedError("This storage backend doesn't support presigned URLs")
+
    def scan(self, path, files=True, directories=False) -> list[str]:
        """
        Scan files and directories in the given path.
@@ -1,5 +1,5 @@
 {
-  "commit": "4310e2d582d25e7de58183f27720afab01e123cf",
-  "openapiSha256": "5827ca930ce38462bfd1b2bef387efbf37eb7ffcaedde4558af2fbaeccbfbc4b",
+  "commit": "a0f50470612cc0b3656f89e4f2435aaf412e6e3b",
+  "openapiSha256": "f18910e9c45a64f0855e0643a7a626fb2889021b4f943458de86c6bd2469facb",
  "repository": "https://github.com/langgenius/knowledge-fs"
 }
@@ -9,6 +9,8 @@ from werkzeug.http import HTTP_STATUS_CODES

 from configs import dify_config
 from core.errors.error import AppInvokeQuotaExceededError
+from core.plugin.impl.exc import PluginRuntimeError
+from extensions.ext_logging import get_request_id
 from libs.flask_restx_compat import install_swagger_compatibility
 from libs.token import build_force_logout_cookie_headers

@@ -100,6 +102,20 @@ def register_external_error_handlers(api: Api, body_formatter: ErrorBodyFormatte
        data = {"code": "too_many_requests", "message": str(e), "status": status_code}
        return _finalize(e, data, status_code), status_code

+    def handle_plugin_runtime_error(e: PluginRuntimeError):
+        got_request_exception.send(current_app, exception=e)
+        status_code = 502
+        details = {"request_id": get_request_id()}
+        if e.lambda_request_id:
+            details["lambda_request_id"] = e.lambda_request_id
+        data = {
+            "code": "plugin_runtime_error",
+            "message": e.description,
+            "details": details,
+            "status": status_code,
+        }
+        return _finalize(e, data, status_code), status_code
+
    def handle_general_exception(e: Exception):
        got_request_exception.send(current_app, exception=e)

@@ -121,6 +137,7 @@ def register_external_error_handlers(api: Api, body_formatter: ErrorBodyFormatte
    api.errorhandler(HTTPException)(handle_http_exception)
    api.errorhandler(ValueError)(handle_value_error)
    api.errorhandler(AppInvokeQuotaExceededError)(handle_quota_exceeded)
+    api.errorhandler(PluginRuntimeError)(handle_plugin_runtime_error)
    api.errorhandler(Exception)(handle_general_exception)


@@ -221,8 +221,11 @@ def current_timestamp() -> int:
 def email(email):
    # Define a regex pattern for email addresses
    pattern = r"^[\w\.!#$%&'*+\-/=?^_`{|}~]+@([\w-]+\.)+[\w-]{2,}$"
-    # Check if the email matches the pattern
-    if re.match(pattern, email) is not None:
+    # Use re.fullmatch instead of re.match to reject trailing newlines.
+    # In Python, '$' matches at end-of-string OR just before a trailing newline,
+    # so re.match accepts "user@example.com\n". re.fullmatch requires the entire
+    # string to match, closing the mail header-injection vector. (#39234)
+    if re.fullmatch(pattern, email) is not None:
        return email

    error = f"{email} is not a valid email."
@@ -23,9 +23,9 @@ class PaginatedResult[T]:

    @property
    def pages(self) -> int:
-        if self.per_page == 0:
+        if self.total == 0 or self.per_page == 0:
            return 0
-        return max(1, math.ceil(self.total / self.per_page))
+        return math.ceil(self.total / self.per_page)

    @property
    def has_next(self) -> bool:
@@ -0,0 +1,77 @@
+"""scope agent debug conversations by draft type
+
+Revision ID: d2825e7b9c10
+Revises: b8c9d0e1f2a3
+Create Date: 2026-07-22 15:00:00.000000
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+import models
+
+# revision identifiers, used by Alembic.
+revision = "d2825e7b9c10"
+down_revision = "b8c9d0e1f2a3"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # Existing pointers have always represented Build chat because the Agent
+    # detail API exposes them as ``debug_conversation_id`` for that surface.
+    op.add_column(
+        "agent_debug_conversations",
+        sa.Column(
+            "draft_type",
+            sa.String(length=32),
+            nullable=False,
+            server_default=sa.text("'debug_build'"),
+        ),
+    )
+    op.drop_constraint(
+        "agent_debug_conversation_agent_account_unique",
+        "agent_debug_conversations",
+        type_="unique",
+    )
+    op.create_unique_constraint(
+        "agent_debug_conversation_agent_account_draft_type_unique",
+        "agent_debug_conversations",
+        ["tenant_id", "agent_id", "account_id", "draft_type"],
+    )
+
+
+def downgrade():
+    debug_conversations = sa.table(
+        "agent_debug_conversations",
+        sa.column("tenant_id", models.types.StringUUID()),
+        sa.column("agent_id", models.types.StringUUID()),
+        sa.column("account_id", models.types.StringUUID()),
+        sa.column("draft_type", sa.String(length=32)),
+    )
+    build_conversations = debug_conversations.alias("build_conversations")
+    op.get_bind().execute(
+        sa.delete(debug_conversations).where(
+            debug_conversations.c.draft_type == "draft",
+            sa.exists(
+                sa.select(sa.literal(1)).where(
+                    build_conversations.c.tenant_id == debug_conversations.c.tenant_id,
+                    build_conversations.c.agent_id == debug_conversations.c.agent_id,
+                    build_conversations.c.account_id == debug_conversations.c.account_id,
+                    build_conversations.c.draft_type == "debug_build",
+                )
+            ),
+        )
+    )
+    op.drop_constraint(
+        "agent_debug_conversation_agent_account_draft_type_unique",
+        "agent_debug_conversations",
+        type_="unique",
+    )
+    op.create_unique_constraint(
+        "agent_debug_conversation_agent_account_unique",
+        "agent_debug_conversations",
+        ["tenant_id", "agent_id", "account_id"],
+    )
+    op.drop_column("agent_debug_conversations", "draft_type")
@@ -222,11 +222,13 @@ class Agent(DefaultFieldsMixin, Base):


 class AgentDebugConversation(DefaultFieldsMixin, Base):
-    """Per-account console debug conversation for an Agent App.
+    """Per-account, per-draft console debug conversation for an Agent App.

    Agent App preview state must be isolated by editor account. The Agent row is
    shared by everyone in the workspace, so this table owns the user-specific
-    conversation pointer used by console debug chat.
+    conversation pointers used by console debug chat. ``draft`` is the Preview
+    conversation and ``debug_build`` is the Build conversation; they must never
+    share persisted messages or runtime sessions.
    """

    __tablename__ = "agent_debug_conversations"
@@ -236,7 +238,8 @@ class AgentDebugConversation(DefaultFieldsMixin, Base):
            "tenant_id",
            "agent_id",
            "account_id",
-            name="agent_debug_conversation_agent_account_unique",
+            "draft_type",
+            name="agent_debug_conversation_agent_account_draft_type_unique",
        ),
        Index("agent_debug_conversation_conversation_idx", "conversation_id"),
        Index("agent_debug_conversation_account_idx", "tenant_id", "account_id"),
@@ -246,6 +249,12 @@ class AgentDebugConversation(DefaultFieldsMixin, Base):
    agent_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    account_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    draft_type: Mapped[AgentConfigDraftType] = mapped_column(
+        EnumText(AgentConfigDraftType, length=32),
+        nullable=False,
+        default=AgentConfigDraftType.DEBUG_BUILD,
+        server_default=sa.text("'debug_build'"),
+    )
    conversation_id: Mapped[str] = mapped_column(StringUUID, nullable=False)


@@ -7,6 +7,7 @@ from typing import Annotated, Any, Final, Literal, Self
 from pydantic import BaseModel, ConfigDict, Field, WithJsonSchema, field_validator, model_validator

 from core.rag.entities.metadata_entities import ConditionValue, SupportedComparisonOperator
+from core.tools.entities.tool_entities import ToolProviderType
 from core.workflow.file_reference import is_canonical_file_reference
 from graphon.file import FileTransferMethod, FileType

@@ -43,18 +44,28 @@ _DECLARED_OUTPUT_CHILDREN_JSON_SCHEMA = {
            },
            "description": {"anyOf": [{"type": "string"}, {"type": "null"}]},
            "required": {"type": "boolean"},
-            "file": {"type": "object", "additionalProperties": True},
+            "file": {
+                "anyOf": [
+                    {"type": "object", "additionalProperties": True},
+                    {"type": "null"},
+                ]
+            },
            "array_item": {
-                "type": "object",
-                "additionalProperties": True,
-                "properties": {
-                    "type": {
-                        "type": "string",
-                        "enum": [item.value for item in DeclaredOutputType],
+                "anyOf": [
+                    {
+                        "type": "object",
+                        "additionalProperties": True,
+                        "properties": {
+                            "type": {
+                                "type": "string",
+                                "enum": [item.value for item in DeclaredOutputType],
+                            },
+                            "description": {"anyOf": [{"type": "string"}, {"type": "null"}]},
+                            "children": {"type": "array", "items": {"type": "object", "additionalProperties": True}},
+                        },
                    },
-                    "description": {"anyOf": [{"type": "string"}, {"type": "null"}]},
-                    "children": {"type": "array", "items": {"type": "object", "additionalProperties": True}},
-                },
+                    {"type": "null"},
+                ]
            },
            "children": {"type": "array", "items": {"type": "object", "additionalProperties": True}},
        },
@@ -653,11 +664,7 @@ class AgentSoulDifyToolConfig(BaseModel):
    model_config = ConfigDict(extra="ignore")

    enabled: bool = True
-    # ``plugin`` remains the default for legacy Agent Soul payloads. The runtime
-    # now also accepts ``builtin`` / ``api`` / ``workflow`` / ``mcp`` here and
-    # routes them through ``dify.core.tools``; keeping the default narrow still
-    # makes a missing field resolve against the plugin provider table.
-    provider_type: str = "plugin"
+    provider_type: ToolProviderType
    provider_id: str | None = Field(default=None, max_length=255)
    plugin_id: str | None = Field(default=None, max_length=255)
    provider: str | None = Field(default=None, max_length=255)
@@ -56,6 +56,7 @@ from .provider_ids import GenericProviderID
 from .types import EnumText, LongText, StringUUID

 if TYPE_CHECKING:
+    from .agent import Agent
    from .workflow import Workflow


@@ -501,25 +502,42 @@ class App(Base):
        Resolved via ``Agent.app_id`` so the console can open the Composer in
        roster-detail mode from the app id. ``None`` for non-agent apps.
        """
+        agent = self.agent_app_binding_with_session(session=session)
+        return agent.id if agent else None
+
+    def agent_app_binding_with_session(self, *, session: Session, include_archived: bool = False) -> Agent | None:
+        """For an Agent App (mode=agent), the Agent bound to it.
+
+        A roster Agent is bound through ``Agent.app_id``; a workflow-only Agent
+        is bound to its hidden runtime backing App through
+        ``Agent.backing_app_id``. Callers branch on ``Agent.scope`` to tell the
+        public roster Agent App apart from the hidden backing App. Archived
+        Agents are excluded unless ``include_archived`` is set (authorization
+        gates must keep covering an Agent App after its Agent is archived).
+        ``None`` for non-agent apps and unbound agent apps.
+        """
        if self.mode != AppMode.AGENT:
            return None
        from .agent import APP_BACKED_AGENT_SOURCES, Agent, AgentScope, AgentStatus

-        agent = session.scalar(
-            select(Agent).where(
-                Agent.tenant_id == self.tenant_id,
-                sa.or_(
-                    sa.and_(
-                        Agent.app_id == self.id,
-                        Agent.scope == AgentScope.ROSTER,
-                        Agent.source.in_(APP_BACKED_AGENT_SOURCES),
-                    ),
-                    Agent.backing_app_id == self.id,
+        conditions = [
+            Agent.tenant_id == self.tenant_id,
+            sa.or_(
+                sa.and_(
+                    Agent.app_id == self.id,
+                    Agent.scope == AgentScope.ROSTER,
+                    Agent.source.in_(APP_BACKED_AGENT_SOURCES),
                ),
-                Agent.status == AgentStatus.ACTIVE,
-            )
-        )
-        return agent.id if agent else None
+                sa.and_(
+                    Agent.backing_app_id == self.id,
+                    Agent.scope == AgentScope.WORKFLOW_ONLY,
+                ),
+            ),
+        ]
+        if not include_archived:
+            conditions.append(Agent.status == AgentStatus.ACTIVE)
+
+        return session.scalar(select(Agent).where(*conditions).limit(1))

    @property
    def api_base_url(self) -> str:
@@ -260,7 +260,12 @@ Get account avatar url
 | 200 | Success | **application/json**: [AccountResponse](#accountresponse)<br> |

 ### [POST] /activate
+**Accept an invitation without letting an existing session act for another account**
+
 Activate account with invitation token
+Token-only activation remains available for legacy clients. When the request already
+carries a console session, that session must belong to the account encoded in the
+invitation before the token is consumed or tenant membership is changed.

 #### Request Body

@@ -531,6 +536,7 @@ Run a build-draft Agent App turn that asks the agent to push config updates
 | Code | Description | Schema |
 | ---- | ----------- | ------ |
 | 200 | Agent build draft | **application/json**: [AgentBuildDraftResponse](#agentbuilddraftresponse)<br> |
+| 404 | Agent build draft not found |  |

 ### [PUT] /agent/{agent_id}/build-draft
 #### Parameters
@@ -958,6 +964,12 @@ Stop a running Agent App chat message generation
 | ---- | ---------- | ----------- | -------- | ------ |
 | agent_id | path |  | Yes | string (uuid) |

+#### Request Body
+
+| Required | Schema |
+| -------- | ------ |
+|  No | **application/json**: [AgentDebugConversationRefreshPayload](#agentdebugconversationrefreshpayload)<br> |
+
 #### Responses

 | Code | Description | Schema |
@@ -9471,7 +9483,8 @@ data required for dashboard initialization.

 Authentication would create circular dependency (can't login without dashboard loading).

-Only non-sensitive configuration data should be returned by this endpoint.
+Only non-sensitive configuration data should be returned by this endpoint. Authenticated
+license detail is served separately by SystemFeatureLicenseApi.

 #### Responses

@@ -9479,6 +9492,19 @@ Only non-sensitive configuration data should be returned by this endpoint.
 | ---- | ----------- | ------ |
 | 200 | Success | **application/json**: [SystemFeatureModel](#systemfeaturemodel)<br> |

+### [GET] /system-features/license
+**Get full license detail (status, expiry, workspace/seat usage)**
+
+Get license status and usage detail
+Authenticated counterpart to the license *status* exposed on the public
+system-features endpoint.
+
+#### Responses
+
+| Code | Description | Schema |
+| ---- | ----------- | ------ |
+| 200 | Success | **application/json**: [LicenseModel](#licensemodel)<br> |
+
 ### [POST] /tag-bindings
 #### Request Body

@@ -11091,6 +11117,19 @@ Returns permission flags that control workspace features like member invitations
 | ---- | ----------- | ------ |
 | 200 | Success | **application/json**: [PluginInstallTaskStartResponse](#plugininstalltaskstartresponse)<br> |

+### [GET] /workspaces/current/plugin/installed-ids
+#### Parameters
+
+| Name | Located in | Description | Required | Schema |
+| ---- | ---------- | ----------- | -------- | ------ |
+| category | query | Plugin category to include | Yes | string, <br>**Available values:** "agent-strategy", "datasource", "extension", "model", "tool", "trigger" |
+
+#### Responses
+
+| Code | Description | Schema |
+| ---- | ----------- | ------ |
+| 200 | Success | **application/json**: [PluginInstalledIdsResponse](#plugininstalledidsresponse)<br> |
+
 ### [GET] /workspaces/current/plugin/list
 #### Parameters

@@ -11332,6 +11371,12 @@ Returns permission flags that control workspace features like member invitations
 | 200 | Success | **application/json**: [PluginDecodeResponse](#plugindecoderesponse)<br> |

 ### [POST] /workspaces/current/plugin/upload/pkg
+#### Request Body
+
+| Required | Schema |
+| -------- | ------ |
+|  Yes | **multipart/form-data**: { **"pkg"**: binary }<br> |
+
 #### Responses

 | Code | Description | Schema |
@@ -11343,8 +11388,11 @@ Returns permission flags that control workspace features like member invitations

 | Name | Located in | Description | Required | Schema |
 | ---- | ---------- | ----------- | -------- | ------ |
+| language | query | Language used for localized label and description search | No | string, <br>**Available values:** "en_US", "ja_JP", "pt_BR", "zh_Hans", <br>**Default:** en_US |
 | page | query | Page number | No | integer, <br>**Default:** 1 |
 | page_size | query | Page size (1-256) | No | integer, <br>**Default:** 256 |
+| query | query | Case-insensitive search query | No | string |
+| tags | query | Match any plugin tag | No | [ string ] |
 | category | path |  | Yes | string |

 #### Responses
@@ -13278,7 +13326,7 @@ Model class for AI model.
 | maintainer | string |  | No |
 | max_active_requests | integer |  | No |
 | mode | string |  | Yes |
-| model_config | [ModelConfig](#modelconfig) |  | No |
+| model_config | [AppModelConfigResponse](#appmodelconfigresponse) |  | No |
 | name | string |  | Yes |
 | permission_keys | [ string ] |  | No |
 | role | string |  | No |
@@ -13899,6 +13947,12 @@ Stable Agent Soul reference to one normalized skill archive.
 | date | string |  | Yes |
 | message_count | integer |  | Yes |

+#### AgentDebugConversationRefreshPayload
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| draft_type | [AgentConfigDraftType](#agentconfigdrafttype) | Agent draft surface whose conversation should be refreshed | No |
+
 #### AgentDebugConversationRefreshResponse

 | Name | Type | Description | Required |
@@ -14737,7 +14791,7 @@ should send ``plugin_id`` + ``provider`` when available.
 | plugin_id | string |  | No |
 | provider | string |  | No |
 | provider_id | string |  | No |
-| provider_type | string, <br>**Default:** plugin |  | No |
+| provider_type | [ToolProviderType](#toolprovidertype) |  | Yes |
 | runtime_parameters | object |  | No |
 | tool_name | string |  | No |

@@ -15348,7 +15402,6 @@ This class is used to store the schema information of an api based tool.
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | access_mode | string |  | No |
-| app_model_config | [ModelConfig](#modelconfig) |  | No |
 | created_at | integer |  | No |
 | created_by | string |  | No |
 | description | string |  | No |
@@ -15358,7 +15411,8 @@ This class is used to store the schema information of an api based tool.
 | icon_background | string |  | No |
 | id | string |  | Yes |
 | maintainer | string |  | No |
-| mode_compatible_with_agent | string |  | Yes |
+| mode | string |  | Yes |
+| model_config | [AppModelConfigResponse](#appmodelconfigresponse) |  | No |
 | name | string |  | Yes |
 | permission_keys | [ string ] |  | No |
 | tags | [ [Tag](#tag) ] |  | No |
@@ -15420,7 +15474,7 @@ This class is used to store the schema information of an api based tool.
 | maintainer | string |  | No |
 | max_active_requests | integer |  | No |
 | mode | string |  | Yes |
-| model_config | [ModelConfig](#modelconfig) |  | No |
+| model_config | [AppModelConfigResponse](#appmodelconfigresponse) |  | No |
 | name | string |  | Yes |
 | permission_keys | [ string ] |  | No |
 | site | [AppDetailSiteResponse](#appdetailsiteresponse) |  | No |
@@ -15519,6 +15573,35 @@ AppMCPServer Status Enum
 | ---- | ---- | ----------- | -------- |
 | AppMCPServerStatus | string | AppMCPServer Status Enum |  |

+#### AppModelConfigResponse
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| agent_mode |  |  | No |
+| annotation_reply |  |  | No |
+| chat_prompt_config |  |  | No |
+| completion_prompt_config |  |  | No |
+| created_at | integer |  | No |
+| created_by | string |  | No |
+| dataset_configs |  |  | No |
+| dataset_query_variable | string |  | No |
+| external_data_tools |  |  | No |
+| file_upload |  |  | No |
+| model |  |  | No |
+| more_like_this |  |  | No |
+| opening_statement | string |  | No |
+| pre_prompt | string |  | No |
+| prompt_type | string |  | No |
+| retriever_resource |  |  | No |
+| sensitive_word_avoidance |  |  | No |
+| speech_to_text |  |  | No |
+| suggested_questions |  |  | No |
+| suggested_questions_after_answer |  |  | No |
+| text_to_speech |  |  | No |
+| updated_at | integer |  | No |
+| updated_by | string |  | No |
+| user_input_form |  |  | No |
+
 #### AppNamePayload

 | Name | Type | Description | Required |
@@ -17147,7 +17230,7 @@ about. Stage 4 §4.2.

 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| children | [ { **"array_item"**: { **"children"**: [ object ], **"description"**: , **"type"**: string, <br>**Available values:** "array", "boolean", "file", "number", "object", "string" }, **"children"**: [ object ], **"description"**: , **"file"**: object, **"name"**: string, **"required"**: boolean, **"type"**: string, <br>**Available values:** "array", "boolean", "file", "number", "object", "string" } ] |  | No |
+| children | [ { **"array_item"**: , **"children"**: [ object ], **"description"**: , **"file"**: , **"name"**: string, **"required"**: boolean, **"type"**: string, <br>**Available values:** "array", "boolean", "file", "number", "object", "string" } ] |  | No |
 | description | string |  | No |
 | type | [DeclaredOutputType](#declaredoutputtype) |  | Yes |

@@ -17176,7 +17259,7 @@ code can call ``output.failure_strategy.on_failure`` without None-guards.
 | ---- | ---- | ----------- | -------- |
 | array_item | [DeclaredArrayItem](#declaredarrayitem) |  | No |
 | check | [DeclaredOutputCheckConfig](#declaredoutputcheckconfig) |  | No |
-| children | [ { **"array_item"**: { **"children"**: [ object ], **"description"**: , **"type"**: string, <br>**Available values:** "array", "boolean", "file", "number", "object", "string" }, **"children"**: [ object ], **"description"**: , **"file"**: object, **"name"**: string, **"required"**: boolean, **"type"**: string, <br>**Available values:** "array", "boolean", "file", "number", "object", "string" } ] |  | No |
+| children | [ { **"array_item"**: , **"children"**: [ object ], **"description"**: , **"file"**: , **"name"**: string, **"required"**: boolean, **"type"**: string, <br>**Available values:** "array", "boolean", "file", "number", "object", "string" } ] |  | No |
 | description | string |  | No |
 | failure_strategy | [DeclaredOutputFailureStrategy](#declaredoutputfailurestrategy) |  | No |
 | file | [DeclaredOutputFileConfig](#declaredoutputfileconfig) |  | No |
@@ -17265,6 +17348,14 @@ Default model entity.
 | tool_name | string |  | Yes |
 | type | string |  | Yes |

+#### DeploymentEdition
+
+Enum representing the deployment edition of the platform.
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| DeploymentEdition | string | Enum representing the deployment edition of the platform. |  |
+
 #### DismissNotificationPayload

 | Name | Type | Description | Required |
@@ -18762,6 +18853,12 @@ Enum class for large language model mode.
 | ---- | ---- | ----------- | -------- |
 | LicenseStatus | string |  |  |

+#### LicenseStatusModel
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| status | [LicenseStatus](#licensestatus) |  | Yes |
+
 #### LimitationModel

 | Name | Type | Description | Required |
@@ -19506,6 +19603,7 @@ Coarse node-level status used by Inspector to pick a banner.
 | ---- | ---- | ----------- | -------- |
 | avatar | string |  | No |
 | email | string |  | Yes |
+| id | string |  | Yes |
 | interface_language | string |  | Yes |
 | name | string |  | Yes |
 | timezone | string |  | Yes |
@@ -20230,8 +20328,11 @@ Shared permission levels for resources (datasets, credentials, etc.)

 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
+| language | string, <br>**Available values:** "en_US", "ja_JP", "pt_BR", "zh_Hans", <br>**Default:** en_US | Language used for localized label and description search<br>*Enum:* `"en_US"`, `"ja_JP"`, `"pt_BR"`, `"zh_Hans"` | No |
 | page | integer, <br>**Default:** 1 | Page number | No |
 | page_size | integer, <br>**Default:** 256 | Page size (1-256) | No |
+| query | string | Case-insensitive search query | No |
+| tags | [ string ] | Match any plugin tag | No |

 #### PluginCategoryListResponse

@@ -20432,6 +20533,18 @@ Shared permission levels for resources (datasets, credentials, etc.)
 | ---- | ---- | ----------- | -------- |
 | plugins | [ [PluginInstallationItemResponse](#plugininstallationitemresponse) ] |  | Yes |

+#### PluginInstalledIdsQuery
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| category | [PluginCategory](#plugincategory) | Plugin category to include | Yes |
+
+#### PluginInstalledIdsResponse
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| plugin_ids | [ string ] |  | Yes |
+
 #### PluginListResponse

 | Name | Type | Description | Required |
@@ -21954,9 +22067,9 @@ The subscription constructor of the trigger provider

 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| hash | string |  | No |
-| result | string |  | No |
-| updated_at | string |  | No |
+| hash | string |  | Yes |
+| result | string |  | Yes |
+| updated_at | integer |  | Yes |

 #### SystemConfigurationResponse

@@ -21973,6 +22086,7 @@ Model class for provider system configuration response.
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | branding | [BrandingModel](#brandingmodel) |  | Yes |
+| deployment_edition | [DeploymentEdition](#deploymentedition) |  | Yes |
 | enable_app_deploy | boolean |  | Yes |
 | enable_change_email | boolean, <br>**Default:** true |  | Yes |
 | enable_collaboration_mode | boolean, <br>**Default:** true |  | Yes |
@@ -21988,7 +22102,8 @@ Model class for provider system configuration response.
 | is_allow_create_workspace | boolean |  | Yes |
 | is_allow_register | boolean |  | Yes |
 | is_email_setup | boolean |  | Yes |
-| license | [LicenseModel](#licensemodel) |  | Yes |
+| knowledge_fs_enabled | boolean |  | Yes |
+| license | [LicenseStatusModel](#licensestatusmodel) |  | Yes |
 | max_plugin_package_size | integer, <br>**Default:** 15728640 |  | Yes |
 | plugin_installation_permission | [PluginInstallationPermissionModel](#plugininstallationpermissionmodel) |  | Yes |
 | plugin_manager | [PluginManagerModel](#pluginmanagermodel) |  | Yes |
@@ -22281,7 +22396,7 @@ Tool label

 #### ToolParameter

-Overrides type
+Tool-specific parameter declaration and invocation-value normalization.

 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
@@ -22294,6 +22409,7 @@ Overrides type
 | llm_description | string |  | No |
 | max | number<br>integer |  | No |
 | min | number<br>integer |  | No |
+| multiple | boolean | Whether the parameter is multiple select, only valid for select or dynamic-select type | No |
 | name | string | The name of the parameter | Yes |
 | options | [ [PluginParameterOption](#pluginparameteroption) ] |  | No |
 | placeholder | [I18nObject](#i18nobject) | The placeholder presented to the user | No |
@@ -1058,6 +1058,14 @@ Button styles for user actions.
 | auto_generate | boolean | Automatically generate the conversation name. When `true`, the `name` field is ignored. | No |
 | name | string | Conversation name. Required when `auto_generate` is `false`. | No |

+#### DeploymentEdition
+
+Enum representing the deployment edition of the platform.
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| DeploymentEdition | string | Enum representing the deployment edition of the platform. |  |
+
 #### EmailCodeLoginSendPayload

 | Name | Type | Description | Required |
@@ -1284,33 +1292,18 @@ Parsed multipart form fields for HITL uploads.
 | ---- | ---- | ----------- | -------- |
 | JsonValue |  |  |  |

-#### LicenseLimitationModel
-
- enabled: whether this limit is enforced
- size: current usage count
- limit: maximum allowed count; 0 means unlimited
-
-| Name | Type | Description | Required |
-| ---- | ---- | ----------- | -------- |
-| enabled | boolean | Whether this limit is currently active | Yes |
-| limit | integer | Maximum number of resources allowed; 0 means no limit | Yes |
-| size | integer | Number of resources already consumed | Yes |
-
-#### LicenseModel
-
-| Name | Type | Description | Required |
-| ---- | ---- | ----------- | -------- |
-| expired_at | string |  | Yes |
-| seats | [LicenseLimitationModel](#licenselimitationmodel) |  | Yes |
-| status | [LicenseStatus](#licensestatus) |  | Yes |
-| workspaces | [LicenseLimitationModel](#licenselimitationmodel) |  | Yes |
-
 #### LicenseStatus

 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | LicenseStatus | string |  |  |

+#### LicenseStatusModel
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| status | [LicenseStatus](#licensestatus) |  | Yes |
+
 #### LoginPayload

 | Name | Type | Description | Required |
@@ -1567,6 +1560,7 @@ Default configuration for form inputs.
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | branding | [BrandingModel](#brandingmodel) |  | Yes |
+| deployment_edition | [DeploymentEdition](#deploymentedition) |  | Yes |
 | enable_app_deploy | boolean |  | Yes |
 | enable_change_email | boolean, <br>**Default:** true |  | Yes |
 | enable_collaboration_mode | boolean, <br>**Default:** true |  | Yes |
@@ -1582,7 +1576,8 @@ Default configuration for form inputs.
 | is_allow_create_workspace | boolean |  | Yes |
 | is_allow_register | boolean |  | Yes |
 | is_email_setup | boolean |  | Yes |
-| license | [LicenseModel](#licensemodel) |  | Yes |
+| knowledge_fs_enabled | boolean |  | Yes |
+| license | [LicenseStatusModel](#licensestatusmodel) |  | Yes |
 | max_plugin_package_size | integer, <br>**Default:** 15728640 |  | Yes |
 | plugin_installation_permission | [PluginInstallationPermissionModel](#plugininstallationpermissionmodel) |  | Yes |
 | plugin_manager | [PluginManagerModel](#pluginmanagermodel) |  | Yes |
@@ -1733,7 +1728,7 @@ in form definiton, or a variable while the workflow is running.
 | icon | string |  | No |
 | icon_background | string |  | No |
 | icon_type | string |  | No |
-| icon_url | string |  | Yes |
+| icon_url | string |  | No |
 | input_placeholder | string |  | No |
 | privacy_policy | string |  | No |
 | prompt_public | boolean |  | No |
@@ -442,9 +442,9 @@ class AccountService:

        # A licensed seat is one Account row, deployment-wide; joining an existing
        # account into another workspace does not pass through here and costs no seat.
-        # is_authenticated=True: server-side enforcement needs the full license payload,
-        # which the enterprise fill withholds from unauthenticated (browser-facing) calls.
-        if not FeatureService.get_system_features(is_authenticated=True).license.seats.is_available():
+        # get_license() carries the full license payload that server-side enforcement needs;
+        # the public system-features endpoint exposes only license status.
+        if not FeatureService.get_license().seats.is_available():
            raise SeatsLimitExceededError("licensed seats limit exceeded")

        if dify_config.BILLING_ENABLED and BillingService.is_email_in_freeze(email):
@@ -1229,12 +1229,12 @@ class AccountService:
            if hour_limit_count >= 1:
                redis_client.setex(freeze_key, 60 * 60, 1)
                return True
-            else:
-                redis_client.setex(hour_limit_key, 60 * 10, hour_limit_count + 1)  # first time limit 10 minutes

-            # add hour limit count
-            redis_client.incr(hour_limit_key)
-            redis_client.expire(hour_limit_key, 60 * 60)
+            # First strike claims a 10-minute window atomically; a concurrent
+            # over-limit request that loses the claim is the second strike and
+            # freezes the IP for an hour.
+            if not redis_client.set(hour_limit_key, 1, ex=60 * 10, nx=True):
+                redis_client.setex(freeze_key, 60 * 60, 1)

            return True

@@ -1332,7 +1332,7 @@ class TenantService:
        ):
            raise WorkSpaceNotAllowedCreateError()

-        workspaces = FeatureService.get_system_features().license.workspaces
+        workspaces = FeatureService.get_license().workspaces
        if not workspaces.is_available():
            raise WorkspacesLimitExceededError()

@@ -2012,7 +2012,7 @@ class RegisterService:
            if (
                FeatureService.get_system_features().is_allow_create_workspace
                and create_workspace_required
-                and FeatureService.get_system_features().license.workspaces.is_available()
+                and FeatureService.get_license().workspaces.is_available()
            ):
                try:
                    TenantService.create_owner_tenant(account, session=session)
@@ -7,6 +7,7 @@ from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 from sqlalchemy.sql.elements import ColumnElement

+from core.agent.publish_visibility import agent_has_workflow_callable_active_snapshot
 from libs.helper import to_timestamp
 from models import Account
 from models.agent import (
@@ -271,6 +272,8 @@ class AgentComposerService:
        source_snapshot_id: str | None = None,
        idempotency_key: str | None = None,
    ) -> dict[str, Any]:
+        """Copy a callable roster Agent snapshot into a workflow-owned inline Agent."""
+
        workflow = cls._get_draft_workflow(session=session, tenant_id=tenant_id, app_id=app_id)
        binding = cls._require_binding(
            cls._get_workflow_binding(session=session, tenant_id=tenant_id, workflow_id=workflow.id, node_id=node_id)
@@ -296,6 +299,8 @@ class AgentComposerService:
        source_agent = cls._require_agent(session=session, tenant_id=tenant_id, agent_id=source_agent_id)
        if source_agent.scope != AgentScope.ROSTER or source_agent.status != AgentStatus.ACTIVE:
            raise InvalidComposerConfigError("Source agent must be an active roster agent.")
+        if not agent_has_workflow_callable_active_snapshot(session=session, agent=source_agent):
+            raise InvalidComposerConfigError("Source agent must have a published config snapshot.")
        source_version = cls._require_version(
            session=session,
            tenant_id=tenant_id,
@@ -529,39 +534,11 @@ class AgentComposerService:
        )
        if not active_version:
            return False
-        if agent.source in APP_BACKED_AGENT_SOURCES and not cls._has_publish_visible_revision(
-            session=session,
-            tenant_id=tenant_id,
-            agent_id=agent.id,
-            snapshot_id=agent.active_config_snapshot_id,
-        ):
+        if not agent_has_workflow_callable_active_snapshot(session=session, agent=agent):
            return False

        return _agent_soul_config_json(agent_soul) == _agent_soul_config_json(active_version.config_snapshot_dict)

-    @classmethod
-    def _has_publish_visible_revision(
-        cls, *, session: Session, tenant_id: str, agent_id: str, snapshot_id: str
-    ) -> bool:
-        revisions = session.scalars(
-            select(AgentConfigRevision.operation).where(
-                AgentConfigRevision.tenant_id == tenant_id,
-                AgentConfigRevision.agent_id == agent_id,
-                AgentConfigRevision.current_snapshot_id == snapshot_id,
-            )
-        ).all()
-
-        return any(
-            operation
-            in {
-                AgentConfigRevisionOperation.PUBLISH_DRAFT,
-                AgentConfigRevisionOperation.SAVE_NEW_VERSION,
-                AgentConfigRevisionOperation.SAVE_TO_ROSTER,
-                AgentConfigRevisionOperation.RESTORE_VERSION,
-            }
-            for operation in revisions
-        )
-
    @classmethod
    def publish_agent_app_draft(
        cls, *, session: Session, tenant_id: str, agent_id: str, account_id: str, version_note: str | None = None
@@ -1164,6 +1141,20 @@ class AgentComposerService:
                agent.active_config_is_published = True
                agent.updated_by = account_id
                binding.current_snapshot_id = version.id
+                normal_draft = cls._get_agent_draft(
+                    session=session,
+                    tenant_id=tenant_id,
+                    agent_id=agent.id,
+                    draft_type=AgentConfigDraftType.DRAFT,
+                    account_id=None,
+                )
+                if normal_draft is not None and cls._rebase_workflow_only_normal_draft(
+                    agent=agent,
+                    draft=normal_draft,
+                    snapshot=version,
+                    updated_by=account_id,
+                ):
+                    session.flush()
            binding.updated_by = account_id
            return binding

@@ -1748,6 +1739,47 @@ class AgentComposerService:
            stmt = stmt.where(AgentConfigDraft.account_id.is_(None))
        return session.scalar(stmt.order_by(AgentConfigDraft.updated_at.desc()).limit(1))

+    @classmethod
+    def get_or_create_normal_agent_draft(
+        cls,
+        *,
+        session: Session,
+        tenant_id: str,
+        agent: Agent,
+        created_by: str | None,
+    ) -> AgentConfigDraft:
+        """Resolve the shared Preview draft, rebasing inline agents when needed."""
+        return cls._get_or_create_agent_draft(
+            session=session,
+            tenant_id=tenant_id,
+            agent=agent,
+            draft_type=AgentConfigDraftType.DRAFT,
+            account_id=None,
+            created_by=created_by,
+        )
+
+    @staticmethod
+    def _rebase_workflow_only_normal_draft(
+        *,
+        agent: Agent,
+        draft: AgentConfigDraft,
+        snapshot: AgentConfigSnapshot,
+        updated_by: str | None,
+    ) -> bool:
+        if (
+            agent.scope != AgentScope.WORKFLOW_ONLY
+            or draft.draft_type != AgentConfigDraftType.DRAFT
+            or draft.account_id is not None
+            or not agent.active_config_snapshot_id
+            or draft.base_snapshot_id == agent.active_config_snapshot_id
+            or snapshot.id != agent.active_config_snapshot_id
+        ):
+            return False
+        draft.base_snapshot_id = snapshot.id
+        draft.config_snapshot = AgentSoulConfig.model_validate(snapshot.config_snapshot_dict)
+        draft.updated_by = updated_by
+        return True
+
    @classmethod
    def _get_or_create_agent_draft(
        cls,
@@ -1767,6 +1799,26 @@ class AgentComposerService:
            account_id=account_id,
        )
        if draft is not None:
+            if (
+                agent.scope == AgentScope.WORKFLOW_ONLY
+                and draft_type == AgentConfigDraftType.DRAFT
+                and draft.account_id is None
+                and agent.active_config_snapshot_id
+                and draft.base_snapshot_id != agent.active_config_snapshot_id
+            ):
+                active_snapshot = cls._get_version_if_present(
+                    session=session,
+                    tenant_id=tenant_id,
+                    agent_id=agent.id,
+                    version_id=agent.active_config_snapshot_id,
+                )
+                if active_snapshot is not None and cls._rebase_workflow_only_normal_draft(
+                    agent=agent,
+                    draft=draft,
+                    snapshot=active_snapshot,
+                    updated_by=agent.updated_by or agent.created_by,
+                ):
+                    session.flush()
            return draft
        base_snapshot = cls._get_version_if_present(
            session=session,
@@ -1,8 +1,11 @@
 from __future__ import annotations

+import json
+from collections.abc import Mapping
 from dataclasses import dataclass
 from datetime import datetime
 from decimal import Decimal
+from enum import Enum
 from typing import Any

 import sqlalchemy as sa
@@ -11,6 +14,7 @@ from sqlalchemy.orm import aliased

 from configs import dify_config
 from core.app.entities.app_invoke_entities import InvokeFrom
+from graphon.enums import WorkflowNodeExecutionStatus
 from libs.helper import convert_datetime_to_date, escape_like_pattern, to_timestamp
 from models.agent import WorkflowAgentNodeBinding
 from models.enums import CreatorUserRole, MessageStatus
@@ -194,11 +198,12 @@ class AgentObservabilityService:
        self, *, app: App, agent_id: str, conversation_id: str, params: AgentLogQueryParams
    ) -> dict[str, Any]:
        source_filters = self.resolve_source_filters(params.sources)
-        rows: list[Message] = []
+        rows: list[dict[str, Any]] = []
        for source_filter in source_filters:
            if source_filter.kind in {"all", "webapp"}:
                rows.extend(
-                    self._list_webapp_messages(
+                    self.serialize_log_message(message)
+                    for message in self._list_webapp_messages(
                        app=app,
                        conversation_id=conversation_id,
                        params=params,
@@ -216,18 +221,18 @@ class AgentObservabilityService:
                    )
                )

-        deduped = {message.id: message for message in rows}
-        sort_column = Message.created_at if params.sort_by == "created_at" else Message.updated_at
+        deduped = {row["id"]: row for row in rows}
+        sort_key = "created_at" if params.sort_by == "created_at" else "updated_at"
        sorted_rows = sorted(
            deduped.values(),
-            key=lambda message: (getattr(message, sort_column.key), message.id),
+            key=lambda row: (row[sort_key] or 0, row["id"]),
            reverse=params.sort_order != "asc",
        )
        total = len(sorted_rows)
        start = (params.page - 1) * params.limit
        end = start + params.limit
        return {
-            "data": [self.serialize_log_message(message) for message in sorted_rows[start:end]],
+            "data": sorted_rows[start:end],
            "page": params.page,
            "limit": params.limit,
            "total": total,
@@ -284,22 +289,20 @@ class AgentObservabilityService:
        workflow_app = aliased(App)
        stmt = (
            select(
-                Conversation,
+                WorkflowNodeExecutionModel.id.label("node_execution_id"),
+                WorkflowNodeExecutionModel.title.label("node_title"),
+                WorkflowNodeExecutionModel.status.label("node_status"),
+                WorkflowNodeExecutionModel.created_by_role.label("node_created_by_role"),
+                WorkflowNodeExecutionModel.created_by.label("node_created_by"),
+                WorkflowNodeExecutionModel.created_at.label("node_created_at"),
+                WorkflowNodeExecutionModel.finished_at.label("node_finished_at"),
                workflow_app,
                WorkflowAgentNodeBinding.workflow_id,
                WorkflowAgentNodeBinding.workflow_version,
                WorkflowAgentNodeBinding.node_id,
-                func.count(sa.distinct(Message.id)).label("message_count"),
-                func.max(Message.created_at).label("created_at"),
-                func.max(Message.updated_at).label("updated_at"),
-                func.sum(sa.case((Message.status == MessageStatus.PAUSED, 1), else_=0)).label("paused_count"),
-                func.sum(
-                    sa.case((or_(Message.error.is_not(None), Message.status == MessageStatus.ERROR), 1), else_=0)
-                ).label("failed_count"),
            )
-            .select_from(Message)
-            .join(Conversation, Conversation.id == Message.conversation_id)
-            .join(WorkflowRun, WorkflowRun.id == Message.workflow_run_id)
+            .select_from(WorkflowNodeExecutionModel)
+            .join(WorkflowRun, WorkflowRun.id == WorkflowNodeExecutionModel.workflow_run_id)
            .join(
                WorkflowAgentNodeBinding,
                and_(
@@ -310,40 +313,32 @@ class AgentObservabilityService:
                    WorkflowAgentNodeBinding.workflow_version == WorkflowRun.version,
                ),
            )
-            .join(
-                WorkflowNodeExecutionModel,
-                and_(
-                    WorkflowNodeExecutionModel.workflow_run_id == WorkflowRun.id,
-                    WorkflowNodeExecutionModel.node_id == WorkflowAgentNodeBinding.node_id,
-                ),
-            )
            .join(workflow_app, workflow_app.id == WorkflowAgentNodeBinding.app_id)
-            .where(Message.workflow_run_id.is_not(None), Conversation.app_id == WorkflowAgentNodeBinding.app_id)
-            .group_by(
-                Conversation.id,
-                workflow_app.id,
-                WorkflowAgentNodeBinding.workflow_id,
-                WorkflowAgentNodeBinding.workflow_version,
-                WorkflowAgentNodeBinding.node_id,
+            .where(
+                WorkflowNodeExecutionModel.tenant_id == app.tenant_id,
+                WorkflowNodeExecutionModel.app_id == WorkflowAgentNodeBinding.app_id,
+                WorkflowNodeExecutionModel.workflow_id == WorkflowAgentNodeBinding.workflow_id,
+                WorkflowNodeExecutionModel.node_id == WorkflowAgentNodeBinding.node_id,
            )
        )
-        stmt = self._apply_observability_filters(stmt, params=params, source_filter=source_filter)
+        stmt = self._apply_workflow_node_filters(stmt, params=params, workflow_app=workflow_app)
        stmt = self._apply_workflow_source_filter(stmt, source_filter)
        rows = list(self._session.execute(stmt).all())
        return [
-            self._serialize_conversation_log(
-                conversation=row[0],
-                message_count=row.message_count,
-                paused_count=row.paused_count,
-                failed_count=row.failed_count,
+            self._serialize_workflow_execution_log(
+                node_execution_id=row.node_execution_id,
+                title=row.node_title,
+                status=row.node_status,
+                created_by_role=row.node_created_by_role,
+                created_by=row.node_created_by,
+                created_at=row.node_created_at,
+                finished_at=row.node_finished_at,
                source=self._serialize_workflow_source(
-                    app=row[1],
+                    app=row[7],
                    workflow_id=row.workflow_id,
                    workflow_version=row.workflow_version,
                    node_id=row.node_id,
                ),
-                created_at=row.created_at,
-                updated_at=row.updated_at,
            )
            for row in rows
        ]
@@ -363,10 +358,12 @@ class AgentObservabilityService:
        conversation_id: str,
        params: AgentLogQueryParams,
        source_filter: AgentSourceFilter,
-    ) -> list[Message]:
+    ) -> list[dict[str, Any]]:
+        workflow_app = aliased(App)
        stmt = (
-            select(Message)
-            .join(WorkflowRun, WorkflowRun.id == Message.workflow_run_id)
+            select(WorkflowNodeExecutionModel)
+            .select_from(WorkflowNodeExecutionModel)
+            .join(WorkflowRun, WorkflowRun.id == WorkflowNodeExecutionModel.workflow_run_id)
            .join(
                WorkflowAgentNodeBinding,
                and_(
@@ -377,18 +374,23 @@ class AgentObservabilityService:
                    WorkflowAgentNodeBinding.workflow_version == WorkflowRun.version,
                ),
            )
-            .join(
-                WorkflowNodeExecutionModel,
-                and_(
-                    WorkflowNodeExecutionModel.workflow_run_id == WorkflowRun.id,
-                    WorkflowNodeExecutionModel.node_id == WorkflowAgentNodeBinding.node_id,
-                ),
+            .join(workflow_app, workflow_app.id == WorkflowAgentNodeBinding.app_id)
+            .where(
+                WorkflowNodeExecutionModel.id == conversation_id,
+                WorkflowNodeExecutionModel.tenant_id == app.tenant_id,
+                WorkflowNodeExecutionModel.app_id == WorkflowAgentNodeBinding.app_id,
+                WorkflowNodeExecutionModel.workflow_id == WorkflowAgentNodeBinding.workflow_id,
+                WorkflowNodeExecutionModel.node_id == WorkflowAgentNodeBinding.node_id,
            )
-            .where(Message.conversation_id == conversation_id)
        )
-        stmt = self._apply_message_filters(stmt, params=params, source_filter=source_filter)
+        stmt = self._apply_workflow_node_filters(stmt, params=params, workflow_app=workflow_app)
        stmt = self._apply_workflow_source_filter(stmt, source_filter)
-        return list(self._session.scalars(stmt.order_by(Message.created_at.desc(), Message.id.desc())).all())
+        executions = list(
+            self._session.scalars(
+                stmt.order_by(WorkflowNodeExecutionModel.created_at.desc(), WorkflowNodeExecutionModel.id.desc())
+            ).all()
+        )
+        return [self.serialize_workflow_node_message(execution) for execution in executions]

    def _list_workflow_sources(self, *, app: App, agent_id: str) -> list[dict[str, Any]]:
        workflow_app = aliased(App)
@@ -443,6 +445,62 @@ class AgentObservabilityService:
            stmt = cls._apply_status_filter(stmt, params.statuses)
        return stmt

+    @classmethod
+    def _apply_workflow_node_filters(cls, stmt, *, params: AgentLogQueryParams, workflow_app):
+        if params.start:
+            stmt = stmt.where(WorkflowNodeExecutionModel.created_at >= params.start)
+        if params.end:
+            stmt = stmt.where(WorkflowNodeExecutionModel.created_at < params.end)
+        if params.keyword:
+            escaped_keyword = escape_like_pattern(params.keyword)
+            pattern = f"%{escaped_keyword}%"
+            stmt = stmt.where(
+                or_(
+                    WorkflowNodeExecutionModel.inputs.ilike(pattern, escape="\\"),
+                    WorkflowNodeExecutionModel.outputs.ilike(pattern, escape="\\"),
+                    WorkflowNodeExecutionModel.error.ilike(pattern, escape="\\"),
+                    WorkflowNodeExecutionModel.title.ilike(pattern, escape="\\"),
+                    workflow_app.name.ilike(pattern, escape="\\"),
+                )
+            )
+        if params.statuses:
+            stmt = cls._apply_workflow_node_status_filter(stmt, params.statuses)
+        return stmt
+
+    @staticmethod
+    def _apply_workflow_node_status_filter(stmt, statuses: tuple[str, ...]):
+        conditions = []
+        for status in statuses:
+            normalized = status.strip().lower()
+            if normalized in {"success", "normal"}:
+                conditions.append(WorkflowNodeExecutionModel.status == WorkflowNodeExecutionStatus.SUCCEEDED)
+            elif normalized in {"failed", "error"}:
+                conditions.append(
+                    WorkflowNodeExecutionModel.status.in_(
+                        (
+                            WorkflowNodeExecutionStatus.FAILED,
+                            WorkflowNodeExecutionStatus.EXCEPTION,
+                            WorkflowNodeExecutionStatus.STOPPED,
+                        )
+                    )
+                )
+            elif normalized == "paused":
+                conditions.append(
+                    WorkflowNodeExecutionModel.status.in_(
+                        (
+                            WorkflowNodeExecutionStatus.PAUSED,
+                            WorkflowNodeExecutionStatus.PENDING,
+                            WorkflowNodeExecutionStatus.RUNNING,
+                            WorkflowNodeExecutionStatus.RETRY,
+                        )
+                    )
+                )
+            else:
+                raise ValueError(f"Unsupported status: {status}")
+        if not conditions:
+            return stmt
+        return stmt.where(or_(*conditions))
+
    @staticmethod
    def _apply_workflow_source_filter(stmt, source_filter: AgentSourceFilter):
        if source_filter.app_id:
@@ -505,6 +563,148 @@ class AgentObservabilityService:
            "updated_at": to_timestamp(updated_at or conversation.updated_at),
        }

+    @classmethod
+    def _serialize_workflow_execution_log(
+        cls,
+        *,
+        node_execution_id: str,
+        title: str,
+        status: object,
+        created_by_role: object,
+        created_by: str,
+        created_at: datetime,
+        finished_at: datetime | None,
+        source: dict[str, Any],
+    ) -> dict[str, Any]:
+        created_by_role_value = cls._enum_value(created_by_role)
+        return {
+            "id": node_execution_id,
+            "conversation_id": node_execution_id,
+            "title": title,
+            "end_user_id": created_by if created_by_role_value == CreatorUserRole.END_USER.value else None,
+            "message_count": 1,
+            "user_rate": None,
+            "operation_rate": None,
+            "unread": False,
+            "source": source,
+            "status": cls._workflow_node_status(status),
+            "created_at": to_timestamp(created_at),
+            "updated_at": to_timestamp(finished_at or created_at),
+        }
+
+    @classmethod
+    def serialize_workflow_node_message(cls, node_execution: WorkflowNodeExecutionModel) -> dict[str, Any]:
+        inputs = cls._json_mapping(node_execution.inputs)
+        outputs = cls._json_mapping(node_execution.outputs)
+        metadata = cls._json_mapping(node_execution.execution_metadata)
+        agent_log = cls._mapping_value(metadata, "agent_log")
+        agent_backend = cls._mapping_value(agent_log, "agent_backend")
+        usage = cls._mapping_value(agent_backend, "usage")
+
+        prompt_tokens = cls._int_value(usage.get("prompt_tokens"))
+        completion_tokens = cls._int_value(usage.get("completion_tokens"))
+        total_tokens = cls._int_value(usage.get("total_tokens") or metadata.get("total_tokens"))
+        if not total_tokens:
+            total_tokens = prompt_tokens + completion_tokens
+        created_by_role = cls._enum_value(node_execution.created_by_role)
+
+        return {
+            "id": node_execution.id,
+            "message_id": node_execution.id,
+            "conversation_id": node_execution.id,
+            "query": cls._workflow_node_query(inputs, fallback=node_execution.title),
+            "answer": cls._workflow_node_answer(outputs),
+            "status": cls._workflow_node_status(node_execution.status),
+            "error": node_execution.error,
+            "from_end_user_id": (
+                node_execution.created_by if created_by_role == CreatorUserRole.END_USER.value else None
+            ),
+            "from_account_id": (
+                node_execution.created_by if created_by_role == CreatorUserRole.ACCOUNT.value else None
+            ),
+            "message_tokens": prompt_tokens,
+            "answer_tokens": completion_tokens,
+            "total_tokens": total_tokens,
+            "total_price": str(usage.get("total_price") or metadata.get("total_price") or Decimal(0)),
+            "currency": str(usage.get("currency") or metadata.get("currency") or ""),
+            "latency": float(usage.get("latency") or node_execution.elapsed_time or 0),
+            "created_at": to_timestamp(node_execution.created_at),
+            "updated_at": to_timestamp(node_execution.finished_at or node_execution.created_at),
+        }
+
+    @staticmethod
+    def _json_mapping(value: object) -> Mapping[str, Any]:
+        if isinstance(value, Mapping):
+            return value
+        if not isinstance(value, str) or not value:
+            return {}
+        try:
+            parsed = json.loads(value)
+        except (TypeError, ValueError):
+            return {}
+        return parsed if isinstance(parsed, Mapping) else {}
+
+    @staticmethod
+    def _mapping_value(value: Mapping[str, Any], key: str) -> Mapping[str, Any]:
+        nested = value.get(key)
+        return nested if isinstance(nested, Mapping) else {}
+
+    @staticmethod
+    def _enum_value(value: object) -> str:
+        return str(value.value) if isinstance(value, Enum) else str(value)
+
+    @staticmethod
+    def _int_value(value: object) -> int:
+        if not isinstance(value, (str, int, float, Decimal)):
+            return 0
+        try:
+            return int(value)
+        except (TypeError, ValueError):
+            return 0
+
+    @classmethod
+    def _workflow_node_query(cls, inputs: Mapping[str, Any], *, fallback: str) -> str:
+        request_data = cls._mapping_value(inputs, "agent_backend_request")
+        composition = cls._mapping_value(request_data, "composition")
+        layers = composition.get("layers")
+        prompts: list[str] = []
+        if isinstance(layers, list):
+            for layer_name in ("workflow_node_job_prompt", "workflow_user_prompt"):
+                for layer in layers:
+                    if not isinstance(layer, Mapping) or layer.get("name") != layer_name:
+                        continue
+                    config = cls._mapping_value(layer, "config")
+                    user_prompt = config.get("user")
+                    if isinstance(user_prompt, str) and user_prompt.strip():
+                        prompts.append(user_prompt.strip())
+        return "\n\n".join(prompts) or fallback
+
+    @staticmethod
+    def _workflow_node_answer(outputs: Mapping[str, Any]) -> str:
+        for key in ("output", "text", "answer"):
+            value = outputs.get(key)
+            if isinstance(value, str):
+                return value
+        return json.dumps(outputs, ensure_ascii=False) if outputs else ""
+
+    @classmethod
+    def _workflow_node_status(cls, status: object) -> str:
+        value = cls._enum_value(status)
+        if value in {
+            WorkflowNodeExecutionStatus.FAILED.value,
+            WorkflowNodeExecutionStatus.EXCEPTION.value,
+            WorkflowNodeExecutionStatus.STOPPED.value,
+        }:
+            return "failed"
+        if value in {
+            WorkflowNodeExecutionStatus.PAUSED.value,
+            WorkflowNodeExecutionStatus.PENDING.value,
+            WorkflowNodeExecutionStatus.RUNNING.value,
+            WorkflowNodeExecutionStatus.RETRY.value,
+        }:
+            return "paused"
+        return "success"
+
    @staticmethod
    def _conversation_status(*, paused_count: int, failed_count: int) -> str:
        if paused_count:
@@ -6,6 +6,7 @@ from sqlalchemy.exc import IntegrityError

 from clients.agent_backend.session_cleanup import AgentBackendSessionCleanupPayload
 from constants.model_template import default_app_templates
+from core.agent.publish_visibility import workflow_callable_active_snapshot_filter
 from core.app.apps.agent_app.session_store import AgentAppRuntimeSessionStore
 from core.app.entities.app_invoke_entities import InvokeFrom
 from libs.datetime_utils import naive_utc_now
@@ -225,8 +226,11 @@ class AgentRosterService:
    def list_invite_options(
        self, *, tenant_id: str, page: int = 1, limit: int = 20, keyword: str | None = None, app_id: str | None = None
    ) -> dict[str, Any]:
+        """List active roster Agents whose published snapshot can be called by Workflow."""
+
        stmt = self._build_roster_agents_stmt(tenant_id=tenant_id, keyword=keyword).where(
-            Agent.active_config_has_model.is_(True)
+            Agent.active_config_has_model.is_(True),
+            workflow_callable_active_snapshot_filter(),
        )
        total = self._session.scalar(select(func.count()).select_from(stmt.subquery())) or 0
        agents = list(self._session.scalars(stmt.offset((page - 1) * limit).limit(limit)).all())
@@ -430,7 +434,11 @@ class AgentRosterService:
        agent.active_config_has_model = agent_soul_has_model(soul)
        agent.active_config_is_published = False
        self._session.flush()
-        self._get_or_create_agent_app_debug_conversation(agent=agent, account_id=account_id)
+        self._get_or_create_agent_app_debug_conversation(
+            agent=agent,
+            account_id=account_id,
+            draft_type=AgentConfigDraftType.DEBUG_BUILD,
+        )
        return agent

    def create_hidden_backing_app_for_workflow_agent(
@@ -527,7 +535,11 @@ class AgentRosterService:
        self._session.flush()
        return backing_app.id

-    def _get_or_create_agent_app_debug_conversation(self, *, agent: Agent, account_id: str) -> str:
+    def _get_or_create_agent_app_debug_conversation(
+        self, *, agent: Agent, account_id: str, draft_type: AgentConfigDraftType
+    ) -> str:
+        """Return the editor's conversation for one Agent draft surface."""
+
        backing_app_id = self._ensure_workflow_agent_backing_app(agent=agent, account_id=account_id)
        if not backing_app_id:
            raise AgentNotFoundError()
@@ -537,6 +549,7 @@ class AgentRosterService:
                AgentDebugConversation.tenant_id == agent.tenant_id,
                AgentDebugConversation.agent_id == agent.id,
                AgentDebugConversation.account_id == account_id,
+                AgentDebugConversation.draft_type == draft_type,
            )
        )
        if mapping is not None:
@@ -570,6 +583,7 @@ class AgentRosterService:
                agent_id=agent.id,
                app_id=backing_app_id,
                account_id=account_id,
+                draft_type=draft_type,
                conversation_id=conversation_id,
            )
        )
@@ -577,9 +591,15 @@ class AgentRosterService:
        return conversation_id

    def get_or_create_agent_app_debug_conversation_id(
-        self, *, tenant_id: str, agent_id: str, account_id: str, commit: bool = True
+        self,
+        *,
+        tenant_id: str,
+        agent_id: str,
+        account_id: str,
+        draft_type: AgentConfigDraftType = AgentConfigDraftType.DEBUG_BUILD,
+        commit: bool = True,
    ) -> str:
-        """Return the current editor's debug conversation for an Agent App."""
+        """Return the current editor's Build or Preview conversation for an Agent App."""

        agent = self._session.scalar(
            select(Agent).where(
@@ -591,13 +611,24 @@ class AgentRosterService:
        if agent is None:
            raise AgentNotFoundError()

-        conversation_id = self._get_or_create_agent_app_debug_conversation(agent=agent, account_id=account_id)
+        conversation_id = self._get_or_create_agent_app_debug_conversation(
+            agent=agent,
+            account_id=account_id,
+            draft_type=draft_type,
+        )
        if commit:
            self._session.commit()
        return conversation_id

-    def load_agent_app_debug_conversation_id(self, *, tenant_id: str, agent_id: str, account_id: str) -> str | None:
-        """Return the current editor's existing debug conversation without creating or repairing rows."""
+    def load_agent_app_debug_conversation_id(
+        self,
+        *,
+        tenant_id: str,
+        agent_id: str,
+        account_id: str,
+        draft_type: AgentConfigDraftType = AgentConfigDraftType.DEBUG_BUILD,
+    ) -> str | None:
+        """Return the editor's existing scoped conversation without creating or repairing rows."""

        return self._session.scalar(
            select(Conversation.id)
@@ -606,6 +637,7 @@ class AgentRosterService:
                AgentDebugConversation.tenant_id == tenant_id,
                AgentDebugConversation.agent_id == agent_id,
                AgentDebugConversation.account_id == account_id,
+                AgentDebugConversation.draft_type == draft_type,
                AgentDebugConversation.app_id == Conversation.app_id,
                Conversation.from_source == ConversationFromSource.CONSOLE,
                Conversation.from_account_id == account_id,
@@ -626,16 +658,24 @@ class AgentRosterService:
        )

    def refresh_agent_app_debug_conversation_id(
-        self, *, tenant_id: str, agent_id: str, account_id: str, commit: bool = True
+        self,
+        *,
+        tenant_id: str,
+        agent_id: str,
+        account_id: str,
+        draft_type: AgentConfigDraftType = AgentConfigDraftType.DEBUG_BUILD,
    ) -> str:
-        """Start a new console debug conversation for the current Agent App editor.
+        """Start a new scoped console conversation for the current Agent App editor.

-        If this account already has a debug conversation mapping, the previous
-        conversation is abandoned first: any ACTIVE conversation-owned Agent
-        runtime sessions for that old conversation are sent through best-effort
-        backend cleanup and then retired locally even when enqueueing fails.
-        The debug mapping is then repointed to the freshly created
-        conversation.
+        If this account already has a mapping for the requested draft surface, the previous
+        conversation is abandoned after the replacement mapping is committed: any ACTIVE
+        conversation-owned Agent runtime sessions for that old conversation are sent through
+        best-effort backend cleanup and then retired locally even when enqueueing fails. This
+        order prevents a failed database commit from retiring the still-current runtime session.
+        The other draft surface is left untouched.
+
+        A user and draft surface own one current mapping. If new-conversation requests overlap,
+        the last committed rotation becomes current and earlier response IDs cannot be continued.
        """

        agent = self._session.scalar(
@@ -658,11 +698,13 @@ class AgentRosterService:
            app_id=backing_app_id,
            account_id=account_id,
        )
+        previous_conversation: tuple[str, str] | None = None
        mapping = self._session.scalar(
            select(AgentDebugConversation).where(
                AgentDebugConversation.tenant_id == tenant_id,
                AgentDebugConversation.agent_id == agent_id,
                AgentDebugConversation.account_id == account_id,
+                AgentDebugConversation.draft_type == draft_type,
            )
        )
        if mapping is None:
@@ -672,6 +714,7 @@ class AgentRosterService:
                    agent_id=agent_id,
                    app_id=backing_app_id,
                    account_id=account_id,
+                    draft_type=draft_type,
                    conversation_id=conversation_id,
                )
            )
@@ -679,18 +722,22 @@ class AgentRosterService:
            previous_app_id = mapping.app_id
            previous_conversation_id = mapping.conversation_id
            if previous_conversation_id:
-                self._cleanup_debug_conversation_runtime_sessions(
-                    tenant_id=tenant_id,
-                    agent_id=agent_id,
-                    account_id=account_id,
-                    app_id=previous_app_id or backing_app_id,
-                    conversation_id=previous_conversation_id,
-                )
+                previous_conversation = (previous_app_id or backing_app_id, previous_conversation_id)
            mapping.app_id = backing_app_id
            mapping.conversation_id = conversation_id
        self._session.flush()
-        if commit:
-            self._session.commit()
+        self._session.commit()
+
+        if previous_conversation:
+            previous_app_id, previous_conversation_id = previous_conversation
+            self._cleanup_debug_conversation_runtime_sessions(
+                tenant_id=tenant_id,
+                agent_id=agent_id,
+                account_id=account_id,
+                draft_type=draft_type,
+                app_id=previous_app_id,
+                conversation_id=previous_conversation_id,
+            )
        return conversation_id

    def _cleanup_debug_conversation_runtime_sessions(
@@ -699,11 +746,12 @@ class AgentRosterService:
        tenant_id: str,
        agent_id: str,
        account_id: str,
+        draft_type: AgentConfigDraftType,
        app_id: str,
        conversation_id: str,
    ) -> None:
-        session_store = AgentAppRuntimeSessionStore()
        try:
+            session_store = AgentAppRuntimeSessionStore()
            stored_sessions = session_store.list_active_sessions_for_conversation(
                tenant_id=tenant_id,
                app_id=app_id,
@@ -727,7 +775,8 @@ class AgentRosterService:
                        session_snapshot=stored_session.session_snapshot,
                        runtime_layer_specs=stored_session.runtime_layer_specs,
                        idempotency_key=(
-                            f"{tenant_id}:{agent_id}:{account_id}:{conversation_id}:debug-session-cleanup:"
+                            f"{tenant_id}:{agent_id}:{account_id}:{draft_type.value}:{conversation_id}:"
+                            "debug-session-cleanup:"
                            f"{stored_session.scope.agent_id}:"
                            f"{stored_session.scope.agent_config_snapshot_id or 'no-config'}:"
                            f"{stored_session.backend_run_id or 'no-run'}"
@@ -738,6 +787,7 @@ class AgentRosterService:
                            "conversation_id": stored_session.scope.conversation_id,
                            "agent_id": stored_session.scope.agent_id,
                            "agent_config_snapshot_id": stored_session.scope.agent_config_snapshot_id,
+                            "draft_type": draft_type.value,
                            "previous_agent_backend_run_id": stored_session.backend_run_id,
                        },
                    )
@@ -772,9 +822,14 @@ class AgentRosterService:
                    )

    def load_or_create_agent_app_debug_conversation_ids_by_agent_id(
-        self, *, tenant_id: str, agents: list[Agent], account_id: str
+        self,
+        *,
+        tenant_id: str,
+        agents: list[Agent],
+        account_id: str,
+        draft_type: AgentConfigDraftType = AgentConfigDraftType.DEBUG_BUILD,
    ) -> dict[str, str]:
-        """Return per-account debug conversations for a page of Agent Apps."""
+        """Return per-account scoped conversations for a page of Agent Apps."""

        conversation_ids_by_agent_id: dict[str, str] = {}
        changed = False
@@ -784,6 +839,7 @@ class AgentRosterService:
            conversation_ids_by_agent_id[agent.id] = self._get_or_create_agent_app_debug_conversation(
                agent=agent,
                account_id=account_id,
+                draft_type=draft_type,
            )
            changed = True
        if changed:
@@ -866,16 +922,14 @@ class AgentRosterService:
            raise AgentNotFoundError()
        return app

-    def get_agent_runtime_app_model(self, *, tenant_id: str, agent_id: str) -> App:
-        """Resolve the App that backs an Agent runtime surface.
+    def _get_runtime_resolvable_agent(self, *, tenant_id: str, agent_id: str) -> Agent | None:
+        """Load an Agent that is eligible to resolve to a runtime backing App.

-        Roster Agents use their public Agent App. Workflow-only Agents use a
-        hidden Agent App stored in ``backing_app_id`` so console chat/logs can
-        reuse the app runtime without exposing the resource in workspace app
-        lists.
+        Shared by the runtime resolver and the read-only authorization resolver
+        so both agree on what counts as a resolvable Agent.
        """

-        agent = self._session.scalar(
+        return self._session.scalar(
            select(Agent)
            .where(
                Agent.tenant_id == tenant_id,
@@ -893,6 +947,34 @@ class AgentRosterService:
            )
            .limit(1)
        )
+
+    def peek_authz_app_id(self, *, tenant_id: str, agent_id: str) -> str | None:
+        """Resolve the App id whose access policy governs an Agent.
+
+        Roster Agents are governed by their own Agent App, while workflow-only
+        Agents are governed by their parent workflow App: the hidden runtime
+        backing App never receives a resource access policy, so it must not be
+        used for authorization. Stays read-only — unlike
+        :meth:`get_agent_runtime_app_model`, this never materializes the hidden
+        backing App. Returns ``None`` when the Agent does not resolve, leaving
+        the caller to decide how to treat it.
+        """
+
+        agent = self._get_runtime_resolvable_agent(tenant_id=tenant_id, agent_id=agent_id)
+        if agent is None:
+            return None
+        return agent.app_id
+
+    def get_agent_runtime_app_model(self, *, tenant_id: str, agent_id: str) -> App:
+        """Resolve the App that backs an Agent runtime surface.
+
+        Roster Agents use their public Agent App. Workflow-only Agents use a
+        hidden Agent App stored in ``backing_app_id`` so console chat/logs can
+        reuse the app runtime without exposing the resource in workspace app
+        lists.
+        """
+
+        agent = self._get_runtime_resolvable_agent(tenant_id=tenant_id, agent_id=agent_id)
        if agent is None:
            raise AgentNotFoundError()
        should_commit_backing_app = agent.scope == AgentScope.WORKFLOW_ONLY and not agent.backing_app_id
@@ -8,6 +8,7 @@ from pydantic import ValidationError
 from sqlalchemy import select
 from sqlalchemy.orm import Session

+from core.agent.publish_visibility import workflow_callable_active_snapshot_filter
 from core.workflow.nodes.agent_v2.validators import WorkflowAgentNodeValidationError, WorkflowAgentNodeValidator
 from models.agent import (
    Agent,
@@ -447,6 +448,8 @@ class WorkflowAgentPublishService:
        node_id: str,
        agent_id: str,
    ) -> tuple[Agent, str]:
+        """Resolve an active roster Agent whose published snapshot is callable."""
+
        agent = session.scalar(
            select(Agent)
            .where(
@@ -454,11 +457,12 @@ class WorkflowAgentPublishService:
                Agent.id == agent_id,
                Agent.scope == AgentScope.ROSTER,
                Agent.status == AgentStatus.ACTIVE,
+                workflow_callable_active_snapshot_filter(),
            )
            .limit(1)
        )
        if agent is None:
-            raise ValueError(f"Workflow Agent node {node_id} references an unavailable roster agent.")
+            raise ValueError(f"Workflow Agent node {node_id} references an unavailable or unpublished roster agent.")
        if agent.scope != AgentScope.ROSTER:
            raise ValueError(f"Workflow Agent node {node_id} roster_agent binding must reference a roster agent.")
        if not agent.active_config_snapshot_id:
@@ -9,6 +9,7 @@ import contexts
 from core.app.app_config.easy_ui_based_app.agent.manager import AgentConfigManager
 from core.plugin.impl.agent import PluginAgentClient
 from core.plugin.impl.exc import PluginDaemonClientSideError
+from core.tools.entities.tool_entities import EmojiIconDict
 from core.tools.tool_manager import ToolManager
 from libs.login import current_user
 from models import Account
@@ -105,11 +106,22 @@ class AgentService:
                tool_output = tool_outputs.get(tool_name, {})
                tool_meta_data = tool_meta.get(tool_name, {})
                tool_config = tool_meta_data.get("tool_config", {})
-                if tool_config.get("tool_provider_type", "") != "dataset-retrieval":
+                tool_provider_type = tool_config.get("tool_provider_type", "")
+                tool_provider_id = tool_config.get("tool_provider", "")
+
+                if not tool_provider_type:
+                    tool_entity = find_agent_tool(tool_name)
+                    if tool_entity:
+                        tool_provider_type = tool_entity.provider_type
+                        tool_provider_id = tool_provider_id or tool_entity.provider_id
+
+                tool_icon: str | EmojiIconDict = ""
+
+                if tool_provider_type and tool_provider_type != "dataset-retrieval":
                    tool_icon = ToolManager.get_tool_icon(
                        tenant_id=app_model.tenant_id,
-                        provider_type=tool_config.get("tool_provider_type", ""),
-                        provider_id=tool_config.get("tool_provider", ""),
+                        provider_type=tool_provider_type,
+                        provider_id=tool_provider_id,
                    )
                    if not tool_icon:
                        tool_entity = find_agent_tool(tool_name)
@@ -19,6 +19,7 @@ from configs import dify_config
 from constants.dsl_version import CURRENT_APP_DSL_VERSION
 from core.file import remote_fetcher
 from core.plugin.entities.plugin import PluginDependency
+from core.rbac import RBACPermission
 from core.trigger.constants import (
    TRIGGER_PLUGIN_NODE_TYPE,
    TRIGGER_SCHEDULE_NODE_TYPE,
@@ -43,7 +44,9 @@ from services.agent.dsl_service import AgentDslService, AgentPackage
 from services.agent.workflow_publish_service import WorkflowAgentPublishService
 from services.dsl_content import DSL_MAX_SIZE, dsl_content_size
 from services.dsl_version import check_version_compatibility
+from services.enterprise.rbac_service import RBACService
 from services.entities.dsl_entities import CheckDependenciesResult, DslImportWarning, ImportMode, ImportStatus
+from services.errors.account import NoPermissionError
 from services.errors.app import WorkflowNotFoundError
 from services.plugin.dependencies_analysis import DependenciesAnalysisService
 from services.workflow_draft_variable_service import WorkflowDraftVariableService
@@ -301,6 +304,9 @@ class AppDslService:
                error=f"Invalid YAML format: {str(e)}",
            )

+        except NoPermissionError:
+            raise
+
        except Exception as e:
            logger.exception("Failed to import app")
            return Import(
@@ -364,6 +370,9 @@ class AppDslService:
                warnings=self._warnings,
            )

+        except NoPermissionError:
+            raise
+
        except Exception as e:
            logger.exception("Error confirming import")
            return Import(
@@ -395,6 +404,21 @@ class AppDslService:
            leaked_dependencies=leaked_dependencies,
        )

+    @staticmethod
+    def _ensure_agent_manage_permission(account: Account) -> None:
+        """Importing an Agent DSL creates a roster Agent, which requires ``agent.manage``."""
+        if not dify_config.RBAC_ENABLED:
+            return
+        if account.current_tenant_id is None:
+            raise ValueError("Current tenant is not set")
+        allowed = RBACService.CheckAccess.check(
+            account.current_tenant_id,
+            account.id,
+            scene=RBACPermission.AGENT_MANAGE,
+        )
+        if not allowed:
+            raise NoPermissionError("Agent management permission is required to import an Agent App")
+
    def _create_or_update_app(
        self,
        *,
@@ -415,6 +439,8 @@ class AppDslService:
        if not app_mode:
            raise ValueError("loss app mode")
        app_mode = AppMode(app_mode)
+        if app_mode == AppMode.AGENT:
+            self._ensure_agent_manage_permission(account)

        # Set icon type
        icon_type_value = icon_type or app_data.get("icon_type")
@@ -217,12 +217,18 @@ class BillingService:
        return _billing_info_adapter.validate_python(billing_info)

    @classmethod
-    def get_vector_space(cls, tenant_id: str) -> _VectorSpaceQuota:
+    def get_vector_space(cls, tenant_id: str, bypass_cache: bool = False) -> _VectorSpaceQuota:
        params = {"tenant_id": tenant_id}
+        if bypass_cache:
+            params["bypass_cache"] = "true"
        return _vector_space_quota_adapter.validate_python(
            cls._send_request("GET", "/subscription/vector-space", params=params)
        )

+    @classmethod
+    def invalidate_vector_space_cache(cls, tenant_id: str) -> None:
+        cls.get_vector_space(tenant_id, bypass_cache=True)
+
    @classmethod
    def get_tenant_feature_plan_usage_info(cls, tenant_id: str):
        """Deprecated: Use get_quota_info instead."""
@@ -1974,7 +1974,10 @@ class DocumentService:
                if data_source_info and "upload_file_id" in data_source_info:
                    file_id = data_source_info["upload_file_id"]
        document_was_deleted.send(
-            document.id, dataset_id=document.dataset_id, doc_form=document.doc_form, file_id=file_id
+            document.id,
+            dataset_id=document.dataset_id,
+            doc_form=document.doc_form,
+            file_id=file_id,
        )

        session.delete(document)
@@ -2013,7 +2016,12 @@ class DocumentService:
        # Dispatch cleanup task after commit to avoid lock contention
        # Task cleans up segments, files, and vector indexes
        if deleted_document_ids and doc_form is not None:
-            batch_clean_document_task.delay(deleted_document_ids, dataset_ref.dataset_id, doc_form, file_ids)
+            batch_clean_document_task.delay(
+                deleted_document_ids,
+                dataset_ref.dataset_id,
+                doc_form,
+                file_ids,
+            )

    @staticmethod
    def rename_document(dataset_id: str, document_id: str, name: str, session: Session) -> Document:
@@ -5,6 +5,7 @@ from typing import Any

 import httpx

+from configs import dify_config
 from core.helper.trace_id_helper import generate_traceparent_header
 from services.errors.enterprise import (
    EnterpriseAPIBadRequestError,
@@ -96,12 +97,14 @@ class BaseRequest:
            logger.debug("Failed to generate traceparent header", exc_info=True)

        with httpx.Client(mounts=mounts) as client:
-            # IMPORTANT:
-            # - In httpx, passing timeout=None disables timeouts (infinite) and overrides the library default.
-            # - To preserve httpx's default timeout behavior for existing call sites, only pass the kwarg when set.
-            request_kwargs: dict[str, Any] = {"json": json, "params": params, "headers": headers}
-            if timeout is not None:
-                request_kwargs["timeout"] = timeout
+            # Callers that pass an explicit timeout keep it; everyone else gets the
+            # configured budget rather than httpx's implicit 5s default.
+            request_kwargs: dict[str, Any] = {
+                "json": json,
+                "params": params,
+                "headers": headers,
+                "timeout": timeout if timeout is not None else dify_config.ENTERPRISE_REQUEST_TIMEOUT,
+            }

            response = client.request(method, url, **request_kwargs)

@@ -206,9 +209,8 @@ class EnterpriseRequest(BaseRequest):
                "json": json,
                "params": params,
                "headers": {"Content-Type": "application/json", cls.secret_key_header: cls.secret_key, **inner_headers},
+                "timeout": timeout if timeout is not None else dify_config.ENTERPRISE_RBAC_REQUEST_TIMEOUT,
            }
-            if timeout is not None:
-                request_kwargs["timeout"] = timeout
            response = client.request(method, url, **request_kwargs)
            if not response.is_success:
                cls._handle_error_response(response)
@@ -330,6 +330,7 @@ _LEGACY_WORKSPACE_OWNER_KEYS: list[str] = [
    "snippets.management",
    "tool.manage",
    "mcp.manage",
+    "agent.manage",
 ]

 _LEGACY_WORKSPACE_ADMIN_KEYS: list[str] = [
@@ -357,6 +358,7 @@ _LEGACY_WORKSPACE_ADMIN_KEYS: list[str] = [
    "snippets.management",
    "tool.manage",
    "mcp.manage",
+    "agent.manage",
 ]

 _LEGACY_WORKSPACE_EDITOR_KEYS: list[str] = [
@@ -371,6 +373,7 @@ _LEGACY_WORKSPACE_EDITOR_KEYS: list[str] = [
    "dataset.external.connect",
    "snippets.create_and_modify",
    "tool.manage",
+    "agent.manage",
 ]

 _LEGACY_WORKSPACE_NORMAL_KEYS: list[str] = [
@@ -93,8 +93,20 @@ class ExternalDatasetService:
                raise ValueError(f"invalid endpoint: {endpoint} must start with http:// or https://")
            else:
                raise ValueError(f"invalid endpoint: {endpoint}")
+        # Send a minimal body shaped like the External Knowledge API retrieval contract so providers
+        # that require a JSON payload (e.g. RAGFlow) accept the validation probe instead of rejecting
+        # a body-less POST. Mirrors the request built in fetch_external_knowledge_retrieval.
+        validation_payload = {
+            "knowledge_id": "",
+            "query": "",
+            "retrieval_setting": {"top_k": 1, "score_threshold": 0.0},
+        }
        try:
-            response = ssrf_proxy.post(endpoint, headers={"Authorization": f"Bearer {api_key}"})
+            response = ssrf_proxy.post(
+                endpoint,
+                headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
+                data=json.dumps(validation_payload),
+            )
        except Exception as e:
            raise ValueError(f"failed to connect to the endpoint: {endpoint}") from e
        if response.status_code == 502:
@@ -5,6 +5,7 @@ from pydantic import BaseModel, ConfigDict, Field
 from configs import dify_config
 from constants.dsl_version import CURRENT_APP_DSL_VERSION
 from enums.cloud_plan import CloudPlan
+from enums.deployment_edition import DeploymentEdition
 from enums.hosted_provider import HostedTrialProvider
 from services.billing_service import BillingInfo, BillingService
 from services.enterprise.enterprise_service import EnterpriseService
@@ -75,8 +76,11 @@ class LicenseStatus(StrEnum):
    LOST = "lost"


-class LicenseModel(FeatureResponseModel):
+class LicenseStatusModel(FeatureResponseModel):
    status: LicenseStatus = LicenseStatus.NONE
+
+
+class LicenseModel(LicenseStatusModel):
    expired_at: str = ""
    workspaces: LicenseLimitationModel = LicenseLimitationModel(enabled=False, size=0, limit=0)
    seats: LicenseLimitationModel = LicenseLimitationModel(enabled=False, size=0, limit=0)
@@ -162,6 +166,7 @@ class PluginManagerModel(FeatureResponseModel):


 class SystemFeatureModel(FeatureResponseModel):
+    deployment_edition: DeploymentEdition
    enable_app_deploy: bool = False
    sso_enforced_for_signin: bool = False
    sso_enforced_for_signin_protocol: str = ""
@@ -174,7 +179,7 @@ class SystemFeatureModel(FeatureResponseModel):
    is_allow_register: bool = False
    is_allow_create_workspace: bool = False
    is_email_setup: bool = False
-    license: LicenseModel = LicenseModel()
+    license: LicenseStatusModel = LicenseStatusModel()
    branding: BrandingModel = BrandingModel()
    webapp_auth: WebAppAuthModel = WebAppAuthModel()
    plugin_installation_permission: PluginInstallationPermissionModel = PluginInstallationPermissionModel()
@@ -186,6 +191,7 @@ class SystemFeatureModel(FeatureResponseModel):
    enable_learn_app: bool = True
    enable_step_by_step_tour: bool = False
    rbac_enabled: bool = False
+    knowledge_fs_enabled: bool = False


 class FeatureService:
@@ -250,8 +256,8 @@ class FeatureService:
        )

    @classmethod
-    def get_system_features(cls, is_authenticated: bool = False) -> SystemFeatureModel:
-        system_features = SystemFeatureModel()
+    def get_system_features(cls) -> SystemFeatureModel:
+        system_features = SystemFeatureModel(deployment_edition=dify_config.DEPLOYMENT_EDITION)
        system_features.rbac_enabled = dify_config.RBAC_ENABLED

        cls._fulfill_system_params_from_env(system_features)
@@ -261,7 +267,7 @@ class FeatureService:
            system_features.webapp_auth.enabled = True
            system_features.enable_change_email = False
            system_features.plugin_manager.enabled = True
-            cls._fulfill_params_from_enterprise(system_features, is_authenticated)
+            cls._fulfill_params_from_enterprise(system_features)

        if dify_config.MARKETPLACE_ENABLED:
            system_features.enable_marketplace = True
@@ -271,6 +277,17 @@ class FeatureService:

        return system_features

+    @classmethod
+    def get_license(cls) -> LicenseModel:
+        """Return full license detail. Enterprise-only; requires an authenticated caller.
+
+        Non-enterprise deployments have no license, so an unconstrained default
+        (unlimited seats/workspaces) is returned.
+        """
+        if not dify_config.ENTERPRISE_ENABLED:
+            return LicenseModel()
+        return cls._build_license(EnterpriseService.get_info())
+
    @classmethod
    def get_app_dsl_version(cls) -> str:
        return CURRENT_APP_DSL_VERSION
@@ -284,11 +301,13 @@ class FeatureService:
        system_features.is_allow_register = dify_config.ALLOW_REGISTER
        system_features.is_allow_create_workspace = dify_config.ALLOW_CREATE_WORKSPACE
        system_features.is_email_setup = dify_config.MAIL_TYPE is not None and dify_config.MAIL_TYPE != ""
+        system_features.enable_change_email = dify_config.ENABLE_CHANGE_EMAIL
        system_features.enable_trial_app = dify_config.ENABLE_TRIAL_APP
        system_features.enable_explore_banner = dify_config.ENABLE_EXPLORE_BANNER
        system_features.enable_learn_app = dify_config.ENABLE_LEARN_APP
        system_features.webapp_auth.allow_public_access = dify_config.WEBAPP_PUBLIC_ACCESS_ENABLED
        system_features.enable_step_by_step_tour = dify_config.ENABLE_STEP_BY_STEP_TOUR
+        system_features.knowledge_fs_enabled = dify_config.KNOWLEDGE_FS_ENABLED

    @classmethod
    def _fulfill_trial_models_from_env(cls) -> list[str]:
@@ -408,7 +427,27 @@ class FeatureService:
        vector_space.limit = billing_info["vector_space"]["limit"]

    @classmethod
-    def _fulfill_params_from_enterprise(cls, features: SystemFeatureModel, is_authenticated: bool = False):
+    def _build_license(cls, enterprise_info: dict) -> LicenseModel:
+        license_model = LicenseModel()
+        if license_info := enterprise_info.get("License"):
+            license_model.status = LicenseStatus(license_info.get("status", LicenseStatus.INACTIVE))
+            license_model.expired_at = license_info.get("expiredAt", "")
+            if workspaces_info := license_info.get("workspaces"):
+                license_model.workspaces = LicenseLimitationModel(
+                    enabled=workspaces_info.get("enabled", False),
+                    limit=workspaces_info.get("limit", 0),
+                    size=workspaces_info.get("used", 0),
+                )
+            if seats_info := license_info.get("licensedSeats"):
+                license_model.seats = LicenseLimitationModel(
+                    enabled=seats_info.get("enabled", False),
+                    limit=seats_info.get("limit", 0),
+                    size=seats_info.get("used", 0),
+                )
+        return license_model
+
+    @classmethod
+    def _fulfill_params_from_enterprise(cls, features: SystemFeatureModel):
        enterprise_info = EnterpriseService.get_info()

        if "SSOEnforcedForSignin" in enterprise_info:
@@ -448,24 +487,14 @@ class FeatureService:
            )
            features.webapp_auth.sso_config.protocol = enterprise_info.get("SSOEnforcedForWebProtocol", "")

-        # SECURITY NOTE: Only license *status* is exposed to unauthenticated callers
-        # so the login page can detect an expired/inactive license after force-logout.
-        # All other license details (expiry date, workspace usage) remain auth-gated.
-        # This behavior reflects prior internal review of information-leakage risks.
+        # SECURITY NOTE: system-features is unauthenticated, so it exposes only license
+        # *status* — enough for the login page to detect an expired/inactive license after
+        # force-logout. Full license detail (expiry, workspace/seat usage) is served
+        # separately by get_license() behind an authenticated endpoint.
        if license_info := enterprise_info.get("License"):
-            features.license.status = LicenseStatus(license_info.get("status", LicenseStatus.INACTIVE))
-
-            if is_authenticated:
-                features.license.expired_at = license_info.get("expiredAt", "")
-                if workspaces_info := license_info.get("workspaces"):
-                    features.license.workspaces.enabled = workspaces_info.get("enabled", False)
-                    features.license.workspaces.limit = workspaces_info.get("limit", 0)
-                    features.license.workspaces.size = workspaces_info.get("used", 0)
-
-                if seats_info := license_info.get("licensedSeats"):
-                    features.license.seats.enabled = seats_info.get("enabled", False)
-                    features.license.seats.limit = seats_info.get("limit", 0)
-                    features.license.seats.size = seats_info.get("used", 0)
+            features.license = LicenseStatusModel(
+                status=LicenseStatus(license_info.get("status", LicenseStatus.INACTIVE))
+            )

        if "PluginInstallationPermission" in enterprise_info:
            plugin_installation_info = enterprise_info["PluginInstallationPermission"]
@@ -141,6 +141,29 @@ class FileService:
        blob = storage.load_once(upload_file_key)
        return base64.b64encode(blob).decode()

+    def get_file_presigned_url(self, *, file_id: str, tenant_id: str) -> str:
+        """Generate a direct storage URL for a tenant-owned upload file."""
+        with self._session_maker(expire_on_commit=False) as session:
+            upload_file = session.scalar(
+                select(UploadFile)
+                .where(
+                    UploadFile.id == file_id,
+                    UploadFile.tenant_id == tenant_id,
+                )
+                .limit(1)
+            )
+            if upload_file is None:
+                raise NotFound("File not found")
+
+            file_key = upload_file.key
+            content_type = upload_file.mime_type
+
+        return storage.generate_presigned_url(
+            file_key,
+            expires_in=dify_config.FILES_ACCESS_TIMEOUT,
+            content_type=content_type,
+        )
+
    def upload_text(self, text: str, text_name: str, user_id: str, tenant_id: str) -> UploadFile:
        if len(text_name) > 200:
            text_name = text_name[:200]
@@ -0,0 +1,500 @@
+"""Product-facing KnowledgeFS operation and authorization declarations.
+
+This registry is Dify's explicit Console surface. Transport concerns live in
+knowledge_fs_proxy so contract review does not require reading proxy mechanics.
+"""
+
+from __future__ import annotations
+
+from typing import Final, Literal, NamedTuple
+
+from core.rbac import RBACPermission
+
+type KnowledgeFSMethod = Literal["DELETE", "GET", "PATCH", "POST", "PUT"]
+type KnowledgeFSResponseKind = Literal["binary", "buffered", "stream"]
+type KnowledgeFSRequiredScope = Literal["knowledge-spaces:read", "knowledge-spaces:write"]
+type KnowledgeFSLegacyRole = Literal["reader", "dataset_editor", "admin"]
+type KnowledgeFSErrorStatusMap = tuple[tuple[int, int], ...]
+
+
+class KnowledgeFSOperation(NamedTuple):
+    operation_id: str
+    method: KnowledgeFSMethod
+    path: str
+    response_kind: KnowledgeFSResponseKind
+    required_scope: KnowledgeFSRequiredScope
+    rbac_permission: RBACPermission
+    legacy_role: KnowledgeFSLegacyRole
+    max_response_bytes: int
+    request_headers: tuple[str, ...]
+    response_headers: tuple[str, ...]
+    response_media_types: tuple[str, ...]
+    error_status_map: KnowledgeFSErrorStatusMap
+
+
+def _console_operation(
+    operation_id: str,
+    method: KnowledgeFSMethod,
+    path: str,
+    *,
+    rbac_permission: RBACPermission,
+    legacy_role: KnowledgeFSLegacyRole,
+    max_response_bytes: int = 1_048_576,
+    request_headers: tuple[str, ...] = ("x-trace-id",),
+    response_kind: KnowledgeFSResponseKind = "buffered",
+    response_media_types: tuple[str, ...] = ("application/json",),
+    error_status_map: KnowledgeFSErrorStatusMap = ((401, 502), (403, 403)),
+) -> KnowledgeFSOperation:
+    """Declare one contract-pinned operation with an explicit Dify authorization policy."""
+    is_read = method == "GET"
+    return KnowledgeFSOperation(
+        operation_id=operation_id,
+        method=method,
+        path=path,
+        response_kind=response_kind,
+        required_scope="knowledge-spaces:read" if is_read else "knowledge-spaces:write",
+        rbac_permission=rbac_permission,
+        legacy_role=legacy_role,
+        max_response_bytes=max_response_bytes,
+        request_headers=request_headers,
+        response_headers=("x-trace-id",),
+        response_media_types=response_media_types,
+        error_status_map=error_status_map,
+    )
+
+
+def _dataset_read_operation(operation_id: str, path: str) -> KnowledgeFSOperation:
+    """Declare a dataset-readable buffered JSON operation."""
+    return _console_operation(
+        operation_id,
+        "GET",
+        path,
+        rbac_permission=RBACPermission.DATASET_READONLY,
+        legacy_role="reader",
+    )
+
+
+def _dataset_edit_operation(
+    operation_id: str,
+    method: KnowledgeFSMethod,
+    path: str,
+    *,
+    request_headers: tuple[str, ...] = ("x-trace-id",),
+) -> KnowledgeFSOperation:
+    """Declare a dataset-editable buffered JSON operation."""
+    return _console_operation(
+        operation_id,
+        method,
+        path,
+        rbac_permission=RBACPermission.DATASET_EDIT,
+        legacy_role="dataset_editor",
+        request_headers=request_headers,
+    )
+
+
+def _external_source_operation(
+    operation_id: str,
+    method: KnowledgeFSMethod,
+    path: str,
+    *,
+    request_headers: tuple[str, ...] = ("x-trace-id",),
+) -> KnowledgeFSOperation:
+    """Declare an external-source operation restricted to dataset editors."""
+    return _console_operation(
+        operation_id,
+        method,
+        path,
+        rbac_permission=RBACPermission.DATASET_EXTERNAL_CONNECT,
+        legacy_role="dataset_editor",
+        request_headers=request_headers,
+    )
+
+
+KNOWLEDGE_FS_CONSOLE_OPERATIONS: Final[tuple[KnowledgeFSOperation, ...]] = (
+    _console_operation(
+        operation_id="listKnowledgeSpaces",
+        method="GET",
+        path="knowledge-spaces",
+        rbac_permission=RBACPermission.DATASET_READONLY,
+        legacy_role="reader",
+    ),
+    _console_operation(
+        operation_id="createKnowledgeSpace",
+        method="POST",
+        path="knowledge-spaces",
+        rbac_permission=RBACPermission.DATASET_CREATE_AND_MANAGEMENT,
+        legacy_role="dataset_editor",
+    ),
+    _console_operation(
+        operation_id="getKnowledgeSpacesById",
+        method="GET",
+        path="knowledge-spaces/{id}",
+        rbac_permission=RBACPermission.DATASET_READONLY,
+        legacy_role="reader",
+    ),
+    _dataset_edit_operation("patchKnowledgeSpacesById", "PATCH", "knowledge-spaces/{id}"),
+    _dataset_edit_operation(
+        "deleteKnowledgeSpacesById",
+        "DELETE",
+        "knowledge-spaces/{id}",
+        request_headers=("idempotency-key", "x-trace-id"),
+    ),
+    _dataset_read_operation("getKnowledgeSpacesByIdStats", "knowledge-spaces/{id}/stats"),
+    _console_operation(
+        operation_id="getKnowledgeSpacesByIdAccessPolicy",
+        method="GET",
+        path="knowledge-spaces/{id}/access-policy",
+        rbac_permission=RBACPermission.DATASET_READONLY,
+        legacy_role="reader",
+    ),
+    _console_operation(
+        operation_id="patchKnowledgeSpacesByIdAccessPolicy",
+        method="PATCH",
+        path="knowledge-spaces/{id}/access-policy",
+        rbac_permission=RBACPermission.DATASET_ACCESS_CONFIG,
+        legacy_role="admin",
+    ),
+    _console_operation(
+        operation_id="getSourceProviders",
+        method="GET",
+        path="source-providers",
+        rbac_permission=RBACPermission.DATASET_EXTERNAL_CONNECT,
+        legacy_role="dataset_editor",
+    ),
+    _console_operation(
+        operation_id="getKnowledgeSpacesByIdSourceConnections",
+        method="GET",
+        path="knowledge-spaces/{id}/source-connections",
+        rbac_permission=RBACPermission.DATASET_EXTERNAL_CONNECT,
+        legacy_role="dataset_editor",
+    ),
+    _console_operation(
+        operation_id="postKnowledgeSpacesByIdSourceConnections",
+        method="POST",
+        path="knowledge-spaces/{id}/source-connections",
+        rbac_permission=RBACPermission.DATASET_EXTERNAL_CONNECT,
+        legacy_role="dataset_editor",
+    ),
+    _external_source_operation(
+        "postKnowledgeSpacesByIdSourceConnectionsOauth",
+        "POST",
+        "knowledge-spaces/{id}/source-connections/oauth",
+    ),
+    _external_source_operation("postSourceOauthCallback", "POST", "source-oauth/callback"),
+    _external_source_operation(
+        "getKnowledgeSpacesByIdSourceConnectionsByConnectionId",
+        "GET",
+        "knowledge-spaces/{id}/source-connections/{connectionId}",
+    ),
+    _external_source_operation(
+        "deleteKnowledgeSpacesByIdSourceConnectionsByConnectionId",
+        "DELETE",
+        "knowledge-spaces/{id}/source-connections/{connectionId}",
+    ),
+    _console_operation(
+        operation_id="postKnowledgeSpacesByIdSourceConnectionsByConnectionIdRefresh",
+        method="POST",
+        path="knowledge-spaces/{id}/source-connections/{connectionId}/refresh",
+        rbac_permission=RBACPermission.DATASET_EXTERNAL_CONNECT,
+        legacy_role="dataset_editor",
+    ),
+    _console_operation(
+        operation_id="getKnowledgeSpacesByIdSources",
+        method="GET",
+        path="knowledge-spaces/{id}/sources",
+        rbac_permission=RBACPermission.DATASET_READONLY,
+        legacy_role="reader",
+    ),
+    _console_operation(
+        operation_id="postKnowledgeSpacesByIdSources",
+        method="POST",
+        path="knowledge-spaces/{id}/sources",
+        rbac_permission=RBACPermission.DATASET_EXTERNAL_CONNECT,
+        legacy_role="dataset_editor",
+    ),
+    _external_source_operation(
+        "getKnowledgeSpacesByIdSourcesBySourceId",
+        "GET",
+        "knowledge-spaces/{id}/sources/{sourceId}",
+    ),
+    _external_source_operation(
+        "patchKnowledgeSpacesByIdSourcesBySourceId",
+        "PATCH",
+        "knowledge-spaces/{id}/sources/{sourceId}",
+    ),
+    _external_source_operation(
+        "deleteKnowledgeSpacesByIdSourcesBySourceId",
+        "DELETE",
+        "knowledge-spaces/{id}/sources/{sourceId}",
+        request_headers=("idempotency-key", "x-trace-id"),
+    ),
+    _external_source_operation(
+        "putKnowledgeSpacesByIdSourcesBySourceIdCredentials",
+        "PUT",
+        "knowledge-spaces/{id}/sources/{sourceId}/credentials",
+    ),
+    _external_source_operation(
+        "deleteKnowledgeSpacesByIdSourcesBySourceIdCredentials",
+        "DELETE",
+        "knowledge-spaces/{id}/sources/{sourceId}/credentials",
+    ),
+    _external_source_operation(
+        "postKnowledgeSpacesByIdSourcesBySourceIdSync",
+        "POST",
+        "knowledge-spaces/{id}/sources/{sourceId}/sync",
+        request_headers=("idempotency-key", "x-trace-id"),
+    ),
+    _console_operation(
+        operation_id="postKnowledgeSpacesByIdSourcesBySourceIdCrawlPreview",
+        method="POST",
+        path="knowledge-spaces/{id}/sources/{sourceId}/crawl-preview",
+        rbac_permission=RBACPermission.DATASET_EXTERNAL_CONNECT,
+        legacy_role="dataset_editor",
+        request_headers=("idempotency-key", "x-trace-id"),
+    ),
+    _external_source_operation(
+        "postKnowledgeSpacesByIdSourcesBySourceIdWorkflowImports",
+        "POST",
+        "knowledge-spaces/{id}/sources/{sourceId}/workflow-imports",
+        request_headers=("idempotency-key", "x-trace-id"),
+    ),
+    _external_source_operation(
+        "getKnowledgeSpacesByIdSourcesBySourceIdPages",
+        "GET",
+        "knowledge-spaces/{id}/sources/{sourceId}/pages",
+    ),
+    _external_source_operation(
+        "getKnowledgeSpacesByIdSourcesBySourceIdFiles",
+        "GET",
+        "knowledge-spaces/{id}/sources/{sourceId}/files",
+    ),
+    _external_source_operation(
+        "postKnowledgeSpacesByIdSourcesBySourceIdCrawl",
+        "POST",
+        "knowledge-spaces/{id}/sources/{sourceId}/crawl",
+    ),
+    _external_source_operation(
+        "postKnowledgeSpacesByIdSourcesBySourceIdImport",
+        "POST",
+        "knowledge-spaces/{id}/sources/{sourceId}/import",
+    ),
+    _external_source_operation(
+        "postKnowledgeSpacesByIdSourcesBySourceIdTest",
+        "POST",
+        "knowledge-spaces/{id}/sources/{sourceId}/test",
+    ),
+    _external_source_operation(
+        "postKnowledgeSpacesByIdSourcesBySourceIdImportFiles",
+        "POST",
+        "knowledge-spaces/{id}/sources/{sourceId}/import-files",
+    ),
+    _external_source_operation(
+        "postKnowledgeSpacesByIdSourcesBulk",
+        "POST",
+        "knowledge-spaces/{id}/sources/bulk",
+        request_headers=("idempotency-key", "x-trace-id"),
+    ),
+    _external_source_operation(
+        "getKnowledgeSpacesByIdSourceWorkflows",
+        "GET",
+        "knowledge-spaces/{id}/source-workflows",
+    ),
+    _console_operation(
+        operation_id="getKnowledgeSpacesByIdSourceWorkflowsByRunId",
+        method="GET",
+        path="knowledge-spaces/{id}/source-workflows/{runId}",
+        rbac_permission=RBACPermission.DATASET_EXTERNAL_CONNECT,
+        legacy_role="dataset_editor",
+    ),
+    _external_source_operation(
+        "getKnowledgeSpacesByIdSourceWorkflowsByRunIdBulkItems",
+        "GET",
+        "knowledge-spaces/{id}/source-workflows/{runId}/bulk-items",
+    ),
+    _console_operation(
+        operation_id="getKnowledgeSpacesByIdSourceWorkflowsByRunIdPages",
+        method="GET",
+        path="knowledge-spaces/{id}/source-workflows/{runId}/pages",
+        rbac_permission=RBACPermission.DATASET_EXTERNAL_CONNECT,
+        legacy_role="dataset_editor",
+    ),
+    _console_operation(
+        operation_id="postKnowledgeSpacesByIdSourceWorkflowsByRunIdCancel",
+        method="POST",
+        path="knowledge-spaces/{id}/source-workflows/{runId}/cancel",
+        rbac_permission=RBACPermission.DATASET_EXTERNAL_CONNECT,
+        legacy_role="dataset_editor",
+    ),
+    _console_operation(
+        operation_id="postKnowledgeSpacesByIdSourceWorkflowsByRunIdRetry",
+        method="POST",
+        path="knowledge-spaces/{id}/source-workflows/{runId}/retry",
+        rbac_permission=RBACPermission.DATASET_EXTERNAL_CONNECT,
+        legacy_role="dataset_editor",
+    ),
+    _console_operation(
+        operation_id="postKnowledgeSpacesByIdSourceWorkflowsByRunIdSelection",
+        method="POST",
+        path="knowledge-spaces/{id}/source-workflows/{runId}/selection",
+        rbac_permission=RBACPermission.DATASET_EXTERNAL_CONNECT,
+        legacy_role="dataset_editor",
+        request_headers=("idempotency-key", "x-trace-id"),
+    ),
+    _console_operation(
+        operation_id="getKnowledgeSpacesByIdSourcesBySourceIdSyncPolicy",
+        method="GET",
+        path="knowledge-spaces/{id}/sources/{sourceId}/sync-policy",
+        rbac_permission=RBACPermission.DATASET_READONLY,
+        legacy_role="reader",
+    ),
+    _external_source_operation(
+        "putKnowledgeSpacesByIdSourcesBySourceIdSyncPolicy",
+        "PUT",
+        "knowledge-spaces/{id}/sources/{sourceId}/sync-policy",
+    ),
+    _dataset_read_operation("getKnowledgeSpacesByIdDocuments", "knowledge-spaces/{id}/documents"),
+    _dataset_edit_operation("postKnowledgeSpacesByIdDocuments", "POST", "knowledge-spaces/{id}/documents"),
+    _dataset_edit_operation(
+        "deleteKnowledgeSpacesByIdDocumentsBulk",
+        "DELETE",
+        "knowledge-spaces/{id}/documents/bulk",
+        request_headers=("idempotency-key", "x-trace-id"),
+    ),
+    _dataset_edit_operation(
+        "postKnowledgeSpacesByIdDocumentsBulk",
+        "POST",
+        "knowledge-spaces/{id}/documents/bulk",
+    ),
+    _dataset_edit_operation(
+        "postKnowledgeSpacesByIdDocumentsBulkReindex",
+        "POST",
+        "knowledge-spaces/{id}/documents/bulk/reindex",
+    ),
+    _dataset_read_operation(
+        "getKnowledgeSpacesByIdDocumentsByDocumentId",
+        "knowledge-spaces/{id}/documents/{documentId}",
+    ),
+    _dataset_edit_operation(
+        "deleteKnowledgeSpacesByIdDocumentsByDocumentId",
+        "DELETE",
+        "knowledge-spaces/{id}/documents/{documentId}",
+        request_headers=("idempotency-key", "x-trace-id"),
+    ),
+    _console_operation(
+        operation_id="getKnowledgeSpacesByIdLogicalDocuments",
+        method="GET",
+        path="knowledge-spaces/{id}/logical-documents",
+        rbac_permission=RBACPermission.DATASET_READONLY,
+        legacy_role="reader",
+    ),
+    _dataset_edit_operation(
+        "deleteKnowledgeSpacesByIdLogicalDocumentsByDocumentId",
+        "DELETE",
+        "knowledge-spaces/{id}/logical-documents/{documentId}",
+        request_headers=("idempotency-key", "x-trace-id"),
+    ),
+    _dataset_read_operation(
+        "getKnowledgeSpacesByIdDocumentsByDocumentIdOutline",
+        "knowledge-spaces/{id}/documents/{documentId}/outline",
+    ),
+    _console_operation(
+        operation_id="getKnowledgeSpacesByIdLogicalDocumentsByDocumentId",
+        method="GET",
+        path="knowledge-spaces/{id}/logical-documents/{documentId}",
+        rbac_permission=RBACPermission.DATASET_READONLY,
+        legacy_role="reader",
+    ),
+    _console_operation(
+        operation_id="getKnowledgeSpacesByIdDocumentsByDocumentIdRevisions",
+        method="GET",
+        path="knowledge-spaces/{id}/documents/{documentId}/revisions",
+        rbac_permission=RBACPermission.DATASET_READONLY,
+        legacy_role="reader",
+    ),
+    _dataset_edit_operation(
+        "postKnowledgeSpacesByIdDocumentsByDocumentIdRevisionsByRevisionRollback",
+        "POST",
+        "knowledge-spaces/{id}/documents/{documentId}/revisions/{revision}/rollback",
+    ),
+    _dataset_edit_operation(
+        "patchKnowledgeSpacesByIdDocumentsByDocumentIdMetadata",
+        "PATCH",
+        "knowledge-spaces/{id}/documents/{documentId}/metadata",
+    ),
+    _console_operation(
+        operation_id="getKnowledgeSpacesByIdDocumentsByDocumentIdRevisionsByRevisionChunks",
+        method="GET",
+        path="knowledge-spaces/{id}/documents/{documentId}/revisions/{revision}/chunks",
+        rbac_permission=RBACPermission.DATASET_READONLY,
+        legacy_role="reader",
+    ),
+    _dataset_read_operation(
+        "getKnowledgeSpacesByIdDocumentsByDocumentIdRevisionsByRevisionChunksByChunkId",
+        "knowledge-spaces/{id}/documents/{documentId}/revisions/{revision}/chunks/{chunkId}",
+    ),
+    _dataset_edit_operation(
+        "postKnowledgeSpacesByIdDocumentsByDocumentIdRevisionsByRevisionChunksByChunkIdState",
+        "POST",
+        "knowledge-spaces/{id}/documents/{documentId}/revisions/{revision}/chunks/{chunkId}/state",
+    ),
+    _console_operation(
+        operation_id="getKnowledgeSpacesByIdProcessingTasks",
+        method="GET",
+        path="knowledge-spaces/{id}/processing-tasks",
+        rbac_permission=RBACPermission.DATASET_READONLY,
+        legacy_role="reader",
+    ),
+    _dataset_read_operation(
+        "getKnowledgeSpacesByIdDocumentsByDocumentIdProcessingTasks",
+        "knowledge-spaces/{id}/documents/{documentId}/processing-tasks",
+    ),
+    _dataset_read_operation(
+        "getKnowledgeSpacesByIdDocumentsByDocumentIdProcessingTasksByTaskId",
+        "knowledge-spaces/{id}/documents/{documentId}/processing-tasks/{taskId}",
+    ),
+    _console_operation(
+        operation_id="getKnowledgeSpacesByIdDocumentsByDocumentIdProcessingTasksByTaskIdEvents",
+        method="GET",
+        path="knowledge-spaces/{id}/documents/{documentId}/processing-tasks/{taskId}/events",
+        rbac_permission=RBACPermission.DATASET_READONLY,
+        legacy_role="reader",
+        max_response_bytes=67_108_864,
+        request_headers=("last-event-id", "x-trace-id"),
+        response_kind="stream",
+        response_media_types=("text/event-stream",),
+    ),
+    _console_operation(
+        operation_id="deleteKnowledgeSpacesByIdDocumentsByDocumentIdProcessingTasksByTaskId",
+        method="DELETE",
+        path="knowledge-spaces/{id}/documents/{documentId}/processing-tasks/{taskId}",
+        rbac_permission=RBACPermission.DATASET_EDIT,
+        legacy_role="dataset_editor",
+    ),
+    _console_operation(
+        operation_id="postKnowledgeSpacesByIdDocumentsByDocumentIdProcessingTasksByTaskIdRetry",
+        method="POST",
+        path="knowledge-spaces/{id}/documents/{documentId}/processing-tasks/{taskId}/retry",
+        rbac_permission=RBACPermission.DATASET_EDIT,
+        legacy_role="dataset_editor",
+    ),
+    _dataset_read_operation(
+        "getKnowledgeSpacesByIdDocumentsByDocumentIdSettings",
+        "knowledge-spaces/{id}/documents/{documentId}/settings",
+    ),
+    _dataset_edit_operation(
+        "putKnowledgeSpacesByIdDocumentsByDocumentIdSettings",
+        "PUT",
+        "knowledge-spaces/{id}/documents/{documentId}/settings",
+    ),
+    _dataset_read_operation("getJobsById", "jobs/{id}"),
+    _dataset_edit_operation("deleteJobsById", "DELETE", "jobs/{id}"),
+    _dataset_edit_operation("postJobsByIdRetry", "POST", "jobs/{id}/retry"),
+    _dataset_read_operation("getDeletionJobsByJobId", "deletion-jobs/{jobId}"),
+    _dataset_edit_operation(
+        "postDeletionJobsByJobIdRetry",
+        "POST",
+        "deletion-jobs/{jobId}/retry",
+        request_headers=("idempotency-key", "x-trace-id"),
+    ),
+    _dataset_read_operation("getBulkJobsById", "bulk-jobs/{id}"),
+)
@@ -1,33 +1,32 @@
-"""Transport-only forwarding for the explicitly enabled KnowledgeFS Console operations.
+"""Authorize and forward the explicitly enabled KnowledgeFS Console operations.

-KnowledgeFS owns the request and response contract. This module binds short-lived
-account and workspace identities, enforces Dify's coarse workspace policy, and
-normalizes transport failures. Dify deliberately maintains a small product-facing
-operation registry instead of exposing the full upstream OpenAPI surface. The
-dedicated request path uses Dify's shared SSRF policy, never follows redirects,
-bounds buffered responses, and rejects compressed responses.
+The dedicated request path uses Dify's shared SSRF policy, never follows redirects,
+bounds buffered responses, and rejects compressed streaming responses.
 """

 from __future__ import annotations

 from collections.abc import Iterable, Mapping
+from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
 from http import HTTPStatus
-from typing import Final, Literal, NamedTuple, Protocol
+from typing import NamedTuple, Protocol

 import httpx
 import jwt

 from configs import dify_config
 from core.helper import ssrf_proxy
-from core.rbac import RBACPermission, RBACResourceScope
+from core.rbac import RBACResourceScope
 from core.tools.errors import ToolSSRFError
 from models import Account
 from services.enterprise.rbac_service import RBACService
-
-type KnowledgeFSMethod = Literal["DELETE", "GET", "PATCH", "POST", "PUT"]
-type KnowledgeFSResponseKind = Literal["binary", "buffered", "stream"]
-type KnowledgeFSRequiredScope = Literal["knowledge-spaces:read", "knowledge-spaces:write"]
+from services.knowledge_fs_operations import (
+    KNOWLEDGE_FS_CONSOLE_OPERATIONS,
+    KnowledgeFSMethod,
+    KnowledgeFSOperation,
+    KnowledgeFSResponseKind,
+)

 _JWT_AUDIENCE = "knowledge-fs"
 _JWT_ISSUER = "dify"
@@ -35,56 +34,55 @@ _JWT_TTL_SECONDS = 60
 _MAX_BUFFERED_RESPONSE_BYTES = 1024 * 1024


-class KnowledgeFSOperation(NamedTuple):
-    operation_id: str
-    method: KnowledgeFSMethod
-    path: str
-    response_kind: KnowledgeFSResponseKind
-    required_scope: KnowledgeFSRequiredScope
-    rbac_permission: RBACPermission
-    requires_dataset_editor: bool
-    max_response_bytes: int
-    request_headers: tuple[str, ...]
-    response_headers: tuple[str, ...]
-    response_media_types: tuple[str, ...]
-
-
-KNOWLEDGE_FS_CONSOLE_OPERATIONS: Final[tuple[KnowledgeFSOperation, ...]] = (
-    KnowledgeFSOperation(
-        operation_id="listKnowledgeSpaces",
-        method="GET",
-        path="knowledge-spaces",
-        response_kind="buffered",
-        required_scope="knowledge-spaces:read",
-        rbac_permission=RBACPermission.DATASET_READONLY,
-        requires_dataset_editor=False,
-        max_response_bytes=1_048_576,
-        request_headers=("x-trace-id",),
-        response_headers=("x-trace-id",),
-        response_media_types=("application/json",),
-    ),
-    KnowledgeFSOperation(
-        operation_id="createKnowledgeSpace",
-        method="POST",
-        path="knowledge-spaces",
-        response_kind="buffered",
-        required_scope="knowledge-spaces:write",
-        rbac_permission=RBACPermission.DATASET_CREATE_AND_MANAGEMENT,
-        requires_dataset_editor=True,
-        max_response_bytes=1_048_576,
-        request_headers=("x-trace-id",),
-        response_headers=("x-trace-id",),
-        response_media_types=("application/json",),
-    ),
-)
-
-
 class KnowledgeFSUpstreamResponse(NamedTuple):
    response: httpx.Response
    response_kind: KnowledgeFSResponseKind
    operation: KnowledgeFSOperation


+_AUTHORIZATION_MARKER = object()
+
+
+@dataclass(eq=False, frozen=True, init=False, slots=True)
+class KnowledgeFSAuthorization:
+    """Single-use forwarding capability created after Dify workspace policy checks.
+
+    Callers obtain this value from :func:`authorize_knowledge_fs_request`. Direct
+    construction and repeated forwarding are rejected before outbound I/O.
+    """
+
+    account_id: str
+    tenant_id: str
+    operation: KnowledgeFSOperation
+    _used: bool
+
+    def __init__(
+        self,
+        account_id: str,
+        tenant_id: str,
+        operation: KnowledgeFSOperation,
+        *,
+        _marker: object | None = None,
+    ) -> None:
+        if _marker is not _AUTHORIZATION_MARKER:
+            raise KnowledgeFSAccessDeniedError("KnowledgeFS authorization must be created by workspace authorization")
+        object.__setattr__(self, "account_id", account_id)
+        object.__setattr__(self, "tenant_id", tenant_id)
+        object.__setattr__(self, "operation", operation)
+        object.__setattr__(self, "_used", False)
+
+    def consume(self) -> tuple[str, str, KnowledgeFSOperation]:
+        """Return the authorized principals and canonical operation exactly once.
+
+        Raises:
+            KnowledgeFSAccessDeniedError: The capability was already consumed.
+        """
+        if self._used:
+            raise KnowledgeFSAccessDeniedError("KnowledgeFS authorization has already been used")
+        object.__setattr__(self, "_used", True)
+        return self.account_id, self.tenant_id, self.operation
+
+
 class _RequestHeaders(Protocol):
    def items(self) -> Iterable[tuple[str, str]]: ...

@@ -113,20 +111,29 @@ def authorize_knowledge_fs_request(
    *,
    account: Account,
    tenant_id: str,
-    operation: KnowledgeFSOperation,
-) -> None:
+    method: KnowledgeFSMethod,
+    path: str,
+) -> KnowledgeFSAuthorization:
    """Enforce Dify's workspace policy before KFS performs resource authorization.

    Args:
        account: Authenticated Dify account with its current workspace role.
        tenant_id: Current Dify workspace identifier.
-        operation: Dify-maintained KnowledgeFS operation and policy metadata.
+        method: Requested upstream HTTP method.
+        path: Requested relative KnowledgeFS path.

    Raises:
+        KnowledgeFSRouteNotAllowedError: The method and path do not resolve to a declared operation.
        KnowledgeFSAccessDeniedError: The account lacks a required legacy or enterprise permission.
+
+    Returns:
+        A request-scoped capability binding the authorized account, workspace, and operation.
    """
-    if operation.requires_dataset_editor and not account.is_dataset_editor:
-        raise KnowledgeFSAccessDeniedError("KnowledgeFS mutations require dataset edit access")
+    operation = get_knowledge_fs_operation(method, path)
+    if operation.legacy_role == "dataset_editor" and not account.is_dataset_editor:
+        raise KnowledgeFSAccessDeniedError("KnowledgeFS operation requires dataset edit access")
+    if operation.legacy_role == "admin" and not account.is_admin_or_owner:
+        raise KnowledgeFSAccessDeniedError("KnowledgeFS operation requires workspace administration access")
    if not RBACService.CheckAccess.check(
        tenant_id,
        account.id,
@@ -134,6 +141,7 @@ def authorize_knowledge_fs_request(
        resource_type=RBACResourceScope.DATASET.value,
    ):
        raise KnowledgeFSAccessDeniedError("KnowledgeFS operation is denied by workspace RBAC")
+    return KnowledgeFSAuthorization(account.id, tenant_id, operation, _marker=_AUTHORIZATION_MARKER)


 def proxy_knowledge_fs_request(
@@ -149,20 +157,62 @@ def proxy_knowledge_fs_request(
    request_headers: _RequestHeaders | None = None,
 ) -> KnowledgeFSUpstreamResponse:
    """Authorize and forward one allowlisted KnowledgeFS request as a single use case."""
-    operation = get_knowledge_fs_operation(method, path)
-    authorize_knowledge_fs_request(
+    authorization = authorize_knowledge_fs_request(
        account=account,
        tenant_id=tenant_id,
-        operation=operation,
+        method=method,
+        path=path,
    )
+
+    return proxy_authorized_knowledge_fs_request(
+        authorization=authorization,
+        accept=accept,
+        content_type=content_type,
+        query=query,
+        body=body,
+        request_headers=request_headers,
+    )
+
+
+def proxy_authorized_knowledge_fs_request(
+    *,
+    authorization: KnowledgeFSAuthorization,
+    accept: str | None = None,
+    content_type: str | None = None,
+    query: bytes | None = None,
+    body: bytes | None = None,
+    request_headers: _RequestHeaders | None = None,
+) -> KnowledgeFSUpstreamResponse:
+    """Forward one request whose operation and workspace policy were already authorized.
+
+    This performs one outbound KnowledgeFS request and does not repeat Dify RBAC checks.
+
+    Args:
+        authorization: Request-scoped capability returned by :func:`authorize_knowledge_fs_request`.
+        accept: Original Accept header, when present.
+        content_type: Original request Content-Type header, when present.
+        query: Original encoded query string from the Console request.
+        body: Original request body, when present.
+        request_headers: Incoming headers; only names declared by the operation are forwarded.
+
+    Returns:
+        The bounded KnowledgeFS response together with its transport metadata.
+
+    Raises:
+        KnowledgeFSConfigurationError: The connection is incomplete or blocked by outbound policy.
+        KnowledgeFSRouteNotAllowedError: A forwarded request header is outside the operation contract.
+        KnowledgeFSTimeoutError: KnowledgeFS exceeds the configured timeout.
+        KnowledgeFSTransportError: The request fails or its response violates transport bounds.
+    """
+    account_id, tenant_id, operation = authorization.consume()
    incoming_request_headers = {name.lower(): value for name, value in (request_headers or {}).items()}
    contract_request_headers = {
        name: incoming_request_headers[name] for name in operation.request_headers if name in incoming_request_headers
    }
    return _forward_knowledge_fs_request(
-        account_id=account.id,
-        method=method,
-        path=path,
+        account_id=account_id,
+        method=operation.method,
+        path=operation.path,
        tenant_id=tenant_id,
        accept=accept,
        content_type=content_type,
--- a/Show More
+++ b/Show More