fix module_loading cached import fallback

.github/actions/setup-web/action.yml

@@ -1,13 +1,33 @@
 name: Setup Web Environment
+description: Setup pnpm, Node.js, and install web dependencies.
+
+inputs:
+  node-version:
+    description: Node.js version to use
+    required: false
+    default: "22"
+  install-dependencies:
+    description: Whether to install web dependencies after setting up Node.js
+    required: false
+    default: "true"
 
 runs:
   using: composite
   steps:
-    - name: Setup Vite+
-      uses: voidzero-dev/setup-vp@b5d848f5a62488f3d3d920f8aa6ac318a60c5f07 # v1
+    - name: Install pnpm
+      uses: pnpm/action-setup@41ff72655975bd51cab0327fa583b6e92b6d3061 # v4.2.0
       with:
-        node-version-file: "./web/.nvmrc"
-        cache: true
-        run-install: |
-          - cwd: ./web
-            args: ['--frozen-lockfile']
+        package_json_file: web/package.json
+        run_install: false
+
+    - name: Setup Node.js
+      uses: actions/setup-node@53b83947a5a98c8d113130e565377fae1a50d02f # v6.3.0
+      with:
+        node-version: ${{ inputs.node-version }}
+        cache: pnpm
+        cache-dependency-path: ./web/pnpm-lock.yaml
+
+    - name: Install dependencies
+      if: ${{ inputs.install-dependencies == 'true' }}
+      shell: bash
+      run: pnpm --dir web install --frozen-lockfile

.github/workflows/api-tests.yml

@@ -27,7 +27,7 @@ jobs:
           persist-credentials: false
 
       - name: Setup UV and Python
-        uses: astral-sh/setup-uv@e06108dd0aef18192324c70427afc47652e63a82 # v7.5.0
+        uses: astral-sh/setup-uv@6ee6290f1cbc4156c0bdd66691b2c144ef8df19a # v7.4.0
         with:
           enable-cache: true
           python-version: ${{ matrix.python-version }}

.github/workflows/autofix.yml

@@ -39,7 +39,7 @@ jobs:
         with:
           python-version: "3.11"
 
-      - uses: astral-sh/setup-uv@e06108dd0aef18192324c70427afc47652e63a82 # v7.5.0
+      - uses: astral-sh/setup-uv@6ee6290f1cbc4156c0bdd66691b2c144ef8df19a # v7.4.0
 
       - name: Generate Docker Compose
         if: steps.docker-compose-changes.outputs.any_changed == 'true'
@@ -102,11 +102,13 @@ jobs:
       - name: Setup web environment
         if: steps.web-changes.outputs.any_changed == 'true'
         uses: ./.github/actions/setup-web
+        with:
+          node-version: "24"
 
       - name: ESLint autofix
         if: steps.web-changes.outputs.any_changed == 'true'
         run: |
           cd web
-          vp exec eslint --concurrency=2 --prune-suppressions --quiet || true
+          pnpm eslint --concurrency=2 --prune-suppressions --quiet || true
 
       - uses: autofix-ci/action@7a166d7532b277f34e16238930461bf77f9d7ed8 # v1.3.3

.github/workflows/build-push.yml

@@ -113,7 +113,7 @@ jobs:
             context: "web"
     steps:
       - name: Download digests
-        uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
+        uses: actions/download-artifact@70fc10c6e5e1ce46ad2ea6f2b72d43f7d47b13c3 # v8.0.0
         with:
           path: /tmp/digests
           pattern: digests-${{ matrix.context }}-*

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

@@ -19,7 +19,7 @@ jobs:
           persist-credentials: false
 
       - name: Setup UV and Python
-        uses: astral-sh/setup-uv@e06108dd0aef18192324c70427afc47652e63a82 # v7.5.0
+        uses: astral-sh/setup-uv@6ee6290f1cbc4156c0bdd66691b2c144ef8df19a # v7.4.0
         with:
           enable-cache: true
           python-version: "3.12"
@@ -69,7 +69,7 @@ jobs:
           persist-credentials: false
 
       - name: Setup UV and Python
-        uses: astral-sh/setup-uv@e06108dd0aef18192324c70427afc47652e63a82 # v7.5.0
+        uses: astral-sh/setup-uv@6ee6290f1cbc4156c0bdd66691b2c144ef8df19a # v7.4.0
         with:
           enable-cache: true
           python-version: "3.12"

.github/workflows/main-ci.yml

@@ -28,7 +28,7 @@ jobs:
       migration-changed: ${{ steps.changes.outputs.migration }}
     steps:
       - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
-      - uses: dorny/paths-filter@fbd0ab8f3e69293af611ebaee6363fc25e6d187d # v4.0.1
+      - uses: dorny/paths-filter@de90cc6fb38fc0963ad72b210f1f284cd68cea36 # v3.0.2
         id: changes
         with:
           filters: |

.github/workflows/pyrefly-diff.yml

@@ -22,7 +22,7 @@ jobs:
           fetch-depth: 0
 
       - name: Setup Python & UV
-        uses: astral-sh/setup-uv@e06108dd0aef18192324c70427afc47652e63a82 # v7.5.0
+        uses: astral-sh/setup-uv@6ee6290f1cbc4156c0bdd66691b2c144ef8df19a # v7.4.0
         with:
           enable-cache: true
 

.github/workflows/style.yml

@@ -33,7 +33,7 @@ jobs:
 
       - name: Setup UV and Python
         if: steps.changed-files.outputs.any_changed == 'true'
-        uses: astral-sh/setup-uv@e06108dd0aef18192324c70427afc47652e63a82 # v7.5.0
+        uses: astral-sh/setup-uv@6ee6290f1cbc4156c0bdd66691b2c144ef8df19a # v7.4.0
         with:
           enable-cache: false
           python-version: "3.12"
@@ -88,7 +88,7 @@ jobs:
         if: steps.changed-files.outputs.any_changed == 'true'
         working-directory: ./web
         run: |
-          vp run lint:ci
+          pnpm run lint:ci
         # pnpm run lint:report
         # continue-on-error: true
 
@@ -102,17 +102,17 @@ jobs:
       - name: Web tsslint
         if: steps.changed-files.outputs.any_changed == 'true'
         working-directory: ./web
-        run: vp run lint:tss
+        run: pnpm run lint:tss
 
       - name: Web type check
         if: steps.changed-files.outputs.any_changed == 'true'
         working-directory: ./web
-        run: vp run type-check
+        run: pnpm run type-check
 
       - name: Web dead code check
         if: steps.changed-files.outputs.any_changed == 'true'
         working-directory: ./web
-        run: vp run knip
+        run: pnpm run knip
 
   superlinter:
     name: SuperLinter

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

@@ -50,6 +50,8 @@ jobs:
 
       - name: Setup web environment
         uses: ./.github/actions/setup-web
+        with:
+          install-dependencies: "false"
 
       - name: Detect changed files and generate diff
         id: detect_changes
@@ -120,7 +122,7 @@ jobs:
 
       - name: Run Claude Code for Translation Sync
         if: steps.detect_changes.outputs.CHANGED_FILES != ''
-        uses: anthropics/claude-code-action@cd77b50d2b0808657f8e6774085c8bf54484351c # v1.0.72
+        uses: anthropics/claude-code-action@26ec041249acb0a944c0a47b6c0c13f05dbc5b44 # v1.0.70
         with:
           anthropic_api_key: ${{ secrets.ANTHROPIC_API_KEY }}
           github_token: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/vdb-tests.yml

@@ -31,7 +31,7 @@ jobs:
           remove_tool_cache: true
 
       - name: Setup UV and Python
-        uses: astral-sh/setup-uv@e06108dd0aef18192324c70427afc47652e63a82 # v7.5.0
+        uses: astral-sh/setup-uv@6ee6290f1cbc4156c0bdd66691b2c144ef8df19a # v7.4.0
         with:
           enable-cache: true
           python-version: ${{ matrix.python-version }}

.github/workflows/web-tests.yml

@@ -26,8 +26,8 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        shardIndex: [1, 2, 3, 4, 5, 6]
-        shardTotal: [6]
+        shardIndex: [1, 2, 3, 4]
+        shardTotal: [4]
     defaults:
       run:
         shell: bash
@@ -43,7 +43,7 @@ jobs:
         uses: ./.github/actions/setup-web
 
       - name: Run tests
-        run: vp test run --reporter=blob --shard=${{ matrix.shardIndex }}/${{ matrix.shardTotal }} --coverage
+        run: pnpm vitest run --reporter=blob --shard=${{ matrix.shardIndex }}/${{ matrix.shardTotal }} --coverage
 
       - name: Upload blob report
         if: ${{ !cancelled() }}
@@ -77,14 +77,14 @@ jobs:
         uses: ./.github/actions/setup-web
 
       - name: Download blob reports
-        uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
+        uses: actions/download-artifact@70fc10c6e5e1ce46ad2ea6f2b72d43f7d47b13c3 # v8.0.0
         with:
           path: web/.vitest-reports
           pattern: blob-report-*
           merge-multiple: true
 
       - name: Merge reports
-        run: vp test --merge-reports --reporter=json --reporter=agent --coverage
+        run: pnpm vitest --merge-reports --reporter=json --reporter=agent --coverage
 
       - name: Check app/components diff coverage
         env:
@@ -447,4 +447,4 @@ jobs:
       - name: Web build check
         if: steps.changed-files.outputs.any_changed == 'true'
         working-directory: ./web
-        run: vp run build
+        run: pnpm run build

.gitignore

@@ -237,6 +237,3 @@ scripts/stress-test/reports/
 # settings
 *.local.json
 *.local.md
-
-# Code Agent Folder
-.qoder/*

api/.env.example

@@ -22,10 +22,10 @@ APP_WEB_URL=http://localhost:3000
 # Files URL
 FILES_URL=http://localhost:5001
 
-# INTERNAL_FILES_URL is used by services running in Docker to reach the API file endpoints.
-# For Docker Desktop (Mac/Windows), use http://host.docker.internal:5001 when the API runs on the host.
-# For Docker Compose on Linux, use http://api:5001 when the API runs inside the Docker network.
-INTERNAL_FILES_URL=http://host.docker.internal:5001
+# INTERNAL_FILES_URL is used for plugin daemon communication within Docker network.
+# Set this to the internal Docker service URL for proper plugin file access.
+# Example: INTERNAL_FILES_URL=http://api:5001
+INTERNAL_FILES_URL=http://127.0.0.1:5001
 
 # TRIGGER URL
 TRIGGER_URL=http://localhost:5001
@@ -180,7 +180,7 @@ CONSOLE_CORS_ALLOW_ORIGINS=http://localhost:3000,*
 COOKIE_DOMAIN=
 
 # Vector database configuration
-# Supported values are `weaviate`, `oceanbase`, `qdrant`, `milvus`, `myscale`, `relyt`, `pgvector`, `pgvecto-rs`, `chroma`, `opensearch`, `oracle`, `tencent`, `elasticsearch`, `elasticsearch-ja`, `analyticdb`, `couchbase`, `vikingdb`,  `opengauss`, `tablestore`,`vastbase`,`tidb`,`tidb_on_qdrant`,`baidu`,`lindorm`,`huawei_cloud`,`upstash`, `matrixone`, `hologres`.
+# Supported values are `weaviate`, `oceanbase`, `qdrant`, `milvus`, `myscale`, `relyt`, `pgvector`, `pgvecto-rs`, `chroma`, `opensearch`, `oracle`, `tencent`, `elasticsearch`, `elasticsearch-ja`, `analyticdb`, `couchbase`, `vikingdb`,  `opengauss`, `tablestore`,`vastbase`,`tidb`,`tidb_on_qdrant`,`baidu`,`lindorm`,`huawei_cloud`,`upstash`, `matrixone`.
 VECTOR_STORE=weaviate
 # Prefix used to create collection name in vector database
 VECTOR_INDEX_NAME_PREFIX=Vector_index
@@ -217,20 +217,6 @@ COUCHBASE_PASSWORD=password
 COUCHBASE_BUCKET_NAME=Embeddings
 COUCHBASE_SCOPE_NAME=_default
 
-# Hologres configuration
-# access_key_id is used as the PG username, access_key_secret is used as the PG password
-HOLOGRES_HOST=
-HOLOGRES_PORT=80
-HOLOGRES_DATABASE=
-HOLOGRES_ACCESS_KEY_ID=
-HOLOGRES_ACCESS_KEY_SECRET=
-HOLOGRES_SCHEMA=public
-HOLOGRES_TOKENIZER=jieba
-HOLOGRES_DISTANCE_METHOD=Cosine
-HOLOGRES_BASE_QUANTIZATION_TYPE=rabitq
-HOLOGRES_MAX_DEGREE=64
-HOLOGRES_EF_CONSTRUCTION=400
-
 # Milvus configuration
 MILVUS_URI=http://127.0.0.1:19530
 MILVUS_TOKEN=

api/.importlinter

@@ -43,6 +43,7 @@ forbidden_modules =
     extensions.ext_redis
 allow_indirect_imports = True
 ignore_imports =
+    dify_graph.nodes.agent.agent_node -> extensions.ext_database
     dify_graph.nodes.llm.node -> extensions.ext_database
     dify_graph.model_runtime.model_providers.__base.ai_model -> extensions.ext_redis
     dify_graph.model_runtime.model_providers.model_provider_factory -> extensions.ext_redis
@@ -89,6 +90,9 @@ forbidden_modules =
     core.trigger
     core.variables
 ignore_imports =
+    dify_graph.nodes.agent.agent_node -> core.model_manager
+    dify_graph.nodes.agent.agent_node -> core.provider_manager
+    dify_graph.nodes.agent.agent_node -> core.tools.tool_manager
     dify_graph.nodes.llm.llm_utils -> core.model_manager
     dify_graph.nodes.llm.protocols -> core.model_manager
     dify_graph.nodes.llm.llm_utils -> dify_graph.model_runtime.model_providers.__base.large_language_model
@@ -96,6 +100,9 @@ ignore_imports =
     dify_graph.nodes.tool.tool_node -> core.callback_handler.workflow_tool_callback_handler
     dify_graph.nodes.tool.tool_node -> core.tools.tool_engine
     dify_graph.nodes.tool.tool_node -> core.tools.tool_manager
+    dify_graph.nodes.agent.agent_node -> core.agent.entities
+    dify_graph.nodes.agent.agent_node -> core.agent.plugin_entities
+    dify_graph.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.app.app_config.entities
     dify_graph.nodes.parameter_extractor.parameter_extractor_node -> core.prompt.advanced_prompt_transform
     dify_graph.nodes.parameter_extractor.parameter_extractor_node -> core.prompt.simple_prompt_transform
     dify_graph.nodes.parameter_extractor.parameter_extractor_node -> dify_graph.model_runtime.model_providers.__base.large_language_model
@@ -103,9 +110,12 @@ ignore_imports =
     dify_graph.nodes.parameter_extractor.parameter_extractor_node -> core.model_manager
     dify_graph.nodes.question_classifier.question_classifier_node -> core.model_manager
     dify_graph.nodes.tool.tool_node -> core.tools.utils.message_transformer
+    dify_graph.nodes.agent.agent_node -> models.model
+    dify_graph.nodes.llm.node -> core.helper.code_executor
     dify_graph.nodes.llm.node -> core.llm_generator.output_parser.errors
     dify_graph.nodes.llm.node -> core.llm_generator.output_parser.structured_output
     dify_graph.nodes.llm.node -> core.model_manager
+    dify_graph.nodes.agent.entities -> core.prompt.entities.advanced_prompt_entities
     dify_graph.nodes.llm.entities -> core.prompt.entities.advanced_prompt_entities
     dify_graph.nodes.llm.node -> core.prompt.entities.advanced_prompt_entities
     dify_graph.nodes.llm.node -> core.prompt.utils.prompt_message_util
@@ -114,12 +124,17 @@ ignore_imports =
     dify_graph.nodes.parameter_extractor.parameter_extractor_node -> core.prompt.utils.prompt_message_util
     dify_graph.nodes.question_classifier.entities -> core.prompt.entities.advanced_prompt_entities
     dify_graph.nodes.question_classifier.question_classifier_node -> core.prompt.utils.prompt_message_util
+    dify_graph.nodes.knowledge_index.entities -> core.rag.retrieval.retrieval_methods
     dify_graph.nodes.llm.node -> models.dataset
+    dify_graph.nodes.agent.agent_node -> core.tools.utils.message_transformer
     dify_graph.nodes.llm.file_saver -> core.tools.signature
     dify_graph.nodes.llm.file_saver -> core.tools.tool_file_manager
     dify_graph.nodes.tool.tool_node -> core.tools.errors
+    dify_graph.nodes.agent.agent_node -> extensions.ext_database
     dify_graph.nodes.llm.node -> extensions.ext_database
+    dify_graph.nodes.agent.agent_node -> models
     dify_graph.nodes.llm.node -> models.model
+    dify_graph.nodes.agent.agent_node -> services
     dify_graph.nodes.tool.tool_node -> services
     dify_graph.model_runtime.model_providers.__base.ai_model -> configs
     dify_graph.model_runtime.model_providers.__base.ai_model -> extensions.ext_redis

api/AGENTS.md

@@ -78,7 +78,7 @@ class UserProfile(TypedDict):
     nickname: NotRequired[str]
 ```
 
-- For classes, declare member variables at the top of the class body (before `__init__`) so the class shape is obvious at a glance:
+- For classes, declare all member variables explicitly with types at the top of the class body (before `__init__`), even when the class is not a dataclass or Pydantic model, so the class shape is obvious at a glance:
 
 ```python
 from datetime import datetime

api/README.md

@@ -42,7 +42,7 @@ The scripts resolve paths relative to their location, so you can run them from a
 
 1. Set up your application by visiting `http://localhost:3000`.
 
-1. Optional: start the worker service (async tasks, runs from `api`).
+1. Start the worker service (async and scheduler tasks, runs from `api`).
 
    ```bash
    ./dev/start-worker
@@ -54,87 +54,6 @@ The scripts resolve paths relative to their location, so you can run them from a
    ./dev/start-beat
    ```
 
-### Manual commands
-
-<details>
-<summary>Show manual setup and run steps</summary>
-
-These commands assume you start from the repository root.
-
-1. Start the docker-compose stack.
-
-   The backend requires middleware, including PostgreSQL, Redis, and Weaviate, which can be started together using `docker-compose`.
-
-   ```bash
-   cp docker/middleware.env.example docker/middleware.env
-   # Use mysql or another vector database profile if you are not using postgres/weaviate.
-   docker compose -f docker/docker-compose.middleware.yaml --profile postgresql --profile weaviate -p dify up -d
-   ```
-
-1. Copy env files.
-
-   ```bash
-   cp api/.env.example api/.env
-   cp web/.env.example web/.env.local
-   ```
-
-1. Install UV if needed.
-
-   ```bash
-   pip install uv
-   # Or on macOS
-   brew install uv
-   ```
-
-1. Install API dependencies.
-
-   ```bash
-   cd api
-   uv sync --group dev
-   ```
-
-1. Install web dependencies.
-
-   ```bash
-   cd web
-   pnpm install
-   cd ..
-   ```
-
-1. Start backend (runs migrations first, in a new terminal).
-
-   ```bash
-   cd api
-   uv run flask db upgrade
-   uv run flask run --host 0.0.0.0 --port=5001 --debug
-   ```
-
-1. Start Dify [web](../web) service (in a new terminal).
-
-   ```bash
-   cd web
-   pnpm dev:inspect
-   ```
-
-1. Set up your application by visiting `http://localhost:3000`.
-
-1. Optional: start the worker service (async tasks, in a new terminal).
-
-   ```bash
-   cd api
-   # Note: enterprise_telemetry queue is only used in Enterprise Edition
-   uv run celery -A app.celery worker -P threads -c 2 --loglevel INFO -Q dataset,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor,retention,enterprise_telemetry
-   ```
-
-1. Optional: start Celery Beat (scheduled tasks, in a new terminal).
-
-   ```bash
-   cd api
-   uv run celery -A app.celery beat
-   ```
-
-</details>
-
 ### Environment notes
 
 > [!IMPORTANT]

api/app_factory.py

@@ -81,7 +81,6 @@ def initialize_extensions(app: DifyApp):
         ext_commands,
         ext_compress,
         ext_database,
-        ext_enterprise_telemetry,
         ext_fastopenapi,
         ext_forward_refs,
         ext_hosting_provider,
@@ -132,7 +131,6 @@ def initialize_extensions(app: DifyApp):
         ext_commands,
         ext_fastopenapi,
         ext_otel,
-        ext_enterprise_telemetry,
         ext_request_logging,
         ext_session_factory,
     ]

api/commands/vector.py

@@ -160,7 +160,6 @@ def migrate_knowledge_vector_database():
     }
     lower_collection_vector_types = {
         VectorType.ANALYTICDB,
-        VectorType.HOLOGRES,
         VectorType.CHROMA,
         VectorType.MYSCALE,
         VectorType.PGVECTO_RS,

api/configs/app_config.py

@@ -8,7 +8,7 @@ from pydantic_settings import BaseSettings, PydanticBaseSettingsSource, Settings
 from libs.file_utils import search_file_upwards
 
 from .deploy import DeploymentConfig
-from .enterprise import EnterpriseFeatureConfig, EnterpriseTelemetryConfig
+from .enterprise import EnterpriseFeatureConfig
 from .extra import ExtraServiceConfig
 from .feature import FeatureConfig
 from .middleware import MiddlewareConfig
@@ -73,8 +73,6 @@ class DifyConfig(
     # Enterprise feature configs
     # **Before using, please contact business@dify.ai by email to inquire about licensing matters.**
     EnterpriseFeatureConfig,
-    # Enterprise telemetry configs
-    EnterpriseTelemetryConfig,
 ):
     model_config = SettingsConfigDict(
         # read from dotenv format config file

api/configs/enterprise/__init__.py

@@ -22,49 +22,3 @@ class EnterpriseFeatureConfig(BaseSettings):
     ENTERPRISE_REQUEST_TIMEOUT: int = Field(
         ge=1, description="Maximum timeout in seconds for enterprise requests", default=5
     )
-
-
-class EnterpriseTelemetryConfig(BaseSettings):
-    """
-    Configuration for enterprise telemetry.
-    """
-
-    ENTERPRISE_TELEMETRY_ENABLED: bool = Field(
-        description="Enable enterprise telemetry collection (also requires ENTERPRISE_ENABLED=true).",
-        default=False,
-    )
-
-    ENTERPRISE_OTLP_ENDPOINT: str = Field(
-        description="Enterprise OTEL collector endpoint.",
-        default="",
-    )
-
-    ENTERPRISE_OTLP_HEADERS: str = Field(
-        description="Auth headers for OTLP export (key=value,key2=value2).",
-        default="",
-    )
-
-    ENTERPRISE_OTLP_PROTOCOL: str = Field(
-        description="OTLP protocol: 'http' or 'grpc' (default: http).",
-        default="http",
-    )
-
-    ENTERPRISE_OTLP_API_KEY: str = Field(
-        description="Bearer token for enterprise OTLP export authentication.",
-        default="",
-    )
-
-    ENTERPRISE_INCLUDE_CONTENT: bool = Field(
-        description="Include input/output content in traces (privacy toggle).",
-        default=True,
-    )
-
-    ENTERPRISE_SERVICE_NAME: str = Field(
-        description="Service name for OTEL resource.",
-        default="dify",
-    )
-
-    ENTERPRISE_OTEL_SAMPLING_RATE: float = Field(
-        description="Sampling rate for enterprise traces (0.0 to 1.0, default 1.0 = 100%).",
-        default=1.0,
-    )

api/configs/middleware/__init__.py

@@ -26,7 +26,6 @@ from .vdb.chroma_config import ChromaConfig
 from .vdb.clickzetta_config import ClickzettaConfig
 from .vdb.couchbase_config import CouchbaseConfig
 from .vdb.elasticsearch_config import ElasticsearchConfig
-from .vdb.hologres_config import HologresConfig
 from .vdb.huawei_cloud_config import HuaweiCloudConfig
 from .vdb.iris_config import IrisVectorConfig
 from .vdb.lindorm_config import LindormConfig
@@ -348,7 +347,6 @@ class MiddlewareConfig(
     AnalyticdbConfig,
     ChromaConfig,
     ClickzettaConfig,
-    HologresConfig,
     HuaweiCloudConfig,
     IrisVectorConfig,
     MilvusConfig,

api/configs/middleware/cache/redis_pubsub_config.py

@@ -1,4 +1,4 @@
-from typing import Literal, Protocol
+from typing import Literal, Protocol, cast
 from urllib.parse import quote_plus, urlunparse
 
 from pydantic import AliasChoices, Field
@@ -12,16 +12,13 @@ class RedisConfigDefaults(Protocol):
     REDIS_PASSWORD: str | None
     REDIS_DB: int
     REDIS_USE_SSL: bool
-    REDIS_USE_SENTINEL: bool | None
-    REDIS_USE_CLUSTERS: bool
 
 
-class RedisConfigDefaultsMixin:
-    def _redis_defaults(self: RedisConfigDefaults) -> RedisConfigDefaults:
-        return self
+def _redis_defaults(config: object) -> RedisConfigDefaults:
+    return cast(RedisConfigDefaults, config)
 
 
-class RedisPubSubConfig(BaseSettings, RedisConfigDefaultsMixin):
+class RedisPubSubConfig(BaseSettings):
     """
     Configuration settings for event transport between API and workers.
 
@@ -74,7 +71,7 @@ class RedisPubSubConfig(BaseSettings, RedisConfigDefaultsMixin):
     )
 
     def _build_default_pubsub_url(self) -> str:
-        defaults = self._redis_defaults()
+        defaults = _redis_defaults(self)
         if not defaults.REDIS_HOST or not defaults.REDIS_PORT:
             raise ValueError("PUBSUB_REDIS_URL must be set when default Redis URL cannot be constructed")
 
@@ -91,11 +88,9 @@ class RedisPubSubConfig(BaseSettings, RedisConfigDefaultsMixin):
         if userinfo:
             userinfo = f"{userinfo}@"
 
-        host = defaults.REDIS_HOST
-        port = defaults.REDIS_PORT
         db = defaults.REDIS_DB
 
-        netloc = f"{userinfo}{host}:{port}"
+        netloc = f"{userinfo}{defaults.REDIS_HOST}:{defaults.REDIS_PORT}"
         return urlunparse((scheme, netloc, f"/{db}", "", "", ""))
 
     @property

api/configs/middleware/vdb/hologres_config.py

@@ -1,68 +0,0 @@
-from holo_search_sdk.types import BaseQuantizationType, DistanceType, TokenizerType
-from pydantic import Field
-from pydantic_settings import BaseSettings
-
-
-class HologresConfig(BaseSettings):
-    """
-    Configuration settings for Hologres vector database.
-
-    Hologres is compatible with PostgreSQL protocol.
-    access_key_id is used as the PostgreSQL username,
-    and access_key_secret is used as the PostgreSQL password.
-    """
-
-    HOLOGRES_HOST: str | None = Field(
-        description="Hostname or IP address of the Hologres instance.",
-        default=None,
-    )
-
-    HOLOGRES_PORT: int = Field(
-        description="Port number for connecting to the Hologres instance.",
-        default=80,
-    )
-
-    HOLOGRES_DATABASE: str | None = Field(
-        description="Name of the Hologres database to connect to.",
-        default=None,
-    )
-
-    HOLOGRES_ACCESS_KEY_ID: str | None = Field(
-        description="Alibaba Cloud AccessKey ID, also used as the PostgreSQL username.",
-        default=None,
-    )
-
-    HOLOGRES_ACCESS_KEY_SECRET: str | None = Field(
-        description="Alibaba Cloud AccessKey Secret, also used as the PostgreSQL password.",
-        default=None,
-    )
-
-    HOLOGRES_SCHEMA: str = Field(
-        description="Schema name in the Hologres database.",
-        default="public",
-    )
-
-    HOLOGRES_TOKENIZER: TokenizerType = Field(
-        description="Tokenizer for full-text search index (e.g., 'jieba', 'ik', 'standard', 'simple').",
-        default="jieba",
-    )
-
-    HOLOGRES_DISTANCE_METHOD: DistanceType = Field(
-        description="Distance method for vector index (e.g., 'Cosine', 'Euclidean', 'InnerProduct').",
-        default="Cosine",
-    )
-
-    HOLOGRES_BASE_QUANTIZATION_TYPE: BaseQuantizationType = Field(
-        description="Base quantization type for vector index (e.g., 'rabitq', 'sq8', 'fp16', 'fp32').",
-        default="rabitq",
-    )
-
-    HOLOGRES_MAX_DEGREE: int = Field(
-        description="Max degree (M) parameter for HNSW vector index.",
-        default=64,
-    )
-
-    HOLOGRES_EF_CONSTRUCTION: int = Field(
-        description="ef_construction parameter for HNSW vector index.",
-        default=400,
-    )

api/controllers/console/app/app.py

@@ -25,8 +25,7 @@ from controllers.console.wraps import (
 )
 from core.ops.ops_trace_manager import OpsTraceManager
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
-from core.trigger.constants import TRIGGER_NODE_TYPES
-from dify_graph.enums import WorkflowExecutionStatus
+from dify_graph.enums import NodeType, WorkflowExecutionStatus
 from dify_graph.file import helpers as file_helpers
 from extensions.ext_database import db
 from libs.login import current_account_with_tenant, login_required
@@ -509,7 +508,11 @@ class AppListApi(Resource):
                 .scalars()
                 .all()
             )
-            trigger_node_types = TRIGGER_NODE_TYPES
+            trigger_node_types = {
+                NodeType.TRIGGER_WEBHOOK,
+                NodeType.TRIGGER_SCHEDULE,
+                NodeType.TRIGGER_PLUGIN,
+            }
             for workflow in draft_workflows:
                 node_id = None
                 try:

api/controllers/console/app/mcp_server.py

@@ -1,4 +1,5 @@
 import json
+from enum import StrEnum
 
 from flask_restx import Resource, marshal_with
 from pydantic import BaseModel, Field
@@ -10,7 +11,6 @@ from controllers.console.wraps import account_initialization_required, edit_perm
 from extensions.ext_database import db
 from fields.app_fields import app_server_fields
 from libs.login import current_account_with_tenant, login_required
-from models.enums import AppMCPServerStatus
 from models.model import AppMCPServer
 
 DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
@@ -19,6 +19,11 @@ DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 app_server_model = console_ns.model("AppServer", app_server_fields)
 
 
+class AppMCPServerStatus(StrEnum):
+    ACTIVE = "active"
+    INACTIVE = "inactive"
+
+
 class MCPServerCreatePayload(BaseModel):
     description: str | None = Field(default=None, description="Server description")
     parameters: dict = Field(..., description="Server parameters configuration")
@@ -112,10 +117,9 @@ class AppMCPServerController(Resource):
 
         server.parameters = json.dumps(payload.parameters, ensure_ascii=False)
         if payload.status:
-            try:
-                server.status = AppMCPServerStatus(payload.status)
-            except ValueError:
+            if payload.status not in [status.value for status in AppMCPServerStatus]:
                 raise ValueError("Invalid status")
+            server.status = payload.status
         db.session.commit()
         return server
 

api/controllers/console/app/workflow.py

@@ -22,7 +22,6 @@ from core.app.apps.workflow.app_generator import SKIP_PREPARE_USER_INPUTS_KEY
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.helper.trace_id_helper import get_external_trace_id
 from core.plugin.impl.exc import PluginInvokeError
-from core.trigger.constants import TRIGGER_SCHEDULE_NODE_TYPE
 from core.trigger.debug.event_selectors import (
     TriggerDebugEvent,
     TriggerDebugEventPoller,
@@ -1210,7 +1209,7 @@ class DraftWorkflowTriggerNodeApi(Resource):
         node_type: NodeType = draft_workflow.get_node_type_from_node_config(node_config)
         event: TriggerDebugEvent | None = None
         # for schedule trigger, when run single node, just execute directly
-        if node_type == TRIGGER_SCHEDULE_NODE_TYPE:
+        if node_type == NodeType.TRIGGER_SCHEDULE:
             event = TriggerDebugEvent(
                 workflow_args={},
                 node_id=node_id,

api/controllers/console/datasets/datasets.py

@@ -263,7 +263,6 @@ def _get_retrieval_methods_by_vector_type(vector_type: str | None, is_mock: bool
         VectorType.BAIDU,
         VectorType.ALIBABACLOUD_MYSQL,
         VectorType.IRIS,
-        VectorType.HOLOGRES,
     }
 
     semantic_methods = {"retrieval_method": [RetrievalMethod.SEMANTIC_SEARCH.value]}

api/controllers/console/workspace/account.py

@@ -43,7 +43,6 @@ from libs.datetime_utils import naive_utc_now
 from libs.helper import EmailStr, TimestampField, extract_remote_ip, timezone
 from libs.login import current_account_with_tenant, login_required
 from models import AccountIntegrate, InvitationCode
-from models.account import AccountStatus, InvitationCodeStatus
 from services.account_service import AccountService
 from services.billing_service import BillingService
 from services.errors.account import CurrentPasswordIncorrectError as ServiceCurrentPasswordIncorrectError
@@ -216,7 +215,7 @@ class AccountInitApi(Resource):
                 db.session.query(InvitationCode)
                 .where(
                     InvitationCode.code == args.invitation_code,
-                    InvitationCode.status == InvitationCodeStatus.UNUSED,
+                    InvitationCode.status == "unused",
                 )
                 .first()
             )
@@ -224,7 +223,7 @@ class AccountInitApi(Resource):
             if not invitation_code:
                 raise InvalidInvitationCodeError()
 
-            invitation_code.status = InvitationCodeStatus.USED
+            invitation_code.status = "used"
             invitation_code.used_at = naive_utc_now()
             invitation_code.used_by_tenant_id = account.current_tenant_id
             invitation_code.used_by_account_id = account.id
@@ -232,7 +231,7 @@ class AccountInitApi(Resource):
         account.interface_language = args.interface_language
         account.timezone = args.timezone
         account.interface_theme = "light"
-        account.status = AccountStatus.ACTIVE
+        account.status = "active"
         account.initialized_at = naive_utc_now()
         db.session.commit()
 

api/controllers/console/workspace/plugin.py

@@ -5,7 +5,6 @@ from typing import Any, Literal
 from flask import request, send_file
 from flask_restx import Resource
 from pydantic import BaseModel, Field
-from werkzeug.datastructures import FileStorage
 from werkzeug.exceptions import Forbidden
 
 from configs import dify_config
@@ -170,20 +169,6 @@ register_enum_models(
 )
 
 
-def _read_upload_content(file: FileStorage, max_size: int) -> bytes:
-    """
-    Read the uploaded file and validate its actual size before delegating to the plugin service.
-
-    FileStorage.content_length is not reliable for multipart test uploads and may be zero even when
-    content exists, so the controllers validate against the loaded bytes instead.
-    """
-    content = file.read()
-    if len(content) > max_size:
-        raise ValueError("File size exceeds the maximum allowed size")
-
-    return content
-
-
 @console_ns.route("/workspaces/current/plugin/debugging-key")
 class PluginDebuggingKeyApi(Resource):
     @setup_required
@@ -299,7 +284,12 @@ class PluginUploadFromPkgApi(Resource):
         _, tenant_id = current_account_with_tenant()
 
         file = request.files["pkg"]
-        content = _read_upload_content(file, dify_config.PLUGIN_MAX_PACKAGE_SIZE)
+
+        # check file size
+        if file.content_length > dify_config.PLUGIN_MAX_PACKAGE_SIZE:
+            raise ValueError("File size exceeds the maximum allowed size")
+
+        content = file.read()
         try:
             response = PluginService.upload_pkg(tenant_id, content)
         except PluginDaemonClientSideError as e:
@@ -338,7 +328,12 @@ class PluginUploadFromBundleApi(Resource):
         _, tenant_id = current_account_with_tenant()
 
         file = request.files["bundle"]
-        content = _read_upload_content(file, dify_config.PLUGIN_MAX_BUNDLE_SIZE)
+
+        # check file size
+        if file.content_length > dify_config.PLUGIN_MAX_BUNDLE_SIZE:
+            raise ValueError("File size exceeds the maximum allowed size")
+
+        content = file.read()
         try:
             response = PluginService.upload_bundle(tenant_id, content)
         except PluginDaemonClientSideError as e:

api/controllers/mcp/mcp.py

@@ -6,13 +6,13 @@ from pydantic import BaseModel, Field, ValidationError
 from sqlalchemy.orm import Session
 
 from controllers.common.schema import register_schema_model
+from controllers.console.app.mcp_server import AppMCPServerStatus
 from controllers.mcp import mcp_ns
 from core.mcp import types as mcp_types
 from core.mcp.server.streamable_http import handle_mcp_request
 from dify_graph.variables.input_entities import VariableEntity
 from extensions.ext_database import db
 from libs import helper
-from models.enums import AppMCPServerStatus
 from models.model import App, AppMCPServer, AppMode, EndUser
 
 

api/controllers/service_api/wraps.py

@@ -3,7 +3,7 @@ import time
 from collections.abc import Callable
 from enum import StrEnum, auto
 from functools import wraps
-from typing import Concatenate, ParamSpec, TypeVar, cast, overload
+from typing import Concatenate, ParamSpec, TypeVar, cast
 
 from flask import current_app, request
 from flask_login import user_logged_in
@@ -44,22 +44,10 @@ class FetchUserArg(BaseModel):
     required: bool = False
 
 
-@overload
-def validate_app_token(view: Callable[P, R]) -> Callable[P, R]: ...
-
-
-@overload
-def validate_app_token(
-    view: None = None, *, fetch_user_arg: FetchUserArg | None = None
-) -> Callable[[Callable[P, R]], Callable[P, R]]: ...
-
-
-def validate_app_token(
-    view: Callable[P, R] | None = None, *, fetch_user_arg: FetchUserArg | None = None
-) -> Callable[P, R] | Callable[[Callable[P, R]], Callable[P, R]]:
-    def decorator(view_func: Callable[P, R]) -> Callable[P, R]:
+def validate_app_token(view: Callable[P, R] | None = None, *, fetch_user_arg: FetchUserArg | None = None):
+    def decorator(view_func: Callable[P, R]):
         @wraps(view_func)
-        def decorated_view(*args: P.args, **kwargs: P.kwargs) -> R:
+        def decorated_view(*args: P.args, **kwargs: P.kwargs):
             api_token = validate_and_get_api_token("app")
 
             app_model = db.session.query(App).where(App.id == api_token.app_id).first()
@@ -225,20 +213,10 @@ def cloud_edition_billing_rate_limit_check(resource: str, api_token_type: str):
     return interceptor
 
 
-@overload
-def validate_dataset_token(view: Callable[Concatenate[T, P], R]) -> Callable[P, R]: ...
-
-
-@overload
-def validate_dataset_token(view: None = None) -> Callable[[Callable[Concatenate[T, P], R]], Callable[P, R]]: ...
-
-
-def validate_dataset_token(
-    view: Callable[Concatenate[T, P], R] | None = None,
-) -> Callable[P, R] | Callable[[Callable[Concatenate[T, P], R]], Callable[P, R]]:
-    def decorator(view_func: Callable[Concatenate[T, P], R]) -> Callable[P, R]:
-        @wraps(view_func)
-        def decorated(*args: P.args, **kwargs: P.kwargs) -> R:
+def validate_dataset_token(view: Callable[Concatenate[T, P], R] | None = None):
+    def decorator(view: Callable[Concatenate[T, P], R]):
+        @wraps(view)
+        def decorated(*args: P.args, **kwargs: P.kwargs):
             api_token = validate_and_get_api_token("dataset")
 
             # get url path dataset_id from positional args or kwargs
@@ -309,7 +287,7 @@ def validate_dataset_token(
                     raise Unauthorized("Tenant owner account does not exist.")
             else:
                 raise Unauthorized("Tenant does not exist.")
-            return view_func(api_token.tenant_id, *args, **kwargs)  # type: ignore[arg-type]
+            return view(api_token.tenant_id, *args, **kwargs)
 
         return decorated
 

api/core/agent/cot_agent_runner.py

@@ -6,7 +6,6 @@ from typing import Any
 
 from core.agent.base_agent_runner import BaseAgentRunner
 from core.agent.entities import AgentScratchpadUnit
-from core.agent.errors import AgentMaxIterationError
 from core.agent.output_parser.cot_output_parser import CotAgentOutputParser
 from core.app.apps.base_app_queue_manager import PublishFrom
 from core.app.entities.queue_entities import QueueAgentThoughtEvent, QueueMessageEndEvent, QueueMessageFileEvent
@@ -23,6 +22,7 @@ from dify_graph.model_runtime.entities.message_entities import (
     ToolPromptMessage,
     UserPromptMessage,
 )
+from dify_graph.nodes.agent.exc import AgentMaxIterationError
 from models.model import Message
 
 logger = logging.getLogger(__name__)

api/core/agent/errors.py

@@ -1,9 +0,0 @@
-class AgentMaxIterationError(Exception):
-    """Raised when an agent runner exceeds the configured max iteration count."""
-
-    def __init__(self, max_iteration: int):
-        self.max_iteration = max_iteration
-        super().__init__(
-            f"Agent exceeded the maximum iteration limit of {max_iteration}. "
-            f"The agent was unable to complete the task within the allowed number of iterations."
-        )

api/core/agent/fc_agent_runner.py

@@ -5,7 +5,6 @@ from copy import deepcopy
 from typing import Any, Union
 
 from core.agent.base_agent_runner import BaseAgentRunner
-from core.agent.errors import AgentMaxIterationError
 from core.app.apps.base_app_queue_manager import PublishFrom
 from core.app.entities.queue_entities import QueueAgentThoughtEvent, QueueMessageEndEvent, QueueMessageFileEvent
 from core.prompt.agent_history_prompt_transform import AgentHistoryPromptTransform
@@ -26,6 +25,7 @@ from dify_graph.model_runtime.entities import (
     UserPromptMessage,
 )
 from dify_graph.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
+from dify_graph.nodes.agent.exc import AgentMaxIterationError
 from models.model import Message
 
 logger = logging.getLogger(__name__)

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

@@ -69,7 +69,7 @@ from dify_graph.entities.pause_reason import HumanInputRequired
 from dify_graph.enums import WorkflowExecutionStatus
 from dify_graph.model_runtime.entities.llm_entities import LLMUsage
 from dify_graph.model_runtime.utils.encoders import jsonable_encoder
-from dify_graph.nodes import BuiltinNodeTypes
+from dify_graph.nodes import NodeType
 from dify_graph.repositories.draft_variable_repository import DraftVariableSaverFactory
 from dify_graph.runtime import GraphRuntimeState
 from dify_graph.system_variable import SystemVariable
@@ -357,7 +357,7 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
     ) -> Generator[StreamResponse, None, None]:
         """Handle node succeeded events."""
         # Record files if it's an answer node or end node
-        if event.node_type in [BuiltinNodeTypes.ANSWER, BuiltinNodeTypes.END, BuiltinNodeTypes.LLM]:
+        if event.node_type in [NodeType.ANSWER, NodeType.END, NodeType.LLM]:
             self._recorded_files.extend(
                 self._workflow_response_converter.fetch_files_from_node_outputs(event.outputs or {})
             )

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

@@ -48,13 +48,12 @@ from core.app.entities.task_entities import (
 from core.plugin.impl.datasource import PluginDatasourceManager
 from core.tools.entities.tool_entities import ToolProviderType
 from core.tools.tool_manager import ToolManager
-from core.trigger.constants import TRIGGER_PLUGIN_NODE_TYPE
 from core.trigger.trigger_manager import TriggerManager
 from core.workflow.workflow_entry import WorkflowEntry
 from dify_graph.entities.pause_reason import HumanInputRequired
 from dify_graph.entities.workflow_start_reason import WorkflowStartReason
 from dify_graph.enums import (
-    BuiltinNodeTypes,
+    NodeType,
     SystemVariableKey,
     WorkflowExecutionStatus,
     WorkflowNodeExecutionMetadataKey,
@@ -443,7 +442,7 @@ class WorkflowResponseConverter:
         event: QueueNodeStartedEvent,
         task_id: str,
     ) -> NodeStartStreamResponse | None:
-        if event.node_type in {BuiltinNodeTypes.ITERATION, BuiltinNodeTypes.LOOP}:
+        if event.node_type in {NodeType.ITERATION, NodeType.LOOP}:
             return None
         run_id = self._ensure_workflow_run_id()
         snapshot = self._store_snapshot(event)
@@ -465,13 +464,13 @@ class WorkflowResponseConverter:
         )
 
         try:
-            if event.node_type == BuiltinNodeTypes.TOOL:
+            if event.node_type == NodeType.TOOL:
                 response.data.extras["icon"] = ToolManager.get_tool_icon(
                     tenant_id=self._application_generate_entity.app_config.tenant_id,
                     provider_type=ToolProviderType(event.provider_type),
                     provider_id=event.provider_id,
                 )
-            elif event.node_type == BuiltinNodeTypes.DATASOURCE:
+            elif event.node_type == NodeType.DATASOURCE:
                 manager = PluginDatasourceManager()
                 provider_entity = manager.fetch_datasource_provider(
                     self._application_generate_entity.app_config.tenant_id,
@@ -480,7 +479,7 @@ class WorkflowResponseConverter:
                 response.data.extras["icon"] = provider_entity.declaration.identity.generate_datasource_icon_url(
                     self._application_generate_entity.app_config.tenant_id
                 )
-            elif event.node_type == TRIGGER_PLUGIN_NODE_TYPE:
+            elif event.node_type == NodeType.TRIGGER_PLUGIN:
                 response.data.extras["icon"] = TriggerManager.get_trigger_plugin_icon(
                     self._application_generate_entity.app_config.tenant_id,
                     event.provider_id,
@@ -497,7 +496,7 @@ class WorkflowResponseConverter:
         event: QueueNodeSucceededEvent | QueueNodeFailedEvent | QueueNodeExceptionEvent,
         task_id: str,
     ) -> NodeFinishStreamResponse | None:
-        if event.node_type in {BuiltinNodeTypes.ITERATION, BuiltinNodeTypes.LOOP}:
+        if event.node_type in {NodeType.ITERATION, NodeType.LOOP}:
             return None
         run_id = self._ensure_workflow_run_id()
         snapshot = self._pop_snapshot(event.node_execution_id)
@@ -555,7 +554,7 @@ class WorkflowResponseConverter:
         event: QueueNodeRetryEvent,
         task_id: str,
     ) -> NodeRetryStreamResponse | None:
-        if event.node_type in {BuiltinNodeTypes.ITERATION, BuiltinNodeTypes.LOOP}:
+        if event.node_type in {NodeType.ITERATION, NodeType.LOOP}:
             return None
         run_id = self._ensure_workflow_run_id()
 
@@ -613,7 +612,7 @@ class WorkflowResponseConverter:
             data=IterationNodeStartStreamResponse.Data(
                 id=event.node_id,
                 node_id=event.node_id,
-                node_type=event.node_type,
+                node_type=event.node_type.value,
                 title=event.node_title,
                 created_at=int(time.time()),
                 extras={},
@@ -636,7 +635,7 @@ class WorkflowResponseConverter:
             data=IterationNodeNextStreamResponse.Data(
                 id=event.node_id,
                 node_id=event.node_id,
-                node_type=event.node_type,
+                node_type=event.node_type.value,
                 title=event.node_title,
                 index=event.index,
                 created_at=int(time.time()),
@@ -663,7 +662,7 @@ class WorkflowResponseConverter:
             data=IterationNodeCompletedStreamResponse.Data(
                 id=event.node_id,
                 node_id=event.node_id,
-                node_type=event.node_type,
+                node_type=event.node_type.value,
                 title=event.node_title,
                 outputs=new_outputs,
                 outputs_truncated=outputs_truncated,
@@ -693,7 +692,7 @@ class WorkflowResponseConverter:
             data=LoopNodeStartStreamResponse.Data(
                 id=event.node_id,
                 node_id=event.node_id,
-                node_type=event.node_type,
+                node_type=event.node_type.value,
                 title=event.node_title,
                 created_at=int(time.time()),
                 extras={},
@@ -716,7 +715,7 @@ class WorkflowResponseConverter:
             data=LoopNodeNextStreamResponse.Data(
                 id=event.node_id,
                 node_id=event.node_id,
-                node_type=event.node_type,
+                node_type=event.node_type.value,
                 title=event.node_title,
                 index=event.index,
                 # The `pre_loop_output` field is not utilized by the frontend.
@@ -745,7 +744,7 @@ class WorkflowResponseConverter:
             data=LoopNodeCompletedStreamResponse.Data(
                 id=event.node_id,
                 node_id=event.node_id,
-                node_type=event.node_type,
+                node_type=event.node_type.value,
                 title=event.node_title,
                 outputs=new_outputs,
                 outputs_truncated=outputs_truncated,

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

@@ -12,7 +12,7 @@ from core.app.entities.app_invoke_entities import (
     build_dify_run_context,
 )
 from core.app.workflow.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
-from core.workflow.node_factory import DifyNodeFactory, get_default_root_node_id
+from core.workflow.node_factory import DifyNodeFactory
 from core.workflow.workflow_entry import WorkflowEntry
 from dify_graph.entities.graph_init_params import GraphInitParams
 from dify_graph.enums import WorkflowType
@@ -274,8 +274,6 @@ class PipelineRunner(WorkflowBasedAppRunner):
             graph_init_params=graph_init_params,
             graph_runtime_state=graph_runtime_state,
         )
-        if start_node_id is None:
-            start_node_id = get_default_root_node_id(graph_config)
         graph = Graph.init(graph_config=graph_config, node_factory=node_factory, root_node_id=start_node_id)
 
         if not graph:

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

@@ -159,12 +159,9 @@ class WorkflowAppGenerator(BaseAppGenerator):
 
         inputs: Mapping[str, Any] = args["inputs"]
 
-        extras: dict[str, Any] = {
+        extras = {
             **extract_external_trace_id_from_args(args),
         }
-        parent_trace_context = args.get("_parent_trace_context")
-        if parent_trace_context:
-            extras["parent_trace_context"] = parent_trace_context
         workflow_run_id = str(workflow_run_id or uuid.uuid4())
         # FIXME (Yeuoly): we need to remove the SKIP_PREPARE_USER_INPUTS_KEY from the args
         # trigger shouldn't prepare user inputs

api/core/app/apps/workflow_app_runner.py

@@ -3,10 +3,7 @@ import time
 from collections.abc import Mapping, Sequence
 from typing import Any, cast
 
-from pydantic import ValidationError
-
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
-from core.app.entities.agent_strategy import AgentStrategyInfo
 from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom, build_dify_run_context
 from core.app.entities.queue_entities import (
     AppQueueEvent,
@@ -32,8 +29,7 @@ from core.app.entities.queue_entities import (
     QueueWorkflowStartedEvent,
     QueueWorkflowSucceededEvent,
 )
-from core.rag.entities.citation_metadata import RetrievalSourceMetadata
-from core.workflow.node_factory import DifyNodeFactory, get_default_root_node_id, resolve_workflow_node_class
+from core.workflow.node_factory import DifyNodeFactory
 from core.workflow.workflow_entry import WorkflowEntry
 from dify_graph.entities import GraphInitParams
 from dify_graph.entities.graph_config import NodeConfigDictAdapter
@@ -67,6 +63,7 @@ from dify_graph.graph_events import (
     NodeRunSucceededEvent,
 )
 from dify_graph.graph_events.graph import GraphRunAbortedEvent
+from dify_graph.nodes.node_mapping import NODE_TYPE_CLASSES_MAPPING
 from dify_graph.runtime import GraphRuntimeState, VariablePool
 from dify_graph.system_variable import SystemVariable
 from dify_graph.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader, load_into_variable_pool
@@ -140,9 +137,6 @@ class WorkflowBasedAppRunner:
             graph_runtime_state=graph_runtime_state,
         )
 
-        if root_node_id is None:
-            root_node_id = get_default_root_node_id(graph_config)
-
         # init graph
         graph = Graph.init(graph_config=graph_config, node_factory=node_factory, root_node_id=root_node_id)
 
@@ -314,7 +308,7 @@ class WorkflowBasedAppRunner:
         # Get node class
         node_type = target_node_config["data"].type
         node_version = str(target_node_config["data"].version)
-        node_cls = resolve_workflow_node_class(node_type=node_type, node_version=node_version)
+        node_cls = NODE_TYPE_CLASSES_MAPPING[node_type][node_version]
 
         # Use the variable pool from graph_runtime_state instead of creating a new one
         variable_pool = graph_runtime_state.variable_pool
@@ -342,18 +336,6 @@ class WorkflowBasedAppRunner:
 
         return graph, variable_pool
 
-    @staticmethod
-    def _build_agent_strategy_info(event: NodeRunStartedEvent) -> AgentStrategyInfo | None:
-        raw_agent_strategy = event.extras.get("agent_strategy")
-        if raw_agent_strategy is None:
-            return None
-
-        try:
-            return AgentStrategyInfo.model_validate(raw_agent_strategy)
-        except ValidationError:
-            logger.warning("Invalid agent strategy payload for node %s", event.node_id, exc_info=True)
-            return None
-
     def _handle_event(self, workflow_entry: WorkflowEntry, event: GraphEngineEvent):
         """
         Handle event
@@ -439,7 +421,7 @@ class WorkflowBasedAppRunner:
                     start_at=event.start_at,
                     in_iteration_id=event.in_iteration_id,
                     in_loop_id=event.in_loop_id,
-                    agent_strategy=self._build_agent_strategy_info(event),
+                    agent_strategy=event.agent_strategy,
                     provider_type=event.provider_type,
                     provider_id=event.provider_id,
                 )
@@ -508,9 +490,7 @@ class WorkflowBasedAppRunner:
         elif isinstance(event, NodeRunRetrieverResourceEvent):
             self._publish_event(
                 QueueRetrieverResourcesEvent(
-                    retriever_resources=[
-                        RetrievalSourceMetadata.model_validate(resource) for resource in event.retriever_resources
-                    ],
+                    retriever_resources=event.retriever_resources,
                     in_iteration_id=event.in_iteration_id,
                     in_loop_id=event.in_loop_id,
                 )

api/core/app/entities/__init__.py

@@ -1,3 +0,0 @@
-from .agent_strategy import AgentStrategyInfo
-
-__all__ = ["AgentStrategyInfo"]

api/core/app/entities/agent_strategy.py

@@ -1,8 +0,0 @@
-from pydantic import BaseModel, ConfigDict
-
-
-class AgentStrategyInfo(BaseModel):
-    name: str
-    icon: str | None = None
-
-    model_config = ConfigDict(extra="forbid")

api/core/app/entities/queue_entities.py

@@ -5,12 +5,13 @@ from typing import Any
 
 from pydantic import BaseModel, ConfigDict, Field
 
-from core.app.entities.agent_strategy import AgentStrategyInfo
 from core.rag.entities.citation_metadata import RetrievalSourceMetadata
+from dify_graph.entities import AgentNodeStrategyInit
 from dify_graph.entities.pause_reason import PauseReason
 from dify_graph.entities.workflow_start_reason import WorkflowStartReason
-from dify_graph.enums import NodeType, WorkflowNodeExecutionMetadataKey
+from dify_graph.enums import WorkflowNodeExecutionMetadataKey
 from dify_graph.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk
+from dify_graph.nodes import NodeType
 
 
 class QueueEvent(StrEnum):
@@ -313,7 +314,7 @@ class QueueNodeStartedEvent(AppQueueEvent):
     in_iteration_id: str | None = None
     in_loop_id: str | None = None
     start_at: datetime
-    agent_strategy: AgentStrategyInfo | None = None
+    agent_strategy: AgentNodeStrategyInit | None = None
 
     # FIXME(-LAN-): only for ToolNode, need to refactor
     provider_type: str  # should be a core.tools.entities.tool_entities.ToolProviderType

api/core/app/entities/task_entities.py

@@ -4,8 +4,8 @@ from typing import Any
 
 from pydantic import BaseModel, ConfigDict, Field
 
-from core.app.entities.agent_strategy import AgentStrategyInfo
 from core.rag.entities.citation_metadata import RetrievalSourceMetadata
+from dify_graph.entities import AgentNodeStrategyInit
 from dify_graph.entities.workflow_start_reason import WorkflowStartReason
 from dify_graph.enums import WorkflowExecutionStatus, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
 from dify_graph.model_runtime.entities.llm_entities import LLMResult, LLMUsage
@@ -349,7 +349,7 @@ class NodeStartStreamResponse(StreamResponse):
         extras: dict[str, object] = Field(default_factory=dict)
         iteration_id: str | None = None
         loop_id: str | None = None
-        agent_strategy: AgentStrategyInfo | None = None
+        agent_strategy: AgentNodeStrategyInit | None = None
 
     event: StreamEvent = StreamEvent.NODE_STARTED
     workflow_run_id: str

api/core/app/layers/conversation_variable_persist_layer.py

@@ -2,7 +2,7 @@ import logging
 
 from dify_graph.constants import CONVERSATION_VARIABLE_NODE_ID
 from dify_graph.conversation_variable_updater import ConversationVariableUpdater
-from dify_graph.enums import BuiltinNodeTypes
+from dify_graph.enums import NodeType
 from dify_graph.graph_engine.layers.base import GraphEngineLayer
 from dify_graph.graph_events import GraphEngineEvent, NodeRunSucceededEvent
 from dify_graph.nodes.variable_assigner.common import helpers as common_helpers
@@ -22,7 +22,7 @@ class ConversationVariablePersistenceLayer(GraphEngineLayer):
     def on_event(self, event: GraphEngineEvent) -> None:
         if not isinstance(event, NodeRunSucceededEvent):
             return
-        if event.node_type != BuiltinNodeTypes.VARIABLE_ASSIGNER:
+        if event.node_type != NodeType.VARIABLE_ASSIGNER:
             return
         if self.graph_runtime_state is None:
             return

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

@@ -12,7 +12,7 @@ from typing_extensions import override
 from core.app.llm import deduct_llm_quota, ensure_llm_quota_available
 from core.errors.error import QuotaExceededError
 from core.model_manager import ModelInstance
-from dify_graph.enums import BuiltinNodeTypes
+from dify_graph.enums import NodeType
 from dify_graph.graph_engine.entities.commands import AbortCommand, CommandType
 from dify_graph.graph_engine.layers.base import GraphEngineLayer
 from dify_graph.graph_events import GraphEngineEvent, GraphNodeEventBase
@@ -113,11 +113,11 @@ class LLMQuotaLayer(GraphEngineLayer):
     def _extract_model_instance(node: Node) -> ModelInstance | None:
         try:
             match node.node_type:
-                case BuiltinNodeTypes.LLM:
+                case NodeType.LLM:
                     return cast("LLMNode", node).model_instance
-                case BuiltinNodeTypes.PARAMETER_EXTRACTOR:
+                case NodeType.PARAMETER_EXTRACTOR:
                     return cast("ParameterExtractorNode", node).model_instance
-                case BuiltinNodeTypes.QUESTION_CLASSIFIER:
+                case NodeType.QUESTION_CLASSIFIER:
                     return cast("QuestionClassifierNode", node).model_instance
                 case _:
                     return None

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

@@ -16,7 +16,7 @@ from opentelemetry.trace import Span, SpanKind, Tracer, get_tracer, set_span_in_
 from typing_extensions import override
 
 from configs import dify_config
-from dify_graph.enums import BuiltinNodeTypes, NodeType
+from dify_graph.enums import NodeType
 from dify_graph.graph_engine.layers.base import GraphEngineLayer
 from dify_graph.graph_events import GraphNodeEventBase
 from dify_graph.nodes.base.node import Node
@@ -74,13 +74,16 @@ class ObservabilityLayer(GraphEngineLayer):
     def _build_parser_registry(self) -> None:
         """Initialize parser registry for node types."""
         self._parsers = {
-            BuiltinNodeTypes.TOOL: ToolNodeOTelParser(),
-            BuiltinNodeTypes.LLM: LLMNodeOTelParser(),
-            BuiltinNodeTypes.KNOWLEDGE_RETRIEVAL: RetrievalNodeOTelParser(),
+            NodeType.TOOL: ToolNodeOTelParser(),
+            NodeType.LLM: LLMNodeOTelParser(),
+            NodeType.KNOWLEDGE_RETRIEVAL: RetrievalNodeOTelParser(),
         }
 
     def _get_parser(self, node: Node) -> NodeOTelParser:
-        return self._parsers.get(node.node_type, self._default_parser)
+        node_type = getattr(node, "node_type", None)
+        if isinstance(node_type, NodeType):
+            return self._parsers.get(node_type, self._default_parser)
+        return self._default_parser
 
     @override
     def on_graph_start(self) -> None:

api/core/callback_handler/index_tool_callback_handler.py

@@ -12,7 +12,6 @@ from core.rag.models.document import Document
 from extensions.ext_database import db
 from models.dataset import ChildChunk, DatasetQuery, DocumentSegment
 from models.dataset import Document as DatasetDocument
-from models.enums import CreatorUserRole
 
 _logger = logging.getLogger(__name__)
 
@@ -39,9 +38,7 @@ class DatasetIndexToolCallbackHandler:
             source="app",
             source_app_id=self._app_id,
             created_by_role=(
-                CreatorUserRole.ACCOUNT
-                if self._invoke_from in {InvokeFrom.EXPLORE, InvokeFrom.DEBUGGER}
-                else CreatorUserRole.END_USER
+                "account" if self._invoke_from in {InvokeFrom.EXPLORE, InvokeFrom.DEBUGGER} else "end_user"
             ),
             created_by=self._user_id,
         )

api/core/datasource/datasource_manager.py

@@ -24,12 +24,12 @@ from core.datasource.utils.message_transformer import DatasourceFileMessageTrans
 from core.datasource.website_crawl.website_crawl_provider import WebsiteCrawlDatasourcePluginProviderController
 from core.db.session_factory import session_factory
 from core.plugin.impl.datasource import PluginDatasourceManager
-from core.workflow.nodes.datasource.entities import DatasourceParameter, OnlineDriveDownloadFileParam
 from dify_graph.entities.workflow_node_execution import WorkflowNodeExecutionStatus
 from dify_graph.enums import WorkflowNodeExecutionMetadataKey
 from dify_graph.file import File
 from dify_graph.file.enums import FileTransferMethod, FileType
 from dify_graph.node_events import NodeRunResult, StreamChunkEvent, StreamCompletedEvent
+from dify_graph.repositories.datasource_manager_protocol import DatasourceParameter, OnlineDriveDownloadFileParam
 from factories import file_factory
 from models.model import UploadFile
 from models.tools import ToolFile

api/core/llm_generator/entities.py

@@ -9,7 +9,6 @@ class RuleGeneratePayload(BaseModel):
     instruction: str = Field(..., description="Rule generation instruction")
     model_config_data: ModelConfig = Field(..., alias="model_config", description="Model configuration")
     no_variable: bool = Field(default=False, description="Whether to exclude variables")
-    app_id: str | None = Field(default=None, description="App ID for prompt generation tracing")
 
 
 class RuleCodeGeneratePayload(RuleGeneratePayload):
@@ -19,4 +18,3 @@ class RuleCodeGeneratePayload(RuleGeneratePayload):
 class RuleStructuredOutputPayload(BaseModel):
     instruction: str = Field(..., description="Structured output generation instruction")
     model_config_data: ModelConfig = Field(..., alias="model_config", description="Model configuration")
-    app_id: str | None = Field(default=None, description="App ID for prompt generation tracing")

api/core/ops/aliyun_trace/aliyun_trace.py

@@ -58,7 +58,7 @@ from core.ops.entities.trace_entity import (
 )
 from core.repositories import DifyCoreRepositoryFactory
 from dify_graph.entities import WorkflowNodeExecution
-from dify_graph.enums import BuiltinNodeTypes, WorkflowNodeExecutionMetadataKey
+from dify_graph.enums import NodeType, WorkflowNodeExecutionMetadataKey
 from extensions.ext_database import db
 from models import WorkflowNodeExecutionTriggeredFrom
 
@@ -302,11 +302,11 @@ class AliyunDataTrace(BaseTraceInstance):
         self, node_execution: WorkflowNodeExecution, trace_info: WorkflowTraceInfo, trace_metadata: TraceMetadata
     ):
         try:
-            if node_execution.node_type == BuiltinNodeTypes.LLM:
+            if node_execution.node_type == NodeType.LLM:
                 node_span = self.build_workflow_llm_span(trace_info, node_execution, trace_metadata)
-            elif node_execution.node_type == BuiltinNodeTypes.KNOWLEDGE_RETRIEVAL:
+            elif node_execution.node_type == NodeType.KNOWLEDGE_RETRIEVAL:
                 node_span = self.build_workflow_retrieval_span(trace_info, node_execution, trace_metadata)
-            elif node_execution.node_type == BuiltinNodeTypes.TOOL:
+            elif node_execution.node_type == NodeType.TOOL:
                 node_span = self.build_workflow_tool_span(trace_info, node_execution, trace_metadata)
             else:
                 node_span = self.build_workflow_task_span(trace_info, node_execution, trace_metadata)

api/core/ops/arize_phoenix_trace/arize_phoenix_trace.py

@@ -155,8 +155,8 @@ def wrap_span_metadata(metadata, **kwargs):
     return metadata
 
 
-# Mapping from built-in node type strings to OpenInference span kinds.
-# Node types not listed here default to CHAIN.
+# Mapping from NodeType string values to OpenInference span kinds.
+# NodeType values not listed here default to CHAIN.
 _NODE_TYPE_TO_SPAN_KIND: dict[str, OpenInferenceSpanKindValues] = {
     "llm": OpenInferenceSpanKindValues.LLM,
     "knowledge-retrieval": OpenInferenceSpanKindValues.RETRIEVER,
@@ -168,7 +168,7 @@ _NODE_TYPE_TO_SPAN_KIND: dict[str, OpenInferenceSpanKindValues] = {
 def _get_node_span_kind(node_type: str) -> OpenInferenceSpanKindValues:
     """Return the OpenInference span kind for a given workflow node type.
 
-    Covers every built-in node type string. Nodes that do not have a
+    Covers every ``NodeType`` enum value.  Nodes that do not have a
     specialised span kind (e.g. ``start``, ``end``, ``if-else``,
     ``code``, ``loop``, ``iteration``, etc.) are mapped to ``CHAIN``.
     """

api/core/ops/entities/trace_entity.py

@@ -9,8 +9,8 @@ from pydantic import BaseModel, ConfigDict, field_serializer, field_validator
 class BaseTraceInfo(BaseModel):
     message_id: str | None = None
     message_data: Any | None = None
-    inputs: Union[str, dict[str, Any], list[Any]] | None = None
-    outputs: Union[str, dict[str, Any], list[Any]] | None = None
+    inputs: Union[str, dict[str, Any], list] | None = None
+    outputs: Union[str, dict[str, Any], list] | None = None
     start_time: datetime | None = None
     end_time: datetime | None = None
     metadata: dict[str, Any]
@@ -18,7 +18,7 @@ class BaseTraceInfo(BaseModel):
 
     @field_validator("inputs", "outputs")
     @classmethod
-    def ensure_type(cls, v: str | dict[str, Any] | list[Any] | None) -> str | dict[str, Any] | list[Any] | None:
+    def ensure_type(cls, v):
         if v is None:
             return None
         if isinstance(v, str | dict | list):
@@ -27,48 +27,6 @@ class BaseTraceInfo(BaseModel):
 
     model_config = ConfigDict(protected_namespaces=())
 
-    @property
-    def resolved_trace_id(self) -> str | None:
-        """Get trace_id with intelligent fallback.
-
-        Priority:
-        1. External trace_id (from X-Trace-Id header)
-        2. workflow_run_id (if this trace type has it)
-        3. message_id (as final fallback)
-        """
-        if self.trace_id:
-            return self.trace_id
-
-        # Try workflow_run_id (only exists on workflow-related traces)
-        workflow_run_id = getattr(self, "workflow_run_id", None)
-        if workflow_run_id:
-            return workflow_run_id
-
-        # Final fallback to message_id
-        return str(self.message_id) if self.message_id else None
-
-    @property
-    def resolved_parent_context(self) -> tuple[str | None, str | None]:
-        """Resolve cross-workflow parent linking from metadata.
-
-        Extracts typed parent IDs from the untyped ``parent_trace_context``
-        metadata dict (set by tool_node when invoking nested workflows).
-
-        Returns:
-            (trace_correlation_override, parent_span_id_source) where
-            trace_correlation_override is the outer workflow_run_id and
-            parent_span_id_source is the outer node_execution_id.
-        """
-        parent_ctx = self.metadata.get("parent_trace_context")
-        if not isinstance(parent_ctx, dict):
-            return None, None
-        trace_override = parent_ctx.get("parent_workflow_run_id")
-        parent_span = parent_ctx.get("parent_node_execution_id")
-        return (
-            trace_override if isinstance(trace_override, str) else None,
-            parent_span if isinstance(parent_span, str) else None,
-        )
-
     @field_serializer("start_time", "end_time")
     def serialize_datetime(self, dt: datetime | None) -> str | None:
         if dt is None:
@@ -90,10 +48,7 @@ class WorkflowTraceInfo(BaseTraceInfo):
     workflow_run_version: str
     error: str | None = None
     total_tokens: int
-    prompt_tokens: int | None = None
-    completion_tokens: int | None = None
     file_list: list[str]
-    invoked_by: str | None = None
     query: str
     metadata: dict[str, Any]
 
@@ -104,7 +59,7 @@ class MessageTraceInfo(BaseTraceInfo):
     answer_tokens: int
     total_tokens: int
     error: str | None = None
-    file_list: Union[str, dict[str, Any], list[Any]] | None = None
+    file_list: Union[str, dict[str, Any], list] | None = None
     message_file_data: Any | None = None
     conversation_mode: str
     gen_ai_server_time_to_first_token: float | None = None
@@ -151,7 +106,7 @@ class ToolTraceInfo(BaseTraceInfo):
     tool_config: dict[str, Any]
     time_cost: Union[int, float]
     tool_parameters: dict[str, Any]
-    file_url: Union[str, None, list[str]] = None
+    file_url: Union[str, None, list] = None
 
 
 class GenerateNameTraceInfo(BaseTraceInfo):
@@ -159,79 +114,6 @@ class GenerateNameTraceInfo(BaseTraceInfo):
     tenant_id: str
 
 
-class PromptGenerationTraceInfo(BaseTraceInfo):
-    """Trace information for prompt generation operations (rule-generate, code-generate, etc.)."""
-
-    tenant_id: str
-    user_id: str
-    app_id: str | None = None
-
-    operation_type: str
-    instruction: str
-
-    prompt_tokens: int
-    completion_tokens: int
-    total_tokens: int
-
-    model_provider: str
-    model_name: str
-
-    latency: float
-
-    total_price: float | None = None
-    currency: str | None = None
-
-    error: str | None = None
-
-    model_config = ConfigDict(protected_namespaces=())
-
-
-class WorkflowNodeTraceInfo(BaseTraceInfo):
-    workflow_id: str
-    workflow_run_id: str
-    tenant_id: str
-    node_execution_id: str
-    node_id: str
-    node_type: str
-    title: str
-
-    status: str
-    error: str | None = None
-    elapsed_time: float
-
-    index: int
-    predecessor_node_id: str | None = None
-
-    total_tokens: int = 0
-    total_price: float = 0.0
-    currency: str | None = None
-
-    model_provider: str | None = None
-    model_name: str | None = None
-    prompt_tokens: int | None = None
-    completion_tokens: int | None = None
-
-    tool_name: str | None = None
-
-    iteration_id: str | None = None
-    iteration_index: int | None = None
-    loop_id: str | None = None
-    loop_index: int | None = None
-    parallel_id: str | None = None
-
-    node_inputs: Mapping[str, Any] | None = None
-    node_outputs: Mapping[str, Any] | None = None
-    process_data: Mapping[str, Any] | None = None
-
-    invoked_by: str | None = None
-
-    model_config = ConfigDict(protected_namespaces=())
-
-
-class DraftNodeExecutionTrace(WorkflowNodeTraceInfo):
-    pass
-
-
 class TaskData(BaseModel):
     app_id: str
     trace_info_type: str
@@ -246,31 +128,11 @@ trace_info_info_map = {
     "DatasetRetrievalTraceInfo": DatasetRetrievalTraceInfo,
     "ToolTraceInfo": ToolTraceInfo,
     "GenerateNameTraceInfo": GenerateNameTraceInfo,
-    "PromptGenerationTraceInfo": PromptGenerationTraceInfo,
-    "WorkflowNodeTraceInfo": WorkflowNodeTraceInfo,
-    "DraftNodeExecutionTrace": DraftNodeExecutionTrace,
 }
 
 
-class OperationType(StrEnum):
-    """Operation type for token metric labels.
-
-    Used as a metric attribute on ``dify.tokens.input`` / ``dify.tokens.output``
-    counters so consumers can break down token usage by operation.
-    """
-
-    WORKFLOW = "workflow"
-    NODE_EXECUTION = "node_execution"
-    MESSAGE = "message"
-    RULE_GENERATE = "rule_generate"
-    CODE_GENERATE = "code_generate"
-    STRUCTURED_OUTPUT = "structured_output"
-    INSTRUCTION_MODIFY = "instruction_modify"
-
-
 class TraceTaskName(StrEnum):
     CONVERSATION_TRACE = "conversation"
-    DRAFT_NODE_EXECUTION_TRACE = "draft_node_execution"
     WORKFLOW_TRACE = "workflow"
     MESSAGE_TRACE = "message"
     MODERATION_TRACE = "moderation"
@@ -278,6 +140,4 @@ class TraceTaskName(StrEnum):
     DATASET_RETRIEVAL_TRACE = "dataset_retrieval"
     TOOL_TRACE = "tool"
     GENERATE_NAME_TRACE = "generate_conversation_name"
-    PROMPT_GENERATION_TRACE = "prompt_generation"
-    NODE_EXECUTION_TRACE = "node_execution"
     DATASOURCE_TRACE = "datasource"

api/core/ops/langfuse_trace/langfuse_trace.py

@@ -28,7 +28,7 @@ from core.ops.langfuse_trace.entities.langfuse_trace_entity import (
 )
 from core.ops.utils import filter_none_values
 from core.repositories import DifyCoreRepositoryFactory
-from dify_graph.enums import BuiltinNodeTypes
+from dify_graph.enums import NodeType
 from extensions.ext_database import db
 from models import EndUser, WorkflowNodeExecutionTriggeredFrom
 from models.enums import MessageStatus
@@ -141,7 +141,7 @@ class LangFuseDataTrace(BaseTraceInstance):
             node_name = node_execution.title
             node_type = node_execution.node_type
             status = node_execution.status
-            if node_type == BuiltinNodeTypes.LLM:
+            if node_type == NodeType.LLM:
                 inputs = node_execution.process_data.get("prompts", {}) if node_execution.process_data else {}
             else:
                 inputs = node_execution.inputs or {}

api/core/ops/langsmith_trace/langsmith_trace.py

@@ -28,7 +28,7 @@ from core.ops.langsmith_trace.entities.langsmith_trace_entity import (
 )
 from core.ops.utils import filter_none_values, generate_dotted_order
 from core.repositories import DifyCoreRepositoryFactory
-from dify_graph.enums import BuiltinNodeTypes, WorkflowNodeExecutionMetadataKey
+from dify_graph.enums import NodeType, WorkflowNodeExecutionMetadataKey
 from extensions.ext_database import db
 from models import EndUser, MessageFile, WorkflowNodeExecutionTriggeredFrom
 
@@ -163,7 +163,7 @@ class LangSmithDataTrace(BaseTraceInstance):
             node_name = node_execution.title
             node_type = node_execution.node_type
             status = node_execution.status
-            if node_type == BuiltinNodeTypes.LLM:
+            if node_type == NodeType.LLM:
                 inputs = node_execution.process_data.get("prompts", {}) if node_execution.process_data else {}
             else:
                 inputs = node_execution.inputs or {}
@@ -197,7 +197,7 @@ class LangSmithDataTrace(BaseTraceInstance):
                         "ls_model_name": process_data.get("model_name", ""),
                     }
                 )
-            elif node_type == BuiltinNodeTypes.KNOWLEDGE_RETRIEVAL:
+            elif node_type == NodeType.KNOWLEDGE_RETRIEVAL:
                 run_type = LangSmithRunType.retriever
             else:
                 run_type = LangSmithRunType.tool

api/core/ops/mlflow_trace/mlflow_trace.py

@@ -23,7 +23,7 @@ from core.ops.entities.trace_entity import (
     TraceTaskName,
     WorkflowTraceInfo,
 )
-from dify_graph.enums import BuiltinNodeTypes
+from dify_graph.enums import NodeType
 from extensions.ext_database import db
 from models import EndUser
 from models.workflow import WorkflowNodeExecutionModel
@@ -145,10 +145,10 @@ class MLflowDataTrace(BaseTraceInstance):
                     "app_name": node.title,
                 }
 
-                if node.node_type in (BuiltinNodeTypes.LLM, BuiltinNodeTypes.QUESTION_CLASSIFIER):
+                if node.node_type in (NodeType.LLM, NodeType.QUESTION_CLASSIFIER):
                     inputs, llm_attributes = self._parse_llm_inputs_and_attributes(node)
                     attributes.update(llm_attributes)
-                elif node.node_type == BuiltinNodeTypes.HTTP_REQUEST:
+                elif node.node_type == NodeType.HTTP_REQUEST:
                     inputs = node.process_data  # contains request URL
 
                 if not inputs:
@@ -180,9 +180,9 @@ class MLflowDataTrace(BaseTraceInstance):
                 # End node span
                 finished_at = node.created_at + timedelta(seconds=node.elapsed_time)
                 outputs = json.loads(node.outputs) if node.outputs else {}
-                if node.node_type == BuiltinNodeTypes.KNOWLEDGE_RETRIEVAL:
+                if node.node_type == NodeType.KNOWLEDGE_RETRIEVAL:
                     outputs = self._parse_knowledge_retrieval_outputs(outputs)
-                elif node.node_type == BuiltinNodeTypes.LLM:
+                elif node.node_type == NodeType.LLM:
                     outputs = outputs.get("text", outputs)
                 node_span.end(
                     outputs=outputs,
@@ -471,13 +471,13 @@ class MLflowDataTrace(BaseTraceInstance):
     def _get_node_span_type(self, node_type: str) -> str:
         """Map Dify node types to MLflow span types"""
         node_type_mapping = {
-            BuiltinNodeTypes.LLM: SpanType.LLM,
-            BuiltinNodeTypes.QUESTION_CLASSIFIER: SpanType.LLM,
-            BuiltinNodeTypes.KNOWLEDGE_RETRIEVAL: SpanType.RETRIEVER,
-            BuiltinNodeTypes.TOOL: SpanType.TOOL,
-            BuiltinNodeTypes.CODE: SpanType.TOOL,
-            BuiltinNodeTypes.HTTP_REQUEST: SpanType.TOOL,
-            BuiltinNodeTypes.AGENT: SpanType.AGENT,
+            NodeType.LLM: SpanType.LLM,
+            NodeType.QUESTION_CLASSIFIER: SpanType.LLM,
+            NodeType.KNOWLEDGE_RETRIEVAL: SpanType.RETRIEVER,
+            NodeType.TOOL: SpanType.TOOL,
+            NodeType.CODE: SpanType.TOOL,
+            NodeType.HTTP_REQUEST: SpanType.TOOL,
+            NodeType.AGENT: SpanType.AGENT,
         }
         return node_type_mapping.get(node_type, "CHAIN")  # type: ignore[arg-type,call-overload]
 

api/core/ops/opik_trace/opik_trace.py

@@ -23,7 +23,7 @@ from core.ops.entities.trace_entity import (
     WorkflowTraceInfo,
 )
 from core.repositories import DifyCoreRepositoryFactory
-from dify_graph.enums import BuiltinNodeTypes, WorkflowNodeExecutionMetadataKey
+from dify_graph.enums import NodeType, WorkflowNodeExecutionMetadataKey
 from extensions.ext_database import db
 from models import EndUser, MessageFile, WorkflowNodeExecutionTriggeredFrom
 
@@ -187,7 +187,7 @@ class OpikDataTrace(BaseTraceInstance):
             node_name = node_execution.title
             node_type = node_execution.node_type
             status = node_execution.status
-            if node_type == BuiltinNodeTypes.LLM:
+            if node_type == NodeType.LLM:
                 inputs = node_execution.process_data.get("prompts", {}) if node_execution.process_data else {}
             else:
                 inputs = node_execution.inputs or {}

api/core/ops/ops_trace_manager.py

@@ -15,32 +15,22 @@ from sqlalchemy import select
 from sqlalchemy.orm import Session, sessionmaker
 
 from core.helper.encrypter import batch_decrypt_token, encrypt_token, obfuscated_token
-from core.ops.entities.config_entity import (
-    OPS_FILE_PATH,
-    TracingProviderEnum,
-)
+from core.ops.entities.config_entity import OPS_FILE_PATH, TracingProviderEnum
 from core.ops.entities.trace_entity import (
     DatasetRetrievalTraceInfo,
-    DraftNodeExecutionTrace,
     GenerateNameTraceInfo,
     MessageTraceInfo,
     ModerationTraceInfo,
-    PromptGenerationTraceInfo,
     SuggestedQuestionTraceInfo,
     TaskData,
     ToolTraceInfo,
     TraceTaskName,
-    WorkflowNodeTraceInfo,
     WorkflowTraceInfo,
 )
 from core.ops.utils import get_message_data
-from extensions.ext_database import db
 from extensions.ext_storage import storage
-from models.account import Tenant
-from models.dataset import Dataset
+from models.engine import db
 from models.model import App, AppModelConfig, Conversation, Message, MessageFile, TraceAppConfig
-from models.provider import Provider, ProviderCredential, ProviderModel, ProviderModelCredential, ProviderType
-from models.tools import ApiToolProvider, BuiltinToolProvider, MCPToolProvider, WorkflowToolProvider
 from models.workflow import WorkflowAppLog
 from tasks.ops_trace_task import process_trace_tasks
 
@@ -50,142 +40,9 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
-def _lookup_app_and_workspace_names(app_id: str | None, tenant_id: str | None) -> tuple[str, str]:
-    """Return (app_name, workspace_name) for the given IDs. Falls back to empty strings."""
-    app_name = ""
-    workspace_name = ""
-    if not app_id and not tenant_id:
-        return app_name, workspace_name
-    with Session(db.engine) as session:
-        if app_id:
-            name = session.scalar(select(App.name).where(App.id == app_id))
-            if name:
-                app_name = name
-        if tenant_id:
-            name = session.scalar(select(Tenant.name).where(Tenant.id == tenant_id))
-            if name:
-                workspace_name = name
-    return app_name, workspace_name
-
-
-_PROVIDER_TYPE_TO_MODEL: dict[str, type] = {
-    "builtin": BuiltinToolProvider,
-    "plugin": BuiltinToolProvider,
-    "api": ApiToolProvider,
-    "workflow": WorkflowToolProvider,
-    "mcp": MCPToolProvider,
-}
-
-
-def _lookup_credential_name(credential_id: str | None, provider_type: str | None) -> str:
-    if not credential_id:
-        return ""
-    model_cls = _PROVIDER_TYPE_TO_MODEL.get(provider_type or "")
-    if not model_cls:
-        return ""
-    with Session(db.engine) as session:
-        name = session.scalar(select(model_cls.name).where(model_cls.id == credential_id))  # type: ignore[attr-defined]
-        return str(name) if name else ""
-
-
-def _lookup_llm_credential_info(
-    tenant_id: str | None, provider: str | None, model: str | None, model_type: str | None = "llm"
-) -> tuple[str | None, str]:
-    """
-    Lookup LLM credential ID and name for the given provider and model.
-    Returns (credential_id, credential_name).
-
-    Handles async timing issues gracefully - if credential is deleted between lookups,
-    returns the ID but empty name rather than failing.
-    """
-    if not tenant_id or not provider:
-        return None, ""
-
-    try:
-        with Session(db.engine) as session:
-            # Try to find provider-level or model-level configuration
-            provider_record = session.scalar(
-                select(Provider).where(
-                    Provider.tenant_id == tenant_id,
-                    Provider.provider_name == provider,
-                    Provider.provider_type == ProviderType.CUSTOM,
-                )
-            )
-
-            if not provider_record:
-                return None, ""
-
-            # Check if there's a model-specific config
-            credential_id = None
-            credential_name = ""
-            is_model_level = False
-
-            if model:
-                # Try model-level first
-                model_record = session.scalar(
-                    select(ProviderModel).where(
-                        ProviderModel.tenant_id == tenant_id,
-                        ProviderModel.provider_name == provider,
-                        ProviderModel.model_name == model,
-                        ProviderModel.model_type == model_type,
-                    )
-                )
-
-                if model_record and model_record.credential_id:
-                    credential_id = model_record.credential_id
-                    is_model_level = True
-
-            if not credential_id and provider_record.credential_id:
-                # Fall back to provider-level credential
-                credential_id = provider_record.credential_id
-                is_model_level = False
-
-            # Lookup credential_name if we have credential_id
-            if credential_id:
-                try:
-                    if is_model_level:
-                        # Query ProviderModelCredential
-                        cred_name = session.scalar(
-                            select(ProviderModelCredential.credential_name).where(
-                                ProviderModelCredential.id == credential_id
-                            )
-                        )
-                    else:
-                        # Query ProviderCredential
-                        cred_name = session.scalar(
-                            select(ProviderCredential.credential_name).where(ProviderCredential.id == credential_id)
-                        )
-
-                    if cred_name:
-                        credential_name = str(cred_name)
-                except Exception as e:
-                    # Credential might have been deleted between lookups (async timing)
-                    # Return ID but empty name rather than failing
-                    logger.warning(
-                        "Failed to lookup credential name for credential_id=%s (provider=%s, model=%s): %s",
-                        credential_id,
-                        provider,
-                        model,
-                        str(e),
-                    )
-
-            return credential_id, credential_name
-    except Exception as e:
-        # Database query failed or other unexpected error
-        # Return empty rather than propagating error to telemetry emission
-        logger.warning(
-            "Failed to lookup LLM credential info for tenant_id=%s, provider=%s, model=%s: %s",
-            tenant_id,
-            provider,
-            model,
-            str(e),
-        )
-        return None, ""
-
-
 class OpsTraceProviderConfigMap(collections.UserDict[str, dict[str, Any]]):
-    def __getitem__(self, provider: str) -> dict[str, Any]:
-        match provider:
+    def __getitem__(self, key: str) -> dict[str, Any]:
+        match key:
             case TracingProviderEnum.LANGFUSE:
                 from core.ops.entities.config_entity import LangfuseConfig
                 from core.ops.langfuse_trace.langfuse_trace import LangFuseDataTrace
@@ -292,7 +149,7 @@ class OpsTraceProviderConfigMap(collections.UserDict[str, dict[str, Any]]):
                 }
 
             case _:
-                raise KeyError(f"Unsupported tracing provider: {provider}")
+                raise KeyError(f"Unsupported tracing provider: {key}")
 
 
 provider_config_map = OpsTraceProviderConfigMap()
@@ -457,10 +314,6 @@ class OpsTraceManager:
         if app_id is None:
             return None
 
-        # Handle storage_id format (tenant-{uuid}) - not a real app_id
-        if isinstance(app_id, str) and app_id.startswith("tenant-"):
-            return None
-
         app: App | None = db.session.query(App).where(App.id == app_id).first()
 
         if app is None:
@@ -613,6 +466,8 @@ class TraceTask:
 
     @classmethod
     def _get_workflow_run_repo(cls):
+        from repositories.factory import DifyAPIRepositoryFactory
+
         if cls._workflow_run_repo is None:
             with cls._repo_lock:
                 if cls._workflow_run_repo is None:
@@ -623,56 +478,6 @@ class TraceTask:
                     cls._workflow_run_repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker)
         return cls._workflow_run_repo
 
-    @classmethod
-    def _get_user_id_from_metadata(cls, metadata: dict[str, Any]) -> str:
-        """Extract user ID from metadata, prioritizing end_user over account.
-
-        Returns the actual user ID (end_user or account) who invoked the workflow,
-        regardless of invoke_from context.
-        """
-        # Priority 1: End user (external users via API/WebApp)
-        if user_id := metadata.get("from_end_user_id"):
-            return f"end_user:{user_id}"
-
-        # Priority 2: Account user (internal users via console/debugger)
-        if user_id := metadata.get("from_account_id"):
-            return f"account:{user_id}"
-
-        # Priority 3: User (internal users via console/debugger)
-        if user_id := metadata.get("user_id"):
-            return f"user:{user_id}"
-
-        return "anonymous"
-
-    @classmethod
-    def _calculate_workflow_token_split(cls, workflow_run_id: str, tenant_id: str) -> tuple[int, int]:
-        from dify_graph.enums import WorkflowNodeExecutionMetadataKey
-        from models.workflow import WorkflowNodeExecutionModel
-
-        with Session(db.engine) as session:
-            node_executions = session.scalars(
-                select(WorkflowNodeExecutionModel).where(
-                    WorkflowNodeExecutionModel.tenant_id == tenant_id,
-                    WorkflowNodeExecutionModel.workflow_run_id == workflow_run_id,
-                )
-            ).all()
-
-            total_prompt = 0
-            total_completion = 0
-
-            for node_exec in node_executions:
-                metadata = node_exec.execution_metadata_dict
-
-                prompt = metadata.get(WorkflowNodeExecutionMetadataKey.PROMPT_TOKENS)
-                if prompt is not None:
-                    total_prompt += prompt
-
-                completion = metadata.get(WorkflowNodeExecutionMetadataKey.COMPLETION_TOKENS)
-                if completion is not None:
-                    total_completion += completion
-
-            return (total_prompt, total_completion)
-
     def __init__(
         self,
         trace_type: Any,
@@ -693,8 +498,6 @@ class TraceTask:
         self.app_id = None
         self.trace_id = None
         self.kwargs = kwargs
-        if user_id is not None and "user_id" not in self.kwargs:
-            self.kwargs["user_id"] = user_id
         external_trace_id = kwargs.get("external_trace_id")
         if external_trace_id:
             self.trace_id = external_trace_id
@@ -708,7 +511,7 @@ class TraceTask:
             TraceTaskName.WORKFLOW_TRACE: lambda: self.workflow_trace(
                 workflow_run_id=self.workflow_run_id, conversation_id=self.conversation_id, user_id=self.user_id
             ),
-            TraceTaskName.MESSAGE_TRACE: lambda: self.message_trace(message_id=self.message_id, **self.kwargs),
+            TraceTaskName.MESSAGE_TRACE: lambda: self.message_trace(message_id=self.message_id),
             TraceTaskName.MODERATION_TRACE: lambda: self.moderation_trace(
                 message_id=self.message_id, timer=self.timer, **self.kwargs
             ),
@@ -724,9 +527,6 @@ class TraceTask:
             TraceTaskName.GENERATE_NAME_TRACE: lambda: self.generate_name_trace(
                 conversation_id=self.conversation_id, timer=self.timer, **self.kwargs
             ),
-            TraceTaskName.PROMPT_GENERATION_TRACE: lambda: self.prompt_generation_trace(**self.kwargs),
-            TraceTaskName.NODE_EXECUTION_TRACE: lambda: self.node_execution_trace(**self.kwargs),
-            TraceTaskName.DRAFT_NODE_EXECUTION_TRACE: lambda: self.draft_node_execution_trace(**self.kwargs),
         }
 
         return preprocess_map.get(self.trace_type, lambda: None)()
@@ -762,10 +562,6 @@ class TraceTask:
 
         total_tokens = workflow_run.total_tokens
 
-        prompt_tokens, completion_tokens = self._calculate_workflow_token_split(
-            workflow_run_id=workflow_run_id, tenant_id=tenant_id
-        )
-
         file_list = workflow_run_inputs.get("sys.file") or []
         query = workflow_run_inputs.get("query") or workflow_run_inputs.get("sys.query") or ""
 
@@ -786,14 +582,7 @@ class TraceTask:
                 )
                 message_id = session.scalar(message_data_stmt)
 
-        from core.telemetry.gateway import is_enterprise_telemetry_enabled
-
-        if is_enterprise_telemetry_enabled():
-            app_name, workspace_name = _lookup_app_and_workspace_names(workflow_run.app_id, tenant_id)
-        else:
-            app_name, workspace_name = "", ""
-
-        metadata: dict[str, Any] = {
+        metadata = {
             "workflow_id": workflow_id,
             "conversation_id": conversation_id,
             "workflow_run_id": workflow_run_id,
@@ -806,14 +595,8 @@ class TraceTask:
             "triggered_from": workflow_run.triggered_from,
             "user_id": user_id,
             "app_id": workflow_run.app_id,
-            "app_name": app_name,
-            "workspace_name": workspace_name,
         }
 
-        parent_trace_context = self.kwargs.get("parent_trace_context")
-        if parent_trace_context:
-            metadata["parent_trace_context"] = parent_trace_context
-
         workflow_trace_info = WorkflowTraceInfo(
             trace_id=self.trace_id,
             workflow_data=workflow_run.to_dict(),
@@ -828,8 +611,6 @@ class TraceTask:
             workflow_run_version=workflow_run_version,
             error=error,
             total_tokens=total_tokens,
-            prompt_tokens=prompt_tokens,
-            completion_tokens=completion_tokens,
             file_list=file_list,
             query=query,
             metadata=metadata,
@@ -837,21 +618,20 @@ class TraceTask:
             message_id=message_id,
             start_time=workflow_run.created_at,
             end_time=workflow_run.finished_at,
-            invoked_by=self._get_user_id_from_metadata(metadata),
         )
         return workflow_trace_info
 
-    def message_trace(self, message_id: str | None, **kwargs):
+    def message_trace(self, message_id: str | None):
         if not message_id:
             return {}
         message_data = get_message_data(message_id)
         if not message_data:
             return {}
         conversation_mode_stmt = select(Conversation.mode).where(Conversation.id == message_data.conversation_id)
-        conversation_modes = db.session.scalars(conversation_mode_stmt).all()
-        if not conversation_modes or len(conversation_modes) == 0:
+        conversation_mode = db.session.scalars(conversation_mode_stmt).all()
+        if not conversation_mode or len(conversation_mode) == 0:
             return {}
-        conversation_mode = conversation_modes[0]
+        conversation_mode = conversation_mode[0]
         created_at = message_data.created_at
         inputs = message_data.message
 
@@ -864,19 +644,6 @@ class TraceTask:
 
         streaming_metrics = self._extract_streaming_metrics(message_data)
 
-        tenant_id = ""
-        with Session(db.engine) as session:
-            tid = session.scalar(select(App.tenant_id).where(App.id == message_data.app_id))
-            if tid:
-                tenant_id = str(tid)
-
-        from core.telemetry.gateway import is_enterprise_telemetry_enabled
-
-        if is_enterprise_telemetry_enabled():
-            app_name, workspace_name = _lookup_app_and_workspace_names(message_data.app_id, tenant_id)
-        else:
-            app_name, workspace_name = "", ""
-
         metadata = {
             "conversation_id": message_data.conversation_id,
             "ls_provider": message_data.model_provider,
@@ -888,14 +655,7 @@ class TraceTask:
             "workflow_run_id": message_data.workflow_run_id,
             "from_source": message_data.from_source,
             "message_id": message_id,
-            "tenant_id": tenant_id,
-            "app_id": message_data.app_id,
-            "user_id": message_data.from_end_user_id or message_data.from_account_id,
-            "app_name": app_name,
-            "workspace_name": workspace_name,
         }
-        if node_execution_id := kwargs.get("node_execution_id"):
-            metadata["node_execution_id"] = node_execution_id
 
         message_tokens = message_data.message_tokens
 
@@ -912,9 +672,7 @@ class TraceTask:
             outputs=message_data.answer,
             file_list=file_list,
             start_time=created_at,
-            end_time=message_data.updated_at
-            if message_data.updated_at and message_data.updated_at > created_at
-            else created_at + timedelta(seconds=message_data.provider_response_latency),
+            end_time=created_at + timedelta(seconds=message_data.provider_response_latency),
             metadata=metadata,
             message_file_data=message_file_data,
             conversation_mode=conversation_mode,
@@ -939,8 +697,6 @@ class TraceTask:
             "preset_response": moderation_result.preset_response,
             "query": moderation_result.query,
         }
-        if node_execution_id := kwargs.get("node_execution_id"):
-            metadata["node_execution_id"] = node_execution_id
 
         # get workflow_app_log_id
         workflow_app_log_id = None
@@ -982,8 +738,6 @@ class TraceTask:
             "workflow_run_id": message_data.workflow_run_id,
             "from_source": message_data.from_source,
         }
-        if node_execution_id := kwargs.get("node_execution_id"):
-            metadata["node_execution_id"] = node_execution_id
 
         # get workflow_app_log_id
         workflow_app_log_id = None
@@ -1023,52 +777,6 @@ class TraceTask:
         if not message_data:
             return {}
 
-        tenant_id = ""
-        with Session(db.engine) as session:
-            tid = session.scalar(select(App.tenant_id).where(App.id == message_data.app_id))
-            if tid:
-                tenant_id = str(tid)
-
-        from core.telemetry.gateway import is_enterprise_telemetry_enabled
-
-        if is_enterprise_telemetry_enabled():
-            app_name, workspace_name = _lookup_app_and_workspace_names(message_data.app_id, tenant_id)
-        else:
-            app_name, workspace_name = "", ""
-
-        doc_list = [doc.model_dump() for doc in documents] if documents else []
-        dataset_ids: set[str] = set()
-        for doc in doc_list:
-            doc_meta = doc.get("metadata") or {}
-            did = doc_meta.get("dataset_id")
-            if did:
-                dataset_ids.add(did)
-
-        embedding_models: dict[str, dict[str, str]] = {}
-        if dataset_ids:
-            with Session(db.engine) as session:
-                rows = session.execute(
-                    select(Dataset.id, Dataset.embedding_model, Dataset.embedding_model_provider).where(
-                        Dataset.id.in_(list(dataset_ids))
-                    )
-                ).all()
-                for row in rows:
-                    embedding_models[str(row[0])] = {
-                        "embedding_model": row[1] or "",
-                        "embedding_model_provider": row[2] or "",
-                    }
-
-        # Extract rerank model info from retrieval_model kwargs
-        rerank_model_provider = ""
-        rerank_model_name = ""
-        if "retrieval_model" in kwargs:
-            retrieval_model = kwargs["retrieval_model"]
-            if isinstance(retrieval_model, dict):
-                reranking_model = retrieval_model.get("reranking_model")
-                if isinstance(reranking_model, dict):
-                    rerank_model_provider = reranking_model.get("reranking_provider_name", "")
-                    rerank_model_name = reranking_model.get("reranking_model_name", "")
-
         metadata = {
             "message_id": message_id,
             "ls_provider": message_data.model_provider,
@@ -1079,23 +787,13 @@ class TraceTask:
             "agent_based": message_data.agent_based,
             "workflow_run_id": message_data.workflow_run_id,
             "from_source": message_data.from_source,
-            "tenant_id": tenant_id,
-            "app_id": message_data.app_id,
-            "user_id": message_data.from_end_user_id or message_data.from_account_id,
-            "app_name": app_name,
-            "workspace_name": workspace_name,
-            "embedding_models": embedding_models,
-            "rerank_model_provider": rerank_model_provider,
-            "rerank_model_name": rerank_model_name,
         }
-        if node_execution_id := kwargs.get("node_execution_id"):
-            metadata["node_execution_id"] = node_execution_id
 
         dataset_retrieval_trace_info = DatasetRetrievalTraceInfo(
             trace_id=self.trace_id,
             message_id=message_id,
             inputs=message_data.query or message_data.inputs,
-            documents=doc_list,
+            documents=[doc.model_dump() for doc in documents] if documents else [],
             start_time=timer.get("start"),
             end_time=timer.get("end"),
             metadata=metadata,
@@ -1138,10 +836,6 @@ class TraceTask:
             "error": error,
             "tool_parameters": tool_parameters,
         }
-        if message_data.workflow_run_id:
-            metadata["workflow_run_id"] = message_data.workflow_run_id
-        if node_execution_id := kwargs.get("node_execution_id"):
-            metadata["node_execution_id"] = node_execution_id
 
         file_url = ""
         message_file_data = db.session.query(MessageFile).filter_by(message_id=message_id).first()
@@ -1196,8 +890,6 @@ class TraceTask:
             "conversation_id": conversation_id,
             "tenant_id": tenant_id,
         }
-        if node_execution_id := kwargs.get("node_execution_id"):
-            metadata["node_execution_id"] = node_execution_id
 
         generate_name_trace_info = GenerateNameTraceInfo(
             trace_id=self.trace_id,
@@ -1212,182 +904,6 @@ class TraceTask:
 
         return generate_name_trace_info
 
-    def prompt_generation_trace(self, **kwargs) -> PromptGenerationTraceInfo | dict:
-        tenant_id = kwargs.get("tenant_id", "")
-        user_id = kwargs.get("user_id", "")
-        app_id = kwargs.get("app_id")
-        operation_type = kwargs.get("operation_type", "")
-        instruction = kwargs.get("instruction", "")
-        generated_output = kwargs.get("generated_output", "")
-
-        prompt_tokens = kwargs.get("prompt_tokens", 0)
-        completion_tokens = kwargs.get("completion_tokens", 0)
-        total_tokens = kwargs.get("total_tokens", 0)
-
-        model_provider = kwargs.get("model_provider", "")
-        model_name = kwargs.get("model_name", "")
-
-        latency = kwargs.get("latency", 0.0)
-
-        timer = kwargs.get("timer")
-        start_time = timer.get("start") if timer else None
-        end_time = timer.get("end") if timer else None
-
-        total_price = kwargs.get("total_price")
-        currency = kwargs.get("currency")
-
-        error = kwargs.get("error")
-
-        app_name = None
-        workspace_name = None
-        if app_id:
-            app_name, workspace_name = _lookup_app_and_workspace_names(app_id, tenant_id)
-
-        metadata = {
-            "tenant_id": tenant_id,
-            "user_id": user_id,
-            "app_id": app_id or "",
-            "app_name": app_name,
-            "workspace_name": workspace_name,
-            "operation_type": operation_type,
-            "model_provider": model_provider,
-            "model_name": model_name,
-        }
-        if node_execution_id := kwargs.get("node_execution_id"):
-            metadata["node_execution_id"] = node_execution_id
-
-        return PromptGenerationTraceInfo(
-            trace_id=self.trace_id,
-            inputs=instruction,
-            outputs=generated_output,
-            start_time=start_time,
-            end_time=end_time,
-            metadata=metadata,
-            tenant_id=tenant_id,
-            user_id=user_id,
-            app_id=app_id,
-            operation_type=operation_type,
-            instruction=instruction,
-            prompt_tokens=prompt_tokens,
-            completion_tokens=completion_tokens,
-            total_tokens=total_tokens,
-            model_provider=model_provider,
-            model_name=model_name,
-            latency=latency,
-            total_price=total_price,
-            currency=currency,
-            error=error,
-        )
-
-    def node_execution_trace(self, **kwargs) -> WorkflowNodeTraceInfo | dict:
-        node_data: dict = kwargs.get("node_execution_data", {})
-        if not node_data:
-            return {}
-
-        from core.telemetry.gateway import is_enterprise_telemetry_enabled
-
-        if is_enterprise_telemetry_enabled():
-            app_name, workspace_name = _lookup_app_and_workspace_names(
-                node_data.get("app_id"), node_data.get("tenant_id")
-            )
-        else:
-            app_name, workspace_name = "", ""
-
-        # Try tool credential lookup first
-        credential_id = node_data.get("credential_id")
-        if is_enterprise_telemetry_enabled():
-            credential_name = _lookup_credential_name(credential_id, node_data.get("credential_provider_type"))
-            # If no credential_id found (e.g., LLM nodes), try LLM credential lookup
-            if not credential_id:
-                llm_cred_id, llm_cred_name = _lookup_llm_credential_info(
-                    tenant_id=node_data.get("tenant_id"),
-                    provider=node_data.get("model_provider"),
-                    model=node_data.get("model_name"),
-                    model_type="llm",
-                )
-                if llm_cred_id:
-                    credential_id = llm_cred_id
-                    credential_name = llm_cred_name
-        else:
-            credential_name = ""
-        metadata: dict[str, Any] = {
-            "tenant_id": node_data.get("tenant_id"),
-            "app_id": node_data.get("app_id"),
-            "app_name": app_name,
-            "workspace_name": workspace_name,
-            "user_id": node_data.get("user_id"),
-            "invoke_from": node_data.get("invoke_from"),
-            "credential_id": credential_id,
-            "credential_name": credential_name,
-            "dataset_ids": node_data.get("dataset_ids"),
-            "dataset_names": node_data.get("dataset_names"),
-            "plugin_name": node_data.get("plugin_name"),
-        }
-
-        parent_trace_context = node_data.get("parent_trace_context")
-        if parent_trace_context:
-            metadata["parent_trace_context"] = parent_trace_context
-
-        message_id: str | None = None
-        conversation_id = node_data.get("conversation_id")
-        workflow_execution_id = node_data.get("workflow_execution_id")
-        if conversation_id and workflow_execution_id and not parent_trace_context:
-            with Session(db.engine) as session:
-                msg_id = session.scalar(
-                    select(Message.id).where(
-                        Message.conversation_id == conversation_id,
-                        Message.workflow_run_id == workflow_execution_id,
-                    )
-                )
-                if msg_id:
-                    message_id = str(msg_id)
-                    metadata["message_id"] = message_id
-            if conversation_id:
-                metadata["conversation_id"] = conversation_id
-
-        return WorkflowNodeTraceInfo(
-            trace_id=self.trace_id,
-            message_id=message_id,
-            start_time=node_data.get("created_at"),
-            end_time=node_data.get("finished_at"),
-            metadata=metadata,
-            workflow_id=node_data.get("workflow_id", ""),
-            workflow_run_id=node_data.get("workflow_execution_id", ""),
-            tenant_id=node_data.get("tenant_id", ""),
-            node_execution_id=node_data.get("node_execution_id", ""),
-            node_id=node_data.get("node_id", ""),
-            node_type=node_data.get("node_type", ""),
-            title=node_data.get("title", ""),
-            status=node_data.get("status", ""),
-            error=node_data.get("error"),
-            elapsed_time=node_data.get("elapsed_time", 0.0),
-            index=node_data.get("index", 0),
-            predecessor_node_id=node_data.get("predecessor_node_id"),
-            total_tokens=node_data.get("total_tokens", 0),
-            total_price=node_data.get("total_price", 0.0),
-            currency=node_data.get("currency"),
-            model_provider=node_data.get("model_provider"),
-            model_name=node_data.get("model_name"),
-            prompt_tokens=node_data.get("prompt_tokens"),
-            completion_tokens=node_data.get("completion_tokens"),
-            tool_name=node_data.get("tool_name"),
-            iteration_id=node_data.get("iteration_id"),
-            iteration_index=node_data.get("iteration_index"),
-            loop_id=node_data.get("loop_id"),
-            loop_index=node_data.get("loop_index"),
-            parallel_id=node_data.get("parallel_id"),
-            node_inputs=node_data.get("node_inputs"),
-            node_outputs=node_data.get("node_outputs"),
-            process_data=node_data.get("process_data"),
-            invoked_by=self._get_user_id_from_metadata(metadata),
-        )
-
-    def draft_node_execution_trace(self, **kwargs) -> DraftNodeExecutionTrace | dict:
-        node_trace = self.node_execution_trace(**kwargs)
-        if not isinstance(node_trace, WorkflowNodeTraceInfo):
-            return node_trace
-        return DraftNodeExecutionTrace(**node_trace.model_dump())
-
     def _extract_streaming_metrics(self, message_data) -> dict:
         if not message_data.message_metadata:
             return {}
@@ -1421,17 +937,13 @@ class TraceQueueManager:
         self.user_id = user_id
         self.trace_instance = OpsTraceManager.get_ops_trace_instance(app_id)
         self.flask_app = current_app._get_current_object()  # type: ignore
-
-        from core.telemetry.gateway import is_enterprise_telemetry_enabled
-
-        self._enterprise_telemetry_enabled = is_enterprise_telemetry_enabled()
         if trace_manager_timer is None:
             self.start_timer()
 
     def add_trace_task(self, trace_task: TraceTask):
         global trace_manager_timer, trace_manager_queue
         try:
-            if self._enterprise_telemetry_enabled or self.trace_instance:
+            if self.trace_instance:
                 trace_task.app_id = self.app_id
                 trace_manager_queue.put(trace_task)
         except Exception:
@@ -1467,27 +979,20 @@ class TraceQueueManager:
     def send_to_celery(self, tasks: list[TraceTask]):
         with self.flask_app.app_context():
             for task in tasks:
-                storage_id = task.app_id
-                if storage_id is None:
-                    tenant_id = task.kwargs.get("tenant_id")
-                    if tenant_id:
-                        storage_id = f"tenant-{tenant_id}"
-                    else:
-                        logger.warning("Skipping trace without app_id or tenant_id, trace_type: %s", task.trace_type)
-                        continue
-
+                if task.app_id is None:
+                    continue
                 file_id = uuid4().hex
                 trace_info = task.execute()
 
                 task_data = TaskData(
-                    app_id=storage_id,
+                    app_id=task.app_id,
                     trace_info_type=type(trace_info).__name__,
                     trace_info=trace_info.model_dump() if trace_info else None,
                 )
-                file_path = f"{OPS_FILE_PATH}{storage_id}/{file_id}.json"
+                file_path = f"{OPS_FILE_PATH}{task.app_id}/{file_id}.json"
                 storage.save(file_path, task_data.model_dump_json().encode("utf-8"))
                 file_info = {
                     "file_id": file_id,
-                    "app_id": storage_id,
+                    "app_id": task.app_id,
                 }
                 process_trace_tasks.delay(file_info)  # type: ignore

api/core/ops/tencent_trace/tencent_trace.py

@@ -27,7 +27,7 @@ from core.repositories import SQLAlchemyWorkflowNodeExecutionRepository
 from dify_graph.entities.workflow_node_execution import (
     WorkflowNodeExecution,
 )
-from dify_graph.nodes import BuiltinNodeTypes
+from dify_graph.nodes import NodeType
 from extensions.ext_database import db
 from models import Account, App, TenantAccountJoin, WorkflowNodeExecutionTriggeredFrom
 
@@ -179,7 +179,7 @@ class TencentDataTrace(BaseTraceInstance):
                     if node_span:
                         self.trace_client.add_span(node_span)
 
-                        if node_execution.node_type == BuiltinNodeTypes.LLM:
+                        if node_execution.node_type == NodeType.LLM:
                             self._record_llm_metrics(node_execution)
                 except Exception:
                     logger.exception("[Tencent APM] Failed to process node execution: %s", node_execution.id)
@@ -192,15 +192,15 @@ class TencentDataTrace(BaseTraceInstance):
     ) -> SpanData | None:
         """Build span for different node types"""
         try:
-            if node_execution.node_type == BuiltinNodeTypes.LLM:
+            if node_execution.node_type == NodeType.LLM:
                 return TencentSpanBuilder.build_workflow_llm_span(
                     trace_id, workflow_span_id, trace_info, node_execution
                 )
-            elif node_execution.node_type == BuiltinNodeTypes.KNOWLEDGE_RETRIEVAL:
+            elif node_execution.node_type == NodeType.KNOWLEDGE_RETRIEVAL:
                 return TencentSpanBuilder.build_workflow_retrieval_span(
                     trace_id, workflow_span_id, trace_info, node_execution
                 )
-            elif node_execution.node_type == BuiltinNodeTypes.TOOL:
+            elif node_execution.node_type == NodeType.TOOL:
                 return TencentSpanBuilder.build_workflow_tool_span(
                     trace_id, workflow_span_id, trace_info, node_execution
                 )

api/core/ops/weave_trace/weave_trace.py

@@ -31,7 +31,7 @@ from core.ops.entities.trace_entity import (
 )
 from core.ops.weave_trace.entities.weave_trace_entity import WeaveTraceModel
 from core.repositories import DifyCoreRepositoryFactory
-from dify_graph.enums import BuiltinNodeTypes, WorkflowNodeExecutionMetadataKey
+from dify_graph.enums import NodeType, WorkflowNodeExecutionMetadataKey
 from extensions.ext_database import db
 from models import EndUser, MessageFile, WorkflowNodeExecutionTriggeredFrom
 
@@ -175,7 +175,7 @@ class WeaveDataTrace(BaseTraceInstance):
             node_name = node_execution.title
             node_type = node_execution.node_type
             status = node_execution.status
-            if node_type == BuiltinNodeTypes.LLM:
+            if node_type == NodeType.LLM:
                 inputs = node_execution.process_data.get("prompts", {}) if node_execution.process_data else {}
             else:
                 inputs = node_execution.inputs or {}

api/core/plugin/backwards_invocation/node.py

@@ -1,5 +1,5 @@
 from core.plugin.backwards_invocation.base import BaseBackwardsInvocation
-from dify_graph.enums import BuiltinNodeTypes
+from dify_graph.enums import NodeType
 from dify_graph.nodes.parameter_extractor.entities import (
     ModelConfig as ParameterExtractorModelConfig,
 )
@@ -52,7 +52,7 @@ class PluginNodeBackwardsInvocation(BaseBackwardsInvocation):
             instruction=instruction,  # instruct with variables are not supported
         )
         node_data_dict = node_data.model_dump()
-        node_data_dict["type"] = BuiltinNodeTypes.PARAMETER_EXTRACTOR
+        node_data_dict["type"] = NodeType.PARAMETER_EXTRACTOR
         execution = workflow_service.run_free_workflow_node(
             node_data_dict,
             tenant_id=tenant_id,

api/core/provider_manager.py

@@ -305,7 +305,9 @@ class ProviderManager:
             available_models = provider_configurations.get_models(model_type=model_type, only_active=True)
 
             if available_models:
-                available_model = available_models[0]
+                available_model = next(
+                    (model for model in available_models if model.model == "gpt-4"), available_models[0]
+                )
 
                 default_model = TenantDefaultModel(
                     tenant_id=tenant_id,
@@ -625,7 +627,7 @@ class ProviderManager:
                                 tenant_id=tenant_id,
                                 # TODO: Use provider name with prefix after the data migration.
                                 provider_name=ModelProviderID(provider_name).provider_name,
-                                provider_type=ProviderType.SYSTEM,
+                                provider_type=ProviderType.SYSTEM.value,
                                 quota_type=quota.quota_type,
                                 quota_limit=0,  # type: ignore
                                 quota_used=0,

api/core/rag/datasource/vdb/hologres/hologres_vector.py

@@ -1,361 +0,0 @@
-import json
-import logging
-import time
-from typing import Any
-
-import holo_search_sdk as holo  # type: ignore
-from holo_search_sdk.types import BaseQuantizationType, DistanceType, TokenizerType
-from psycopg import sql as psql
-from pydantic import BaseModel, model_validator
-
-from configs import dify_config
-from core.rag.datasource.vdb.vector_base import BaseVector
-from core.rag.datasource.vdb.vector_factory import AbstractVectorFactory
-from core.rag.datasource.vdb.vector_type import VectorType
-from core.rag.embedding.embedding_base import Embeddings
-from core.rag.models.document import Document
-from extensions.ext_redis import redis_client
-from models.dataset import Dataset
-
-logger = logging.getLogger(__name__)
-
-
-class HologresVectorConfig(BaseModel):
-    """
-    Configuration for Hologres vector database connection.
-
-    In Hologres, access_key_id is used as the PostgreSQL username,
-    and access_key_secret is used as the PostgreSQL password.
-    """
-
-    host: str
-    port: int = 80
-    database: str
-    access_key_id: str
-    access_key_secret: str
-    schema_name: str = "public"
-    tokenizer: TokenizerType = "jieba"
-    distance_method: DistanceType = "Cosine"
-    base_quantization_type: BaseQuantizationType = "rabitq"
-    max_degree: int = 64
-    ef_construction: int = 400
-
-    @model_validator(mode="before")
-    @classmethod
-    def validate_config(cls, values: dict):
-        if not values.get("host"):
-            raise ValueError("config HOLOGRES_HOST is required")
-        if not values.get("database"):
-            raise ValueError("config HOLOGRES_DATABASE is required")
-        if not values.get("access_key_id"):
-            raise ValueError("config HOLOGRES_ACCESS_KEY_ID is required")
-        if not values.get("access_key_secret"):
-            raise ValueError("config HOLOGRES_ACCESS_KEY_SECRET is required")
-        return values
-
-
-class HologresVector(BaseVector):
-    """
-    Hologres vector storage implementation using holo-search-sdk.
-
-    Supports semantic search (vector), full-text search, and hybrid search.
-    """
-
-    def __init__(self, collection_name: str, config: HologresVectorConfig):
-        super().__init__(collection_name)
-        self._config = config
-        self._client = self._init_client(config)
-        self.table_name = f"embedding_{collection_name}".lower()
-
-    def _init_client(self, config: HologresVectorConfig):
-        """Initialize and return a holo-search-sdk client."""
-        client = holo.connect(
-            host=config.host,
-            port=config.port,
-            database=config.database,
-            access_key_id=config.access_key_id,
-            access_key_secret=config.access_key_secret,
-            schema=config.schema_name,
-        )
-        client.connect()
-        return client
-
-    def get_type(self) -> str:
-        return VectorType.HOLOGRES
-
-    def create(self, texts: list[Document], embeddings: list[list[float]], **kwargs):
-        """Create collection table with vector and full-text indexes, then add texts."""
-        dimension = len(embeddings[0])
-        self._create_collection(dimension)
-        self.add_texts(texts, embeddings)
-
-    def add_texts(self, documents: list[Document], embeddings: list[list[float]], **kwargs):
-        """Add texts with embeddings to the collection using batch upsert."""
-        if not documents:
-            return []
-
-        pks: list[str] = []
-        batch_size = 100
-        for i in range(0, len(documents), batch_size):
-            batch_docs = documents[i : i + batch_size]
-            batch_embeddings = embeddings[i : i + batch_size]
-
-            values = []
-            column_names = ["id", "text", "meta", "embedding"]
-
-            for j, doc in enumerate(batch_docs):
-                doc_id = doc.metadata.get("doc_id", "") if doc.metadata else ""
-                pks.append(doc_id)
-                values.append(
-                    [
-                        doc_id,
-                        doc.page_content,
-                        json.dumps(doc.metadata or {}),
-                        batch_embeddings[j],
-                    ]
-                )
-
-            table = self._client.open_table(self.table_name)
-            table.upsert_multi(
-                index_column="id",
-                values=values,
-                column_names=column_names,
-                update=True,
-                update_columns=["text", "meta", "embedding"],
-            )
-
-        return pks
-
-    def text_exists(self, id: str) -> bool:
-        """Check if a text with the given doc_id exists in the collection."""
-        if not self._client.check_table_exist(self.table_name):
-            return False
-
-        result = self._client.execute(
-            psql.SQL("SELECT 1 FROM {} WHERE id = {} LIMIT 1").format(
-                psql.Identifier(self.table_name), psql.Literal(id)
-            ),
-            fetch_result=True,
-        )
-        return bool(result)
-
-    def get_ids_by_metadata_field(self, key: str, value: str) -> list[str] | None:
-        """Get document IDs by metadata field key and value."""
-        result = self._client.execute(
-            psql.SQL("SELECT id FROM {} WHERE meta->>{} = {}").format(
-                psql.Identifier(self.table_name), psql.Literal(key), psql.Literal(value)
-            ),
-            fetch_result=True,
-        )
-        if result:
-            return [row[0] for row in result]
-        return None
-
-    def delete_by_ids(self, ids: list[str]):
-        """Delete documents by their doc_id list."""
-        if not ids:
-            return
-        if not self._client.check_table_exist(self.table_name):
-            return
-
-        self._client.execute(
-            psql.SQL("DELETE FROM {} WHERE id IN ({})").format(
-                psql.Identifier(self.table_name),
-                psql.SQL(", ").join(psql.Literal(id) for id in ids),
-            )
-        )
-
-    def delete_by_metadata_field(self, key: str, value: str):
-        """Delete documents by metadata field key and value."""
-        if not self._client.check_table_exist(self.table_name):
-            return
-
-        self._client.execute(
-            psql.SQL("DELETE FROM {} WHERE meta->>{} = {}").format(
-                psql.Identifier(self.table_name), psql.Literal(key), psql.Literal(value)
-            )
-        )
-
-    def search_by_vector(self, query_vector: list[float], **kwargs: Any) -> list[Document]:
-        """Search for documents by vector similarity."""
-        if not self._client.check_table_exist(self.table_name):
-            return []
-
-        top_k = kwargs.get("top_k", 4)
-        score_threshold = float(kwargs.get("score_threshold") or 0.0)
-
-        table = self._client.open_table(self.table_name)
-        query = (
-            table.search_vector(
-                vector=query_vector,
-                column="embedding",
-                distance_method=self._config.distance_method,
-                output_name="distance",
-            )
-            .select(["id", "text", "meta"])
-            .limit(top_k)
-        )
-
-        # Apply document_ids_filter if provided
-        document_ids_filter = kwargs.get("document_ids_filter")
-        if document_ids_filter:
-            filter_sql = psql.SQL("meta->>'document_id' IN ({})").format(
-                psql.SQL(", ").join(psql.Literal(id) for id in document_ids_filter)
-            )
-            query = query.where(filter_sql)
-
-        results = query.fetchall()
-        return self._process_vector_results(results, score_threshold)
-
-    def _process_vector_results(self, results: list, score_threshold: float) -> list[Document]:
-        """Process vector search results into Document objects."""
-        docs = []
-        for row in results:
-            # row format: (distance, id, text, meta)
-            # distance is first because search_vector() adds the computed column before selected columns
-            distance = row[0]
-            text = row[2]
-            meta = row[3]
-
-            if isinstance(meta, str):
-                meta = json.loads(meta)
-
-            # Convert distance to similarity score (consistent with pgvector)
-            score = 1 - distance
-            meta["score"] = score
-
-            if score >= score_threshold:
-                docs.append(Document(page_content=text, metadata=meta))
-
-        return docs
-
-    def search_by_full_text(self, query: str, **kwargs: Any) -> list[Document]:
-        """Search for documents by full-text search."""
-        if not self._client.check_table_exist(self.table_name):
-            return []
-
-        top_k = kwargs.get("top_k", 4)
-
-        table = self._client.open_table(self.table_name)
-        search_query = table.search_text(
-            column="text",
-            expression=query,
-            return_score=True,
-            return_score_name="score",
-            return_all_columns=True,
-        ).limit(top_k)
-
-        # Apply document_ids_filter if provided
-        document_ids_filter = kwargs.get("document_ids_filter")
-        if document_ids_filter:
-            filter_sql = psql.SQL("meta->>'document_id' IN ({})").format(
-                psql.SQL(", ").join(psql.Literal(id) for id in document_ids_filter)
-            )
-            search_query = search_query.where(filter_sql)
-
-        results = search_query.fetchall()
-        return self._process_full_text_results(results)
-
-    def _process_full_text_results(self, results: list) -> list[Document]:
-        """Process full-text search results into Document objects."""
-        docs = []
-        for row in results:
-            # row format: (id, text, meta, embedding, score)
-            text = row[1]
-            meta = row[2]
-            score = row[-1]  # score is the last column from return_score
-
-            if isinstance(meta, str):
-                meta = json.loads(meta)
-
-            meta["score"] = score
-            docs.append(Document(page_content=text, metadata=meta))
-
-        return docs
-
-    def delete(self):
-        """Delete the entire collection table."""
-        if self._client.check_table_exist(self.table_name):
-            self._client.drop_table(self.table_name)
-
-    def _create_collection(self, dimension: int):
-        """Create the collection table with vector and full-text indexes."""
-        lock_name = f"vector_indexing_lock_{self._collection_name}"
-        with redis_client.lock(lock_name, timeout=20):
-            collection_exist_cache_key = f"vector_indexing_{self._collection_name}"
-            if redis_client.get(collection_exist_cache_key):
-                return
-
-            if not self._client.check_table_exist(self.table_name):
-                # Create table via SQL with CHECK constraint for vector dimension
-                create_table_sql = psql.SQL("""
-                    CREATE TABLE IF NOT EXISTS {} (
-                        id TEXT PRIMARY KEY,
-                        text TEXT NOT NULL,
-                        meta JSONB NOT NULL,
-                        embedding float4[] NOT NULL
-                            CHECK (array_ndims(embedding) = 1
-                                   AND array_length(embedding, 1) = {})
-                    );
-                """).format(psql.Identifier(self.table_name), psql.Literal(dimension))
-                self._client.execute(create_table_sql)
-
-                # Wait for table to be fully ready before creating indexes
-                max_wait_seconds = 30
-                poll_interval = 2
-                for _ in range(max_wait_seconds // poll_interval):
-                    if self._client.check_table_exist(self.table_name):
-                        break
-                    time.sleep(poll_interval)
-                else:
-                    raise RuntimeError(f"Table {self.table_name} was not ready after {max_wait_seconds}s")
-
-                # Open table and set vector index
-                table = self._client.open_table(self.table_name)
-                table.set_vector_index(
-                    column="embedding",
-                    distance_method=self._config.distance_method,
-                    base_quantization_type=self._config.base_quantization_type,
-                    max_degree=self._config.max_degree,
-                    ef_construction=self._config.ef_construction,
-                    use_reorder=self._config.base_quantization_type == "rabitq",
-                )
-
-                # Create full-text search index
-                table.create_text_index(
-                    index_name=f"ft_idx_{self._collection_name}",
-                    column="text",
-                    tokenizer=self._config.tokenizer,
-                )
-
-            redis_client.set(collection_exist_cache_key, 1, ex=3600)
-
-
-class HologresVectorFactory(AbstractVectorFactory):
-    """Factory class for creating HologresVector instances."""
-
-    def init_vector(self, dataset: Dataset, attributes: list, embeddings: Embeddings) -> HologresVector:
-        if dataset.index_struct_dict:
-            class_prefix: str = dataset.index_struct_dict["vector_store"]["class_prefix"]
-            collection_name = class_prefix
-        else:
-            dataset_id = dataset.id
-            collection_name = Dataset.gen_collection_name_by_id(dataset_id)
-            dataset.index_struct = json.dumps(self.gen_index_struct_dict(VectorType.HOLOGRES, collection_name))
-
-        return HologresVector(
-            collection_name=collection_name,
-            config=HologresVectorConfig(
-                host=dify_config.HOLOGRES_HOST or "",
-                port=dify_config.HOLOGRES_PORT,
-                database=dify_config.HOLOGRES_DATABASE or "",
-                access_key_id=dify_config.HOLOGRES_ACCESS_KEY_ID or "",
-                access_key_secret=dify_config.HOLOGRES_ACCESS_KEY_SECRET or "",
-                schema_name=dify_config.HOLOGRES_SCHEMA,
-                tokenizer=dify_config.HOLOGRES_TOKENIZER,
-                distance_method=dify_config.HOLOGRES_DISTANCE_METHOD,
-                base_quantization_type=dify_config.HOLOGRES_BASE_QUANTIZATION_TYPE,
-                max_degree=dify_config.HOLOGRES_MAX_DEGREE,
-                ef_construction=dify_config.HOLOGRES_EF_CONSTRUCTION,
-            ),
-        )

api/core/rag/datasource/vdb/pgvecto_rs/pgvecto_rs.py

@@ -135,8 +135,8 @@ class PGVectoRS(BaseVector):
     def get_ids_by_metadata_field(self, key: str, value: str):
         result = None
         with Session(self._client) as session:
-            select_statement = sql_text(f"SELECT id FROM {self._collection_name} WHERE meta->>:key = :value")
-            result = session.execute(select_statement, {"key": key, "value": value}).fetchall()
+            select_statement = sql_text(f"SELECT id FROM {self._collection_name} WHERE meta->>'{key}' = '{value}'; ")
+            result = session.execute(select_statement).fetchall()
         if result:
             return [item[0] for item in result]
         else:
@@ -172,9 +172,9 @@ class PGVectoRS(BaseVector):
     def text_exists(self, id: str) -> bool:
         with Session(self._client) as session:
             select_statement = sql_text(
-                f"SELECT id FROM {self._collection_name} WHERE meta->>'doc_id' = :doc_id limit 1"
+                f"SELECT id FROM {self._collection_name} WHERE meta->>'doc_id' = '{id}' limit 1; "
             )
-            result = session.execute(select_statement, {"doc_id": id}).fetchall()
+            result = session.execute(select_statement).fetchall()
         return len(result) > 0
 
     def search_by_vector(self, query_vector: list[float], **kwargs: Any) -> list[Document]:

api/core/rag/datasource/vdb/relyt/relyt_vector.py

@@ -154,8 +154,10 @@ class RelytVector(BaseVector):
     def get_ids_by_metadata_field(self, key: str, value: str):
         result = None
         with Session(self.client) as session:
-            select_statement = sql_text(f"""SELECT id FROM "{self._collection_name}" WHERE metadata->>:key = :value""")
-            result = session.execute(select_statement, {"key": key, "value": value}).fetchall()
+            select_statement = sql_text(
+                f"""SELECT id FROM "{self._collection_name}" WHERE metadata->>'{key}' = '{value}'; """
+            )
+            result = session.execute(select_statement).fetchall()
         if result:
             return [item[0] for item in result]
         else:
@@ -199,10 +201,11 @@ class RelytVector(BaseVector):
 
     def delete_by_ids(self, ids: list[str]):
         with Session(self.client) as session:
+            ids_str = ",".join(f"'{doc_id}'" for doc_id in ids)
             select_statement = sql_text(
-                f"""SELECT id FROM "{self._collection_name}" WHERE metadata->>'doc_id' = ANY(:doc_ids)"""
+                f"""SELECT id FROM "{self._collection_name}" WHERE metadata->>'doc_id' in ({ids_str}); """
             )
-            result = session.execute(select_statement, {"doc_ids": ids}).fetchall()
+            result = session.execute(select_statement).fetchall()
         if result:
             ids = [item[0] for item in result]
             self.delete_by_uuids(ids)
@@ -215,9 +218,9 @@ class RelytVector(BaseVector):
     def text_exists(self, id: str) -> bool:
         with Session(self.client) as session:
             select_statement = sql_text(
-                f"""SELECT id FROM "{self._collection_name}" WHERE metadata->>'doc_id' = :doc_id limit 1"""
+                f"""SELECT id FROM "{self._collection_name}" WHERE metadata->>'doc_id' = '{id}' limit 1; """
             )
-            result = session.execute(select_statement, {"doc_id": id}).fetchall()
+            result = session.execute(select_statement).fetchall()
         return len(result) > 0
 
     def search_by_vector(self, query_vector: list[float], **kwargs: Any) -> list[Document]:

api/core/rag/datasource/vdb/vector_factory.py

@@ -38,7 +38,7 @@ class AbstractVectorFactory(ABC):
 class Vector:
     def __init__(self, dataset: Dataset, attributes: list | None = None):
         if attributes is None:
-            attributes = ["doc_id", "dataset_id", "document_id", "doc_hash", "doc_type"]
+            attributes = ["doc_id", "dataset_id", "document_id", "doc_hash"]
         self._dataset = dataset
         self._embeddings = self._get_embeddings()
         self._attributes = attributes
@@ -191,10 +191,6 @@ class Vector:
                 from core.rag.datasource.vdb.iris.iris_vector import IrisVectorFactory
 
                 return IrisVectorFactory
-            case VectorType.HOLOGRES:
-                from core.rag.datasource.vdb.hologres.hologres_vector import HologresVectorFactory
-
-                return HologresVectorFactory
             case _:
                 raise ValueError(f"Vector store {vector_type} is not supported.")
 

api/core/rag/datasource/vdb/vector_type.py

@@ -34,4 +34,3 @@ class VectorType(StrEnum):
     MATRIXONE = "matrixone"
     CLICKZETTA = "clickzetta"
     IRIS = "iris"
-    HOLOGRES = "hologres"

api/core/rag/datasource/vdb/weaviate/weaviate_vector.py

@@ -196,7 +196,6 @@ class WeaviateVector(BaseVector):
                             ),
                             wc.Property(name="document_id", data_type=wc.DataType.TEXT),
                             wc.Property(name="doc_id", data_type=wc.DataType.TEXT),
-                            wc.Property(name="doc_type", data_type=wc.DataType.TEXT),
                             wc.Property(name="chunk_index", data_type=wc.DataType.INT),
                         ],
                         vector_config=wc.Configure.Vectors.self_provided(),
@@ -226,8 +225,6 @@ class WeaviateVector(BaseVector):
             to_add.append(wc.Property(name="document_id", data_type=wc.DataType.TEXT))
         if "doc_id" not in existing:
             to_add.append(wc.Property(name="doc_id", data_type=wc.DataType.TEXT))
-        if "doc_type" not in existing:
-            to_add.append(wc.Property(name="doc_type", data_type=wc.DataType.TEXT))
         if "chunk_index" not in existing:
             to_add.append(wc.Property(name="chunk_index", data_type=wc.DataType.INT))
 

api/core/rag/index_processor/index_processor.py

@@ -9,8 +9,8 @@ from flask import current_app
 from sqlalchemy import delete, func, select
 
 from core.db.session_factory import session_factory
-from core.workflow.nodes.knowledge_index.exc import KnowledgeIndexNodeError
-from core.workflow.nodes.knowledge_index.protocols import Preview, PreviewItem, QaPreview
+from dify_graph.nodes.knowledge_index.exc import KnowledgeIndexNodeError
+from dify_graph.repositories.index_processor_protocol import Preview, PreviewItem, QaPreview
 from models.dataset import Dataset, Document, DocumentSegment
 
 from .index_processor_factory import IndexProcessorFactory

api/core/rag/index_processor/index_processor_base.py

@@ -294,7 +294,7 @@ class BaseIndexProcessor(ABC):
             logging.warning("Error downloading image from %s: %s", image_url, str(e))
             return None
         except Exception:
-            logging.warning("Unexpected error downloading image from %s", image_url, exc_info=True)
+            logging.exception("Unexpected error downloading image from %s", image_url)
             return None
 
     def _download_tool_file(self, tool_file_id: str, current_user: Account) -> str | None:

api/core/rag/retrieval/dataset_retrieval.py

@@ -56,18 +56,18 @@ from core.rag.retrieval.template_prompts import (
 )
 from core.tools.signature import sign_upload_file
 from core.tools.utils.dataset_retriever.dataset_retriever_base_tool import DatasetRetrieverBaseTool
-from core.workflow.nodes.knowledge_retrieval import exc
-from core.workflow.nodes.knowledge_retrieval.retrieval import (
-    KnowledgeRetrievalRequest,
-    Source,
-    SourceChildChunk,
-    SourceMetadata,
-)
 from dify_graph.file import File, FileTransferMethod, FileType
 from dify_graph.model_runtime.entities.llm_entities import LLMMode, LLMResult, LLMUsage
 from dify_graph.model_runtime.entities.message_entities import PromptMessage, PromptMessageRole, PromptMessageTool
 from dify_graph.model_runtime.entities.model_entities import ModelFeature, ModelType
 from dify_graph.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
+from dify_graph.nodes.knowledge_retrieval import exc
+from dify_graph.repositories.rag_retrieval_protocol import (
+    KnowledgeRetrievalRequest,
+    Source,
+    SourceChildChunk,
+    SourceMetadata,
+)
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from libs.json_in_md_parser import parse_and_check_json_markdown
@@ -83,7 +83,6 @@ from models.dataset import (
 )
 from models.dataset import Document as DatasetDocument
 from models.dataset import Document as DocumentModel
-from models.enums import CreatorUserRole
 from services.external_knowledge_service import ExternalDatasetService
 from services.feature_service import FeatureService
 
@@ -1010,7 +1009,7 @@ class DatasetRetrieval:
                     content=json.dumps(contents),
                     source="app",
                     source_app_id=app_id,
-                    created_by_role=CreatorUserRole(user_from),
+                    created_by_role=user_from,
                     created_by=user_id,
                 )
                 dataset_queries.append(dataset_query)

api/core/repositories/sqlalchemy_workflow_execution_repository.py

@@ -146,9 +146,7 @@ class SQLAlchemyWorkflowExecutionRepository(WorkflowExecutionRepository):
 
         # No sequence number generation needed anymore
 
-        from models.workflow import WorkflowType as ModelWorkflowType
-
-        db_model.type = ModelWorkflowType(domain_model.workflow_type.value)
+        db_model.type = domain_model.workflow_type
         db_model.version = domain_model.workflow_version
         db_model.graph = json.dumps(domain_model.graph) if domain_model.graph else None
         db_model.inputs = json.dumps(domain_model.inputs) if domain_model.inputs else None

api/core/repositories/sqlalchemy_workflow_node_execution_repository.py

@@ -18,7 +18,7 @@ from tenacity import before_sleep_log, retry, retry_if_exception, stop_after_att
 
 from configs import dify_config
 from dify_graph.entities import WorkflowNodeExecution
-from dify_graph.enums import WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
+from dify_graph.enums import NodeType, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
 from dify_graph.model_runtime.utils.encoders import jsonable_encoder
 from dify_graph.repositories.workflow_node_execution_repository import OrderConfig, WorkflowNodeExecutionRepository
 from dify_graph.workflow_type_encoder import WorkflowRuntimeTypeConverter
@@ -146,7 +146,7 @@ class SQLAlchemyWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository)
             index=db_model.index,
             predecessor_node_id=db_model.predecessor_node_id,
             node_id=db_model.node_id,
-            node_type=db_model.node_type,
+            node_type=NodeType(db_model.node_type),
             title=db_model.title,
             inputs=inputs,
             process_data=process_data,

api/core/telemetry/__init__.py

@@ -1,43 +0,0 @@
-"""Telemetry facade.
-
-Thin public API for emitting telemetry events.  All routing logic
-lives in ``core.telemetry.gateway`` which is shared by both CE and EE.
-"""
-
-from __future__ import annotations
-
-from typing import TYPE_CHECKING
-
-from core.ops.entities.trace_entity import TraceTaskName
-from core.telemetry.events import TelemetryContext, TelemetryEvent
-from core.telemetry.gateway import emit as gateway_emit
-from core.telemetry.gateway import get_trace_task_to_case
-
-if TYPE_CHECKING:
-    from core.ops.ops_trace_manager import TraceQueueManager
-
-
-def emit(event: TelemetryEvent, trace_manager: TraceQueueManager | None = None) -> None:
-    """Emit a telemetry event.
-
-    Translates the ``TelemetryEvent`` (keyed by ``TraceTaskName``) into a
-    ``TelemetryCase`` and delegates to ``core.telemetry.gateway.emit()``.
-    """
-    case = get_trace_task_to_case().get(event.name)
-    if case is None:
-        return
-
-    context: dict[str, object] = {
-        "tenant_id": event.context.tenant_id,
-        "user_id": event.context.user_id,
-        "app_id": event.context.app_id,
-    }
-    gateway_emit(case, context, event.payload, trace_manager)
-
-
-__all__ = [
-    "TelemetryContext",
-    "TelemetryEvent",
-    "TraceTaskName",
-    "emit",
-]

api/core/telemetry/events.py

@@ -1,21 +0,0 @@
-from __future__ import annotations
-
-from dataclasses import dataclass
-from typing import TYPE_CHECKING, Any
-
-if TYPE_CHECKING:
-    from core.ops.entities.trace_entity import TraceTaskName
-
-
-@dataclass(frozen=True)
-class TelemetryContext:
-    tenant_id: str | None = None
-    user_id: str | None = None
-    app_id: str | None = None
-
-
-@dataclass(frozen=True)
-class TelemetryEvent:
-    name: TraceTaskName
-    context: TelemetryContext
-    payload: dict[str, Any]

api/core/telemetry/gateway.py

@@ -1,239 +0,0 @@
-"""Telemetry gateway — single routing layer for all editions.
-
-Maps ``TelemetryCase`` → ``CaseRoute`` and dispatches events to either
-the CE/EE trace pipeline (``TraceQueueManager``) or the enterprise-only
-metric/log Celery queue.
-
-This module lives in ``core/`` so both CE and EE share one routing table
-and one ``emit()`` entry point.  No separate enterprise gateway module is
-needed — enterprise-specific dispatch (Celery task, payload offloading)
-is handled here behind lazy imports that no-op in CE.
-"""
-
-from __future__ import annotations
-
-import json
-import logging
-import uuid
-from typing import TYPE_CHECKING, Any
-
-from core.ops.entities.trace_entity import TraceTaskName
-from enterprise.telemetry.contracts import SignalType
-from extensions.ext_storage import storage
-
-if TYPE_CHECKING:
-    from core.ops.ops_trace_manager import TraceQueueManager
-    from enterprise.telemetry.contracts import TelemetryCase
-
-logger = logging.getLogger(__name__)
-
-PAYLOAD_SIZE_THRESHOLD_BYTES = 1 * 1024 * 1024
-
-# ---------------------------------------------------------------------------
-# Routing table — authoritative mapping for all editions
-# ---------------------------------------------------------------------------
-
-_case_to_trace_task: dict | None = None
-_case_routing: dict | None = None
-
-
-def _get_case_to_trace_task() -> dict:
-    global _case_to_trace_task
-    if _case_to_trace_task is None:
-        from enterprise.telemetry.contracts import TelemetryCase
-
-        _case_to_trace_task = {
-            TelemetryCase.WORKFLOW_RUN: TraceTaskName.WORKFLOW_TRACE,
-            TelemetryCase.MESSAGE_RUN: TraceTaskName.MESSAGE_TRACE,
-            TelemetryCase.NODE_EXECUTION: TraceTaskName.NODE_EXECUTION_TRACE,
-            TelemetryCase.DRAFT_NODE_EXECUTION: TraceTaskName.DRAFT_NODE_EXECUTION_TRACE,
-            TelemetryCase.PROMPT_GENERATION: TraceTaskName.PROMPT_GENERATION_TRACE,
-            TelemetryCase.TOOL_EXECUTION: TraceTaskName.TOOL_TRACE,
-            TelemetryCase.MODERATION_CHECK: TraceTaskName.MODERATION_TRACE,
-            TelemetryCase.SUGGESTED_QUESTION: TraceTaskName.SUGGESTED_QUESTION_TRACE,
-            TelemetryCase.DATASET_RETRIEVAL: TraceTaskName.DATASET_RETRIEVAL_TRACE,
-            TelemetryCase.GENERATE_NAME: TraceTaskName.GENERATE_NAME_TRACE,
-        }
-    return _case_to_trace_task
-
-
-def get_trace_task_to_case() -> dict:
-    """Return TraceTaskName → TelemetryCase (inverse of _get_case_to_trace_task)."""
-    return {v: k for k, v in _get_case_to_trace_task().items()}
-
-
-def _get_case_routing() -> dict:
-    global _case_routing
-    if _case_routing is None:
-        from enterprise.telemetry.contracts import CaseRoute, SignalType, TelemetryCase
-
-        _case_routing = {
-            # TRACE — CE-eligible (flow in both CE and EE)
-            TelemetryCase.WORKFLOW_RUN: CaseRoute(signal_type=SignalType.TRACE, ce_eligible=True),
-            TelemetryCase.MESSAGE_RUN: CaseRoute(signal_type=SignalType.TRACE, ce_eligible=True),
-            TelemetryCase.TOOL_EXECUTION: CaseRoute(signal_type=SignalType.TRACE, ce_eligible=True),
-            TelemetryCase.MODERATION_CHECK: CaseRoute(signal_type=SignalType.TRACE, ce_eligible=True),
-            TelemetryCase.SUGGESTED_QUESTION: CaseRoute(signal_type=SignalType.TRACE, ce_eligible=True),
-            TelemetryCase.DATASET_RETRIEVAL: CaseRoute(signal_type=SignalType.TRACE, ce_eligible=True),
-            TelemetryCase.GENERATE_NAME: CaseRoute(signal_type=SignalType.TRACE, ce_eligible=True),
-            # TRACE — enterprise-only
-            TelemetryCase.NODE_EXECUTION: CaseRoute(signal_type=SignalType.TRACE, ce_eligible=False),
-            TelemetryCase.DRAFT_NODE_EXECUTION: CaseRoute(signal_type=SignalType.TRACE, ce_eligible=False),
-            TelemetryCase.PROMPT_GENERATION: CaseRoute(signal_type=SignalType.TRACE, ce_eligible=False),
-            # METRIC_LOG — enterprise-only (signal-driven, not trace)
-            TelemetryCase.APP_CREATED: CaseRoute(signal_type=SignalType.METRIC_LOG, ce_eligible=False),
-            TelemetryCase.APP_UPDATED: CaseRoute(signal_type=SignalType.METRIC_LOG, ce_eligible=False),
-            TelemetryCase.APP_DELETED: CaseRoute(signal_type=SignalType.METRIC_LOG, ce_eligible=False),
-            TelemetryCase.FEEDBACK_CREATED: CaseRoute(signal_type=SignalType.METRIC_LOG, ce_eligible=False),
-        }
-    return _case_routing
-
-
-def __getattr__(name: str) -> dict:
-    """Lazy module-level access to routing tables."""
-    if name == "CASE_ROUTING":
-        return _get_case_routing()
-    if name == "CASE_TO_TRACE_TASK":
-        return _get_case_to_trace_task()
-    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
-
-
-# ---------------------------------------------------------------------------
-# Helpers
-# ---------------------------------------------------------------------------
-
-
-def is_enterprise_telemetry_enabled() -> bool:
-    try:
-        from enterprise.telemetry.exporter import is_enterprise_telemetry_enabled
-
-        return is_enterprise_telemetry_enabled()
-    except Exception:
-        return False
-
-
-def _handle_payload_sizing(
-    payload: dict[str, Any],
-    tenant_id: str,
-    event_id: str,
-) -> tuple[dict[str, Any], str | None]:
-    """Inline or offload payload based on size.
-
-    Returns ``(payload_for_envelope, storage_key | None)``.  Payloads
-    exceeding ``PAYLOAD_SIZE_THRESHOLD_BYTES`` are written to object
-    storage and replaced with an empty dict in the envelope.
-    """
-    try:
-        payload_json = json.dumps(payload)
-        payload_size = len(payload_json.encode("utf-8"))
-    except (TypeError, ValueError):
-        logger.warning("Failed to serialize payload for sizing: event_id=%s", event_id)
-        return payload, None
-
-    if payload_size <= PAYLOAD_SIZE_THRESHOLD_BYTES:
-        return payload, None
-
-    storage_key = f"telemetry/{tenant_id}/{event_id}.json"
-    try:
-        storage.save(storage_key, payload_json.encode("utf-8"))
-        logger.debug("Stored large payload to storage: key=%s, size=%d", storage_key, payload_size)
-        return {}, storage_key
-    except Exception:
-        logger.warning("Failed to store large payload, inlining instead: event_id=%s", event_id, exc_info=True)
-        return payload, None
-
-
-# ---------------------------------------------------------------------------
-# Public API
-# ---------------------------------------------------------------------------
-
-
-def emit(
-    case: TelemetryCase,
-    context: dict[str, Any],
-    payload: dict[str, Any],
-    trace_manager: TraceQueueManager | None = None,
-) -> None:
-    """Route a telemetry event to the correct pipeline.
-
-    TRACE events are enqueued into ``TraceQueueManager`` (works in both CE
-    and EE).  Enterprise-only traces are silently dropped when EE is
-    disabled.
-
-    METRIC_LOG events are dispatched to the enterprise Celery queue;
-    silently dropped when enterprise telemetry is unavailable.
-    """
-    route = _get_case_routing().get(case)
-    if route is None:
-        logger.warning("Unknown telemetry case: %s, dropping event", case)
-        return
-
-    if not route.ce_eligible and not is_enterprise_telemetry_enabled():
-        logger.debug("Dropping EE-only event: case=%s (EE disabled)", case)
-        return
-
-    if route.signal_type == SignalType.TRACE:
-        _emit_trace(case, context, payload, trace_manager)
-    else:
-        _emit_metric_log(case, context, payload)
-
-
-def _emit_trace(
-    case: TelemetryCase,
-    context: dict[str, Any],
-    payload: dict[str, Any],
-    trace_manager: TraceQueueManager | None,
-) -> None:
-    from core.ops.ops_trace_manager import TraceQueueManager as LocalTraceQueueManager
-    from core.ops.ops_trace_manager import TraceTask
-
-    trace_task_name = _get_case_to_trace_task().get(case)
-    if trace_task_name is None:
-        logger.warning("No TraceTaskName mapping for case: %s", case)
-        return
-
-    queue_manager = trace_manager or LocalTraceQueueManager(
-        app_id=context.get("app_id"),
-        user_id=context.get("user_id"),
-    )
-    queue_manager.add_trace_task(TraceTask(trace_task_name, user_id=context.get("user_id"), **payload))
-    logger.debug("Enqueued trace task: case=%s, app_id=%s", case, context.get("app_id"))
-
-
-def _emit_metric_log(
-    case: TelemetryCase,
-    context: dict[str, Any],
-    payload: dict[str, Any],
-) -> None:
-    """Build envelope and dispatch to enterprise Celery queue.
-
-    No-ops when the enterprise telemetry task is not importable (CE mode).
-    """
-    try:
-        from tasks.enterprise_telemetry_task import process_enterprise_telemetry
-    except ImportError:
-        logger.debug("Enterprise metric/log dispatch unavailable, dropping: case=%s", case)
-        return
-
-    tenant_id = context.get("tenant_id") or ""
-    event_id = str(uuid.uuid4())
-
-    payload_for_envelope, payload_ref = _handle_payload_sizing(payload, tenant_id, event_id)
-
-    from enterprise.telemetry.contracts import TelemetryEnvelope
-
-    envelope = TelemetryEnvelope(
-        case=case,
-        tenant_id=tenant_id,
-        event_id=event_id,
-        payload=payload_for_envelope,
-        metadata={"payload_ref": payload_ref} if payload_ref else None,
-    )
-
-    process_enterprise_telemetry.delay(envelope.model_dump_json())
-    logger.debug(
-        "Enqueued metric/log event: case=%s, tenant_id=%s, event_id=%s",
-        case,
-        tenant_id,
-        event_id,
-    )

api/core/tools/utils/configuration.py

@@ -116,7 +116,6 @@ class ToolParameterConfigurationManager:
 
         return a deep copy of parameters with decrypted values
         """
-        parameters = self._deep_copy(parameters)
 
         cache = ToolParameterCache(
             tenant_id=self.tenant_id,

api/core/tools/utils/workflow_configuration_sync.py

@@ -3,7 +3,7 @@ from typing import Any
 
 from core.tools.entities.tool_entities import WorkflowToolParameterConfiguration
 from core.tools.errors import WorkflowToolHumanInputNotSupportedError
-from dify_graph.enums import BuiltinNodeTypes
+from dify_graph.enums import NodeType
 from dify_graph.nodes.base.entities import OutputVariableEntity
 from dify_graph.variables.input_entities import VariableEntity
 
@@ -51,7 +51,7 @@ class WorkflowToolConfigurationUtils:
     def ensure_no_human_input_nodes(cls, graph: Mapping[str, Any]) -> None:
         nodes = graph.get("nodes", [])
         for node in nodes:
-            if node.get("data", {}).get("type") == BuiltinNodeTypes.HUMAN_INPUT:
+            if node.get("data", {}).get("type") == NodeType.HUMAN_INPUT:
                 raise WorkflowToolHumanInputNotSupportedError()
 
     @classmethod

api/core/trigger/constants.py

@@ -1,18 +0,0 @@
-from typing import Final
-
-TRIGGER_WEBHOOK_NODE_TYPE: Final[str] = "trigger-webhook"
-TRIGGER_SCHEDULE_NODE_TYPE: Final[str] = "trigger-schedule"
-TRIGGER_PLUGIN_NODE_TYPE: Final[str] = "trigger-plugin"
-TRIGGER_INFO_METADATA_KEY: Final[str] = "trigger_info"
-
-TRIGGER_NODE_TYPES: Final[frozenset[str]] = frozenset(
-    {
-        TRIGGER_WEBHOOK_NODE_TYPE,
-        TRIGGER_SCHEDULE_NODE_TYPE,
-        TRIGGER_PLUGIN_NODE_TYPE,
-    }
-)
-
-
-def is_trigger_node_type(node_type: str) -> bool:
-    return node_type in TRIGGER_NODE_TYPES

api/core/trigger/debug/event_selectors.py

@@ -11,11 +11,6 @@ from typing import Any
 from pydantic import BaseModel
 
 from core.plugin.entities.request import TriggerInvokeEventResponse
-from core.trigger.constants import (
-    TRIGGER_PLUGIN_NODE_TYPE,
-    TRIGGER_SCHEDULE_NODE_TYPE,
-    TRIGGER_WEBHOOK_NODE_TYPE,
-)
 from core.trigger.debug.event_bus import TriggerDebugEventBus
 from core.trigger.debug.events import (
     PluginTriggerDebugEvent,
@@ -24,9 +19,10 @@ from core.trigger.debug.events import (
     build_plugin_pool_key,
     build_webhook_pool_key,
 )
-from core.workflow.nodes.trigger_plugin.entities import TriggerEventNodeData
-from core.workflow.nodes.trigger_schedule.entities import ScheduleConfig
 from dify_graph.entities.graph_config import NodeConfigDict
+from dify_graph.enums import NodeType
+from dify_graph.nodes.trigger_plugin.entities import TriggerEventNodeData
+from dify_graph.nodes.trigger_schedule.entities import ScheduleConfig
 from extensions.ext_redis import redis_client
 from libs.datetime_utils import ensure_naive_utc, naive_utc_now
 from libs.schedule_utils import calculate_next_run_at
@@ -210,19 +206,21 @@ def create_event_poller(
     if not node_config:
         raise ValueError("Node data not found for node %s", node_id)
     node_type = draft_workflow.get_node_type_from_node_config(node_config)
-    if node_type == TRIGGER_PLUGIN_NODE_TYPE:
-        return PluginTriggerDebugEventPoller(
-            tenant_id=tenant_id, user_id=user_id, app_id=app_id, node_config=node_config, node_id=node_id
-        )
-    if node_type == TRIGGER_WEBHOOK_NODE_TYPE:
-        return WebhookTriggerDebugEventPoller(
-            tenant_id=tenant_id, user_id=user_id, app_id=app_id, node_config=node_config, node_id=node_id
-        )
-    if node_type == TRIGGER_SCHEDULE_NODE_TYPE:
-        return ScheduleTriggerDebugEventPoller(
-            tenant_id=tenant_id, user_id=user_id, app_id=app_id, node_config=node_config, node_id=node_id
-        )
-    raise ValueError("unable to create event poller for node type %s", node_type)
+    match node_type:
+        case NodeType.TRIGGER_PLUGIN:
+            return PluginTriggerDebugEventPoller(
+                tenant_id=tenant_id, user_id=user_id, app_id=app_id, node_config=node_config, node_id=node_id
+            )
+        case NodeType.TRIGGER_WEBHOOK:
+            return WebhookTriggerDebugEventPoller(
+                tenant_id=tenant_id, user_id=user_id, app_id=app_id, node_config=node_config, node_id=node_id
+            )
+        case NodeType.TRIGGER_SCHEDULE:
+            return ScheduleTriggerDebugEventPoller(
+                tenant_id=tenant_id, user_id=user_id, app_id=app_id, node_config=node_config, node_id=node_id
+            )
+        case _:
+            raise ValueError("unable to create event poller for node type %s", node_type)
 
 
 def select_trigger_debug_events(

api/core/workflow/__init__.py

@@ -1 +1,4 @@
-"""Core workflow package."""
+from .node_factory import DifyNodeFactory
+from .workflow_entry import WorkflowEntry
+
+__all__ = ["DifyNodeFactory", "WorkflowEntry"]

api/core/workflow/node_factory.py

@@ -1,7 +1,4 @@
-import importlib
-import pkgutil
-from collections.abc import Callable, Iterator, Mapping, MutableMapping
-from functools import lru_cache
+from collections.abc import Callable, Mapping
 from typing import TYPE_CHECKING, Any, TypeAlias, cast, final
 
 from sqlalchemy import select
@@ -11,6 +8,7 @@ from typing_extensions import override
 from configs import dify_config
 from core.app.entities.app_invoke_entities import DifyRunContext
 from core.app.llm.model_access import build_dify_model_access
+from core.datasource.datasource_manager import DatasourceManager
 from core.helper.code_executor.code_executor import (
     CodeExecutionError,
     CodeExecutor,
@@ -19,19 +17,15 @@ from core.helper.ssrf_proxy import ssrf_proxy
 from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance
 from core.prompt.entities.advanced_prompt_entities import MemoryConfig
+from core.rag.index_processor.index_processor import IndexProcessor
+from core.rag.retrieval.dataset_retrieval import DatasetRetrieval
+from core.rag.summary_index.summary_index import SummaryIndex
 from core.repositories.human_input_repository import HumanInputFormRepositoryImpl
 from core.tools.tool_file_manager import ToolFileManager
-from core.trigger.constants import TRIGGER_NODE_TYPES
-from core.workflow.nodes.agent.message_transformer import AgentMessageTransformer
-from core.workflow.nodes.agent.plugin_strategy_adapter import (
-    PluginAgentStrategyPresentationProvider,
-    PluginAgentStrategyResolver,
-)
-from core.workflow.nodes.agent.runtime_support import AgentRuntimeSupport
 from dify_graph.entities.base_node_data import BaseNodeData
 from dify_graph.entities.graph_config import NodeConfigDict, NodeConfigDictAdapter
 from dify_graph.entities.graph_init_params import DIFY_RUN_CONTEXT_KEY
-from dify_graph.enums import BuiltinNodeTypes, NodeType, SystemVariableKey
+from dify_graph.enums import NodeType, SystemVariableKey
 from dify_graph.file.file_manager import file_manager
 from dify_graph.graph.graph import NodeFactory
 from dify_graph.model_runtime.entities.model_entities import ModelType
@@ -45,7 +39,7 @@ from dify_graph.nodes.document_extractor import UnstructuredApiConfig
 from dify_graph.nodes.http_request import build_http_request_config
 from dify_graph.nodes.llm.entities import LLMNodeData
 from dify_graph.nodes.llm.exc import LLMModeRequiredError, ModelNotExistError
-from dify_graph.nodes.llm.protocols import TemplateRenderer
+from dify_graph.nodes.node_mapping import LATEST_VERSION, NODE_TYPE_CLASSES_MAPPING
 from dify_graph.nodes.parameter_extractor.entities import ParameterExtractorNodeData
 from dify_graph.nodes.question_classifier.entities import QuestionClassifierNodeData
 from dify_graph.nodes.template_transform.template_renderer import (
@@ -59,135 +53,6 @@ if TYPE_CHECKING:
     from dify_graph.entities import GraphInitParams
     from dify_graph.runtime import GraphRuntimeState
 
-LATEST_VERSION = "latest"
-_START_NODE_TYPES: frozenset[NodeType] = frozenset(
-    (BuiltinNodeTypes.START, BuiltinNodeTypes.DATASOURCE, *TRIGGER_NODE_TYPES)
-)
-
-
-def _import_node_package(package_name: str, *, excluded_modules: frozenset[str] = frozenset()) -> None:
-    package = importlib.import_module(package_name)
-    for _, module_name, _ in pkgutil.walk_packages(package.__path__, package.__name__ + "."):
-        if module_name in excluded_modules:
-            continue
-        importlib.import_module(module_name)
-
-
-@lru_cache(maxsize=1)
-def register_nodes() -> None:
-    """Import production node modules so they self-register with ``Node``."""
-    _import_node_package("dify_graph.nodes")
-    _import_node_package("core.workflow.nodes")
-
-
-def get_node_type_classes_mapping() -> Mapping[NodeType, Mapping[str, type[Node]]]:
-    """Return a read-only snapshot of the current production node registry.
-
-    The workflow layer owns node bootstrap because it must compose built-in
-    `dify_graph.nodes.*` implementations with workflow-local nodes under
-    `core.workflow.nodes.*`. Keeping this import side effect here avoids
-    reintroducing registry bootstrapping into lower-level graph primitives.
-    """
-    register_nodes()
-    return Node.get_node_type_classes_mapping()
-
-
-def resolve_workflow_node_class(*, node_type: NodeType, node_version: str) -> type[Node]:
-    node_mapping = get_node_type_classes_mapping().get(node_type)
-    if not node_mapping:
-        raise ValueError(f"No class mapping found for node type: {node_type}")
-
-    latest_node_class = node_mapping.get(LATEST_VERSION)
-    matched_node_class = node_mapping.get(node_version)
-    node_class = matched_node_class or latest_node_class
-    if not node_class:
-        raise ValueError(f"No latest version class found for node type: {node_type}")
-    return node_class
-
-
-def is_start_node_type(node_type: NodeType) -> bool:
-    """Return True when the node type can serve as a workflow entry point."""
-    return node_type in _START_NODE_TYPES
-
-
-def get_default_root_node_id(graph_config: Mapping[str, Any]) -> str:
-    """Resolve the default entry node for a persisted top-level workflow graph.
-
-    This workflow-layer helper depends on start-node semantics defined by
-    `is_start_node_type`, so it intentionally lives next to the node registry
-    instead of in the raw `dify_graph.entities.graph_config` schema module.
-    """
-    nodes = graph_config.get("nodes")
-    if not isinstance(nodes, list):
-        raise ValueError("nodes in workflow graph must be a list")
-
-    for node in nodes:
-        if not isinstance(node, Mapping):
-            continue
-
-        if node.get("type") == "custom-note":
-            continue
-
-        node_id = node.get("id")
-        data = node.get("data")
-        if not isinstance(node_id, str) or not isinstance(data, Mapping):
-            continue
-
-        node_type = data.get("type")
-        if isinstance(node_type, str) and is_start_node_type(node_type):
-            return node_id
-
-    raise ValueError("Unable to determine default root node ID from workflow graph")
-
-
-class _LazyNodeTypeClassesMapping(MutableMapping[NodeType, Mapping[str, type[Node]]]):
-    """Mutable dict-like view over the current node registry."""
-
-    def __init__(self) -> None:
-        self._cached_snapshot: dict[NodeType, Mapping[str, type[Node]]] = {}
-        self._cached_version = -1
-        self._deleted: set[NodeType] = set()
-        self._overrides: dict[NodeType, Mapping[str, type[Node]]] = {}
-
-    def _snapshot(self) -> dict[NodeType, Mapping[str, type[Node]]]:
-        current_version = Node.get_registry_version()
-        if self._cached_version != current_version:
-            self._cached_snapshot = dict(get_node_type_classes_mapping())
-            self._cached_version = current_version
-        if not self._deleted and not self._overrides:
-            return self._cached_snapshot
-
-        snapshot = {key: value for key, value in self._cached_snapshot.items() if key not in self._deleted}
-        snapshot.update(self._overrides)
-        return snapshot
-
-    def __getitem__(self, key: NodeType) -> Mapping[str, type[Node]]:
-        return self._snapshot()[key]
-
-    def __setitem__(self, key: NodeType, value: Mapping[str, type[Node]]) -> None:
-        self._deleted.discard(key)
-        self._overrides[key] = value
-
-    def __delitem__(self, key: NodeType) -> None:
-        if key in self._overrides:
-            del self._overrides[key]
-            return
-        if key in self._cached_snapshot:
-            self._deleted.add(key)
-            return
-        raise KeyError(key)
-
-    def __iter__(self) -> Iterator[NodeType]:
-        return iter(self._snapshot())
-
-    def __len__(self) -> int:
-        return len(self._snapshot())
-
-
-# Keep the canonical node-class mapping in the workflow layer that also bootstraps
-# legacy `core.workflow.nodes.*` registrations.
-NODE_TYPE_CLASSES_MAPPING: MutableMapping[NodeType, Mapping[str, type[Node]]] = _LazyNodeTypeClassesMapping()
-
 
 LLMCompatibleNodeData: TypeAlias = LLMNodeData | QuestionClassifierNodeData | ParameterExtractorNodeData
 
@@ -229,20 +94,13 @@ class DefaultWorkflowCodeExecutor:
         return isinstance(error, CodeExecutionError)
 
 
-class DefaultLLMTemplateRenderer(TemplateRenderer):
-    def render_jinja2(self, *, template: str, inputs: Mapping[str, Any]) -> str:
-        result = CodeExecutor.execute_workflow_code_template(
-            language=CodeLanguage.JINJA2,
-            code=template,
-            inputs=inputs,
-        )
-        return str(result.get("result", ""))
-
-
 @final
 class DifyNodeFactory(NodeFactory):
     """
-    Default implementation of NodeFactory that resolves node classes from the live registry.
+    Default implementation of NodeFactory that uses the traditional node mapping.
+
+    This factory creates nodes by looking up their types in NODE_TYPE_CLASSES_MAPPING
+    and instantiating the appropriate node class.
     """
 
     def __init__(
@@ -265,11 +123,11 @@ class DifyNodeFactory(NodeFactory):
             max_object_array_length=dify_config.CODE_MAX_OBJECT_ARRAY_LENGTH,
         )
         self._template_renderer = CodeExecutorJinja2TemplateRenderer(code_executor=self._code_executor)
-        self._llm_template_renderer: TemplateRenderer = DefaultLLMTemplateRenderer()
         self._template_transform_max_output_length = dify_config.TEMPLATE_TRANSFORM_MAX_LENGTH
         self._http_request_http_client = ssrf_proxy
         self._http_request_tool_file_manager_factory = ToolFileManager
         self._http_request_file_manager = file_manager
+        self._rag_retrieval = DatasetRetrieval()
         self._document_extractor_unstructured_api_config = UnstructuredApiConfig(
             api_url=dify_config.UNSTRUCTURED_API_URL,
             api_key=dify_config.UNSTRUCTURED_API_KEY or "",
@@ -285,10 +143,6 @@ class DifyNodeFactory(NodeFactory):
         )
 
         self._llm_credentials_provider, self._llm_model_factory = build_dify_model_access(self._dify_context.tenant_id)
-        self._agent_strategy_resolver = PluginAgentStrategyResolver()
-        self._agent_strategy_presentation_provider = PluginAgentStrategyPresentationProvider()
-        self._agent_runtime_support = AgentRuntimeSupport()
-        self._agent_message_transformer = AgentMessageTransformer()
 
     @staticmethod
     def _resolve_dify_context(run_context: Mapping[str, Any]) -> DifyRunContext:
@@ -316,51 +170,55 @@ class DifyNodeFactory(NodeFactory):
         node_class = self._resolve_node_class(node_type=node_data.type, node_version=str(node_data.version))
         node_type = node_data.type
         node_init_kwargs_factories: Mapping[NodeType, Callable[[], dict[str, object]]] = {
-            BuiltinNodeTypes.CODE: lambda: {
+            NodeType.CODE: lambda: {
                 "code_executor": self._code_executor,
                 "code_limits": self._code_limits,
             },
-            BuiltinNodeTypes.TEMPLATE_TRANSFORM: lambda: {
+            NodeType.TEMPLATE_TRANSFORM: lambda: {
                 "template_renderer": self._template_renderer,
                 "max_output_length": self._template_transform_max_output_length,
             },
-            BuiltinNodeTypes.HTTP_REQUEST: lambda: {
+            NodeType.HTTP_REQUEST: lambda: {
                 "http_request_config": self._http_request_config,
                 "http_client": self._http_request_http_client,
                 "tool_file_manager_factory": self._http_request_tool_file_manager_factory,
                 "file_manager": self._http_request_file_manager,
             },
-            BuiltinNodeTypes.HUMAN_INPUT: lambda: {
+            NodeType.HUMAN_INPUT: lambda: {
                 "form_repository": HumanInputFormRepositoryImpl(tenant_id=self._dify_context.tenant_id),
             },
-            BuiltinNodeTypes.LLM: lambda: self._build_llm_compatible_node_init_kwargs(
+            NodeType.KNOWLEDGE_INDEX: lambda: {
+                "index_processor": IndexProcessor(),
+                "summary_index_service": SummaryIndex(),
+            },
+            NodeType.LLM: lambda: self._build_llm_compatible_node_init_kwargs(
                 node_class=node_class,
                 node_data=node_data,
                 include_http_client=True,
             ),
-            BuiltinNodeTypes.DOCUMENT_EXTRACTOR: lambda: {
+            NodeType.DATASOURCE: lambda: {
+                "datasource_manager": DatasourceManager,
+            },
+            NodeType.KNOWLEDGE_RETRIEVAL: lambda: {
+                "rag_retrieval": self._rag_retrieval,
+            },
+            NodeType.DOCUMENT_EXTRACTOR: lambda: {
                 "unstructured_api_config": self._document_extractor_unstructured_api_config,
                 "http_client": self._http_request_http_client,
             },
-            BuiltinNodeTypes.QUESTION_CLASSIFIER: lambda: self._build_llm_compatible_node_init_kwargs(
+            NodeType.QUESTION_CLASSIFIER: lambda: self._build_llm_compatible_node_init_kwargs(
                 node_class=node_class,
                 node_data=node_data,
                 include_http_client=True,
             ),
-            BuiltinNodeTypes.PARAMETER_EXTRACTOR: lambda: self._build_llm_compatible_node_init_kwargs(
+            NodeType.PARAMETER_EXTRACTOR: lambda: self._build_llm_compatible_node_init_kwargs(
                 node_class=node_class,
                 node_data=node_data,
                 include_http_client=False,
             ),
-            BuiltinNodeTypes.TOOL: lambda: {
+            NodeType.TOOL: lambda: {
                 "tool_file_manager_factory": self._http_request_tool_file_manager_factory(),
             },
-            BuiltinNodeTypes.AGENT: lambda: {
-                "strategy_resolver": self._agent_strategy_resolver,
-                "presentation_provider": self._agent_strategy_presentation_provider,
-                "runtime_support": self._agent_runtime_support,
-                "message_transformer": self._agent_message_transformer,
-            },
         }
         node_init_kwargs = node_init_kwargs_factories.get(node_type, lambda: {})()
         return node_class(
@@ -380,7 +238,16 @@ class DifyNodeFactory(NodeFactory):
 
     @staticmethod
     def _resolve_node_class(*, node_type: NodeType, node_version: str) -> type[Node]:
-        return resolve_workflow_node_class(node_type=node_type, node_version=node_version)
+        node_mapping = NODE_TYPE_CLASSES_MAPPING.get(node_type)
+        if not node_mapping:
+            raise ValueError(f"No class mapping found for node type: {node_type}")
+
+        latest_node_class = node_mapping.get(LATEST_VERSION)
+        matched_node_class = node_mapping.get(node_version)
+        node_class = matched_node_class or latest_node_class
+        if not node_class:
+            raise ValueError(f"No latest version class found for node type: {node_type}")
+        return node_class
 
     def _build_llm_compatible_node_init_kwargs(
         self,
@@ -403,8 +270,6 @@ class DifyNodeFactory(NodeFactory):
                 model_instance=model_instance,
             ),
         }
-        if validated_node_data.type in {BuiltinNodeTypes.LLM, BuiltinNodeTypes.QUESTION_CLASSIFIER}:
-            node_init_kwargs["template_renderer"] = self._llm_template_renderer
         if include_http_client:
             node_init_kwargs["http_client"] = self._http_request_http_client
         return node_init_kwargs

api/core/workflow/nodes/__init__.py

@@ -1 +0,0 @@
-"""Workflow node implementations that remain under the legacy core.workflow namespace."""

api/core/workflow/nodes/agent/__init__.py

@@ -1,4 +0,0 @@
-from .agent_node import AgentNode
-from .entities import AgentNodeData
-
-__all__ = ["AgentNode", "AgentNodeData"]

api/core/workflow/nodes/agent/agent_node.py

@@ -1,188 +0,0 @@
-from __future__ import annotations
-
-from collections.abc import Generator, Mapping, Sequence
-from typing import TYPE_CHECKING, Any
-
-from dify_graph.entities.graph_config import NodeConfigDict
-from dify_graph.enums import BuiltinNodeTypes, SystemVariableKey, WorkflowNodeExecutionStatus
-from dify_graph.node_events import NodeEventBase, NodeRunResult, StreamCompletedEvent
-from dify_graph.nodes.base.node import Node
-from dify_graph.nodes.base.variable_template_parser import VariableTemplateParser
-
-from .entities import AgentNodeData
-from .exceptions import (
-    AgentInvocationError,
-    AgentMessageTransformError,
-)
-from .message_transformer import AgentMessageTransformer
-from .runtime_support import AgentRuntimeSupport
-from .strategy_protocols import AgentStrategyPresentationProvider, AgentStrategyResolver
-
-if TYPE_CHECKING:
-    from dify_graph.entities import GraphInitParams
-    from dify_graph.runtime import GraphRuntimeState
-
-
-class AgentNode(Node[AgentNodeData]):
-    node_type = BuiltinNodeTypes.AGENT
-
-    _strategy_resolver: AgentStrategyResolver
-    _presentation_provider: AgentStrategyPresentationProvider
-    _runtime_support: AgentRuntimeSupport
-    _message_transformer: AgentMessageTransformer
-
-    def __init__(
-        self,
-        id: str,
-        config: NodeConfigDict,
-        graph_init_params: GraphInitParams,
-        graph_runtime_state: GraphRuntimeState,
-        *,
-        strategy_resolver: AgentStrategyResolver,
-        presentation_provider: AgentStrategyPresentationProvider,
-        runtime_support: AgentRuntimeSupport,
-        message_transformer: AgentMessageTransformer,
-    ) -> None:
-        super().__init__(
-            id=id,
-            config=config,
-            graph_init_params=graph_init_params,
-            graph_runtime_state=graph_runtime_state,
-        )
-        self._strategy_resolver = strategy_resolver
-        self._presentation_provider = presentation_provider
-        self._runtime_support = runtime_support
-        self._message_transformer = message_transformer
-
-    @classmethod
-    def version(cls) -> str:
-        return "1"
-
-    def populate_start_event(self, event) -> None:
-        dify_ctx = self.require_dify_context()
-        event.extras["agent_strategy"] = {
-            "name": self.node_data.agent_strategy_name,
-            "icon": self._presentation_provider.get_icon(
-                tenant_id=dify_ctx.tenant_id,
-                agent_strategy_provider_name=self.node_data.agent_strategy_provider_name,
-            ),
-        }
-
-    def _run(self) -> Generator[NodeEventBase, None, None]:
-        from core.plugin.impl.exc import PluginDaemonClientSideError
-
-        dify_ctx = self.require_dify_context()
-
-        try:
-            strategy = self._strategy_resolver.resolve(
-                tenant_id=dify_ctx.tenant_id,
-                agent_strategy_provider_name=self.node_data.agent_strategy_provider_name,
-                agent_strategy_name=self.node_data.agent_strategy_name,
-            )
-        except Exception as e:
-            yield StreamCompletedEvent(
-                node_run_result=NodeRunResult(
-                    status=WorkflowNodeExecutionStatus.FAILED,
-                    inputs={},
-                    error=f"Failed to get agent strategy: {str(e)}",
-                ),
-            )
-            return
-
-        agent_parameters = strategy.get_parameters()
-
-        parameters = self._runtime_support.build_parameters(
-            agent_parameters=agent_parameters,
-            variable_pool=self.graph_runtime_state.variable_pool,
-            node_data=self.node_data,
-            strategy=strategy,
-            tenant_id=dify_ctx.tenant_id,
-            app_id=dify_ctx.app_id,
-            invoke_from=dify_ctx.invoke_from,
-        )
-        parameters_for_log = self._runtime_support.build_parameters(
-            agent_parameters=agent_parameters,
-            variable_pool=self.graph_runtime_state.variable_pool,
-            node_data=self.node_data,
-            strategy=strategy,
-            tenant_id=dify_ctx.tenant_id,
-            app_id=dify_ctx.app_id,
-            invoke_from=dify_ctx.invoke_from,
-            for_log=True,
-        )
-        credentials = self._runtime_support.build_credentials(parameters=parameters)
-
-        conversation_id = self.graph_runtime_state.variable_pool.get(["sys", SystemVariableKey.CONVERSATION_ID])
-
-        try:
-            message_stream = strategy.invoke(
-                params=parameters,
-                user_id=dify_ctx.user_id,
-                app_id=dify_ctx.app_id,
-                conversation_id=conversation_id.text if conversation_id else None,
-                credentials=credentials,
-            )
-        except Exception as e:
-            error = AgentInvocationError(f"Failed to invoke agent: {str(e)}", original_error=e)
-            yield StreamCompletedEvent(
-                node_run_result=NodeRunResult(
-                    status=WorkflowNodeExecutionStatus.FAILED,
-                    inputs=parameters_for_log,
-                    error=str(error),
-                )
-            )
-            return
-
-        try:
-            yield from self._message_transformer.transform(
-                messages=message_stream,
-                tool_info={
-                    "icon": self._presentation_provider.get_icon(
-                        tenant_id=dify_ctx.tenant_id,
-                        agent_strategy_provider_name=self.node_data.agent_strategy_provider_name,
-                    ),
-                    "agent_strategy": self.node_data.agent_strategy_name,
-                },
-                parameters_for_log=parameters_for_log,
-                user_id=dify_ctx.user_id,
-                tenant_id=dify_ctx.tenant_id,
-                node_type=self.node_type,
-                node_id=self._node_id,
-                node_execution_id=self.id,
-            )
-        except PluginDaemonClientSideError as e:
-            transform_error = AgentMessageTransformError(
-                f"Failed to transform agent message: {str(e)}", original_error=e
-            )
-            yield StreamCompletedEvent(
-                node_run_result=NodeRunResult(
-                    status=WorkflowNodeExecutionStatus.FAILED,
-                    inputs=parameters_for_log,
-                    error=str(transform_error),
-                )
-            )
-
-    @classmethod
-    def _extract_variable_selector_to_variable_mapping(
-        cls,
-        *,
-        graph_config: Mapping[str, Any],
-        node_id: str,
-        node_data: AgentNodeData,
-    ) -> Mapping[str, Sequence[str]]:
-        _ = graph_config  # Explicitly mark as unused
-        result: dict[str, Any] = {}
-        typed_node_data = node_data
-        for parameter_name in typed_node_data.agent_parameters:
-            input = typed_node_data.agent_parameters[parameter_name]
-            match input.type:
-                case "mixed" | "constant":
-                    selectors = VariableTemplateParser(str(input.value)).extract_variable_selectors()
-                    for selector in selectors:
-                        result[selector.variable] = selector.value_selector
-                case "variable":
-                    result[parameter_name] = input.value
-
-        result = {node_id + "." + key: value for key, value in result.items()}
-
-        return result

api/core/workflow/nodes/agent/message_transformer.py

@@ -1,292 +0,0 @@
-from __future__ import annotations
-
-from collections.abc import Generator, Mapping
-from typing import Any, cast
-
-from sqlalchemy import select
-from sqlalchemy.orm import Session
-
-from core.tools.entities.tool_entities import ToolInvokeMessage
-from core.tools.utils.message_transformer import ToolFileMessageTransformer
-from dify_graph.enums import BuiltinNodeTypes, NodeType, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
-from dify_graph.file import File, FileTransferMethod
-from dify_graph.model_runtime.entities.llm_entities import LLMUsage, LLMUsageMetadata
-from dify_graph.model_runtime.utils.encoders import jsonable_encoder
-from dify_graph.node_events import (
-    AgentLogEvent,
-    NodeEventBase,
-    NodeRunResult,
-    StreamChunkEvent,
-    StreamCompletedEvent,
-)
-from dify_graph.variables.segments import ArrayFileSegment
-from extensions.ext_database import db
-from factories import file_factory
-from models import ToolFile
-from services.tools.builtin_tools_manage_service import BuiltinToolManageService
-
-from .exceptions import AgentNodeError, AgentVariableTypeError, ToolFileNotFoundError
-
-
-class AgentMessageTransformer:
-    def transform(
-        self,
-        *,
-        messages: Generator[ToolInvokeMessage, None, None],
-        tool_info: Mapping[str, Any],
-        parameters_for_log: dict[str, Any],
-        user_id: str,
-        tenant_id: str,
-        node_type: NodeType,
-        node_id: str,
-        node_execution_id: str,
-    ) -> Generator[NodeEventBase, None, None]:
-        from core.plugin.impl.plugin import PluginInstaller
-
-        message_stream = ToolFileMessageTransformer.transform_tool_invoke_messages(
-            messages=messages,
-            user_id=user_id,
-            tenant_id=tenant_id,
-            conversation_id=None,
-        )
-
-        text = ""
-        files: list[File] = []
-        json_list: list[dict | list] = []
-
-        agent_logs: list[AgentLogEvent] = []
-        agent_execution_metadata: Mapping[WorkflowNodeExecutionMetadataKey, Any] = {}
-        llm_usage = LLMUsage.empty_usage()
-        variables: dict[str, Any] = {}
-
-        for message in message_stream:
-            if message.type in {
-                ToolInvokeMessage.MessageType.IMAGE_LINK,
-                ToolInvokeMessage.MessageType.BINARY_LINK,
-                ToolInvokeMessage.MessageType.IMAGE,
-            }:
-                assert isinstance(message.message, ToolInvokeMessage.TextMessage)
-
-                url = message.message.text
-                if message.meta:
-                    transfer_method = message.meta.get("transfer_method", FileTransferMethod.TOOL_FILE)
-                else:
-                    transfer_method = FileTransferMethod.TOOL_FILE
-
-                tool_file_id = str(url).split("/")[-1].split(".")[0]
-
-                with Session(db.engine) as session:
-                    stmt = select(ToolFile).where(ToolFile.id == tool_file_id)
-                    tool_file = session.scalar(stmt)
-                    if tool_file is None:
-                        raise ToolFileNotFoundError(tool_file_id)
-
-                mapping = {
-                    "tool_file_id": tool_file_id,
-                    "type": file_factory.get_file_type_by_mime_type(tool_file.mimetype),
-                    "transfer_method": transfer_method,
-                    "url": url,
-                }
-                file = file_factory.build_from_mapping(
-                    mapping=mapping,
-                    tenant_id=tenant_id,
-                )
-                files.append(file)
-            elif message.type == ToolInvokeMessage.MessageType.BLOB:
-                assert isinstance(message.message, ToolInvokeMessage.TextMessage)
-                assert message.meta
-
-                tool_file_id = message.message.text.split("/")[-1].split(".")[0]
-                with Session(db.engine) as session:
-                    stmt = select(ToolFile).where(ToolFile.id == tool_file_id)
-                    tool_file = session.scalar(stmt)
-                    if tool_file is None:
-                        raise ToolFileNotFoundError(tool_file_id)
-
-                mapping = {
-                    "tool_file_id": tool_file_id,
-                    "transfer_method": FileTransferMethod.TOOL_FILE,
-                }
-                files.append(
-                    file_factory.build_from_mapping(
-                        mapping=mapping,
-                        tenant_id=tenant_id,
-                    )
-                )
-            elif message.type == ToolInvokeMessage.MessageType.TEXT:
-                assert isinstance(message.message, ToolInvokeMessage.TextMessage)
-                text += message.message.text
-                yield StreamChunkEvent(
-                    selector=[node_id, "text"],
-                    chunk=message.message.text,
-                    is_final=False,
-                )
-            elif message.type == ToolInvokeMessage.MessageType.JSON:
-                assert isinstance(message.message, ToolInvokeMessage.JsonMessage)
-                if node_type == BuiltinNodeTypes.AGENT:
-                    if isinstance(message.message.json_object, dict):
-                        msg_metadata: dict[str, Any] = message.message.json_object.pop("execution_metadata", {})
-                        llm_usage = LLMUsage.from_metadata(cast(LLMUsageMetadata, msg_metadata))
-                        agent_execution_metadata = {
-                            WorkflowNodeExecutionMetadataKey(key): value
-                            for key, value in msg_metadata.items()
-                            if key in WorkflowNodeExecutionMetadataKey.__members__.values()
-                        }
-                    else:
-                        llm_usage = LLMUsage.empty_usage()
-                        agent_execution_metadata = {}
-                if message.message.json_object:
-                    json_list.append(message.message.json_object)
-            elif message.type == ToolInvokeMessage.MessageType.LINK:
-                assert isinstance(message.message, ToolInvokeMessage.TextMessage)
-                stream_text = f"Link: {message.message.text}\n"
-                text += stream_text
-                yield StreamChunkEvent(
-                    selector=[node_id, "text"],
-                    chunk=stream_text,
-                    is_final=False,
-                )
-            elif message.type == ToolInvokeMessage.MessageType.VARIABLE:
-                assert isinstance(message.message, ToolInvokeMessage.VariableMessage)
-                variable_name = message.message.variable_name
-                variable_value = message.message.variable_value
-                if message.message.stream:
-                    if not isinstance(variable_value, str):
-                        raise AgentVariableTypeError(
-                            "When 'stream' is True, 'variable_value' must be a string.",
-                            variable_name=variable_name,
-                            expected_type="str",
-                            actual_type=type(variable_value).__name__,
-                        )
-                    if variable_name not in variables:
-                        variables[variable_name] = ""
-                    variables[variable_name] += variable_value
-
-                    yield StreamChunkEvent(
-                        selector=[node_id, variable_name],
-                        chunk=variable_value,
-                        is_final=False,
-                    )
-                else:
-                    variables[variable_name] = variable_value
-            elif message.type == ToolInvokeMessage.MessageType.FILE:
-                assert message.meta is not None
-                assert isinstance(message.meta, dict)
-                if "file" not in message.meta:
-                    raise AgentNodeError("File message is missing 'file' key in meta")
-
-                if not isinstance(message.meta["file"], File):
-                    raise AgentNodeError(f"Expected File object but got {type(message.meta['file']).__name__}")
-                files.append(message.meta["file"])
-            elif message.type == ToolInvokeMessage.MessageType.LOG:
-                assert isinstance(message.message, ToolInvokeMessage.LogMessage)
-                if message.message.metadata:
-                    icon = tool_info.get("icon", "")
-                    dict_metadata = dict(message.message.metadata)
-                    if dict_metadata.get("provider"):
-                        manager = PluginInstaller()
-                        plugins = manager.list_plugins(tenant_id)
-                        try:
-                            current_plugin = next(
-                                plugin
-                                for plugin in plugins
-                                if f"{plugin.plugin_id}/{plugin.name}" == dict_metadata["provider"]
-                            )
-                            icon = current_plugin.declaration.icon
-                        except StopIteration:
-                            pass
-                        icon_dark = None
-                        try:
-                            builtin_tool = next(
-                                provider
-                                for provider in BuiltinToolManageService.list_builtin_tools(
-                                    user_id,
-                                    tenant_id,
-                                )
-                                if provider.name == dict_metadata["provider"]
-                            )
-                            icon = builtin_tool.icon
-                            icon_dark = builtin_tool.icon_dark
-                        except StopIteration:
-                            pass
-
-                        dict_metadata["icon"] = icon
-                        dict_metadata["icon_dark"] = icon_dark
-                        message.message.metadata = dict_metadata
-                agent_log = AgentLogEvent(
-                    message_id=message.message.id,
-                    node_execution_id=node_execution_id,
-                    parent_id=message.message.parent_id,
-                    error=message.message.error,
-                    status=message.message.status.value,
-                    data=message.message.data,
-                    label=message.message.label,
-                    metadata=message.message.metadata,
-                    node_id=node_id,
-                )
-
-                for log in agent_logs:
-                    if log.message_id == agent_log.message_id:
-                        log.data = agent_log.data
-                        log.status = agent_log.status
-                        log.error = agent_log.error
-                        log.label = agent_log.label
-                        log.metadata = agent_log.metadata
-                        break
-                else:
-                    agent_logs.append(agent_log)
-
-                yield agent_log
-
-        json_output: list[dict[str, Any] | list[Any]] = []
-        if agent_logs:
-            for log in agent_logs:
-                json_output.append(
-                    {
-                        "id": log.message_id,
-                        "parent_id": log.parent_id,
-                        "error": log.error,
-                        "status": log.status,
-                        "data": log.data,
-                        "label": log.label,
-                        "metadata": log.metadata,
-                        "node_id": log.node_id,
-                    }
-                )
-        if json_list:
-            json_output.extend(json_list)
-        else:
-            json_output.append({"data": []})
-
-        yield StreamChunkEvent(
-            selector=[node_id, "text"],
-            chunk="",
-            is_final=True,
-        )
-
-        for var_name in variables:
-            yield StreamChunkEvent(
-                selector=[node_id, var_name],
-                chunk="",
-                is_final=True,
-            )
-
-        yield StreamCompletedEvent(
-            node_run_result=NodeRunResult(
-                status=WorkflowNodeExecutionStatus.SUCCEEDED,
-                outputs={
-                    "text": text,
-                    "usage": jsonable_encoder(llm_usage),
-                    "files": ArrayFileSegment(value=files),
-                    "json": json_output,
-                    **variables,
-                },
-                metadata={
-                    **agent_execution_metadata,
-                    WorkflowNodeExecutionMetadataKey.TOOL_INFO: tool_info,
-                    WorkflowNodeExecutionMetadataKey.AGENT_LOG: agent_logs,
-                },
-                inputs=parameters_for_log,
-                llm_usage=llm_usage,
-            )
-        )

api/core/workflow/nodes/agent/plugin_strategy_adapter.py

@@ -1,40 +0,0 @@
-from __future__ import annotations
-
-from factories.agent_factory import get_plugin_agent_strategy
-
-from .strategy_protocols import AgentStrategyPresentationProvider, AgentStrategyResolver, ResolvedAgentStrategy
-
-
-class PluginAgentStrategyResolver(AgentStrategyResolver):
-    def resolve(
-        self,
-        *,
-        tenant_id: str,
-        agent_strategy_provider_name: str,
-        agent_strategy_name: str,
-    ) -> ResolvedAgentStrategy:
-        return get_plugin_agent_strategy(
-            tenant_id=tenant_id,
-            agent_strategy_provider_name=agent_strategy_provider_name,
-            agent_strategy_name=agent_strategy_name,
-        )
-
-
-class PluginAgentStrategyPresentationProvider(AgentStrategyPresentationProvider):
-    def get_icon(self, *, tenant_id: str, agent_strategy_provider_name: str) -> str | None:
-        from core.plugin.impl.plugin import PluginInstaller
-
-        manager = PluginInstaller()
-        try:
-            plugins = manager.list_plugins(tenant_id)
-        except Exception:
-            return None
-
-        try:
-            current_plugin = next(
-                plugin for plugin in plugins if f"{plugin.plugin_id}/{plugin.name}" == agent_strategy_provider_name
-            )
-        except StopIteration:
-            return None
-
-        return current_plugin.declaration.icon

api/core/workflow/nodes/agent/runtime_support.py

@@ -1,276 +0,0 @@
-from __future__ import annotations
-
-import json
-from collections.abc import Sequence
-from typing import Any, cast
-
-from packaging.version import Version
-from pydantic import ValidationError
-from sqlalchemy import select
-from sqlalchemy.orm import Session
-
-from core.agent.entities import AgentToolEntity
-from core.agent.plugin_entities import AgentStrategyParameter
-from core.memory.token_buffer_memory import TokenBufferMemory
-from core.model_manager import ModelInstance, ModelManager
-from core.plugin.entities.request import InvokeCredentials
-from core.provider_manager import ProviderManager
-from core.tools.entities.tool_entities import ToolIdentity, ToolParameter, ToolProviderType
-from core.tools.tool_manager import ToolManager
-from dify_graph.enums import SystemVariableKey
-from dify_graph.model_runtime.entities.model_entities import AIModelEntity, ModelType
-from dify_graph.runtime import VariablePool
-from dify_graph.variables.segments import StringSegment
-from extensions.ext_database import db
-from models.model import Conversation
-
-from .entities import AgentNodeData, AgentOldVersionModelFeatures, ParamsAutoGenerated
-from .exceptions import AgentInputTypeError, AgentVariableNotFoundError
-from .strategy_protocols import ResolvedAgentStrategy
-
-
-class AgentRuntimeSupport:
-    def build_parameters(
-        self,
-        *,
-        agent_parameters: Sequence[AgentStrategyParameter],
-        variable_pool: VariablePool,
-        node_data: AgentNodeData,
-        strategy: ResolvedAgentStrategy,
-        tenant_id: str,
-        app_id: str,
-        invoke_from: Any,
-        for_log: bool = False,
-    ) -> dict[str, Any]:
-        agent_parameters_dictionary = {parameter.name: parameter for parameter in agent_parameters}
-
-        result: dict[str, Any] = {}
-        for parameter_name in node_data.agent_parameters:
-            parameter = agent_parameters_dictionary.get(parameter_name)
-            if not parameter:
-                result[parameter_name] = None
-                continue
-
-            agent_input = node_data.agent_parameters[parameter_name]
-            match agent_input.type:
-                case "variable":
-                    variable = variable_pool.get(agent_input.value)  # type: ignore[arg-type]
-                    if variable is None:
-                        raise AgentVariableNotFoundError(str(agent_input.value))
-                    parameter_value = variable.value
-                case "mixed" | "constant":
-                    try:
-                        if not isinstance(agent_input.value, str):
-                            parameter_value = json.dumps(agent_input.value, ensure_ascii=False)
-                        else:
-                            parameter_value = str(agent_input.value)
-                    except TypeError:
-                        parameter_value = str(agent_input.value)
-
-                    segment_group = variable_pool.convert_template(parameter_value)
-                    parameter_value = segment_group.log if for_log else segment_group.text
-                    try:
-                        if not isinstance(agent_input.value, str):
-                            parameter_value = json.loads(parameter_value)
-                    except json.JSONDecodeError:
-                        parameter_value = parameter_value
-                case _:
-                    raise AgentInputTypeError(agent_input.type)
-
-            value = parameter_value
-            if parameter.type == "array[tools]":
-                value = cast(list[dict[str, Any]], value)
-                value = [tool for tool in value if tool.get("enabled", False)]
-                value = self._filter_mcp_type_tool(strategy, value)
-                for tool in value:
-                    if "schemas" in tool:
-                        tool.pop("schemas")
-                    parameters = tool.get("parameters", {})
-                    if all(isinstance(v, dict) for _, v in parameters.items()):
-                        params = {}
-                        for key, param in parameters.items():
-                            if param.get("auto", ParamsAutoGenerated.OPEN) in (
-                                ParamsAutoGenerated.CLOSE,
-                                0,
-                            ):
-                                value_param = param.get("value", {})
-                                if value_param and value_param.get("type", "") == "variable":
-                                    variable_selector = value_param.get("value")
-                                    if not variable_selector:
-                                        raise ValueError("Variable selector is missing for a variable-type parameter.")
-
-                                    variable = variable_pool.get(variable_selector)
-                                    if variable is None:
-                                        raise AgentVariableNotFoundError(str(variable_selector))
-
-                                    params[key] = variable.value
-                                else:
-                                    params[key] = value_param.get("value", "") if value_param is not None else None
-                            else:
-                                params[key] = None
-                        parameters = params
-                    tool["settings"] = {k: v.get("value", None) for k, v in tool.get("settings", {}).items()}
-                    tool["parameters"] = parameters
-
-            if not for_log:
-                if parameter.type == "array[tools]":
-                    value = cast(list[dict[str, Any]], value)
-                    tool_value = []
-                    for tool in value:
-                        provider_type = ToolProviderType(tool.get("type", ToolProviderType.BUILT_IN))
-                        setting_params = tool.get("settings", {})
-                        parameters = tool.get("parameters", {})
-                        manual_input_params = [key for key, value in parameters.items() if value is not None]
-
-                        parameters = {**parameters, **setting_params}
-                        entity = AgentToolEntity(
-                            provider_id=tool.get("provider_name", ""),
-                            provider_type=provider_type,
-                            tool_name=tool.get("tool_name", ""),
-                            tool_parameters=parameters,
-                            plugin_unique_identifier=tool.get("plugin_unique_identifier", None),
-                            credential_id=tool.get("credential_id", None),
-                        )
-
-                        extra = tool.get("extra", {})
-
-                        runtime_variable_pool: VariablePool | None = None
-                        if node_data.version != "1" or node_data.tool_node_version is not None:
-                            runtime_variable_pool = variable_pool
-                        tool_runtime = ToolManager.get_agent_tool_runtime(
-                            tenant_id,
-                            app_id,
-                            entity,
-                            invoke_from,
-                            runtime_variable_pool,
-                        )
-                        if tool_runtime.entity.description:
-                            tool_runtime.entity.description.llm = (
-                                extra.get("description", "") or tool_runtime.entity.description.llm
-                            )
-                        for tool_runtime_params in tool_runtime.entity.parameters:
-                            tool_runtime_params.form = (
-                                ToolParameter.ToolParameterForm.FORM
-                                if tool_runtime_params.name in manual_input_params
-                                else tool_runtime_params.form
-                            )
-                        manual_input_value = {}
-                        if tool_runtime.entity.parameters:
-                            manual_input_value = {
-                                key: value for key, value in parameters.items() if key in manual_input_params
-                            }
-                        runtime_parameters = {
-                            **tool_runtime.runtime.runtime_parameters,
-                            **manual_input_value,
-                        }
-                        tool_value.append(
-                            {
-                                **tool_runtime.entity.model_dump(mode="json"),
-                                "runtime_parameters": runtime_parameters,
-                                "credential_id": tool.get("credential_id", None),
-                                "provider_type": provider_type.value,
-                            }
-                        )
-                    value = tool_value
-                if parameter.type == AgentStrategyParameter.AgentStrategyParameterType.MODEL_SELECTOR:
-                    value = cast(dict[str, Any], value)
-                    model_instance, model_schema = self.fetch_model(tenant_id=tenant_id, value=value)
-                    history_prompt_messages = []
-                    if node_data.memory:
-                        memory = self.fetch_memory(
-                            variable_pool=variable_pool,
-                            app_id=app_id,
-                            model_instance=model_instance,
-                        )
-                        if memory:
-                            prompt_messages = memory.get_history_prompt_messages(
-                                message_limit=node_data.memory.window.size or None
-                            )
-                            history_prompt_messages = [
-                                prompt_message.model_dump(mode="json") for prompt_message in prompt_messages
-                            ]
-                    value["history_prompt_messages"] = history_prompt_messages
-                    if model_schema:
-                        model_schema = self._remove_unsupported_model_features_for_old_version(model_schema)
-                        value["entity"] = model_schema.model_dump(mode="json")
-                    else:
-                        value["entity"] = None
-            result[parameter_name] = value
-
-        return result
-
-    def build_credentials(self, *, parameters: dict[str, Any]) -> InvokeCredentials:
-        credentials = InvokeCredentials()
-        credentials.tool_credentials = {}
-        for tool in parameters.get("tools", []):
-            if not tool.get("credential_id"):
-                continue
-            try:
-                identity = ToolIdentity.model_validate(tool.get("identity", {}))
-            except ValidationError:
-                continue
-            credentials.tool_credentials[identity.provider] = tool.get("credential_id", None)
-        return credentials
-
-    def fetch_memory(
-        self,
-        *,
-        variable_pool: VariablePool,
-        app_id: str,
-        model_instance: ModelInstance,
-    ) -> TokenBufferMemory | None:
-        conversation_id_variable = variable_pool.get(["sys", SystemVariableKey.CONVERSATION_ID])
-        if not isinstance(conversation_id_variable, StringSegment):
-            return None
-        conversation_id = conversation_id_variable.value
-
-        with Session(db.engine, expire_on_commit=False) as session:
-            stmt = select(Conversation).where(Conversation.app_id == app_id, Conversation.id == conversation_id)
-            conversation = session.scalar(stmt)
-            if not conversation:
-                return None
-
-        return TokenBufferMemory(conversation=conversation, model_instance=model_instance)
-
-    def fetch_model(self, *, tenant_id: str, value: dict[str, Any]) -> tuple[ModelInstance, AIModelEntity | None]:
-        provider_manager = ProviderManager()
-        provider_model_bundle = provider_manager.get_provider_model_bundle(
-            tenant_id=tenant_id,
-            provider=value.get("provider", ""),
-            model_type=ModelType.LLM,
-        )
-        model_name = value.get("model", "")
-        model_credentials = provider_model_bundle.configuration.get_current_credentials(
-            model_type=ModelType.LLM,
-            model=model_name,
-        )
-        provider_name = provider_model_bundle.configuration.provider.provider
-        model_type_instance = provider_model_bundle.model_type_instance
-        model_instance = ModelManager().get_model_instance(
-            tenant_id=tenant_id,
-            provider=provider_name,
-            model_type=ModelType(value.get("model_type", "")),
-            model=model_name,
-        )
-        model_schema = model_type_instance.get_model_schema(model_name, model_credentials)
-        return model_instance, model_schema
-
-    @staticmethod
-    def _remove_unsupported_model_features_for_old_version(model_schema: AIModelEntity) -> AIModelEntity:
-        if model_schema.features:
-            for feature in model_schema.features[:]:
-                try:
-                    AgentOldVersionModelFeatures(feature.value)
-                except ValueError:
-                    model_schema.features.remove(feature)
-        return model_schema
-
-    @staticmethod
-    def _filter_mcp_type_tool(
-        strategy: ResolvedAgentStrategy,
-        tools: list[dict[str, Any]],
-    ) -> list[dict[str, Any]]:
-        meta_version = strategy.meta_version
-        if meta_version and Version(meta_version) > Version("0.0.1"):
-            return tools
-        return [tool for tool in tools if tool.get("type") != ToolProviderType.MCP]

api/core/workflow/nodes/agent/strategy_protocols.py

@@ -1,39 +0,0 @@
-from __future__ import annotations
-
-from collections.abc import Generator, Sequence
-from typing import Any, Protocol
-
-from core.agent.plugin_entities import AgentStrategyParameter
-from core.plugin.entities.request import InvokeCredentials
-from core.tools.entities.tool_entities import ToolInvokeMessage
-
-
-class ResolvedAgentStrategy(Protocol):
-    meta_version: str | None
-
-    def get_parameters(self) -> Sequence[AgentStrategyParameter]: ...
-
-    def invoke(
-        self,
-        *,
-        params: dict[str, Any],
-        user_id: str,
-        conversation_id: str | None = None,
-        app_id: str | None = None,
-        message_id: str | None = None,
-        credentials: InvokeCredentials | None = None,
-    ) -> Generator[ToolInvokeMessage, None, None]: ...
-
-
-class AgentStrategyResolver(Protocol):
-    def resolve(
-        self,
-        *,
-        tenant_id: str,
-        agent_strategy_provider_name: str,
-        agent_strategy_name: str,
-    ) -> ResolvedAgentStrategy: ...
-
-
-class AgentStrategyPresentationProvider(Protocol):
-    def get_icon(self, *, tenant_id: str, agent_strategy_provider_name: str) -> str | None: ...

api/core/workflow/nodes/datasource/__init__.py

@@ -1 +0,0 @@
-"""Datasource workflow node package."""

api/core/workflow/nodes/knowledge_index/__init__.py

@@ -1,5 +0,0 @@
-"""Knowledge index workflow node package."""
-
-KNOWLEDGE_INDEX_NODE_TYPE = "knowledge-index"
-
-__all__ = ["KNOWLEDGE_INDEX_NODE_TYPE"]

api/core/workflow/nodes/knowledge_retrieval/__init__.py

@@ -1 +0,0 @@
-"""Knowledge retrieval workflow node package."""

api/core/workflow/nodes/trigger_schedule/__init__.py

@@ -1,3 +0,0 @@
-from .trigger_schedule_node import TriggerScheduleNode
-
-__all__ = ["TriggerScheduleNode"]

api/core/workflow/workflow_entry.py

@@ -8,7 +8,7 @@ from core.app.apps.exc import GenerateTaskStoppedError
 from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom, build_dify_run_context
 from core.app.workflow.layers.llm_quota import LLMQuotaLayer
 from core.app.workflow.layers.observability import ObservabilityLayer
-from core.workflow.node_factory import DifyNodeFactory, resolve_workflow_node_class
+from core.workflow.node_factory import DifyNodeFactory
 from dify_graph.constants import ENVIRONMENT_VARIABLE_NODE_ID
 from dify_graph.entities import GraphInitParams
 from dify_graph.entities.graph_config import NodeConfigDictAdapter
@@ -21,8 +21,9 @@ from dify_graph.graph_engine.layers import DebugLoggingLayer, ExecutionLimitsLay
 from dify_graph.graph_engine.layers.base import GraphEngineLayer
 from dify_graph.graph_engine.protocols.command_channel import CommandChannel
 from dify_graph.graph_events import GraphEngineEvent, GraphNodeEventBase, GraphRunFailedEvent
-from dify_graph.nodes import BuiltinNodeTypes
+from dify_graph.nodes import NodeType
 from dify_graph.nodes.base.node import Node
+from dify_graph.nodes.node_mapping import NODE_TYPE_CLASSES_MAPPING
 from dify_graph.runtime import ChildGraphNotFoundError, GraphRuntimeState, VariablePool
 from dify_graph.system_variable import SystemVariable
 from dify_graph.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader, load_into_variable_pool
@@ -252,7 +253,7 @@ class WorkflowEntry:
             variable_mapping=variable_mapping,
             user_inputs=user_inputs,
         )
-        if node_type != BuiltinNodeTypes.DATASOURCE:
+        if node_type != NodeType.DATASOURCE:
             cls.mapping_user_inputs_to_variable_pool(
                 variable_mapping=variable_mapping,
                 user_inputs=user_inputs,
@@ -302,7 +303,7 @@ class WorkflowEntry:
             "height": node_height,
             "type": "custom",
             "data": {
-                "type": BuiltinNodeTypes.START,
+                "type": NodeType.START,
                 "title": "Start",
                 "desc": "Start",
             },
@@ -338,11 +339,11 @@ class WorkflowEntry:
         # Create a minimal graph for single node execution
         graph_dict = cls._create_single_node_graph(node_id, node_data)
 
-        node_type = node_data.get("type", "")
-        if node_type not in {BuiltinNodeTypes.PARAMETER_EXTRACTOR, BuiltinNodeTypes.QUESTION_CLASSIFIER}:
+        node_type = NodeType(node_data.get("type", ""))
+        if node_type not in {NodeType.PARAMETER_EXTRACTOR, NodeType.QUESTION_CLASSIFIER}:
             raise ValueError(f"Node type {node_type} not supported")
 
-        node_cls = resolve_workflow_node_class(node_type=node_type, node_version="1")
+        node_cls = NODE_TYPE_CLASSES_MAPPING[node_type]["1"]
         if not node_cls:
             raise ValueError(f"Node class not found for node type {node_type}")
 

api/dify_graph/README.md

@@ -113,7 +113,7 @@ The codebase enforces strict layering via import-linter:
 1. Create node class in `nodes/<node_type>/`
 1. Inherit from `BaseNode` or appropriate base class
 1. Implement `_run()` method
-1. Ensure the node module is importable under `nodes/<node_type>/`
+1. Register in `nodes/node_mapping.py`
 1. Add tests in `tests/unit_tests/dify_graph/nodes/`
 
 ### Implementing a Custom Layer

api/dify_graph/entities/__init__.py

@@ -1,9 +1,11 @@
+from .agent import AgentNodeStrategyInit
 from .graph_init_params import GraphInitParams
 from .workflow_execution import WorkflowExecution
 from .workflow_node_execution import WorkflowNodeExecution
 from .workflow_start_reason import WorkflowStartReason
 
 __all__ = [
+    "AgentNodeStrategyInit",
     "GraphInitParams",
     "WorkflowExecution",
     "WorkflowNodeExecution",

api/dify_graph/entities/agent.py

@@ -0,0 +1,8 @@
+from pydantic import BaseModel
+
+
+class AgentNodeStrategyInit(BaseModel):
+    """Agent node strategy initialization data."""
+
+    name: str
+    icon: str | None = None

api/dify_graph/entities/base_node_data.py

@@ -121,8 +121,6 @@ class DefaultValue(BaseModel):
 class BaseNodeData(ABC, BaseModel):
     # Raw graph payloads are first validated through `NodeConfigDictAdapter`, where
     # `node["data"]` is typed as `BaseNodeData` before the concrete node class is known.
-    # `type` therefore accepts downstream string node kinds; unknown node implementations
-    # are rejected later when the node factory resolves the node registry.
     # At that boundary, node-specific fields are still "extra" relative to this shared DTO,
     # and persisted templates/workflows also carry undeclared compatibility keys such as
     # `selected`, `params`, `paramSchemas`, and `datasource_label`. Keep extras permissive

api/dify_graph/entities/workflow_node_execution.py

@@ -48,7 +48,7 @@ class WorkflowNodeExecution(BaseModel):
     index: int  # Sequence number for ordering in trace visualization
     predecessor_node_id: str | None = None  # ID of the node that executed before this one
     node_id: str  # ID of the node being executed
-    node_type: NodeType  # Type of node (e.g., start, llm, downstream response node)
+    node_type: NodeType  # Type of node (e.g., start, llm, knowledge)
     title: str  # Display title of the node
 
     # Execution data

api/dify_graph/enums.py

@@ -1,5 +1,4 @@
 from enum import StrEnum
-from typing import ClassVar, TypeAlias
 
 
 class NodeState(StrEnum):
@@ -34,71 +33,56 @@ class SystemVariableKey(StrEnum):
     INVOKE_FROM = "invoke_from"
 
 
-NodeType: TypeAlias = str
+class NodeType(StrEnum):
+    START = "start"
+    END = "end"
+    ANSWER = "answer"
+    LLM = "llm"
+    KNOWLEDGE_RETRIEVAL = "knowledge-retrieval"
+    KNOWLEDGE_INDEX = "knowledge-index"
+    IF_ELSE = "if-else"
+    CODE = "code"
+    TEMPLATE_TRANSFORM = "template-transform"
+    QUESTION_CLASSIFIER = "question-classifier"
+    HTTP_REQUEST = "http-request"
+    TOOL = "tool"
+    DATASOURCE = "datasource"
+    VARIABLE_AGGREGATOR = "variable-aggregator"
+    LEGACY_VARIABLE_AGGREGATOR = "variable-assigner"  # TODO: Merge this into VARIABLE_AGGREGATOR in the database.
+    LOOP = "loop"
+    LOOP_START = "loop-start"
+    LOOP_END = "loop-end"
+    ITERATION = "iteration"
+    ITERATION_START = "iteration-start"  # Fake start node for iteration.
+    PARAMETER_EXTRACTOR = "parameter-extractor"
+    VARIABLE_ASSIGNER = "assigner"
+    DOCUMENT_EXTRACTOR = "document-extractor"
+    LIST_OPERATOR = "list-operator"
+    AGENT = "agent"
+    TRIGGER_WEBHOOK = "trigger-webhook"
+    TRIGGER_SCHEDULE = "trigger-schedule"
+    TRIGGER_PLUGIN = "trigger-plugin"
+    HUMAN_INPUT = "human-input"
 
+    @property
+    def is_trigger_node(self) -> bool:
+        """Check if this node type is a trigger node."""
+        return self in [
+            NodeType.TRIGGER_WEBHOOK,
+            NodeType.TRIGGER_SCHEDULE,
+            NodeType.TRIGGER_PLUGIN,
+        ]
 
-class BuiltinNodeTypes:
-    """Built-in node type string constants.
-
-    `node_type` values are plain strings throughout the graph runtime. This namespace
-    only exposes the built-in values shipped by `dify_graph`; downstream packages can
-    use additional strings without extending this class.
-    """
-
-    START: ClassVar[NodeType] = "start"
-    END: ClassVar[NodeType] = "end"
-    ANSWER: ClassVar[NodeType] = "answer"
-    LLM: ClassVar[NodeType] = "llm"
-    KNOWLEDGE_RETRIEVAL: ClassVar[NodeType] = "knowledge-retrieval"
-    IF_ELSE: ClassVar[NodeType] = "if-else"
-    CODE: ClassVar[NodeType] = "code"
-    TEMPLATE_TRANSFORM: ClassVar[NodeType] = "template-transform"
-    QUESTION_CLASSIFIER: ClassVar[NodeType] = "question-classifier"
-    HTTP_REQUEST: ClassVar[NodeType] = "http-request"
-    TOOL: ClassVar[NodeType] = "tool"
-    DATASOURCE: ClassVar[NodeType] = "datasource"
-    VARIABLE_AGGREGATOR: ClassVar[NodeType] = "variable-aggregator"
-    LEGACY_VARIABLE_AGGREGATOR: ClassVar[NodeType] = "variable-assigner"
-    LOOP: ClassVar[NodeType] = "loop"
-    LOOP_START: ClassVar[NodeType] = "loop-start"
-    LOOP_END: ClassVar[NodeType] = "loop-end"
-    ITERATION: ClassVar[NodeType] = "iteration"
-    ITERATION_START: ClassVar[NodeType] = "iteration-start"
-    PARAMETER_EXTRACTOR: ClassVar[NodeType] = "parameter-extractor"
-    VARIABLE_ASSIGNER: ClassVar[NodeType] = "assigner"
-    DOCUMENT_EXTRACTOR: ClassVar[NodeType] = "document-extractor"
-    LIST_OPERATOR: ClassVar[NodeType] = "list-operator"
-    AGENT: ClassVar[NodeType] = "agent"
-    HUMAN_INPUT: ClassVar[NodeType] = "human-input"
-
-
-BUILT_IN_NODE_TYPES: tuple[NodeType, ...] = (
-    BuiltinNodeTypes.START,
-    BuiltinNodeTypes.END,
-    BuiltinNodeTypes.ANSWER,
-    BuiltinNodeTypes.LLM,
-    BuiltinNodeTypes.KNOWLEDGE_RETRIEVAL,
-    BuiltinNodeTypes.IF_ELSE,
-    BuiltinNodeTypes.CODE,
-    BuiltinNodeTypes.TEMPLATE_TRANSFORM,
-    BuiltinNodeTypes.QUESTION_CLASSIFIER,
-    BuiltinNodeTypes.HTTP_REQUEST,
-    BuiltinNodeTypes.TOOL,
-    BuiltinNodeTypes.DATASOURCE,
-    BuiltinNodeTypes.VARIABLE_AGGREGATOR,
-    BuiltinNodeTypes.LEGACY_VARIABLE_AGGREGATOR,
-    BuiltinNodeTypes.LOOP,
-    BuiltinNodeTypes.LOOP_START,
-    BuiltinNodeTypes.LOOP_END,
-    BuiltinNodeTypes.ITERATION,
-    BuiltinNodeTypes.ITERATION_START,
-    BuiltinNodeTypes.PARAMETER_EXTRACTOR,
-    BuiltinNodeTypes.VARIABLE_ASSIGNER,
-    BuiltinNodeTypes.DOCUMENT_EXTRACTOR,
-    BuiltinNodeTypes.LIST_OPERATOR,
-    BuiltinNodeTypes.AGENT,
-    BuiltinNodeTypes.HUMAN_INPUT,
-)
+    @property
+    def is_start_node(self) -> bool:
+        """Check if this node type can serve as a workflow entry point."""
+        return self in [
+            NodeType.START,
+            NodeType.DATASOURCE,
+            NodeType.TRIGGER_WEBHOOK,
+            NodeType.TRIGGER_SCHEDULE,
+            NodeType.TRIGGER_PLUGIN,
+        ]
 
 
 class NodeExecutionType(StrEnum):
@@ -248,12 +232,11 @@ class WorkflowNodeExecutionMetadataKey(StrEnum):
     """
 
     TOTAL_TOKENS = "total_tokens"
-    PROMPT_TOKENS = "prompt_tokens"
-    COMPLETION_TOKENS = "completion_tokens"
     TOTAL_PRICE = "total_price"
     CURRENCY = "currency"
     TOOL_INFO = "tool_info"
     AGENT_LOG = "agent_log"
+    TRIGGER_INFO = "trigger_info"
     ITERATION_ID = "iteration_id"
     ITERATION_INDEX = "iteration_index"
     LOOP_ID = "loop_id"

api/dify_graph/graph/graph.py

@@ -83,6 +83,50 @@ class Graph:
 
         return node_configs_map
 
+    @classmethod
+    def _find_root_node_id(
+        cls,
+        node_configs_map: Mapping[str, NodeConfigDict],
+        edge_configs: Sequence[Mapping[str, object]],
+        root_node_id: str | None = None,
+    ) -> str:
+        """
+        Find the root node ID if not specified.
+
+        :param node_configs_map: mapping of node ID to node config
+        :param edge_configs: list of edge configurations
+        :param root_node_id: explicitly specified root node ID
+        :return: determined root node ID
+        """
+        if root_node_id:
+            if root_node_id not in node_configs_map:
+                raise ValueError(f"Root node id {root_node_id} not found in the graph")
+            return root_node_id
+
+        # Find nodes with no incoming edges
+        nodes_with_incoming: set[str] = set()
+        for edge_config in edge_configs:
+            target = edge_config.get("target")
+            if isinstance(target, str):
+                nodes_with_incoming.add(target)
+
+        root_candidates = [nid for nid in node_configs_map if nid not in nodes_with_incoming]
+
+        # Prefer START node if available
+        start_node_id = None
+        for nid in root_candidates:
+            node_data = node_configs_map[nid]["data"]
+            if node_data.type.is_start_node:
+                start_node_id = nid
+                break
+
+        root_node_id = start_node_id or (root_candidates[0] if root_candidates else None)
+
+        if not root_node_id:
+            raise ValueError("Unable to determine root node ID")
+
+        return root_node_id
+
     @classmethod
     def _build_edges(
         cls, edge_configs: list[dict[str, object]]
@@ -257,15 +301,15 @@ class Graph:
         *,
         graph_config: Mapping[str, object],
         node_factory: NodeFactory,
-        root_node_id: str,
+        root_node_id: str | None = None,
         skip_validation: bool = False,
     ) -> Graph:
         """
-        Initialize a graph with an explicit execution entry point.
+        Initialize graph
 
         :param graph_config: graph config containing nodes and edges
         :param node_factory: factory for creating node instances from config data
-        :param root_node_id: active root node id
+        :param root_node_id: root node id
         :return: graph instance
         """
         # Parse configs
@@ -283,8 +327,8 @@ class Graph:
         # Parse node configurations
         node_configs_map = cls._parse_node_configs(node_configs)
 
-        if root_node_id not in node_configs_map:
-            raise ValueError(f"Root node id {root_node_id} not found in the graph")
+        # Find root node
+        root_node_id = cls._find_root_node_id(node_configs_map, edge_configs, root_node_id)
 
         # Build edges
         edges, in_edges, out_edges = cls._build_edges(edge_configs)