test: refactor GenericTable tests to use a helper function for option selection

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

.github/CODEOWNERS

@@ -36,6 +36,7 @@
 /api/core/workflow/graph/ @laipz8200 @QuantumGhost
 /api/core/workflow/graph_events/ @laipz8200 @QuantumGhost
 /api/core/workflow/node_events/ @laipz8200 @QuantumGhost
+/api/dify_graph/model_runtime/ @laipz8200 @QuantumGhost
 
 # Backend - Workflow - Nodes (Agent, Iteration, Loop, LLM)
 /api/core/workflow/nodes/agent/ @Nov1c444

.github/workflows/api-tests.yml

@@ -25,6 +25,7 @@ jobs:
     strategy:
       matrix:
         python-version:
+          - "3.11"
           - "3.12"
 
     steps:

.github/workflows/autofix.yml

@@ -2,9 +2,6 @@ name: autofix.ci
 on:
   pull_request:
     branches: ["main"]
-  merge_group:
-    branches: ["main"]
-    types: [checks_requested]
   push:
     branches: ["main"]
 permissions:
@@ -15,15 +12,9 @@ jobs:
     if: github.repository == 'langgenius/dify'
     runs-on: ubuntu-latest
     steps:
-      - name: Complete merge group check
-        if: github.event_name == 'merge_group'
-        run: echo "autofix.ci updates pull request branches, not merge group refs."
-
-      - if: github.event_name != 'merge_group'
-        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
 
       - name: Check Docker Compose inputs
-        if: github.event_name != 'merge_group'
         id: docker-compose-changes
         uses: tj-actions/changed-files@22103cc46bda19c2b464ffe86db46df6922fd323 # v47.0.5
         with:
@@ -33,34 +24,30 @@ jobs:
             docker/docker-compose-template.yaml
             docker/docker-compose.yaml
       - name: Check web inputs
-        if: github.event_name != 'merge_group'
         id: web-changes
         uses: tj-actions/changed-files@22103cc46bda19c2b464ffe86db46df6922fd323 # v47.0.5
         with:
           files: |
             web/**
       - name: Check api inputs
-        if: github.event_name != 'merge_group'
         id: api-changes
         uses: tj-actions/changed-files@22103cc46bda19c2b464ffe86db46df6922fd323 # v47.0.5
         with:
           files: |
             api/**
-      - if: github.event_name != 'merge_group'
-        uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0
+      - uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0
         with:
           python-version: "3.11"
 
-      - if: github.event_name != 'merge_group'
-        uses: astral-sh/setup-uv@37802adc94f370d6bfd71619e3f0bf239e1f3b78 # v7.6.0
+      - uses: astral-sh/setup-uv@37802adc94f370d6bfd71619e3f0bf239e1f3b78 # v7.6.0
 
       - name: Generate Docker Compose
-        if: github.event_name != 'merge_group' && steps.docker-compose-changes.outputs.any_changed == 'true'
+        if: steps.docker-compose-changes.outputs.any_changed == 'true'
         run: |
           cd docker
           ./generate_docker_compose
 
-      - if: github.event_name != 'merge_group' && steps.api-changes.outputs.any_changed == 'true'
+      - if: steps.api-changes.outputs.any_changed == 'true'
         run: |
           cd api
           uv sync --dev
@@ -72,13 +59,13 @@ jobs:
           uv run ruff format ..
 
       - name: count migration progress
-        if: github.event_name != 'merge_group' && steps.api-changes.outputs.any_changed == 'true'
+        if: steps.api-changes.outputs.any_changed == 'true'
         run: |
           cd api
           ./cnt_base.sh
 
       - name: ast-grep
-        if: github.event_name != 'merge_group' && steps.api-changes.outputs.any_changed == 'true'
+        if: steps.api-changes.outputs.any_changed == 'true'
         run: |
           # ast-grep exits 1 if no matches are found; allow idempotent runs.
           uvx --from ast-grep-cli ast-grep --pattern 'db.session.query($WHATEVER).filter($HERE)' --rewrite 'db.session.query($WHATEVER).where($HERE)' -l py --update-all || true
@@ -108,14 +95,13 @@ jobs:
           find . -name "*.py.bak" -type f -delete
 
       - name: Setup web environment
-        if: github.event_name != 'merge_group' && steps.web-changes.outputs.any_changed == 'true'
+        if: steps.web-changes.outputs.any_changed == 'true'
         uses: ./.github/actions/setup-web
 
       - name: ESLint autofix
-        if: github.event_name != 'merge_group' && steps.web-changes.outputs.any_changed == 'true'
+        if: steps.web-changes.outputs.any_changed == 'true'
         run: |
           cd web
           vp exec eslint --concurrency=2 --prune-suppressions --quiet || true
 
-      - if: github.event_name != 'merge_group'
-        uses: autofix-ci/action@7a166d7532b277f34e16238930461bf77f9d7ed8 # v1.3.3
+      - uses: autofix-ci/action@7a166d7532b277f34e16238930461bf77f9d7ed8 # v1.3.3

.github/workflows/main-ci.yml

@@ -3,14 +3,10 @@ name: Main CI Pipeline
 on:
   pull_request:
     branches: ["main"]
-  merge_group:
-    branches: ["main"]
-    types: [checks_requested]
   push:
     branches: ["main"]
 
 permissions:
-  actions: write
   contents: write
   pull-requests: write
   checks: write
@@ -21,24 +17,9 @@ concurrency:
   cancel-in-progress: true
 
 jobs:
-  pre_job:
-    name: Skip Duplicate Checks
-    runs-on: ubuntu-latest
-    outputs:
-      should_skip: ${{ steps.skip_check.outputs.should_skip || 'false' }}
-    steps:
-      - id: skip_check
-        continue-on-error: true
-        uses: fkirc/skip-duplicate-actions@f75f66ce1886f00957d99748a42c724f4330bdcf # v5.3.1
-        with:
-          cancel_others: 'true'
-          concurrent_skipping: same_content_newer
-
   # Check which paths were changed to determine which tests to run
   check-changes:
     name: Check Changed Files
-    needs: pre_job
-    if: needs.pre_job.outputs.should_skip != 'true'
     runs-on: ubuntu-latest
     outputs:
       api-changed: ${{ steps.changes.outputs.api }}
@@ -53,295 +34,49 @@ jobs:
           filters: |
             api:
               - 'api/**'
+              - 'docker/**'
               - '.github/workflows/api-tests.yml'
-              - '.github/workflows/expose_service_ports.sh'
-              - 'docker/.env.example'
-              - 'docker/middleware.env.example'
-              - 'docker/docker-compose.middleware.yaml'
-              - 'docker/docker-compose-template.yaml'
-              - 'docker/generate_docker_compose'
-              - 'docker/ssrf_proxy/**'
-              - 'docker/volumes/sandbox/conf/**'
             web:
               - 'web/**'
               - '.github/workflows/web-tests.yml'
               - '.github/actions/setup-web/**'
             vdb:
               - 'api/core/rag/datasource/**'
-              - 'api/tests/integration_tests/vdb/**'
+              - 'docker/**'
               - '.github/workflows/vdb-tests.yml'
-              - '.github/workflows/expose_service_ports.sh'
-              - 'docker/.env.example'
-              - 'docker/middleware.env.example'
-              - 'docker/docker-compose.yaml'
-              - 'docker/docker-compose-template.yaml'
-              - 'docker/generate_docker_compose'
-              - 'docker/certbot/**'
-              - 'docker/couchbase-server/**'
-              - 'docker/elasticsearch/**'
-              - 'docker/iris/**'
-              - 'docker/nginx/**'
-              - 'docker/pgvector/**'
-              - 'docker/ssrf_proxy/**'
-              - 'docker/startupscripts/**'
-              - 'docker/tidb/**'
-              - 'docker/volumes/**'
               - 'api/uv.lock'
               - 'api/pyproject.toml'
             migration:
               - 'api/migrations/**'
-              - 'api/.env.example'
               - '.github/workflows/db-migration-test.yml'
-              - '.github/workflows/expose_service_ports.sh'
-              - 'docker/.env.example'
-              - 'docker/middleware.env.example'
-              - 'docker/docker-compose.middleware.yaml'
-              - 'docker/docker-compose-template.yaml'
-              - 'docker/generate_docker_compose'
-              - 'docker/ssrf_proxy/**'
-              - 'docker/volumes/sandbox/conf/**'
 
-  # Run tests in parallel while always emitting stable required checks.
-  api-tests-run:
-    name: Run API Tests
-    needs:
-      - pre_job
-      - check-changes
-    if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.api-changed == 'true'
+  # Run tests in parallel
+  api-tests:
+    name: API Tests
+    needs: check-changes
+    if: needs.check-changes.outputs.api-changed == 'true'
     uses: ./.github/workflows/api-tests.yml
     secrets: inherit
 
-  api-tests-skip:
-    name: Skip API Tests
-    needs:
-      - pre_job
-      - check-changes
-    if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.api-changed != 'true'
-    runs-on: ubuntu-latest
-    steps:
-      - name: Report skipped API tests
-        run: echo "No API-related changes detected; skipping API tests."
-
-  api-tests:
-    name: API Tests
-    if: ${{ always() }}
-    needs:
-      - pre_job
-      - check-changes
-      - api-tests-run
-      - api-tests-skip
-    runs-on: ubuntu-latest
-    steps:
-      - name: Finalize API Tests status
-        env:
-          SHOULD_SKIP_WORKFLOW: ${{ needs.pre_job.outputs.should_skip }}
-          TESTS_CHANGED: ${{ needs.check-changes.outputs.api-changed }}
-          RUN_RESULT: ${{ needs.api-tests-run.result }}
-          SKIP_RESULT: ${{ needs.api-tests-skip.result }}
-        run: |
-          if [[ "$SHOULD_SKIP_WORKFLOW" == 'true' ]]; then
-            echo "API tests were skipped because this workflow run duplicated a successful or newer run."
-            exit 0
-          fi
-
-          if [[ "$TESTS_CHANGED" == 'true' ]]; then
-            if [[ "$RUN_RESULT" == 'success' ]]; then
-              echo "API tests ran successfully."
-              exit 0
-            fi
-
-            echo "API tests were required but finished with result: $RUN_RESULT" >&2
-            exit 1
-          fi
-
-          if [[ "$SKIP_RESULT" == 'success' ]]; then
-            echo "API tests were skipped because no API-related files changed."
-            exit 0
-          fi
-
-          echo "API tests were not required, but the skip job finished with result: $SKIP_RESULT" >&2
-          exit 1
-
-  web-tests-run:
-    name: Run Web Tests
-    needs:
-      - pre_job
-      - check-changes
-    if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.web-changed == 'true'
+  web-tests:
+    name: Web Tests
+    needs: check-changes
+    if: needs.check-changes.outputs.web-changed == 'true'
     uses: ./.github/workflows/web-tests.yml
     secrets: inherit
 
-  web-tests-skip:
-    name: Skip Web Tests
-    needs:
-      - pre_job
-      - check-changes
-    if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.web-changed != 'true'
-    runs-on: ubuntu-latest
-    steps:
-      - name: Report skipped web tests
-        run: echo "No web-related changes detected; skipping web tests."
-
-  web-tests:
-    name: Web Tests
-    if: ${{ always() }}
-    needs:
-      - pre_job
-      - check-changes
-      - web-tests-run
-      - web-tests-skip
-    runs-on: ubuntu-latest
-    steps:
-      - name: Finalize Web Tests status
-        env:
-          SHOULD_SKIP_WORKFLOW: ${{ needs.pre_job.outputs.should_skip }}
-          TESTS_CHANGED: ${{ needs.check-changes.outputs.web-changed }}
-          RUN_RESULT: ${{ needs.web-tests-run.result }}
-          SKIP_RESULT: ${{ needs.web-tests-skip.result }}
-        run: |
-          if [[ "$SHOULD_SKIP_WORKFLOW" == 'true' ]]; then
-            echo "Web tests were skipped because this workflow run duplicated a successful or newer run."
-            exit 0
-          fi
-
-          if [[ "$TESTS_CHANGED" == 'true' ]]; then
-            if [[ "$RUN_RESULT" == 'success' ]]; then
-              echo "Web tests ran successfully."
-              exit 0
-            fi
-
-            echo "Web tests were required but finished with result: $RUN_RESULT" >&2
-            exit 1
-          fi
-
-          if [[ "$SKIP_RESULT" == 'success' ]]; then
-            echo "Web tests were skipped because no web-related files changed."
-            exit 0
-          fi
-
-          echo "Web tests were not required, but the skip job finished with result: $SKIP_RESULT" >&2
-          exit 1
-
   style-check:
     name: Style Check
-    needs: pre_job
-    if: needs.pre_job.outputs.should_skip != 'true'
     uses: ./.github/workflows/style.yml
 
-  vdb-tests-run:
-    name: Run VDB Tests
-    needs:
-      - pre_job
-      - check-changes
-    if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.vdb-changed == 'true'
-    uses: ./.github/workflows/vdb-tests.yml
-
-  vdb-tests-skip:
-    name: Skip VDB Tests
-    needs:
-      - pre_job
-      - check-changes
-    if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.vdb-changed != 'true'
-    runs-on: ubuntu-latest
-    steps:
-      - name: Report skipped VDB tests
-        run: echo "No VDB-related changes detected; skipping VDB tests."
-
   vdb-tests:
     name: VDB Tests
-    if: ${{ always() }}
-    needs:
-      - pre_job
-      - check-changes
-      - vdb-tests-run
-      - vdb-tests-skip
-    runs-on: ubuntu-latest
-    steps:
-      - name: Finalize VDB Tests status
-        env:
-          SHOULD_SKIP_WORKFLOW: ${{ needs.pre_job.outputs.should_skip }}
-          TESTS_CHANGED: ${{ needs.check-changes.outputs.vdb-changed }}
-          RUN_RESULT: ${{ needs.vdb-tests-run.result }}
-          SKIP_RESULT: ${{ needs.vdb-tests-skip.result }}
-        run: |
-          if [[ "$SHOULD_SKIP_WORKFLOW" == 'true' ]]; then
-            echo "VDB tests were skipped because this workflow run duplicated a successful or newer run."
-            exit 0
-          fi
-
-          if [[ "$TESTS_CHANGED" == 'true' ]]; then
-            if [[ "$RUN_RESULT" == 'success' ]]; then
-              echo "VDB tests ran successfully."
-              exit 0
-            fi
-
-            echo "VDB tests were required but finished with result: $RUN_RESULT" >&2
-            exit 1
-          fi
-
-          if [[ "$SKIP_RESULT" == 'success' ]]; then
-            echo "VDB tests were skipped because no VDB-related files changed."
-            exit 0
-          fi
-
-          echo "VDB tests were not required, but the skip job finished with result: $SKIP_RESULT" >&2
-          exit 1
-
-  db-migration-test-run:
-    name: Run DB Migration Test
-    needs:
-      - pre_job
-      - check-changes
-    if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.migration-changed == 'true'
-    uses: ./.github/workflows/db-migration-test.yml
-
-  db-migration-test-skip:
-    name: Skip DB Migration Test
-    needs:
-      - pre_job
-      - check-changes
-    if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.migration-changed != 'true'
-    runs-on: ubuntu-latest
-    steps:
-      - name: Report skipped DB migration tests
-        run: echo "No migration-related changes detected; skipping DB migration tests."
+    needs: check-changes
+    if: needs.check-changes.outputs.vdb-changed == 'true'
+    uses: ./.github/workflows/vdb-tests.yml
 
   db-migration-test:
     name: DB Migration Test
-    if: ${{ always() }}
-    needs:
-      - pre_job
-      - check-changes
-      - db-migration-test-run
-      - db-migration-test-skip
-    runs-on: ubuntu-latest
-    steps:
-      - name: Finalize DB Migration Test status
-        env:
-          SHOULD_SKIP_WORKFLOW: ${{ needs.pre_job.outputs.should_skip }}
-          TESTS_CHANGED: ${{ needs.check-changes.outputs.migration-changed }}
-          RUN_RESULT: ${{ needs.db-migration-test-run.result }}
-          SKIP_RESULT: ${{ needs.db-migration-test-skip.result }}
-        run: |
-          if [[ "$SHOULD_SKIP_WORKFLOW" == 'true' ]]; then
-            echo "DB migration tests were skipped because this workflow run duplicated a successful or newer run."
-            exit 0
-          fi
-
-          if [[ "$TESTS_CHANGED" == 'true' ]]; then
-            if [[ "$RUN_RESULT" == 'success' ]]; then
-              echo "DB migration tests ran successfully."
-              exit 0
-            fi
-
-            echo "DB migration tests were required but finished with result: $RUN_RESULT" >&2
-            exit 1
-          fi
-
-          if [[ "$SKIP_RESULT" == 'success' ]]; then
-            echo "DB migration tests were skipped because no migration-related files changed."
-            exit 0
-          fi
-
-          echo "DB migration tests were not required, but the skip job finished with result: $SKIP_RESULT" >&2
-          exit 1
+    needs: check-changes
+    if: needs.check-changes.outputs.migration-changed == 'true'
+    uses: ./.github/workflows/db-migration-test.yml

.github/workflows/semantic-pull-request.yml

@@ -7,9 +7,6 @@ on:
       - edited
       - reopened
       - synchronize
-  merge_group:
-    branches: ["main"]
-    types: [checks_requested]
 
 jobs:
   lint:
@@ -18,11 +15,7 @@ jobs:
       pull-requests: read
     runs-on: ubuntu-latest
     steps:
-      - name: Complete merge group check
-        if: github.event_name == 'merge_group'
-        run: echo "Semantic PR title validation is handled on pull requests."
       - name: Check title
-        if: github.event_name == 'pull_request'
         uses: amannn/action-semantic-pull-request@48f256284bd46cdaab1048c3721360e808335d50 # v6.1.1
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/vdb-tests.yml

@@ -14,6 +14,7 @@ jobs:
     strategy:
       matrix:
         python-version:
+          - "3.11"
           - "3.12"
 
     steps:

api/.importlinter

@@ -1,14 +1,202 @@
 [importlinter]
 root_packages =
     core
-    constants
-    context
+    dify_graph
     configs
     controllers
     extensions
-    factories
-    libs
     models
     tasks
     services
 include_external_packages = True
+
+[importlinter:contract:workflow]
+name = Workflow
+type=layers
+layers =
+    graph_engine
+    graph_events
+    graph
+    nodes
+    node_events
+    runtime
+    entities
+containers =
+    dify_graph
+ignore_imports =
+    dify_graph.nodes.base.node -> dify_graph.graph_events
+    dify_graph.nodes.iteration.iteration_node -> dify_graph.graph_events
+    dify_graph.nodes.loop.loop_node -> dify_graph.graph_events
+
+    dify_graph.nodes.iteration.iteration_node -> dify_graph.graph_engine
+    dify_graph.nodes.loop.loop_node -> dify_graph.graph_engine
+    # TODO(QuantumGhost): fix the import violation later
+    dify_graph.entities.pause_reason -> dify_graph.nodes.human_input.entities
+
+[importlinter:contract:workflow-infrastructure-dependencies]
+name = Workflow Infrastructure Dependencies
+type = forbidden
+source_modules =
+    dify_graph
+forbidden_modules =
+    extensions.ext_database
+    extensions.ext_redis
+allow_indirect_imports = True
+ignore_imports =
+    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
+
+[importlinter:contract:workflow-external-imports]
+name = Workflow External Imports
+type = forbidden
+source_modules =
+    dify_graph
+forbidden_modules =
+    configs
+    controllers
+    extensions
+    models
+    services
+    tasks
+    core.agent
+    core.app
+    core.base
+    core.callback_handler
+    core.datasource
+    core.db
+    core.entities
+    core.errors
+    core.extension
+    core.external_data_tool
+    core.file
+    core.helper
+    core.hosting_configuration
+    core.indexing_runner
+    core.llm_generator
+    core.logging
+    core.mcp
+    core.memory
+    core.moderation
+    core.ops
+    core.plugin
+    core.prompt
+    core.provider_manager
+    core.rag
+    core.repositories
+    core.schemas
+    core.tools
+    core.trigger
+    core.variables
+ignore_imports =
+    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
+    dify_graph.nodes.llm.node -> core.tools.signature
+    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.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
+    dify_graph.nodes.question_classifier.question_classifier_node -> core.prompt.simple_prompt_transform
+    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.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.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
+    dify_graph.nodes.parameter_extractor.entities -> core.prompt.entities.advanced_prompt_entities
+    dify_graph.nodes.parameter_extractor.parameter_extractor_node -> core.prompt.entities.advanced_prompt_entities
+    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.llm.node -> models.dataset
+    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.llm.node -> extensions.ext_database
+    dify_graph.nodes.llm.node -> models.model
+    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
+    dify_graph.model_runtime.model_providers.__base.large_language_model -> configs
+    dify_graph.model_runtime.model_providers.__base.text_embedding_model -> core.entities.embedding_type
+    dify_graph.model_runtime.model_providers.model_provider_factory -> configs
+    dify_graph.model_runtime.model_providers.model_provider_factory -> extensions.ext_redis
+    dify_graph.model_runtime.model_providers.model_provider_factory -> models.provider_ids
+
+[importlinter:contract:rsc]
+name = RSC
+type = layers
+layers =
+    graph_engine
+    response_coordinator
+containers =
+    dify_graph.graph_engine
+
+[importlinter:contract:worker]
+name = Worker
+type = layers
+layers =
+    graph_engine
+    worker
+containers =
+    dify_graph.graph_engine
+
+[importlinter:contract:graph-engine-architecture]
+name = Graph Engine Architecture
+type = layers
+layers =
+    graph_engine
+    orchestration
+    command_processing
+    event_management
+    error_handler
+    graph_traversal
+    graph_state_manager
+    worker_management
+    domain
+containers =
+    dify_graph.graph_engine
+
+[importlinter:contract:domain-isolation]
+name = Domain Model Isolation
+type = forbidden
+source_modules =
+    dify_graph.graph_engine.domain
+forbidden_modules =
+    dify_graph.graph_engine.worker_management
+    dify_graph.graph_engine.command_channels
+    dify_graph.graph_engine.layers
+    dify_graph.graph_engine.protocols
+
+[importlinter:contract:worker-management]
+name = Worker Management
+type = forbidden
+source_modules =
+    dify_graph.graph_engine.worker_management
+forbidden_modules =
+    dify_graph.graph_engine.orchestration
+    dify_graph.graph_engine.command_processing
+    dify_graph.graph_engine.event_management
+
+
+[importlinter:contract:graph-traversal-components]
+name = Graph Traversal Components
+type = layers
+layers =
+    edge_processor
+    skip_propagator
+containers =
+    dify_graph.graph_engine.graph_traversal
+
+[importlinter:contract:command-channels]
+name = Command Channels Independence
+type = independence
+modules =
+    dify_graph.graph_engine.command_channels.in_memory_channel
+    dify_graph.graph_engine.command_channels.redis_channel

api/.ruff.toml

@@ -100,7 +100,7 @@ ignore = [
 "configs/*" = [
     "N802", # invalid-function-name
 ]
-"graphon/model_runtime/callbacks/base_callback.py" = ["T201"]
+"dify_graph/model_runtime/callbacks/base_callback.py" = ["T201"]
 "core/workflow/callbacks/workflow_logging_callback.py" = ["T201"]
 "libs/gmpy2_pkcs10aep_cipher.py" = [
     "N803", # invalid-argument-name

api/app_factory.py

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

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,52 +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).",
-        # Setting the default value to False to avoid accidentally log PII data in traces.
-        default=False,
-    )
-
-    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,
-        ge=0.0,
-        le=1.0,
-    )

api/context/__init__.py

@@ -1,36 +1,74 @@
 """
-Application-layer context adapters.
+Core Context - Framework-agnostic context management.
 
-Concrete execution-context implementations live here so `graphon` only
-depends on injected context managers rather than framework state capture.
+This module provides context management that is independent of any specific
+web framework. Framework-specific implementations register their context
+capture functions at application initialization time.
+
+This ensures the workflow layer remains completely decoupled from Flask
+or any other web framework.
 """
 
-from context.execution_context import (
-    AppContext,
-    ContextProviderNotFoundError,
+import contextvars
+from collections.abc import Callable
+
+from dify_graph.context.execution_context import (
     ExecutionContext,
-    ExecutionContextBuilder,
     IExecutionContext,
     NullAppContext,
-    capture_current_context,
-    read_context,
-    register_context,
-    register_context_capturer,
-    reset_context_provider,
 )
-from context.models import SandboxContext
+
+# Global capturer function - set by framework-specific modules
+_capturer: Callable[[], IExecutionContext] | None = None
+
+
+def register_context_capturer(capturer: Callable[[], IExecutionContext]) -> None:
+    """
+    Register a context capture function.
+
+    This should be called by framework-specific modules (e.g., Flask)
+    during application initialization.
+
+    Args:
+        capturer: Function that captures current context and returns IExecutionContext
+    """
+    global _capturer
+    _capturer = capturer
+
+
+def capture_current_context() -> IExecutionContext:
+    """
+    Capture current execution context.
+
+    This function uses the registered context capturer. If no capturer
+    is registered, it returns a minimal context with only contextvars
+    (suitable for non-framework environments like tests or standalone scripts).
+
+    Returns:
+        IExecutionContext with captured context
+    """
+    if _capturer is None:
+        # No framework registered - return minimal context
+        return ExecutionContext(
+            app_context=NullAppContext(),
+            context_vars=contextvars.copy_context(),
+        )
+
+    return _capturer()
+
+
+def reset_context_provider() -> None:
+    """
+    Reset the context capturer.
+
+    This is primarily useful for testing to ensure a clean state.
+    """
+    global _capturer
+    _capturer = None
+
 
 __all__ = [
-    "AppContext",
-    "ContextProviderNotFoundError",
-    "ExecutionContext",
-    "ExecutionContextBuilder",
-    "IExecutionContext",
-    "NullAppContext",
-    "SandboxContext",
     "capture_current_context",
-    "read_context",
-    "register_context",
     "register_context_capturer",
     "reset_context_provider",
 ]

api/context/flask_app_context.py

@@ -10,7 +10,11 @@ from typing import Any, final
 
 from flask import Flask, current_app, g
 
-from context.execution_context import AppContext, IExecutionContext, register_context_capturer
+from dify_graph.context import register_context_capturer
+from dify_graph.context.execution_context import (
+    AppContext,
+    IExecutionContext,
+)
 
 
 @final

api/contexts/__init__.py

@@ -6,6 +6,7 @@ from contexts.wrapper import RecyclableContextVar
 
 if TYPE_CHECKING:
     from core.datasource.__base.datasource_provider import DatasourcePluginProviderController
+    from core.plugin.entities.plugin_daemon import PluginModelProviderEntity
     from core.tools.plugin_tool.provider import PluginToolProviderController
     from core.trigger.provider import PluginTriggerProviderController
 
@@ -19,6 +20,14 @@ plugin_tool_providers: RecyclableContextVar[dict[str, "PluginToolProviderControl
 
 plugin_tool_providers_lock: RecyclableContextVar[Lock] = RecyclableContextVar(ContextVar("plugin_tool_providers_lock"))
 
+plugin_model_providers: RecyclableContextVar[list["PluginModelProviderEntity"] | None] = RecyclableContextVar(
+    ContextVar("plugin_model_providers")
+)
+
+plugin_model_providers_lock: RecyclableContextVar[Lock] = RecyclableContextVar(
+    ContextVar("plugin_model_providers_lock")
+)
+
 datasource_plugin_providers: RecyclableContextVar[dict[str, "DatasourcePluginProviderController"]] = (
     RecyclableContextVar(ContextVar("datasource_plugin_providers"))
 )

api/controllers/common/fields.py

@@ -2,9 +2,9 @@ from __future__ import annotations
 
 from typing import Any, TypeAlias
 
-from graphon.file import helpers as file_helpers
 from pydantic import BaseModel, ConfigDict, computed_field
 
+from dify_graph.file import helpers as file_helpers
 from models.model import IconType
 
 JSONValue: TypeAlias = str | int | float | bool | None | dict[str, Any] | list[Any]

api/controllers/console/app/app.py

@@ -5,8 +5,6 @@ from typing import Any, Literal, TypeAlias
 
 from flask import request
 from flask_restx import Resource
-from graphon.enums import WorkflowExecutionStatus
-from graphon.file import helpers as file_helpers
 from pydantic import AliasChoices, BaseModel, ConfigDict, Field, computed_field, field_validator
 from sqlalchemy import select
 from sqlalchemy.orm import Session
@@ -28,6 +26,8 @@ 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.file import helpers as file_helpers
 from extensions.ext_database import db
 from libs.login import current_account_with_tenant, login_required
 from models import App, DatasetPermissionEnum, Workflow

api/controllers/console/app/audio.py

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

api/controllers/console/app/completion.py

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

api/controllers/console/app/generator.py

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

api/controllers/console/app/message.py

@@ -3,7 +3,6 @@ from typing import Literal
 
 from flask import request
 from flask_restx import Resource, fields, marshal_with
-from graphon.model_runtime.errors.invoke import InvokeError
 from pydantic import BaseModel, Field, field_validator
 from sqlalchemy import exists, func, select
 from werkzeug.exceptions import InternalServerError, NotFound
@@ -25,6 +24,7 @@ from controllers.console.wraps import (
 )
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
+from dify_graph.model_runtime.errors.invoke import InvokeError
 from extensions.ext_database import db
 from fields.raws import FilesContainedField
 from libs.helper import TimestampField, uuid_value

api/controllers/console/app/model_config.py

@@ -88,7 +88,6 @@ class ModelConfigResource(Resource):
                         tenant_id=current_tenant_id,
                         app_id=app_model.id,
                         agent_tool=agent_tool_entity,
-                        user_id=current_user.id,
                     )
                     manager = ToolParameterConfigurationManager(
                         tenant_id=current_tenant_id,
@@ -128,7 +127,6 @@ class ModelConfigResource(Resource):
                             tenant_id=current_tenant_id,
                             app_id=app_model.id,
                             agent_tool=agent_tool_entity,
-                            user_id=current_user.id,
                         )
                     except Exception:
                         continue

api/controllers/console/app/workflow.py

@@ -5,10 +5,6 @@ from typing import Any
 
 from flask import abort, request
 from flask_restx import Resource, fields, marshal_with
-from graphon.enums import NodeType
-from graphon.file import File
-from graphon.graph_engine.manager import GraphEngineManager
-from graphon.model_runtime.utils.encoders import jsonable_encoder
 from pydantic import BaseModel, Field, field_validator
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import BadRequest, Forbidden, InternalServerError, NotFound
@@ -24,7 +20,6 @@ from core.app.app_config.features.file_upload.manager import FileUploadConfigMan
 from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.apps.workflow.app_generator import SKIP_PREPARE_USER_INPUTS_KEY
 from core.app.entities.app_invoke_entities import InvokeFrom
-from core.app.file_access import DatabaseFileAccessController
 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
@@ -34,6 +29,10 @@ from core.trigger.debug.event_selectors import (
     create_event_poller,
     select_trigger_debug_events,
 )
+from dify_graph.enums import NodeType
+from dify_graph.file.models import File
+from dify_graph.graph_engine.manager import GraphEngineManager
+from dify_graph.model_runtime.utils.encoders import jsonable_encoder
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from factories import file_factory, variable_factory
@@ -52,7 +51,6 @@ from services.errors.llm import InvokeRateLimitError
 from services.workflow_service import DraftWorkflowDeletionError, WorkflowInUseError, WorkflowService
 
 logger = logging.getLogger(__name__)
-_file_access_controller = DatabaseFileAccessController()
 LISTENING_RETRY_IN = 2000
 DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 RESTORE_SOURCE_WORKFLOW_MUST_BE_PUBLISHED_MESSAGE = "source workflow must be published"
@@ -206,7 +204,6 @@ def _parse_file(workflow: Workflow, files: list[dict] | None = None) -> Sequence
         mappings=files,
         tenant_id=workflow.tenant_id,
         config=file_extra_config,
-        access_controller=_file_access_controller,
     )
     return file_objs
 

api/controllers/console/app/workflow_app_log.py

@@ -3,13 +3,13 @@ from datetime import datetime
 from dateutil.parser import isoparse
 from flask import request
 from flask_restx import Resource, marshal_with
-from graphon.enums import WorkflowExecutionStatus
 from pydantic import BaseModel, Field, field_validator
 from sqlalchemy.orm import Session
 
 from controllers.console import console_ns
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
+from dify_graph.enums import WorkflowExecutionStatus
 from extensions.ext_database import db
 from fields.workflow_app_log_fields import (
     build_workflow_app_log_pagination_model,

api/controllers/console/app/workflow_draft_variable.py

@@ -5,10 +5,6 @@ from typing import Any, NoReturn, ParamSpec, TypeVar
 
 from flask import Response, request
 from flask_restx import Resource, fields, marshal, marshal_with
-from graphon.file import helpers as file_helpers
-from graphon.variables.segment_group import SegmentGroup
-from graphon.variables.segments import ArrayFileSegment, FileSegment, Segment
-from graphon.variables.types import SegmentType
 from pydantic import BaseModel, Field
 from sqlalchemy.orm import Session
 
@@ -19,8 +15,11 @@ from controllers.console.app.error import (
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
 from controllers.web.error import InvalidArgumentError, NotFoundError
-from core.app.file_access import DatabaseFileAccessController
-from core.workflow.variable_prefixes import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
+from dify_graph.constants import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
+from dify_graph.file import helpers as file_helpers
+from dify_graph.variables.segment_group import SegmentGroup
+from dify_graph.variables.segments import ArrayFileSegment, FileSegment, Segment
+from dify_graph.variables.types import SegmentType
 from extensions.ext_database import db
 from factories.file_factory import build_from_mapping, build_from_mappings
 from factories.variable_factory import build_segment_with_type
@@ -31,7 +30,6 @@ from services.workflow_draft_variable_service import WorkflowDraftVariableList,
 from services.workflow_service import WorkflowService
 
 logger = logging.getLogger(__name__)
-_file_access_controller = DatabaseFileAccessController()
 DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
 
@@ -391,21 +389,13 @@ class VariableApi(Resource):
             if variable.value_type == SegmentType.FILE:
                 if not isinstance(raw_value, dict):
                     raise InvalidArgumentError(description=f"expected dict for file, got {type(raw_value)}")
-                raw_value = build_from_mapping(
-                    mapping=raw_value,
-                    tenant_id=app_model.tenant_id,
-                    access_controller=_file_access_controller,
-                )
+                raw_value = build_from_mapping(mapping=raw_value, tenant_id=app_model.tenant_id)
             elif variable.value_type == SegmentType.ARRAY_FILE:
                 if not isinstance(raw_value, list):
                     raise InvalidArgumentError(description=f"expected list for files, got {type(raw_value)}")
                 if len(raw_value) > 0 and not isinstance(raw_value[0], dict):
                     raise InvalidArgumentError(description=f"expected dict for files[0], got {type(raw_value)}")
-                raw_value = build_from_mappings(
-                    mappings=raw_value,
-                    tenant_id=app_model.tenant_id,
-                    access_controller=_file_access_controller,
-                )
+                raw_value = build_from_mappings(mappings=raw_value, tenant_id=app_model.tenant_id)
             new_value = build_segment_with_type(variable.value_type, raw_value)
         draft_var_srv.update_variable(variable, name=new_name, value=new_value)
         db.session.commit()

api/controllers/console/app/workflow_run.py

@@ -1,10 +1,8 @@
 from datetime import UTC, datetime, timedelta
-from typing import Literal, TypedDict, cast
+from typing import Literal, cast
 
 from flask import request
 from flask_restx import Resource, fields, marshal_with
-from graphon.entities.pause_reason import HumanInputRequired
-from graphon.enums import WorkflowExecutionStatus
 from pydantic import BaseModel, Field, field_validator
 from sqlalchemy import select
 from sqlalchemy.orm import sessionmaker
@@ -14,7 +12,8 @@ from controllers.console import console_ns
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
 from controllers.web.error import NotFoundError
-from core.workflow.human_input_forms import load_form_tokens_by_form_id as _load_form_tokens_by_form_id
+from dify_graph.entities.pause_reason import HumanInputRequired
+from dify_graph.enums import WorkflowExecutionStatus
 from extensions.ext_database import db
 from fields.end_user_fields import simple_end_user_fields
 from fields.member_fields import simple_account_fields
@@ -173,23 +172,6 @@ console_ns.schema_model(
 )
 
 
-class HumanInputPauseTypeResponse(TypedDict):
-    type: Literal["human_input"]
-    form_id: str
-    backstage_input_url: str | None
-
-
-class PausedNodeResponse(TypedDict):
-    node_id: str
-    node_title: str
-    pause_type: HumanInputPauseTypeResponse
-
-
-class WorkflowPauseDetailsResponse(TypedDict):
-    paused_at: str | None
-    paused_nodes: list[PausedNodeResponse]
-
-
 @console_ns.route("/apps/<uuid:app_id>/advanced-chat/workflow-runs")
 class AdvancedChatAppWorkflowRunListApi(Resource):
     @console_ns.doc("get_advanced_chat_workflow_runs")
@@ -507,22 +489,18 @@ class ConsoleWorkflowPauseDetailsApi(Resource):
         # Check if workflow is suspended
         is_paused = workflow_run.status == WorkflowExecutionStatus.PAUSED
         if not is_paused:
-            empty_response: WorkflowPauseDetailsResponse = {
+            return {
                 "paused_at": None,
                 "paused_nodes": [],
-            }
-            return empty_response, 200
+            }, 200
 
         pause_entity = workflow_run_repo.get_workflow_pause(workflow_run_id)
         pause_reasons = pause_entity.get_pause_reasons() if pause_entity else []
-        form_tokens_by_form_id = _load_form_tokens_by_form_id(
-            [reason.form_id for reason in pause_reasons if isinstance(reason, HumanInputRequired)]
-        )
 
         # Build response
         paused_at = pause_entity.paused_at if pause_entity else None
-        paused_nodes: list[PausedNodeResponse] = []
-        response: WorkflowPauseDetailsResponse = {
+        paused_nodes = []
+        response = {
             "paused_at": paused_at.isoformat() + "Z" if paused_at else None,
             "paused_nodes": paused_nodes,
         }
@@ -536,9 +514,7 @@ class ConsoleWorkflowPauseDetailsApi(Resource):
                         "pause_type": {
                             "type": "human_input",
                             "form_id": reason.form_id,
-                            "backstage_input_url": _build_backstage_input_url(
-                                form_tokens_by_form_id.get(reason.form_id)
-                            ),
+                            "backstage_input_url": _build_backstage_input_url(reason.form_token),
                         },
                     }
                 )

api/controllers/console/auth/oauth_server.py

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

api/controllers/console/datasets/datasets.py

@@ -2,7 +2,6 @@ from typing import Any, cast
 
 from flask import request
 from flask_restx import Resource, fields, marshal, marshal_with
-from graphon.model_runtime.entities.model_entities import ModelType
 from pydantic import BaseModel, Field, field_validator
 from sqlalchemy import func, select
 from werkzeug.exceptions import Forbidden, NotFound
@@ -26,12 +25,13 @@ from controllers.console.wraps import (
 )
 from core.errors.error import LLMBadRequestError, ProviderTokenNotInitError
 from core.indexing_runner import IndexingRunner
-from core.plugin.impl.model_runtime_factory import create_plugin_provider_manager
+from core.provider_manager import ProviderManager
 from core.rag.datasource.vdb.vector_type import VectorType
 from core.rag.extractor.entity.datasource_type import DatasourceType
 from core.rag.extractor.entity.extract_setting import ExtractSetting, NotionInfo, WebsiteInfo
 from core.rag.index_processor.constant.index_type import IndexTechniqueType
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
+from dify_graph.model_runtime.entities.model_entities import ModelType
 from extensions.ext_database import db
 from fields.app_fields import app_detail_kernel_fields, related_app_list
 from fields.dataset_fields import (
@@ -332,7 +332,7 @@ class DatasetListApi(Resource):
             )
 
         # check embedding setting
-        provider_manager = create_plugin_provider_manager(tenant_id=current_tenant_id)
+        provider_manager = ProviderManager()
         configurations = provider_manager.get_configurations(tenant_id=current_tenant_id)
 
         embedding_models = configurations.get_models(model_type=ModelType.TEXT_EMBEDDING, only_active=True)
@@ -446,7 +446,7 @@ class DatasetApi(Resource):
             data.update({"partial_member_list": part_users_list})
 
         # check embedding setting
-        provider_manager = create_plugin_provider_manager(tenant_id=current_tenant_id)
+        provider_manager = ProviderManager()
         configurations = provider_manager.get_configurations(tenant_id=current_tenant_id)
 
         embedding_models = configurations.get_models(model_type=ModelType.TEXT_EMBEDDING, only_active=True)

api/controllers/console/datasets/datasets_document.py

@@ -9,8 +9,6 @@ from uuid import UUID
 import sqlalchemy as sa
 from flask import request, send_file
 from flask_restx import Resource, fields, marshal, marshal_with
-from graphon.model_runtime.entities.model_entities import ModelType
-from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
 from pydantic import BaseModel, Field
 from sqlalchemy import asc, desc, func, select
 from werkzeug.exceptions import Forbidden, NotFound
@@ -30,6 +28,8 @@ from core.plugin.impl.exc import PluginDaemonClientSideError
 from core.rag.extractor.entity.datasource_type import DatasourceType
 from core.rag.extractor.entity.extract_setting import ExtractSetting, NotionInfo, WebsiteInfo
 from core.rag.index_processor.constant.index_type import IndexTechniqueType
+from dify_graph.model_runtime.entities.model_entities import ModelType
+from dify_graph.model_runtime.errors.invoke import InvokeAuthorizationError
 from extensions.ext_database import db
 from fields.dataset_fields import dataset_fields
 from fields.document_fields import (
@@ -454,7 +454,7 @@ class DatasetInitApi(Resource):
             if knowledge_config.embedding_model is None or knowledge_config.embedding_model_provider is None:
                 raise ValueError("embedding model and embedding model provider are required for high quality indexing.")
             try:
-                model_manager = ModelManager.for_tenant(tenant_id=current_tenant_id)
+                model_manager = ModelManager()
                 model_manager.get_model_instance(
                     tenant_id=current_tenant_id,
                     provider=knowledge_config.embedding_model_provider,

api/controllers/console/datasets/datasets_segments.py

@@ -2,7 +2,6 @@ import uuid
 
 from flask import request
 from flask_restx import Resource, marshal
-from graphon.model_runtime.entities.model_entities import ModelType
 from pydantic import BaseModel, Field
 from sqlalchemy import String, cast, func, or_, select
 from sqlalchemy.dialects.postgresql import JSONB
@@ -28,6 +27,7 @@ from controllers.console.wraps import (
 from core.errors.error import LLMBadRequestError, ProviderTokenNotInitError
 from core.model_manager import ModelManager
 from core.rag.index_processor.constant.index_type import IndexTechniqueType
+from dify_graph.model_runtime.entities.model_entities import ModelType
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from fields.segment_fields import child_chunk_fields, segment_fields
@@ -283,7 +283,7 @@ class DatasetDocumentSegmentApi(Resource):
         if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
             # check embedding model setting
             try:
-                model_manager = ModelManager.for_tenant(tenant_id=current_tenant_id)
+                model_manager = ModelManager()
                 model_manager.get_model_instance(
                     tenant_id=current_tenant_id,
                     provider=dataset.embedding_model_provider,
@@ -336,7 +336,7 @@ class DatasetDocumentSegmentAddApi(Resource):
         # check embedding model setting
         if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
             try:
-                model_manager = ModelManager.for_tenant(tenant_id=current_tenant_id)
+                model_manager = ModelManager()
                 model_manager.get_model_instance(
                     tenant_id=current_tenant_id,
                     provider=dataset.embedding_model_provider,
@@ -387,7 +387,7 @@ class DatasetDocumentSegmentUpdateApi(Resource):
         if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
             # check embedding model setting
             try:
-                model_manager = ModelManager.for_tenant(tenant_id=current_tenant_id)
+                model_manager = ModelManager()
                 model_manager.get_model_instance(
                     tenant_id=current_tenant_id,
                     provider=dataset.embedding_model_provider,
@@ -572,7 +572,7 @@ class ChildChunkAddApi(Resource):
         # check embedding model setting
         if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
             try:
-                model_manager = ModelManager.for_tenant(tenant_id=current_tenant_id)
+                model_manager = ModelManager()
                 model_manager.get_model_instance(
                     tenant_id=current_tenant_id,
                     provider=dataset.embedding_model_provider,

api/controllers/console/datasets/external.py

@@ -25,7 +25,7 @@ from libs.login import current_account_with_tenant, login_required
 from services.dataset_service import DatasetService
 from services.external_knowledge_service import ExternalDatasetService
 from services.hit_testing_service import HitTestingService
-from services.knowledge_service import BedrockRetrievalSetting, ExternalDatasetTestService
+from services.knowledge_service import ExternalDatasetTestService
 
 
 def _build_dataset_detail_model():
@@ -86,7 +86,7 @@ class ExternalHitTestingPayload(BaseModel):
 
 
 class BedrockRetrievalPayload(BaseModel):
-    retrieval_setting: "BedrockRetrievalSetting"
+    retrieval_setting: dict[str, object]
     query: str
     knowledge_id: str
 

api/controllers/console/datasets/hit_testing_base.py

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

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

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

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

@@ -3,7 +3,6 @@ from typing import Any, NoReturn
 
 from flask import Response, request
 from flask_restx import Resource, marshal, marshal_with
-from graphon.variables.types import SegmentType
 from pydantic import BaseModel, Field
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden
@@ -22,8 +21,8 @@ from controllers.console.app.workflow_draft_variable import (
 from controllers.console.datasets.wraps import get_rag_pipeline
 from controllers.console.wraps import account_initialization_required, setup_required
 from controllers.web.error import InvalidArgumentError, NotFoundError
-from core.app.file_access import DatabaseFileAccessController
-from core.workflow.variable_prefixes import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
+from dify_graph.constants import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
+from dify_graph.variables.types import SegmentType
 from extensions.ext_database import db
 from factories.file_factory import build_from_mapping, build_from_mappings
 from factories.variable_factory import build_segment_with_type
@@ -34,7 +33,6 @@ from services.rag_pipeline.rag_pipeline import RagPipelineService
 from services.workflow_draft_variable_service import WorkflowDraftVariableList, WorkflowDraftVariableService
 
 logger = logging.getLogger(__name__)
-_file_access_controller = DatabaseFileAccessController()
 
 
 def _create_pagination_parser():
@@ -225,21 +223,13 @@ class RagPipelineVariableApi(Resource):
             if variable.value_type == SegmentType.FILE:
                 if not isinstance(raw_value, dict):
                     raise InvalidArgumentError(description=f"expected dict for file, got {type(raw_value)}")
-                raw_value = build_from_mapping(
-                    mapping=raw_value,
-                    tenant_id=pipeline.tenant_id,
-                    access_controller=_file_access_controller,
-                )
+                raw_value = build_from_mapping(mapping=raw_value, tenant_id=pipeline.tenant_id)
             elif variable.value_type == SegmentType.ARRAY_FILE:
                 if not isinstance(raw_value, list):
                     raise InvalidArgumentError(description=f"expected list for files, got {type(raw_value)}")
                 if len(raw_value) > 0 and not isinstance(raw_value[0], dict):
                     raise InvalidArgumentError(description=f"expected dict for files[0], got {type(raw_value)}")
-                raw_value = build_from_mappings(
-                    mappings=raw_value,
-                    tenant_id=pipeline.tenant_id,
-                    access_controller=_file_access_controller,
-                )
+                raw_value = build_from_mappings(mappings=raw_value, tenant_id=pipeline.tenant_id)
             new_value = build_segment_with_type(variable.value_type, raw_value)
         draft_var_srv.update_variable(variable, name=new_name, value=new_value)
         db.session.commit()

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

@@ -4,7 +4,6 @@ from typing import Any, Literal, cast
 
 from flask import abort, request
 from flask_restx import Resource, marshal_with  # type: ignore
-from graphon.model_runtime.utils.encoders import jsonable_encoder
 from pydantic import BaseModel, Field
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import BadRequest, Forbidden, InternalServerError, NotFound
@@ -38,6 +37,7 @@ from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpErr
 from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.apps.pipeline.pipeline_generator import PipelineGenerator
 from core.app.entities.app_invoke_entities import InvokeFrom
+from dify_graph.model_runtime.utils.encoders import jsonable_encoder
 from extensions.ext_database import db
 from factories import variable_factory
 from libs import helper

api/controllers/console/explore/audio.py

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

api/controllers/console/explore/completion.py

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

api/controllers/console/explore/message.py

@@ -2,7 +2,6 @@ import logging
 from typing import Literal
 
 from flask import request
-from graphon.model_runtime.errors.invoke import InvokeError
 from pydantic import BaseModel, Field, TypeAdapter
 from werkzeug.exceptions import InternalServerError, NotFound
 
@@ -22,6 +21,7 @@ from controllers.console.explore.error import (
 from controllers.console.explore.wraps import InstalledAppResource
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
+from dify_graph.model_runtime.errors.invoke import InvokeError
 from fields.conversation_fields import ResultResponse
 from fields.message_fields import MessageInfiniteScrollPagination, MessageListItem, SuggestedQuestionsResponse
 from libs import helper

api/controllers/console/explore/trial.py

@@ -3,8 +3,6 @@ from typing import Any, Literal, cast
 
 from flask import request
 from flask_restx import Resource, fields, marshal, marshal_with
-from graphon.graph_engine.manager import GraphEngineManager
-from graphon.model_runtime.errors.invoke import InvokeError
 from pydantic import BaseModel
 from sqlalchemy import select
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
@@ -44,6 +42,8 @@ from core.errors.error import (
     ProviderTokenNotInitError,
     QuotaExceededError,
 )
+from dify_graph.graph_engine.manager import GraphEngineManager
+from dify_graph.model_runtime.errors.invoke import InvokeError
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from fields.app_fields import (

api/controllers/console/explore/workflow.py

@@ -1,8 +1,6 @@
 import logging
 from typing import Any
 
-from graphon.graph_engine.manager import GraphEngineManager
-from graphon.model_runtime.errors.invoke import InvokeError
 from pydantic import BaseModel
 from werkzeug.exceptions import InternalServerError
 
@@ -23,6 +21,8 @@ from core.errors.error import (
     ProviderTokenNotInitError,
     QuotaExceededError,
 )
+from dify_graph.graph_engine.manager import GraphEngineManager
+from dify_graph.model_runtime.errors.invoke import InvokeError
 from extensions.ext_redis import redis_client
 from libs import helper
 from libs.login import current_account_with_tenant

api/controllers/console/human_input_form.py

@@ -15,7 +15,6 @@ from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, setup_required
 from controllers.web.error import InvalidArgumentError, NotFoundError
 from core.app.apps.advanced_chat.app_generator import AdvancedChatAppGenerator
-from core.app.apps.base_app_generator import BaseAppGenerator
 from core.app.apps.common.workflow_response_converter import WorkflowResponseConverter
 from core.app.apps.message_generator import MessageGenerator
 from core.app.apps.workflow.app_generator import WorkflowAppGenerator
@@ -167,7 +166,6 @@ class ConsoleWorkflowEventsApi(Resource):
 
         else:
             msg_generator = MessageGenerator()
-            generator: BaseAppGenerator
             if app.mode == AppMode.ADVANCED_CHAT:
                 generator = AdvancedChatAppGenerator()
             elif app.mode == AppMode.WORKFLOW:
@@ -204,7 +202,7 @@ class ConsoleWorkflowEventsApi(Resource):
         )
 
 
-def _retrieve_app_for_workflow_run(session: Session, workflow_run: WorkflowRun) -> App:
+def _retrieve_app_for_workflow_run(session: Session, workflow_run: WorkflowRun):
     query = select(App).where(
         App.id == workflow_run.app_id,
         App.tenant_id == workflow_run.tenant_id,

api/controllers/console/remote_files.py

@@ -2,7 +2,6 @@ import urllib.parse
 
 import httpx
 from flask_restx import Resource
-from graphon.file import helpers as file_helpers
 from pydantic import BaseModel, Field
 
 import services
@@ -14,6 +13,7 @@ from controllers.common.errors import (
 )
 from controllers.console import console_ns
 from core.helper import ssrf_proxy
+from dify_graph.file import helpers as file_helpers
 from extensions.ext_database import db
 from fields.file_fields import FileWithSignedUrl, RemoteFileInfo
 from libs.login import current_account_with_tenant, login_required

api/controllers/console/workspace/agent_providers.py

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

api/controllers/console/workspace/endpoint.py

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

api/controllers/console/workspace/load_balancing_config.py

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

api/controllers/console/workspace/model_providers.py

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

api/controllers/console/workspace/models.py

@@ -3,14 +3,14 @@ from typing import Any, cast
 
 from flask import request
 from flask_restx import Resource
-from graphon.model_runtime.entities.model_entities import ModelType
-from graphon.model_runtime.errors.validate import CredentialsValidateFailedError
-from graphon.model_runtime.utils.encoders import jsonable_encoder
 from pydantic import BaseModel, Field, field_validator
 
 from controllers.common.schema import register_enum_models, register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, is_admin_or_owner_required, setup_required
+from dify_graph.model_runtime.entities.model_entities import ModelType
+from dify_graph.model_runtime.errors.validate import CredentialsValidateFailedError
+from dify_graph.model_runtime.utils.encoders import jsonable_encoder
 from libs.helper import uuid_value
 from libs.login import current_account_with_tenant, login_required
 from services.model_load_balancing_service import ModelLoadBalancingService
@@ -282,18 +282,14 @@ class ModelProviderModelCredentialApi(Resource):
         )
 
         if args.config_from == "predefined-model":
-            available_credentials = model_provider_service.get_provider_available_credentials(
-                tenant_id=tenant_id,
-                provider=provider,
+            available_credentials = model_provider_service.provider_manager.get_provider_available_credentials(
+                tenant_id=tenant_id, provider_name=provider
             )
         else:
             # Normalize model_type to the origin value stored in DB (e.g., "text-generation" for LLM)
             normalized_model_type = args.model_type.to_origin_model_type()
-            available_credentials = model_provider_service.get_provider_model_available_credentials(
-                tenant_id=tenant_id,
-                provider=provider,
-                model_type=normalized_model_type,
-                model=args.model,
+            available_credentials = model_provider_service.provider_manager.get_provider_model_available_credentials(
+                tenant_id=tenant_id, provider_name=provider, model_type=normalized_model_type, model_name=args.model
             )
 
         return jsonable_encoder(

api/controllers/console/workspace/plugin.py

@@ -4,7 +4,6 @@ from typing import Any, Literal
 
 from flask import request, send_file
 from flask_restx import Resource
-from graphon.model_runtime.utils.encoders import jsonable_encoder
 from pydantic import BaseModel, Field
 from werkzeug.datastructures import FileStorage
 from werkzeug.exceptions import Forbidden
@@ -15,6 +14,7 @@ from controllers.console import console_ns
 from controllers.console.workspace import plugin_permission_required
 from controllers.console.wraps import account_initialization_required, is_admin_or_owner_required, setup_required
 from core.plugin.impl.exc import PluginDaemonClientSideError
+from dify_graph.model_runtime.utils.encoders import jsonable_encoder
 from libs.login import current_account_with_tenant, login_required
 from models.account import TenantPluginAutoUpgradeStrategy, TenantPluginPermission
 from services.plugin.plugin_auto_upgrade_service import PluginAutoUpgradeService
@@ -200,7 +200,7 @@ class PluginDebuggingKeyApi(Resource):
                 "port": dify_config.PLUGIN_REMOTE_INSTALL_PORT,
             }
         except PluginDaemonClientSideError as e:
-            return {"code": "plugin_error", "message": e.description}, 400
+            raise ValueError(e)
 
 
 @console_ns.route("/workspaces/current/plugin/list")
@@ -215,7 +215,7 @@ class PluginListApi(Resource):
         try:
             plugins_with_total = PluginService.list_with_total(tenant_id, args.page, args.page_size)
         except PluginDaemonClientSideError as e:
-            return {"code": "plugin_error", "message": e.description}, 400
+            raise ValueError(e)
 
         return jsonable_encoder({"plugins": plugins_with_total.list, "total": plugins_with_total.total})
 
@@ -232,7 +232,7 @@ class PluginListLatestVersionsApi(Resource):
         try:
             versions = PluginService.list_latest_versions(args.plugin_ids)
         except PluginDaemonClientSideError as e:
-            return {"code": "plugin_error", "message": e.description}, 400
+            raise ValueError(e)
 
         return jsonable_encoder({"versions": versions})
 
@@ -251,7 +251,7 @@ class PluginListInstallationsFromIdsApi(Resource):
         try:
             plugins = PluginService.list_installations_from_ids(tenant_id, args.plugin_ids)
         except PluginDaemonClientSideError as e:
-            return {"code": "plugin_error", "message": e.description}, 400
+            raise ValueError(e)
 
         return jsonable_encoder({"plugins": plugins})
 
@@ -266,7 +266,7 @@ class PluginIconApi(Resource):
         try:
             icon_bytes, mimetype = PluginService.get_asset(args.tenant_id, args.filename)
         except PluginDaemonClientSideError as e:
-            return {"code": "plugin_error", "message": e.description}, 400
+            raise ValueError(e)
 
         icon_cache_max_age = dify_config.TOOL_ICON_CACHE_MAX_AGE
         return send_file(io.BytesIO(icon_bytes), mimetype=mimetype, max_age=icon_cache_max_age)
@@ -286,7 +286,7 @@ class PluginAssetApi(Resource):
             binary = PluginService.extract_asset(tenant_id, args.plugin_unique_identifier, args.file_name)
             return send_file(io.BytesIO(binary), mimetype="application/octet-stream")
         except PluginDaemonClientSideError as e:
-            return {"code": "plugin_error", "message": e.description}, 400
+            raise ValueError(e)
 
 
 @console_ns.route("/workspaces/current/plugin/upload/pkg")
@@ -303,7 +303,7 @@ class PluginUploadFromPkgApi(Resource):
         try:
             response = PluginService.upload_pkg(tenant_id, content)
         except PluginDaemonClientSideError as e:
-            return {"code": "plugin_error", "message": e.description}, 400
+            raise ValueError(e)
 
         return jsonable_encoder(response)
 
@@ -323,7 +323,7 @@ class PluginUploadFromGithubApi(Resource):
         try:
             response = PluginService.upload_pkg_from_github(tenant_id, args.repo, args.version, args.package)
         except PluginDaemonClientSideError as e:
-            return {"code": "plugin_error", "message": e.description}, 400
+            raise ValueError(e)
 
         return jsonable_encoder(response)
 
@@ -361,7 +361,7 @@ class PluginInstallFromPkgApi(Resource):
         try:
             response = PluginService.install_from_local_pkg(tenant_id, args.plugin_unique_identifiers)
         except PluginDaemonClientSideError as e:
-            return {"code": "plugin_error", "message": e.description}, 400
+            raise ValueError(e)
 
         return jsonable_encoder(response)
 
@@ -387,7 +387,7 @@ class PluginInstallFromGithubApi(Resource):
                 args.package,
             )
         except PluginDaemonClientSideError as e:
-            return {"code": "plugin_error", "message": e.description}, 400
+            raise ValueError(e)
 
         return jsonable_encoder(response)
 
@@ -407,7 +407,7 @@ class PluginInstallFromMarketplaceApi(Resource):
         try:
             response = PluginService.install_from_marketplace_pkg(tenant_id, args.plugin_unique_identifiers)
         except PluginDaemonClientSideError as e:
-            return {"code": "plugin_error", "message": e.description}, 400
+            raise ValueError(e)
 
         return jsonable_encoder(response)
 
@@ -433,7 +433,7 @@ class PluginFetchMarketplacePkgApi(Resource):
                 }
             )
         except PluginDaemonClientSideError as e:
-            return {"code": "plugin_error", "message": e.description}, 400
+            raise ValueError(e)
 
 
 @console_ns.route("/workspaces/current/plugin/fetch-manifest")
@@ -453,7 +453,7 @@ class PluginFetchManifestApi(Resource):
                 {"manifest": PluginService.fetch_plugin_manifest(tenant_id, args.plugin_unique_identifier).model_dump()}
             )
         except PluginDaemonClientSideError as e:
-            return {"code": "plugin_error", "message": e.description}, 400
+            raise ValueError(e)
 
 
 @console_ns.route("/workspaces/current/plugin/tasks")
@@ -471,7 +471,7 @@ class PluginFetchInstallTasksApi(Resource):
         try:
             return jsonable_encoder({"tasks": PluginService.fetch_install_tasks(tenant_id, args.page, args.page_size)})
         except PluginDaemonClientSideError as e:
-            return {"code": "plugin_error", "message": e.description}, 400
+            raise ValueError(e)
 
 
 @console_ns.route("/workspaces/current/plugin/tasks/<task_id>")
@@ -486,7 +486,7 @@ class PluginFetchInstallTaskApi(Resource):
         try:
             return jsonable_encoder({"task": PluginService.fetch_install_task(tenant_id, task_id)})
         except PluginDaemonClientSideError as e:
-            return {"code": "plugin_error", "message": e.description}, 400
+            raise ValueError(e)
 
 
 @console_ns.route("/workspaces/current/plugin/tasks/<task_id>/delete")
@@ -501,7 +501,7 @@ class PluginDeleteInstallTaskApi(Resource):
         try:
             return {"success": PluginService.delete_install_task(tenant_id, task_id)}
         except PluginDaemonClientSideError as e:
-            return {"code": "plugin_error", "message": e.description}, 400
+            raise ValueError(e)
 
 
 @console_ns.route("/workspaces/current/plugin/tasks/delete_all")
@@ -516,7 +516,7 @@ class PluginDeleteAllInstallTaskItemsApi(Resource):
         try:
             return {"success": PluginService.delete_all_install_task_items(tenant_id)}
         except PluginDaemonClientSideError as e:
-            return {"code": "plugin_error", "message": e.description}, 400
+            raise ValueError(e)
 
 
 @console_ns.route("/workspaces/current/plugin/tasks/<task_id>/delete/<path:identifier>")
@@ -531,7 +531,7 @@ class PluginDeleteInstallTaskItemApi(Resource):
         try:
             return {"success": PluginService.delete_install_task_item(tenant_id, task_id, identifier)}
         except PluginDaemonClientSideError as e:
-            return {"code": "plugin_error", "message": e.description}, 400
+            raise ValueError(e)
 
 
 @console_ns.route("/workspaces/current/plugin/upgrade/marketplace")
@@ -553,7 +553,7 @@ class PluginUpgradeFromMarketplaceApi(Resource):
                 )
             )
         except PluginDaemonClientSideError as e:
-            return {"code": "plugin_error", "message": e.description}, 400
+            raise ValueError(e)
 
 
 @console_ns.route("/workspaces/current/plugin/upgrade/github")
@@ -580,7 +580,7 @@ class PluginUpgradeFromGithubApi(Resource):
                 )
             )
         except PluginDaemonClientSideError as e:
-            return {"code": "plugin_error", "message": e.description}, 400
+            raise ValueError(e)
 
 
 @console_ns.route("/workspaces/current/plugin/uninstall")
@@ -598,7 +598,7 @@ class PluginUninstallApi(Resource):
         try:
             return {"success": PluginService.uninstall(tenant_id, args.plugin_installation_id)}
         except PluginDaemonClientSideError as e:
-            return {"code": "plugin_error", "message": e.description}, 400
+            raise ValueError(e)
 
 
 @console_ns.route("/workspaces/current/plugin/permission/change")
@@ -674,7 +674,7 @@ class PluginFetchDynamicSelectOptionsApi(Resource):
                 provider_type=args.provider_type,
             )
         except PluginDaemonClientSideError as e:
-            return {"code": "plugin_error", "message": e.description}, 400
+            raise ValueError(e)
 
         return jsonable_encoder({"options": options})
 
@@ -705,7 +705,7 @@ class PluginFetchDynamicSelectOptionsWithCredentialsApi(Resource):
                 credentials=args.credentials,
             )
         except PluginDaemonClientSideError as e:
-            return {"code": "plugin_error", "message": e.description}, 400
+            raise ValueError(e)
 
         return jsonable_encoder({"options": options})
 

api/controllers/console/workspace/tool_providers.py

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

api/controllers/console/workspace/trigger_providers.py

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

api/controllers/files/tool_files.py

@@ -70,25 +70,22 @@ class ToolFileApi(Resource):
         except Exception:
             raise UnsupportedFileTypeError()
 
-        mime_type = tool_file.mime_type
-        filename = tool_file.filename
-
         response = Response(
             stream,
-            mimetype=mime_type,
+            mimetype=tool_file.mimetype,
             direct_passthrough=True,
             headers={},
         )
         if tool_file.size > 0:
             response.headers["Content-Length"] = str(tool_file.size)
-        if args.as_attachment and filename:
-            encoded_filename = quote(filename)
+        if args.as_attachment:
+            encoded_filename = quote(tool_file.name)
             response.headers["Content-Disposition"] = f"attachment; filename*=UTF-8''{encoded_filename}"
 
         enforce_download_for_html(
             response,
-            mime_type=mime_type,
-            filename=filename,
+            mime_type=tool_file.mimetype,
+            filename=tool_file.name,
             extension=extension,
         )
 

api/controllers/files/upload.py

@@ -7,8 +7,8 @@ from pydantic import BaseModel, Field
 from werkzeug.exceptions import Forbidden
 
 import services
-from core.tools.signature import verify_plugin_file_signature
 from core.tools.tool_file_manager import ToolFileManager
+from dify_graph.file.helpers import verify_plugin_file_signature
 from fields.file_fields import FileResponse
 
 from ..common.errors import (

api/controllers/inner_api/__init__.py

@@ -16,14 +16,12 @@ api = ExternalApi(
 inner_api_ns = Namespace("inner_api", description="Internal API operations", path="/")
 
 from . import mail as _mail
-from .app import dsl as _app_dsl
 from .plugin import plugin as _plugin
 from .workspace import workspace as _workspace
 
 api.add_namespace(inner_api_ns)
 
 __all__ = [
-    "_app_dsl",
     "_mail",
     "_plugin",
     "_workspace",

api/controllers/inner_api/app/dsl.py

@@ -1,110 +0,0 @@
-"""Inner API endpoints for app DSL import/export.
-
-Called by the enterprise admin-api service. Import requires ``creator_email``
-to attribute the created app; workspace/membership validation is done by the
-Go admin-api caller.
-"""
-
-from flask import request
-from flask_restx import Resource
-from pydantic import BaseModel, Field
-from sqlalchemy.orm import Session
-
-from controllers.common.schema import register_schema_model
-from controllers.console.wraps import setup_required
-from controllers.inner_api import inner_api_ns
-from controllers.inner_api.wraps import enterprise_inner_api_only
-from extensions.ext_database import db
-from models import Account, App
-from models.account import AccountStatus
-from services.app_dsl_service import AppDslService, ImportMode, ImportStatus
-
-
-class InnerAppDSLImportPayload(BaseModel):
-    yaml_content: str = Field(description="YAML DSL content")
-    creator_email: str = Field(description="Email of the workspace member who will own the imported app")
-    name: str | None = Field(default=None, description="Override app name from DSL")
-    description: str | None = Field(default=None, description="Override app description from DSL")
-
-
-register_schema_model(inner_api_ns, InnerAppDSLImportPayload)
-
-
-@inner_api_ns.route("/enterprise/workspaces/<string:workspace_id>/dsl/import")
-class EnterpriseAppDSLImport(Resource):
-    @setup_required
-    @enterprise_inner_api_only
-    @inner_api_ns.doc("enterprise_app_dsl_import")
-    @inner_api_ns.expect(inner_api_ns.models[InnerAppDSLImportPayload.__name__])
-    @inner_api_ns.doc(
-        responses={
-            200: "Import completed",
-            202: "Import pending (DSL version mismatch requires confirmation)",
-            400: "Import failed (business error)",
-            404: "Creator account not found or inactive",
-        }
-    )
-    def post(self, workspace_id: str):
-        """Import a DSL into a workspace on behalf of a specified creator."""
-        args = InnerAppDSLImportPayload.model_validate(inner_api_ns.payload or {})
-
-        account = _get_active_account(args.creator_email)
-        if account is None:
-            return {"message": f"account '{args.creator_email}' not found or inactive"}, 404
-
-        account.set_tenant_id(workspace_id)
-
-        with Session(db.engine) as session:
-            dsl_service = AppDslService(session)
-            result = dsl_service.import_app(
-                account=account,
-                import_mode=ImportMode.YAML_CONTENT,
-                yaml_content=args.yaml_content,
-                name=args.name,
-                description=args.description,
-            )
-            session.commit()
-
-        if result.status == ImportStatus.FAILED:
-            return result.model_dump(mode="json"), 400
-        if result.status == ImportStatus.PENDING:
-            return result.model_dump(mode="json"), 202
-        return result.model_dump(mode="json"), 200
-
-
-@inner_api_ns.route("/enterprise/apps/<string:app_id>/dsl")
-class EnterpriseAppDSLExport(Resource):
-    @setup_required
-    @enterprise_inner_api_only
-    @inner_api_ns.doc(
-        "enterprise_app_dsl_export",
-        responses={
-            200: "Export successful",
-            404: "App not found",
-        },
-    )
-    def get(self, app_id: str):
-        """Export an app's DSL as YAML."""
-        include_secret = request.args.get("include_secret", "false").lower() == "true"
-
-        app_model = db.session.query(App).filter_by(id=app_id).first()
-        if not app_model:
-            return {"message": "app not found"}, 404
-
-        data = AppDslService.export_dsl(
-            app_model=app_model,
-            include_secret=include_secret,
-        )
-
-        return {"data": data}, 200
-
-
-def _get_active_account(email: str) -> Account | None:
-    """Look up an active account by email.
-
-    Workspace membership is already validated by the Go admin-api caller.
-    """
-    account = db.session.query(Account).filter_by(email=email).first()
-    if account is None or account.status != AccountStatus.ACTIVE:
-        return None
-    return account

api/controllers/inner_api/plugin/plugin.py

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

api/controllers/mcp/mcp.py

@@ -2,7 +2,6 @@ from typing import Any, Union
 
 from flask import Response
 from flask_restx import Resource
-from graphon.variables.input_entities import VariableEntity
 from pydantic import BaseModel, Field, ValidationError
 from sqlalchemy.orm import Session
 
@@ -10,6 +9,7 @@ from controllers.common.schema import register_schema_model
 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

api/controllers/service_api/app/audio.py

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

api/controllers/service_api/app/completion.py

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

api/controllers/service_api/app/file_preview.py

@@ -4,7 +4,6 @@ from urllib.parse import quote
 from flask import Response, request
 from flask_restx import Resource
 from pydantic import BaseModel, Field
-from sqlalchemy import select
 
 from controllers.common.file_response import enforce_download_for_html
 from controllers.common.schema import register_schema_model
@@ -103,27 +102,27 @@ class FilePreviewApi(Resource):
                 raise FileAccessDeniedError("Invalid file or app identifier")
 
             # First, find the MessageFile that references this upload file
-            message_file = db.session.scalar(select(MessageFile).where(MessageFile.upload_file_id == file_id).limit(1))
+            message_file = db.session.query(MessageFile).where(MessageFile.upload_file_id == file_id).first()
 
             if not message_file:
                 raise FileNotFoundError("File not found in message context")
 
             # Get the message and verify it belongs to the requesting app
-            message = db.session.scalar(
-                select(Message).where(Message.id == message_file.message_id, Message.app_id == app_id).limit(1)
+            message = (
+                db.session.query(Message).where(Message.id == message_file.message_id, Message.app_id == app_id).first()
             )
 
             if not message:
                 raise FileAccessDeniedError("File access denied: not owned by requesting app")
 
             # Get the actual upload file record
-            upload_file = db.session.get(UploadFile, file_id)
+            upload_file = db.session.query(UploadFile).where(UploadFile.id == file_id).first()
 
             if not upload_file:
                 raise FileNotFoundError("Upload file record not found")
 
             # Additional security: verify tenant isolation
-            app = db.session.get(App, app_id)
+            app = db.session.query(App).where(App.id == app_id).first()
             if app and upload_file.tenant_id != app.tenant_id:
                 raise FileAccessDeniedError("File access denied: tenant mismatch")
 

api/controllers/service_api/app/site.py

@@ -1,5 +1,4 @@
 from flask_restx import Resource
-from sqlalchemy import select
 from werkzeug.exceptions import Forbidden
 
 from controllers.common.fields import Site as SiteResponse
@@ -29,7 +28,7 @@ class AppSiteApi(Resource):
 
         Returns the site configuration for the application including theme, icons, and text.
         """
-        site = db.session.scalar(select(Site).where(Site.app_id == app_model.id).limit(1))
+        site = db.session.query(Site).where(Site.app_id == app_model.id).first()
 
         if not site:
             raise Forbidden()

api/controllers/service_api/app/workflow.py

@@ -4,9 +4,6 @@ from typing import Any, Literal
 from dateutil.parser import isoparse
 from flask import request
 from flask_restx import Namespace, Resource, fields
-from graphon.enums import WorkflowExecutionStatus
-from graphon.graph_engine.manager import GraphEngineManager
-from graphon.model_runtime.errors.invoke import InvokeError
 from pydantic import BaseModel, Field
 from sqlalchemy.orm import Session, sessionmaker
 from werkzeug.exceptions import BadRequest, InternalServerError, NotFound
@@ -30,6 +27,9 @@ from core.errors.error import (
     QuotaExceededError,
 )
 from core.helper.trace_id_helper import get_external_trace_id
+from dify_graph.enums import WorkflowExecutionStatus
+from dify_graph.graph_engine.manager import GraphEngineManager
+from dify_graph.model_runtime.errors.invoke import InvokeError
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from fields.workflow_app_log_fields import build_workflow_app_log_pagination_model

api/controllers/service_api/dataset/dataset.py

@@ -2,7 +2,6 @@ from typing import Any, Literal, cast
 
 from flask import request
 from flask_restx import marshal
-from graphon.model_runtime.entities.model_entities import ModelType
 from pydantic import BaseModel, Field, TypeAdapter, field_validator
 from werkzeug.exceptions import Forbidden, NotFound
 
@@ -15,8 +14,9 @@ from controllers.service_api.wraps import (
     DatasetApiResource,
     cloud_edition_billing_rate_limit_check,
 )
-from core.plugin.impl.model_runtime_factory import create_plugin_provider_manager
+from core.provider_manager import ProviderManager
 from core.rag.index_processor.constant.index_type import IndexTechniqueType
+from dify_graph.model_runtime.entities.model_entities import ModelType
 from fields.dataset_fields import dataset_detail_fields
 from fields.tag_fields import DataSetTag
 from libs.login import current_user
@@ -140,10 +140,10 @@ class DatasetListApi(DatasetApiResource):
             query.page, query.limit, tenant_id, current_user, query.keyword, query.tag_ids, query.include_all
         )
         # check embedding setting
+        provider_manager = ProviderManager()
         assert isinstance(current_user, Account)
         cid = current_user.current_tenant_id
         assert cid is not None
-        provider_manager = create_plugin_provider_manager(tenant_id=cid)
         configurations = provider_manager.get_configurations(tenant_id=cid)
 
         embedding_models = configurations.get_models(model_type=ModelType.TEXT_EMBEDDING, only_active=True)
@@ -259,10 +259,10 @@ class DatasetApi(DatasetApiResource):
             raise Forbidden(str(e))
         data = cast(dict[str, Any], marshal(dataset, dataset_detail_fields))
         # check embedding setting
+        provider_manager = ProviderManager()
         assert isinstance(current_user, Account)
         cid = current_user.current_tenant_id
         assert cid is not None
-        provider_manager = create_plugin_provider_manager(tenant_id=cid)
         configurations = provider_manager.get_configurations(tenant_id=cid)
 
         embedding_models = configurations.get_models(model_type=ModelType.TEXT_EMBEDDING, only_active=True)

api/controllers/service_api/dataset/document.py

@@ -6,7 +6,7 @@ from uuid import UUID
 from flask import request, send_file
 from flask_restx import marshal
 from pydantic import BaseModel, Field, field_validator, model_validator
-from sqlalchemy import desc, func, select
+from sqlalchemy import desc, select
 from werkzeug.exceptions import Forbidden, NotFound
 
 import services
@@ -155,9 +155,7 @@ class DocumentAddByTextApi(DatasetApiResource):
 
         dataset_id = str(dataset_id)
         tenant_id = str(tenant_id)
-        dataset = db.session.scalar(
-            select(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).limit(1)
-        )
+        dataset = db.session.query(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()
 
         if not dataset:
             raise ValueError("Dataset does not exist.")
@@ -240,9 +238,7 @@ class DocumentUpdateByTextApi(DatasetApiResource):
     def post(self, tenant_id: str, dataset_id: UUID, document_id: UUID):
         """Update document by text."""
         payload = DocumentTextUpdate.model_validate(service_api_ns.payload or {})
-        dataset = db.session.scalar(
-            select(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == str(dataset_id)).limit(1)
-        )
+        dataset = db.session.query(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == str(dataset_id)).first()
         args = payload.model_dump(exclude_none=True)
         if not dataset:
             raise ValueError("Dataset does not exist.")
@@ -319,9 +315,7 @@ class DocumentAddByFileApi(DatasetApiResource):
     @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
     def post(self, tenant_id, dataset_id):
         """Create document by upload file."""
-        dataset = db.session.scalar(
-            select(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).limit(1)
-        )
+        dataset = db.session.query(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()
 
         if not dataset:
             raise ValueError("Dataset does not exist.")
@@ -431,9 +425,7 @@ class DocumentUpdateByFileApi(DatasetApiResource):
     @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
     def post(self, tenant_id, dataset_id, document_id):
         """Update document by upload file."""
-        dataset = db.session.scalar(
-            select(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).limit(1)
-        )
+        dataset = db.session.query(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()
 
         if not dataset:
             raise ValueError("Dataset does not exist.")
@@ -523,9 +515,7 @@ class DocumentListApi(DatasetApiResource):
         dataset_id = str(dataset_id)
         tenant_id = str(tenant_id)
         query_params = DocumentListQuery.model_validate(request.args.to_dict())
-        dataset = db.session.scalar(
-            select(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).limit(1)
-        )
+        dataset = db.session.query(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()
         if not dataset:
             raise NotFound("Dataset not found.")
 
@@ -619,9 +609,7 @@ class DocumentIndexingStatusApi(DatasetApiResource):
         batch = str(batch)
         tenant_id = str(tenant_id)
         # get dataset
-        dataset = db.session.scalar(
-            select(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).limit(1)
-        )
+        dataset = db.session.query(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()
         if not dataset:
             raise NotFound("Dataset not found.")
         # get documents
@@ -631,23 +619,20 @@ class DocumentIndexingStatusApi(DatasetApiResource):
         documents_status = []
         for document in documents:
             completed_segments = (
-                db.session.scalar(
-                    select(func.count(DocumentSegment.id)).where(
-                        DocumentSegment.completed_at.isnot(None),
-                        DocumentSegment.document_id == str(document.id),
-                        DocumentSegment.status != SegmentStatus.RE_SEGMENT,
-                    )
+                db.session.query(DocumentSegment)
+                .where(
+                    DocumentSegment.completed_at.isnot(None),
+                    DocumentSegment.document_id == str(document.id),
+                    DocumentSegment.status != SegmentStatus.RE_SEGMENT,
                 )
-                or 0
+                .count()
             )
             total_segments = (
-                db.session.scalar(
-                    select(func.count(DocumentSegment.id)).where(
-                        DocumentSegment.document_id == str(document.id),
-                        DocumentSegment.status != SegmentStatus.RE_SEGMENT,
-                    )
+                db.session.query(DocumentSegment)
+                .where(
+                    DocumentSegment.document_id == str(document.id), DocumentSegment.status != SegmentStatus.RE_SEGMENT
                 )
-                or 0
+                .count()
             )
             # Create a dictionary with document attributes and additional fields
             document_dict = {
@@ -837,9 +822,7 @@ class DocumentApi(DatasetApiResource):
         tenant_id = str(tenant_id)
 
         # get dataset info
-        dataset = db.session.scalar(
-            select(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).limit(1)
-        )
+        dataset = db.session.query(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()
 
         if not dataset:
             raise ValueError("Dataset does not exist.")

api/controllers/service_api/dataset/segment.py

@@ -2,9 +2,7 @@ from typing import Any
 
 from flask import request
 from flask_restx import marshal
-from graphon.model_runtime.entities.model_entities import ModelType
 from pydantic import BaseModel, Field
-from sqlalchemy import select
 from werkzeug.exceptions import NotFound
 
 from configs import dify_config
@@ -20,6 +18,7 @@ from controllers.service_api.wraps import (
 from core.errors.error import LLMBadRequestError, ProviderTokenNotInitError
 from core.model_manager import ModelManager
 from core.rag.index_processor.constant.index_type import IndexTechniqueType
+from dify_graph.model_runtime.entities.model_entities import ModelType
 from extensions.ext_database import db
 from fields.segment_fields import child_chunk_fields, segment_fields
 from libs.login import current_account_with_tenant
@@ -93,9 +92,7 @@ class SegmentApi(DatasetApiResource):
         _, current_tenant_id = current_account_with_tenant()
         """Create single segment."""
         # check dataset
-        dataset = db.session.scalar(
-            select(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).limit(1)
-        )
+        dataset = db.session.query(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()
         if not dataset:
             raise NotFound("Dataset not found.")
         # check document
@@ -109,7 +106,7 @@ class SegmentApi(DatasetApiResource):
         # check embedding model setting
         if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
             try:
-                model_manager = ModelManager.for_tenant(tenant_id=current_tenant_id)
+                model_manager = ModelManager()
                 model_manager.get_model_instance(
                     tenant_id=current_tenant_id,
                     provider=dataset.embedding_model_provider,
@@ -153,9 +150,7 @@ class SegmentApi(DatasetApiResource):
         # check dataset
         page = request.args.get("page", default=1, type=int)
         limit = request.args.get("limit", default=20, type=int)
-        dataset = db.session.scalar(
-            select(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).limit(1)
-        )
+        dataset = db.session.query(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()
         if not dataset:
             raise NotFound("Dataset not found.")
         # check document
@@ -165,7 +160,7 @@ class SegmentApi(DatasetApiResource):
         # check embedding model setting
         if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
             try:
-                model_manager = ModelManager.for_tenant(tenant_id=current_tenant_id)
+                model_manager = ModelManager()
                 model_manager.get_model_instance(
                     tenant_id=current_tenant_id,
                     provider=dataset.embedding_model_provider,
@@ -225,9 +220,7 @@ class DatasetSegmentApi(DatasetApiResource):
     def delete(self, tenant_id: str, dataset_id: str, document_id: str, segment_id: str):
         _, current_tenant_id = current_account_with_tenant()
         # check dataset
-        dataset = db.session.scalar(
-            select(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).limit(1)
-        )
+        dataset = db.session.query(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()
         if not dataset:
             raise NotFound("Dataset not found.")
         # check user's model setting
@@ -261,9 +254,7 @@ class DatasetSegmentApi(DatasetApiResource):
     def post(self, tenant_id: str, dataset_id: str, document_id: str, segment_id: str):
         _, current_tenant_id = current_account_with_tenant()
         # check dataset
-        dataset = db.session.scalar(
-            select(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).limit(1)
-        )
+        dataset = db.session.query(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()
         if not dataset:
             raise NotFound("Dataset not found.")
         # check user's model setting
@@ -275,7 +266,7 @@ class DatasetSegmentApi(DatasetApiResource):
         if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
             # check embedding model setting
             try:
-                model_manager = ModelManager.for_tenant(tenant_id=current_tenant_id)
+                model_manager = ModelManager()
                 model_manager.get_model_instance(
                     tenant_id=current_tenant_id,
                     provider=dataset.embedding_model_provider,
@@ -310,9 +301,7 @@ class DatasetSegmentApi(DatasetApiResource):
     def get(self, tenant_id: str, dataset_id: str, document_id: str, segment_id: str):
         _, current_tenant_id = current_account_with_tenant()
         # check dataset
-        dataset = db.session.scalar(
-            select(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).limit(1)
-        )
+        dataset = db.session.query(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()
         if not dataset:
             raise NotFound("Dataset not found.")
         # check user's model setting
@@ -355,9 +344,7 @@ class ChildChunkApi(DatasetApiResource):
         _, current_tenant_id = current_account_with_tenant()
         """Create child chunk."""
         # check dataset
-        dataset = db.session.scalar(
-            select(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).limit(1)
-        )
+        dataset = db.session.query(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()
         if not dataset:
             raise NotFound("Dataset not found.")
 
@@ -374,7 +361,7 @@ class ChildChunkApi(DatasetApiResource):
         # check embedding model setting
         if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
             try:
-                model_manager = ModelManager.for_tenant(tenant_id=current_tenant_id)
+                model_manager = ModelManager()
                 model_manager.get_model_instance(
                     tenant_id=current_tenant_id,
                     provider=dataset.embedding_model_provider,
@@ -415,9 +402,7 @@ class ChildChunkApi(DatasetApiResource):
         _, current_tenant_id = current_account_with_tenant()
         """Get child chunks."""
         # check dataset
-        dataset = db.session.scalar(
-            select(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).limit(1)
-        )
+        dataset = db.session.query(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()
         if not dataset:
             raise NotFound("Dataset not found.")
 
@@ -483,9 +468,7 @@ class DatasetChildChunkApi(DatasetApiResource):
         _, current_tenant_id = current_account_with_tenant()
         """Delete child chunk."""
         # check dataset
-        dataset = db.session.scalar(
-            select(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).limit(1)
-        )
+        dataset = db.session.query(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()
         if not dataset:
             raise NotFound("Dataset not found.")
 
@@ -544,9 +527,7 @@ class DatasetChildChunkApi(DatasetApiResource):
         _, current_tenant_id = current_account_with_tenant()
         """Update child chunk."""
         # check dataset
-        dataset = db.session.scalar(
-            select(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).limit(1)
-        )
+        dataset = db.session.query(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()
         if not dataset:
             raise NotFound("Dataset not found.")
 

api/controllers/service_api/workspace/models.py

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

api/controllers/service_api/wraps.py

@@ -9,7 +9,6 @@ from flask import current_app, request
 from flask_login import user_logged_in
 from flask_restx import Resource
 from pydantic import BaseModel
-from sqlalchemy import select
 from werkzeug.exceptions import Forbidden, NotFound, Unauthorized
 
 from enums.cloud_plan import CloudPlan
@@ -63,7 +62,7 @@ def validate_app_token(
         def decorated_view(*args: P.args, **kwargs: P.kwargs) -> R:
             api_token = validate_and_get_api_token("app")
 
-            app_model = db.session.get(App, api_token.app_id)
+            app_model = db.session.query(App).where(App.id == api_token.app_id).first()
             if not app_model:
                 raise Forbidden("The app no longer exists.")
 
@@ -73,7 +72,7 @@ def validate_app_token(
             if not app_model.enable_api:
                 raise Forbidden("The app's API service has been disabled.")
 
-            tenant = db.session.get(Tenant, app_model.tenant_id)
+            tenant = db.session.query(Tenant).where(Tenant.id == app_model.tenant_id).first()
             if tenant is None:
                 raise ValueError("Tenant does not exist.")
             if tenant.status == TenantStatus.ARCHIVE:
@@ -107,8 +106,8 @@ def validate_app_token(
             else:
                 # For service API without end-user context, ensure an Account is logged in
                 # so services relying on current_account_with_tenant() work correctly.
-                tenant_owner_info = db.session.execute(
-                    select(Tenant, Account)
+                tenant_owner_info = (
+                    db.session.query(Tenant, Account)
                     .join(TenantAccountJoin, Tenant.id == TenantAccountJoin.tenant_id)
                     .join(Account, TenantAccountJoin.account_id == Account.id)
                     .where(
@@ -116,7 +115,8 @@ def validate_app_token(
                         TenantAccountJoin.role == "owner",
                         Tenant.status == TenantStatus.NORMAL,
                     )
-                ).one_or_none()
+                    .one_or_none()
+                )
 
                 if tenant_owner_info:
                     tenant_model, account = tenant_owner_info
@@ -277,28 +277,29 @@ def validate_dataset_token(
             # Validate dataset if dataset_id is provided
             if dataset_id:
                 dataset_id = str(dataset_id)
-                dataset = db.session.scalar(
-                    select(Dataset)
+                dataset = (
+                    db.session.query(Dataset)
                     .where(
                         Dataset.id == dataset_id,
                         Dataset.tenant_id == api_token.tenant_id,
                     )
-                    .limit(1)
+                    .first()
                 )
                 if not dataset:
                     raise NotFound("Dataset not found.")
                 if not dataset.enable_api:
                     raise Forbidden("Dataset api access is not enabled.")
-            tenant_account_join = db.session.execute(
-                select(Tenant, TenantAccountJoin)
+            tenant_account_join = (
+                db.session.query(Tenant, TenantAccountJoin)
                 .where(Tenant.id == api_token.tenant_id)
                 .where(TenantAccountJoin.tenant_id == Tenant.id)
                 .where(TenantAccountJoin.role.in_(["owner"]))
                 .where(Tenant.status == TenantStatus.NORMAL)
-            ).one_or_none()  # TODO: only owner information is required, so only one is returned.
+                .one_or_none()
+            )  # TODO: only owner information is required, so only one is returned.
             if tenant_account_join:
                 tenant, ta = tenant_account_join
-                account = db.session.get(Account, ta.account_id)
+                account = db.session.query(Account).where(Account.id == ta.account_id).first()
                 # Login admin
                 if account:
                     account.current_tenant = tenant
@@ -359,9 +360,7 @@ class DatasetApiResource(Resource):
     method_decorators = [validate_dataset_token]
 
     def get_dataset(self, dataset_id: str, tenant_id: str) -> Dataset:
-        dataset = db.session.scalar(
-            select(Dataset).where(Dataset.id == dataset_id, Dataset.tenant_id == tenant_id).limit(1)
-        )
+        dataset = db.session.query(Dataset).where(Dataset.id == dataset_id, Dataset.tenant_id == tenant_id).first()
 
         if not dataset:
             raise NotFound("Dataset not found.")

api/controllers/web/audio.py

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

api/controllers/web/completion.py

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

api/controllers/web/message.py

@@ -2,7 +2,6 @@ import logging
 from typing import Literal
 
 from flask import request
-from graphon.model_runtime.errors.invoke import InvokeError
 from pydantic import BaseModel, Field, TypeAdapter, field_validator
 from werkzeug.exceptions import InternalServerError, NotFound
 
@@ -21,6 +20,7 @@ from controllers.web.error import (
 from controllers.web.wraps import WebApiResource
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
+from dify_graph.model_runtime.errors.invoke import InvokeError
 from fields.conversation_fields import ResultResponse
 from fields.message_fields import SuggestedQuestionsResponse, WebMessageInfiniteScrollPagination, WebMessageListItem
 from libs import helper

api/controllers/web/remote_files.py

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

api/controllers/web/workflow.py

@@ -1,8 +1,6 @@
 import logging
 from typing import Any
 
-from graphon.graph_engine.manager import GraphEngineManager
-from graphon.model_runtime.errors.invoke import InvokeError
 from pydantic import BaseModel, Field
 from werkzeug.exceptions import InternalServerError
 
@@ -24,6 +22,8 @@ from core.errors.error import (
     ProviderTokenNotInitError,
     QuotaExceededError,
 )
+from dify_graph.graph_engine.manager import GraphEngineManager
+from dify_graph.model_runtime.errors.invoke import InvokeError
 from extensions.ext_redis import redis_client
 from libs import helper
 from models.model import App, AppMode, EndUser

api/core/agent/base_agent_runner.py

@@ -4,20 +4,6 @@ import uuid
 from decimal import Decimal
 from typing import Union, cast
 
-from graphon.file import file_manager
-from graphon.model_runtime.entities import (
-    AssistantPromptMessage,
-    LLMUsage,
-    PromptMessage,
-    PromptMessageTool,
-    SystemPromptMessage,
-    TextPromptMessageContent,
-    ToolPromptMessage,
-    UserPromptMessage,
-)
-from graphon.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
-from graphon.model_runtime.entities.model_entities import ModelFeature
-from graphon.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
 from sqlalchemy import select
 
 from core.agent.entities import AgentEntity, AgentToolEntity
@@ -29,7 +15,6 @@ from core.app.entities.app_invoke_entities import (
     AgentChatAppGenerateEntity,
     ModelConfigWithCredentialsEntity,
 )
-from core.app.file_access import DatabaseFileAccessController
 from core.callback_handler.agent_tool_callback_handler import DifyAgentCallbackHandler
 from core.callback_handler.index_tool_callback_handler import DatasetIndexToolCallbackHandler
 from core.memory.token_buffer_memory import TokenBufferMemory
@@ -41,13 +26,26 @@ from core.tools.entities.tool_entities import (
 )
 from core.tools.tool_manager import ToolManager
 from core.tools.utils.dataset_retriever_tool import DatasetRetrieverTool
+from dify_graph.file import file_manager
+from dify_graph.model_runtime.entities import (
+    AssistantPromptMessage,
+    LLMUsage,
+    PromptMessage,
+    PromptMessageTool,
+    SystemPromptMessage,
+    TextPromptMessageContent,
+    ToolPromptMessage,
+    UserPromptMessage,
+)
+from dify_graph.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
+from dify_graph.model_runtime.entities.model_entities import ModelFeature
+from dify_graph.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
 from extensions.ext_database import db
 from factories import file_factory
 from models.enums import CreatorUserRole
 from models.model import Conversation, Message, MessageAgentThought, MessageFile
 
 logger = logging.getLogger(__name__)
-_file_access_controller = DatabaseFileAccessController()
 
 
 class BaseAgentRunner(AppRunner):
@@ -140,7 +138,6 @@ class BaseAgentRunner(AppRunner):
             tenant_id=self.tenant_id,
             app_id=self.app_config.app_id,
             agent_tool=tool,
-            user_id=self.user_id,
             invoke_from=self.application_generate_entity.invoke_from,
         )
         assert tool_entity.entity.description
@@ -527,10 +524,7 @@ class BaseAgentRunner(AppRunner):
         image_detail_config = image_detail_config or ImagePromptMessageContent.DETAIL.LOW
 
         file_objs = file_factory.build_from_message_files(
-            message_files=files,
-            tenant_id=self.tenant_id,
-            config=file_extra_config,
-            access_controller=_file_access_controller,
+            message_files=files, tenant_id=self.tenant_id, config=file_extra_config
         )
         if not file_objs:
             return UserPromptMessage(content=message.query)

api/core/agent/cot_agent_runner.py

@@ -4,15 +4,6 @@ from abc import ABC, abstractmethod
 from collections.abc import Generator, Mapping, Sequence
 from typing import Any
 
-from graphon.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk, LLMResultChunkDelta, LLMUsage
-from graphon.model_runtime.entities.message_entities import (
-    AssistantPromptMessage,
-    PromptMessage,
-    PromptMessageTool,
-    ToolPromptMessage,
-    UserPromptMessage,
-)
-
 from core.agent.base_agent_runner import BaseAgentRunner
 from core.agent.entities import AgentScratchpadUnit
 from core.agent.errors import AgentMaxIterationError
@@ -24,6 +15,14 @@ from core.prompt.agent_history_prompt_transform import AgentHistoryPromptTransfo
 from core.tools.__base.tool import Tool
 from core.tools.entities.tool_entities import ToolInvokeMeta
 from core.tools.tool_engine import ToolEngine
+from dify_graph.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk, LLMResultChunkDelta, LLMUsage
+from dify_graph.model_runtime.entities.message_entities import (
+    AssistantPromptMessage,
+    PromptMessage,
+    PromptMessageTool,
+    ToolPromptMessage,
+    UserPromptMessage,
+)
 from models.model import Message
 
 logger = logging.getLogger(__name__)
@@ -123,6 +122,7 @@ class CotAgentRunner(BaseAgentRunner, ABC):
                 tools=[],
                 stop=app_generate_entity.model_conf.stop,
                 stream=True,
+                user=self.user_id,
                 callbacks=[],
             )
 

api/core/agent/cot_chat_agent_runner.py

@@ -1,17 +1,16 @@
 import json
 
-from graphon.file import file_manager
-from graphon.model_runtime.entities import (
+from core.agent.cot_agent_runner import CotAgentRunner
+from dify_graph.file import file_manager
+from dify_graph.model_runtime.entities import (
     AssistantPromptMessage,
     PromptMessage,
     SystemPromptMessage,
     TextPromptMessageContent,
     UserPromptMessage,
 )
-from graphon.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
-from graphon.model_runtime.utils.encoders import jsonable_encoder
-
-from core.agent.cot_agent_runner import CotAgentRunner
+from dify_graph.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
+from dify_graph.model_runtime.utils.encoders import jsonable_encoder
 
 
 class CotChatAgentRunner(CotAgentRunner):

api/core/agent/cot_completion_agent_runner.py

@@ -1,14 +1,13 @@
 import json
 
-from graphon.model_runtime.entities.message_entities import (
+from core.agent.cot_agent_runner import CotAgentRunner
+from dify_graph.model_runtime.entities.message_entities import (
     AssistantPromptMessage,
     PromptMessage,
     TextPromptMessageContent,
     UserPromptMessage,
 )
-from graphon.model_runtime.utils.encoders import jsonable_encoder
-
-from core.agent.cot_agent_runner import CotAgentRunner
+from dify_graph.model_runtime.utils.encoders import jsonable_encoder
 
 
 class CotCompletionAgentRunner(CotAgentRunner):

api/core/agent/fc_agent_runner.py

@@ -4,8 +4,15 @@ from collections.abc import Generator
 from copy import deepcopy
 from typing import Any, Union
 
-from graphon.file import file_manager
-from graphon.model_runtime.entities import (
+from core.agent.base_agent_runner import BaseAgentRunner
+from core.agent.errors import AgentMaxIterationError
+from core.app.apps.base_app_queue_manager import PublishFrom
+from core.app.entities.queue_entities import QueueAgentThoughtEvent, QueueMessageEndEvent, QueueMessageFileEvent
+from core.prompt.agent_history_prompt_transform import AgentHistoryPromptTransform
+from core.tools.entities.tool_entities import ToolInvokeMeta
+from core.tools.tool_engine import ToolEngine
+from dify_graph.file import file_manager
+from dify_graph.model_runtime.entities import (
     AssistantPromptMessage,
     LLMResult,
     LLMResultChunk,
@@ -18,15 +25,7 @@ from graphon.model_runtime.entities import (
     ToolPromptMessage,
     UserPromptMessage,
 )
-from graphon.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
-
-from core.agent.base_agent_runner import BaseAgentRunner
-from core.agent.errors import AgentMaxIterationError
-from core.app.apps.base_app_queue_manager import PublishFrom
-from core.app.entities.queue_entities import QueueAgentThoughtEvent, QueueMessageEndEvent, QueueMessageFileEvent
-from core.prompt.agent_history_prompt_transform import AgentHistoryPromptTransform
-from core.tools.entities.tool_entities import ToolInvokeMeta
-from core.tools.tool_engine import ToolEngine
+from dify_graph.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
 from models.model import Message
 
 logger = logging.getLogger(__name__)
@@ -97,6 +96,7 @@ class FunctionCallAgentRunner(BaseAgentRunner):
                 tools=prompt_messages_tools,
                 stop=app_generate_entity.model_conf.stop,
                 stream=self.stream_tool_call,
+                user=self.user_id,
                 callbacks=[],
             )
 

api/core/agent/output_parser/cot_output_parser.py

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

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

@@ -1,14 +1,13 @@
 from typing import cast
 
-from graphon.model_runtime.entities.llm_entities import LLMMode
-from graphon.model_runtime.entities.model_entities import ModelPropertyKey, ModelType
-from graphon.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
-
 from core.app.app_config.entities import EasyUIBasedAppConfig
 from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
 from core.entities.model_entities import ModelStatus
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
-from core.plugin.impl.model_runtime_factory import create_plugin_provider_manager
+from core.provider_manager import ProviderManager
+from dify_graph.model_runtime.entities.llm_entities import LLMMode
+from dify_graph.model_runtime.entities.model_entities import ModelPropertyKey, ModelType
+from dify_graph.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
 
 
 class ModelConfigConverter:
@@ -22,7 +21,7 @@ class ModelConfigConverter:
         """
         model_config = app_config.model
 
-        provider_manager = create_plugin_provider_manager(tenant_id=app_config.tenant_id)
+        provider_manager = ProviderManager()
         provider_model_bundle = provider_manager.get_provider_model_bundle(
             tenant_id=app_config.tenant_id, provider=model_config.provider, model_type=ModelType.LLM
         )

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

@@ -1,10 +1,10 @@
 from collections.abc import Mapping
 from typing import Any
 
-from graphon.model_runtime.entities.model_entities import ModelPropertyKey, ModelType
-
 from core.app.app_config.entities import ModelConfigEntity
-from core.plugin.impl.model_runtime_factory import create_plugin_model_assembly
+from core.provider_manager import ProviderManager
+from dify_graph.model_runtime.entities.model_entities import ModelPropertyKey, ModelType
+from dify_graph.model_runtime.model_providers.model_provider_factory import ModelProviderFactory
 from models.model import AppModelConfigDict
 from models.provider_ids import ModelProviderID
 
@@ -54,12 +54,9 @@ class ModelConfigManager:
         if not isinstance(config["model"], dict):
             raise ValueError("model must be of object type")
 
-        # Keep provider discovery and provider-backed model listing on the same
-        # request-scoped runtime so caller scope and provider caches stay aligned.
-        assembly = create_plugin_model_assembly(tenant_id=tenant_id)
-
         # model.provider
-        provider_entities = assembly.model_provider_factory.get_providers()
+        model_provider_factory = ModelProviderFactory(tenant_id)
+        provider_entities = model_provider_factory.get_providers()
         model_provider_names = [provider.provider for provider in provider_entities]
         if "provider" not in config["model"]:
             raise ValueError(f"model.provider is required and must be in {str(model_provider_names)}")
@@ -74,7 +71,8 @@ class ModelConfigManager:
         if "name" not in config["model"]:
             raise ValueError("model.name is required")
 
-        models = assembly.provider_manager.get_configurations(tenant_id).get_models(
+        provider_manager = ProviderManager()
+        models = provider_manager.get_configurations(tenant_id).get_models(
             provider=config["model"]["provider"], model_type=ModelType.LLM
         )
 

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

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

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

@@ -1,10 +1,9 @@
 import re
 from typing import cast
 
-from graphon.variables.input_entities import VariableEntity, VariableEntityType
-
 from core.app.app_config.entities import ExternalDataVariableEntity
 from core.external_data_tool.factory import ExternalDataToolFactory
+from dify_graph.variables.input_entities import VariableEntity, VariableEntityType
 from models.model import AppModelConfigDict
 
 _ALLOWED_VARIABLE_ENTITY_TYPE = frozenset(

api/core/app/app_config/entities.py

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

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

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

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

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

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

@@ -5,7 +5,7 @@ import logging
 import threading
 import uuid
 from collections.abc import Generator, Mapping, Sequence
-from typing import TYPE_CHECKING, Any, Literal, Union, overload
+from typing import TYPE_CHECKING, Any, Literal, TypeVar, Union, overload
 
 from flask import Flask, current_app
 from pydantic import ValidationError
@@ -18,23 +18,12 @@ from constants import UUID_NIL
 
 if TYPE_CHECKING:
     from controllers.console.app.workflow import LoopNodeRunPayload
-from graphon.graph_engine.layers import GraphEngineLayer
-from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
-from graphon.runtime import GraphRuntimeState
-from graphon.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader
-
 from core.app.app_config.features.file_upload.manager import FileUploadConfigManager
 from core.app.apps.advanced_chat.app_config_manager import AdvancedChatAppConfigManager
 from core.app.apps.advanced_chat.app_runner import AdvancedChatAppRunner
 from core.app.apps.advanced_chat.generate_response_converter import AdvancedChatAppGenerateResponseConverter
-from core.app.apps.advanced_chat.generate_task_pipeline import (
-    AdvancedChatAppGenerateTaskPipeline,
-    ConversationSnapshot,
-    MessageSnapshot,
-    WorkflowSnapshot,
-)
+from core.app.apps.advanced_chat.generate_task_pipeline import AdvancedChatAppGenerateTaskPipeline
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
-from core.app.apps.draft_variable_saver import DraftVariableSaverFactory
 from core.app.apps.exc import GenerateTaskStoppedError
 from core.app.apps.message_based_app_generator import MessageBasedAppGenerator
 from core.app.apps.message_based_app_queue_manager import MessageBasedAppQueueManager
@@ -45,11 +34,20 @@ from core.helper.trace_id_helper import extract_external_trace_id_from_args
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.prompt.utils.get_thread_messages_length import get_thread_messages_length
 from core.repositories import DifyCoreRepositoryFactory
-from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
+from dify_graph.graph_engine.layers.base import GraphEngineLayer
+from dify_graph.model_runtime.errors.invoke import InvokeAuthorizationError
+from dify_graph.repositories.draft_variable_repository import (
+    DraftVariableSaverFactory,
+)
+from dify_graph.repositories.workflow_execution_repository import WorkflowExecutionRepository
+from dify_graph.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
+from dify_graph.runtime import GraphRuntimeState
+from dify_graph.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader
 from extensions.ext_database import db
 from factories import file_factory
 from libs.flask_utils import preserve_flask_contexts
 from models import Account, App, Conversation, EndUser, Message, Workflow, WorkflowNodeExecutionTriggeredFrom
+from models.base import Base
 from models.enums import WorkflowRunTriggeredFrom
 from services.conversation_service import ConversationService
 from services.workflow_draft_variable_service import (
@@ -152,87 +150,85 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         #
         # For implementation reference, see the `_parse_file` function and
         # `DraftWorkflowNodeRunApi` class which handle this properly.
-        with self._bind_file_access_scope(tenant_id=app_model.tenant_id, user=user, invoke_from=invoke_from):
-            files = args["files"] if args.get("files") else []
-            file_extra_config = FileUploadConfigManager.convert(workflow.features_dict, is_vision=False)
-            if file_extra_config:
-                file_objs = file_factory.build_from_mappings(
-                    mappings=files,
-                    tenant_id=app_model.tenant_id,
-                    config=file_extra_config,
-                    access_controller=self._file_access_controller,
-                )
-            else:
-                file_objs = []
-
-            # convert to app config
-            app_config = AdvancedChatAppConfigManager.get_app_config(app_model=app_model, workflow=workflow)
-
-            # get tracing instance
-            trace_manager = TraceQueueManager(
-                app_id=app_model.id, user_id=user.id if isinstance(user, Account) else user.session_id
+        files = args["files"] if args.get("files") else []
+        file_extra_config = FileUploadConfigManager.convert(workflow.features_dict, is_vision=False)
+        if file_extra_config:
+            file_objs = file_factory.build_from_mappings(
+                mappings=files,
+                tenant_id=app_model.tenant_id,
+                config=file_extra_config,
             )
+        else:
+            file_objs = []
 
-            if invoke_from == InvokeFrom.DEBUGGER:
-                # always enable retriever resource in debugger mode
-                app_config.additional_features.show_retrieve_source = True  # type: ignore
+        # convert to app config
+        app_config = AdvancedChatAppConfigManager.get_app_config(app_model=app_model, workflow=workflow)
 
-            # init application generate entity
-            application_generate_entity = AdvancedChatAppGenerateEntity(
-                task_id=str(uuid.uuid4()),
-                app_config=app_config,
-                file_upload_config=file_extra_config,
-                conversation_id=conversation.id if conversation else None,
-                inputs=self._prepare_user_inputs(
-                    user_inputs=inputs, variables=app_config.variables, tenant_id=app_model.tenant_id
-                ),
-                query=query,
-                files=list(file_objs),
-                parent_message_id=args.get("parent_message_id") if invoke_from != InvokeFrom.SERVICE_API else UUID_NIL,
-                user_id=user.id,
-                stream=streaming,
-                invoke_from=invoke_from,
-                extras=extras,
-                trace_manager=trace_manager,
-                workflow_run_id=str(workflow_run_id),
-            )
-            contexts.plugin_tool_providers.set({})
-            contexts.plugin_tool_providers_lock.set(threading.Lock())
+        # get tracing instance
+        trace_manager = TraceQueueManager(
+            app_id=app_model.id, user_id=user.id if isinstance(user, Account) else user.session_id
+        )
 
-            # Create repositories
-            #
-            # Create session factory
-            session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
-            # Create workflow execution(aka workflow run) repository
-            if invoke_from == InvokeFrom.DEBUGGER:
-                workflow_triggered_from = WorkflowRunTriggeredFrom.DEBUGGING
-            else:
-                workflow_triggered_from = WorkflowRunTriggeredFrom.APP_RUN
-            workflow_execution_repository = DifyCoreRepositoryFactory.create_workflow_execution_repository(
-                session_factory=session_factory,
-                user=user,
-                app_id=application_generate_entity.app_config.app_id,
-                triggered_from=workflow_triggered_from,
-            )
-            # Create workflow node execution repository
-            workflow_node_execution_repository = DifyCoreRepositoryFactory.create_workflow_node_execution_repository(
-                session_factory=session_factory,
-                user=user,
-                app_id=application_generate_entity.app_config.app_id,
-                triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
-            )
+        if invoke_from == InvokeFrom.DEBUGGER:
+            # always enable retriever resource in debugger mode
+            app_config.additional_features.show_retrieve_source = True  # type: ignore
 
-            return self._generate(
-                workflow=workflow,
-                user=user,
-                invoke_from=invoke_from,
-                application_generate_entity=application_generate_entity,
-                workflow_execution_repository=workflow_execution_repository,
-                workflow_node_execution_repository=workflow_node_execution_repository,
-                conversation=conversation,
-                stream=streaming,
-                pause_state_config=pause_state_config,
-            )
+        # init application generate entity
+        application_generate_entity = AdvancedChatAppGenerateEntity(
+            task_id=str(uuid.uuid4()),
+            app_config=app_config,
+            file_upload_config=file_extra_config,
+            conversation_id=conversation.id if conversation else None,
+            inputs=self._prepare_user_inputs(
+                user_inputs=inputs, variables=app_config.variables, tenant_id=app_model.tenant_id
+            ),
+            query=query,
+            files=list(file_objs),
+            parent_message_id=args.get("parent_message_id") if invoke_from != InvokeFrom.SERVICE_API else UUID_NIL,
+            user_id=user.id,
+            stream=streaming,
+            invoke_from=invoke_from,
+            extras=extras,
+            trace_manager=trace_manager,
+            workflow_run_id=str(workflow_run_id),
+        )
+        contexts.plugin_tool_providers.set({})
+        contexts.plugin_tool_providers_lock.set(threading.Lock())
+
+        # Create repositories
+        #
+        # Create session factory
+        session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
+        # Create workflow execution(aka workflow run) repository
+        if invoke_from == InvokeFrom.DEBUGGER:
+            workflow_triggered_from = WorkflowRunTriggeredFrom.DEBUGGING
+        else:
+            workflow_triggered_from = WorkflowRunTriggeredFrom.APP_RUN
+        workflow_execution_repository = DifyCoreRepositoryFactory.create_workflow_execution_repository(
+            session_factory=session_factory,
+            user=user,
+            app_id=application_generate_entity.app_config.app_id,
+            triggered_from=workflow_triggered_from,
+        )
+        # Create workflow node execution repository
+        workflow_node_execution_repository = DifyCoreRepositoryFactory.create_workflow_node_execution_repository(
+            session_factory=session_factory,
+            user=user,
+            app_id=application_generate_entity.app_config.app_id,
+            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
+        )
+
+        return self._generate(
+            workflow=workflow,
+            user=user,
+            invoke_from=invoke_from,
+            application_generate_entity=application_generate_entity,
+            workflow_execution_repository=workflow_execution_repository,
+            workflow_node_execution_repository=workflow_node_execution_repository,
+            conversation=conversation,
+            stream=streaming,
+            pause_state_config=pause_state_config,
+        )
 
     def resume(
         self,
@@ -464,91 +460,94 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         :param conversation: conversation
         :param stream: is stream
         """
-        with self._bind_file_access_scope(
-            tenant_id=application_generate_entity.app_config.tenant_id,
-            user=user,
-            invoke_from=invoke_from,
-        ):
-            is_first_conversation = conversation is None
+        is_first_conversation = conversation is None
 
-            if conversation is not None and message is not None:
-                pass
-            else:
-                conversation, message = self._init_generate_records(application_generate_entity, conversation)
+        if conversation is not None and message is not None:
+            pass
+        else:
+            conversation, message = self._init_generate_records(application_generate_entity, conversation)
 
-            if is_first_conversation:
-                # update conversation features
-                conversation.override_model_configs = workflow.features
-                db.session.commit()
-                db.session.refresh(conversation)
+        if is_first_conversation:
+            # update conversation features
+            conversation.override_model_configs = workflow.features
+            db.session.commit()
+            db.session.refresh(conversation)
 
-            # get conversation dialogue count
-            # NOTE: dialogue_count should not start from 0,
-            # because during the first conversation, dialogue_count should be 1.
-            self._dialogue_count = get_thread_messages_length(conversation.id) + 1
+        # get conversation dialogue count
+        # NOTE: dialogue_count should not start from 0,
+        # because during the first conversation, dialogue_count should be 1.
+        self._dialogue_count = get_thread_messages_length(conversation.id) + 1
 
-            # init queue manager
-            queue_manager = MessageBasedAppQueueManager(
-                task_id=application_generate_entity.task_id,
-                user_id=application_generate_entity.user_id,
-                invoke_from=application_generate_entity.invoke_from,
-                conversation_id=conversation.id,
-                app_mode=conversation.mode,
-                message_id=message.id,
-            )
+        # init queue manager
+        queue_manager = MessageBasedAppQueueManager(
+            task_id=application_generate_entity.task_id,
+            user_id=application_generate_entity.user_id,
+            invoke_from=application_generate_entity.invoke_from,
+            conversation_id=conversation.id,
+            app_mode=conversation.mode,
+            message_id=message.id,
+        )
 
-            graph_layers: list[GraphEngineLayer] = list(graph_engine_layers)
-            if pause_state_config is not None:
-                graph_layers.append(
-                    PauseStatePersistenceLayer(
-                        session_factory=pause_state_config.session_factory,
-                        generate_entity=application_generate_entity,
-                        state_owner_user_id=pause_state_config.state_owner_user_id,
-                    )
+        graph_layers: list[GraphEngineLayer] = list(graph_engine_layers)
+        if pause_state_config is not None:
+            graph_layers.append(
+                PauseStatePersistenceLayer(
+                    session_factory=pause_state_config.session_factory,
+                    generate_entity=application_generate_entity,
+                    state_owner_user_id=pause_state_config.state_owner_user_id,
                 )
-
-            # new thread with request context and contextvars
-            context = contextvars.copy_context()
-
-            worker_thread = threading.Thread(
-                target=self._generate_worker,
-                kwargs={
-                    "flask_app": current_app._get_current_object(),  # type: ignore
-                    "application_generate_entity": application_generate_entity,
-                    "queue_manager": queue_manager,
-                    "conversation_id": conversation.id,
-                    "message_id": message.id,
-                    "context": context,
-                    "variable_loader": variable_loader,
-                    "workflow_execution_repository": workflow_execution_repository,
-                    "workflow_node_execution_repository": workflow_node_execution_repository,
-                    "graph_engine_layers": tuple(graph_layers),
-                    "graph_runtime_state": graph_runtime_state,
-                },
             )
 
-            worker_thread.start()
+        # new thread with request context and contextvars
+        context = contextvars.copy_context()
 
-            # Capture the scalar fields needed by the response pipeline before
-            # releasing the request-scoped SQLAlchemy session.
-            workflow_snapshot = WorkflowSnapshot.from_workflow(workflow)
-            conversation_snapshot = ConversationSnapshot.from_conversation(conversation)
-            message_snapshot = MessageSnapshot.from_message(message)
-            db.session.close()
+        worker_thread = threading.Thread(
+            target=self._generate_worker,
+            kwargs={
+                "flask_app": current_app._get_current_object(),  # type: ignore
+                "application_generate_entity": application_generate_entity,
+                "queue_manager": queue_manager,
+                "conversation_id": conversation.id,
+                "message_id": message.id,
+                "context": context,
+                "variable_loader": variable_loader,
+                "workflow_execution_repository": workflow_execution_repository,
+                "workflow_node_execution_repository": workflow_node_execution_repository,
+                "graph_engine_layers": tuple(graph_layers),
+                "graph_runtime_state": graph_runtime_state,
+            },
+        )
 
-            # return response or stream generator
-            response = self._handle_advanced_chat_response(
-                application_generate_entity=application_generate_entity,
-                workflow=workflow_snapshot,
-                queue_manager=queue_manager,
-                conversation=conversation_snapshot,
-                message=message_snapshot,
-                user=user,
-                stream=stream,
-                draft_var_saver_factory=self._get_draft_var_saver_factory(invoke_from, account=user),
-            )
+        worker_thread.start()
 
-            return AdvancedChatAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
+        # release database connection, because the following new thread operations may take a long time
+        with Session(bind=db.engine, expire_on_commit=False) as session:
+            workflow = _refresh_model(session, workflow)
+            message = _refresh_model(session, message)
+        #     workflow_ = session.get(Workflow, workflow.id)
+        #     assert workflow_ is not None
+        #     workflow = workflow_
+        #     message_ = session.get(Message, message.id)
+        #     assert message_ is not None
+        #     message = message_
+        # db.session.refresh(workflow)
+        # db.session.refresh(message)
+        # db.session.refresh(user)
+        db.session.close()
+
+        # return response or stream generator
+        response = self._handle_advanced_chat_response(
+            application_generate_entity=application_generate_entity,
+            workflow=workflow,
+            queue_manager=queue_manager,
+            conversation=conversation,
+            message=message,
+            user=user,
+            stream=stream,
+            draft_var_saver_factory=self._get_draft_var_saver_factory(invoke_from, account=user),
+        )
+
+        return AdvancedChatAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
 
     def _generate_worker(
         self,
@@ -649,10 +648,10 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         self,
         *,
         application_generate_entity: AdvancedChatAppGenerateEntity,
-        workflow: WorkflowSnapshot,
+        workflow: Workflow,
         queue_manager: AppQueueManager,
-        conversation: ConversationSnapshot,
-        message: MessageSnapshot,
+        conversation: Conversation,
+        message: Message,
         user: Union[Account, EndUser],
         draft_var_saver_factory: DraftVariableSaverFactory,
         stream: bool = False,
@@ -689,3 +688,13 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
             else:
                 logger.exception("Failed to process generate task pipeline, conversation_id: %s", conversation.id)
                 raise e
+
+
+_T = TypeVar("_T", bound=Base)
+
+
+def _refresh_model(session, model: _T) -> _T:
+    with Session(bind=db.engine, expire_on_commit=False) as session:
+        detach_model = session.get(type(model), model.id)
+        assert detach_model is not None
+        return detach_model

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

@@ -3,12 +3,6 @@ import time
 from collections.abc import Mapping, Sequence
 from typing import Any, cast
 
-from graphon.enums import WorkflowType
-from graphon.graph_engine.command_channels import RedisChannel
-from graphon.graph_engine.layers import GraphEngineLayer
-from graphon.runtime import GraphRuntimeState, VariablePool
-from graphon.variable_loader import VariableLoader
-from graphon.variables.variables import Variable
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
@@ -31,15 +25,16 @@ from core.app.workflow.layers.persistence import PersistenceWorkflowInfo, Workfl
 from core.db.session_factory import session_factory
 from core.moderation.base import ModerationError
 from core.moderation.input_moderation import InputModeration
-from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
-from core.workflow.node_factory import get_default_root_node_id
-from core.workflow.system_variables import (
-    build_bootstrap_variables,
-    build_system_variables,
-    system_variables_to_mapping,
-)
-from core.workflow.variable_pool_initializer import add_node_inputs_to_pool, add_variables_to_pool
 from core.workflow.workflow_entry import WorkflowEntry
+from dify_graph.enums import WorkflowType
+from dify_graph.graph_engine.command_channels.redis_channel import RedisChannel
+from dify_graph.graph_engine.layers.base import GraphEngineLayer
+from dify_graph.repositories.workflow_execution_repository import WorkflowExecutionRepository
+from dify_graph.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
+from dify_graph.runtime import GraphRuntimeState, VariablePool
+from dify_graph.system_variable import SystemVariable
+from dify_graph.variable_loader import VariableLoader
+from dify_graph.variables.variables import Variable
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from extensions.otel import WorkflowAppRunnerHandler, trace_span
@@ -95,7 +90,7 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
         app_config = self.application_generate_entity.app_config
         app_config = cast(AdvancedChatAppConfig, app_config)
 
-        system_inputs = build_system_variables(
+        system_inputs = SystemVariable(
             query=self.application_generate_entity.query,
             files=self.application_generate_entity.files,
             conversation_id=self.conversation.id,
@@ -137,7 +132,6 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
                 workflow=self._workflow,
                 single_iteration_run=self.application_generate_entity.single_iteration_run,
                 single_loop_run=self.application_generate_entity.single_loop_run,
-                user_id=self.application_generate_entity.user_id,
             )
         else:
             inputs = self.application_generate_entity.inputs
@@ -156,10 +150,7 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
 
             self.application_generate_entity.inputs = new_inputs
             self.application_generate_entity.query = new_query
-            system_inputs = build_system_variables(
-                system_variables_to_mapping(system_inputs),
-                query=new_query,
-            )
+            system_inputs.query = new_query
 
             # annotation reply
             if self.handle_annotation_reply(
@@ -175,17 +166,14 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
 
             # Create a variable pool.
             # init variable pool
-            variable_pool = VariablePool()
-            add_variables_to_pool(
-                variable_pool,
-                build_bootstrap_variables(
-                    system_variables=system_inputs,
-                    environment_variables=self._workflow.environment_variables,
-                    conversation_variables=conversation_variables,
-                ),
+            variable_pool = VariablePool(
+                system_variables=system_inputs,
+                user_inputs=new_inputs,
+                environment_variables=self._workflow.environment_variables,
+                # Based on the definition of `Variable`,
+                # `VariableBase` instances can be safely used as `Variable` since they are compatible.
+                conversation_variables=conversation_variables,
             )
-            root_node_id = get_default_root_node_id(self._workflow.graph_dict)
-            add_node_inputs_to_pool(variable_pool, node_id=root_node_id, inputs=new_inputs)
 
             # init graph
             graph_runtime_state = GraphRuntimeState(variable_pool=variable_pool, start_at=time.time())
@@ -197,7 +185,6 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
                 user_id=self.application_generate_entity.user_id,
                 user_from=user_from,
                 invoke_from=invoke_from,
-                root_node_id=root_node_id,
             )
 
         db.session.close()

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

@@ -4,17 +4,9 @@ import re
 import time
 from collections.abc import Callable, Generator, Mapping
 from contextlib import contextmanager
-from dataclasses import dataclass
-from datetime import datetime
 from threading import Thread
 from typing import Any, Union
 
-from graphon.entities.pause_reason import HumanInputRequired
-from graphon.enums import WorkflowExecutionStatus
-from graphon.model_runtime.entities.llm_entities import LLMUsage
-from graphon.model_runtime.utils.encoders import jsonable_encoder
-from graphon.nodes import BuiltinNodeTypes
-from graphon.runtime import GraphRuntimeState
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
@@ -22,7 +14,6 @@ from constants.tts_auto_play_timeout import TTS_AUTO_PLAY_TIMEOUT, TTS_AUTO_PLAY
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
 from core.app.apps.common.graph_runtime_state_support import GraphRuntimeStateSupport
 from core.app.apps.common.workflow_response_converter import WorkflowResponseConverter
-from core.app.apps.draft_variable_saver import DraftVariableSaverFactory
 from core.app.entities.app_invoke_entities import (
     AdvancedChatAppGenerateEntity,
     InvokeFrom,
@@ -74,66 +65,24 @@ from core.app.task_pipeline.message_cycle_manager import MessageCycleManager
 from core.base.tts import AppGeneratorTTSPublisher, AudioTrunk
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.repositories.human_input_repository import HumanInputFormRepositoryImpl
-from core.workflow.file_reference import resolve_file_record_id
-from core.workflow.system_variables import build_system_variables
+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.repositories.draft_variable_repository import DraftVariableSaverFactory
+from dify_graph.runtime import GraphRuntimeState
+from dify_graph.system_variable import SystemVariable
 from extensions.ext_database import db
 from libs.datetime_utils import naive_utc_now
 from models import Account, Conversation, EndUser, Message, MessageFile
 from models.enums import CreatorUserRole, MessageFileBelongsTo, MessageStatus
 from models.execution_extra_content import HumanInputContent
-from models.model import AppMode
 from models.workflow import Workflow
 
 logger = logging.getLogger(__name__)
 
 
-@dataclass(frozen=True, slots=True)
-class WorkflowSnapshot:
-    id: str
-    tenant_id: str
-    features_dict: Mapping[str, Any]
-
-    @classmethod
-    def from_workflow(cls, workflow: Workflow) -> "WorkflowSnapshot":
-        return cls(
-            id=workflow.id,
-            tenant_id=workflow.tenant_id,
-            features_dict=dict(workflow.features_dict),
-        )
-
-
-@dataclass(frozen=True, slots=True)
-class ConversationSnapshot:
-    id: str
-    mode: AppMode
-
-    @classmethod
-    def from_conversation(cls, conversation: Conversation) -> "ConversationSnapshot":
-        return cls(
-            id=conversation.id,
-            mode=conversation.mode,
-        )
-
-
-@dataclass(frozen=True, slots=True)
-class MessageSnapshot:
-    id: str
-    query: str
-    created_at: datetime
-    status: MessageStatus
-    answer: str
-
-    @classmethod
-    def from_message(cls, message: Message) -> "MessageSnapshot":
-        return cls(
-            id=message.id,
-            query=message.query,
-            created_at=message.created_at,
-            status=message.status,
-            answer=message.answer,
-        )
-
-
 class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
     """
     AdvancedChatAppGenerateTaskPipeline is a class that generate stream output and state management for Application.
@@ -142,10 +91,10 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
     def __init__(
         self,
         application_generate_entity: AdvancedChatAppGenerateEntity,
-        workflow: WorkflowSnapshot,
+        workflow: Workflow,
         queue_manager: AppQueueManager,
-        conversation: ConversationSnapshot,
-        message: MessageSnapshot,
+        conversation: Conversation,
+        message: Message,
         user: Union[Account, EndUser],
         stream: bool,
         dialogue_count: int,
@@ -168,7 +117,7 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         else:
             raise NotImplementedError(f"User type not supported: {type(user)}")
 
-        self._workflow_system_variables = build_system_variables(
+        self._workflow_system_variables = SystemVariable(
             query=message.query,
             files=application_generate_entity.files,
             conversation_id=conversation.id,
@@ -206,7 +155,7 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         self._message_saved_on_pause = False
         self._seed_graph_runtime_state_from_queue_manager()
 
-    def _seed_task_state_from_message(self, message: MessageSnapshot) -> None:
+    def _seed_task_state_from_message(self, message: Message) -> None:
         if message.status == MessageStatus.PAUSED and message.answer:
             self._task_state.answer = message.answer
 
@@ -792,9 +741,8 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
     def _load_human_input_form_id(self, *, node_id: str) -> str | None:
         form_repository = HumanInputFormRepositoryImpl(
             tenant_id=self._workflow_tenant_id,
-            workflow_execution_id=self._workflow_run_id,
         )
-        form = form_repository.get_form(node_id)
+        form = form_repository.get_form(self._workflow_run_id, node_id)
         if form is None:
             return None
         return form.id
@@ -985,23 +933,21 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
 
         metadata = self._task_state.metadata.model_dump()
         message.message_metadata = json.dumps(jsonable_encoder(metadata))
-        message_files: list[MessageFile] = []
-        for file in self._recorded_files:
-            reference = file.get("reference") or file.get("related_id")
-            message_files.append(
-                MessageFile(
-                    message_id=message.id,
-                    type=file["type"],
-                    transfer_method=file["transfer_method"],
-                    url=file["remote_url"],
-                    belongs_to=MessageFileBelongsTo.ASSISTANT,
-                    upload_file_id=resolve_file_record_id(reference if isinstance(reference, str) else None),
-                    created_by_role=CreatorUserRole.ACCOUNT
-                    if message.invoke_from in {InvokeFrom.EXPLORE, InvokeFrom.DEBUGGER}
-                    else CreatorUserRole.END_USER,
-                    created_by=message.from_account_id or message.from_end_user_id or "",
-                )
+        message_files = [
+            MessageFile(
+                message_id=message.id,
+                type=file["type"],
+                transfer_method=file["transfer_method"],
+                url=file["remote_url"],
+                belongs_to=MessageFileBelongsTo.ASSISTANT,
+                upload_file_id=file["related_id"],
+                created_by_role=CreatorUserRole.ACCOUNT
+                if message.invoke_from in {InvokeFrom.EXPLORE, InvokeFrom.DEBUGGER}
+                else CreatorUserRole.END_USER,
+                created_by=message.from_account_id or message.from_end_user_id or "",
             )
+            for file in self._recorded_files
+        ]
         session.add_all(message_files)
 
     def _seed_graph_runtime_state_from_queue_manager(self) -> None:
@@ -1057,11 +1003,13 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         return message
 
     def _save_output_for_event(self, event: QueueNodeSucceededEvent | QueueNodeExceptionEvent, node_execution_id: str):
-        saver = self._draft_var_saver_factory(
-            app_id=self._application_generate_entity.app_config.app_id,
-            node_id=event.node_id,
-            node_type=event.node_type,
-            node_execution_id=node_execution_id,
-            enclosing_node_id=event.in_loop_id or event.in_iteration_id,
-        )
-        saver.save(event.process_data, event.outputs)
+        with Session(db.engine) as session, session.begin():
+            saver = self._draft_var_saver_factory(
+                session=session,
+                app_id=self._application_generate_entity.app_config.app_id,
+                node_id=event.node_id,
+                node_type=event.node_type,
+                node_execution_id=node_execution_id,
+                enclosing_node_id=event.in_loop_id or event.in_iteration_id,
+            )
+            saver.save(event.process_data, event.outputs)

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

@@ -6,7 +6,6 @@ from collections.abc import Generator, Mapping
 from typing import Any, Literal, Union, overload
 
 from flask import Flask, current_app
-from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
 from pydantic import ValidationError
 
 from configs import dify_config
@@ -22,6 +21,7 @@ from core.app.apps.message_based_app_generator import MessageBasedAppGenerator
 from core.app.apps.message_based_app_queue_manager import MessageBasedAppQueueManager
 from core.app.entities.app_invoke_entities import AgentChatAppGenerateEntity, InvokeFrom
 from core.ops.ops_trace_manager import TraceQueueManager
+from dify_graph.model_runtime.errors.invoke import InvokeAuthorizationError
 from extensions.ext_database import db
 from factories import file_factory
 from libs.flask_utils import preserve_flask_contexts
@@ -129,93 +129,89 @@ class AgentChatAppGenerator(MessageBasedAppGenerator):
         #
         # For implementation reference, see the `_parse_file` function and
         # `DraftWorkflowNodeRunApi` class which handle this properly.
-        with self._bind_file_access_scope(tenant_id=app_model.tenant_id, user=user, invoke_from=invoke_from):
-            files = args.get("files") or []
-            file_extra_config = FileUploadConfigManager.convert(
-                override_model_config_dict or app_model_config.to_dict()
+        files = args.get("files") or []
+        file_extra_config = FileUploadConfigManager.convert(override_model_config_dict or app_model_config.to_dict())
+        if file_extra_config:
+            file_objs = file_factory.build_from_mappings(
+                mappings=files,
+                tenant_id=app_model.tenant_id,
+                config=file_extra_config,
             )
-            if file_extra_config:
-                file_objs = file_factory.build_from_mappings(
-                    mappings=files,
-                    tenant_id=app_model.tenant_id,
-                    config=file_extra_config,
-                    access_controller=self._file_access_controller,
-                )
-            else:
-                file_objs = []
+        else:
+            file_objs = []
 
-            # convert to app config
-            app_config = AgentChatAppConfigManager.get_app_config(
-                app_model=app_model,
-                app_model_config=app_model_config,
-                conversation=conversation,
-                override_config_dict=override_model_config_dict,
-            )
+        # convert to app config
+        app_config = AgentChatAppConfigManager.get_app_config(
+            app_model=app_model,
+            app_model_config=app_model_config,
+            conversation=conversation,
+            override_config_dict=override_model_config_dict,
+        )
 
-            # get tracing instance
-            trace_manager = TraceQueueManager(app_model.id, user.id if isinstance(user, Account) else user.session_id)
+        # get tracing instance
+        trace_manager = TraceQueueManager(app_model.id, user.id if isinstance(user, Account) else user.session_id)
 
-            # init application generate entity
-            application_generate_entity = AgentChatAppGenerateEntity(
-                task_id=str(uuid.uuid4()),
-                app_config=app_config,
-                model_conf=ModelConfigConverter.convert(app_config),
-                file_upload_config=file_extra_config,
-                conversation_id=conversation.id if conversation else None,
-                inputs=self._prepare_user_inputs(
-                    user_inputs=inputs, variables=app_config.variables, tenant_id=app_model.tenant_id
-                ),
-                query=query,
-                files=list(file_objs),
-                parent_message_id=args.get("parent_message_id") if invoke_from != InvokeFrom.SERVICE_API else UUID_NIL,
-                user_id=user.id,
-                stream=streaming,
-                invoke_from=invoke_from,
-                extras=extras,
-                call_depth=0,
-                trace_manager=trace_manager,
-            )
+        # init application generate entity
+        application_generate_entity = AgentChatAppGenerateEntity(
+            task_id=str(uuid.uuid4()),
+            app_config=app_config,
+            model_conf=ModelConfigConverter.convert(app_config),
+            file_upload_config=file_extra_config,
+            conversation_id=conversation.id if conversation else None,
+            inputs=self._prepare_user_inputs(
+                user_inputs=inputs, variables=app_config.variables, tenant_id=app_model.tenant_id
+            ),
+            query=query,
+            files=list(file_objs),
+            parent_message_id=args.get("parent_message_id") if invoke_from != InvokeFrom.SERVICE_API else UUID_NIL,
+            user_id=user.id,
+            stream=streaming,
+            invoke_from=invoke_from,
+            extras=extras,
+            call_depth=0,
+            trace_manager=trace_manager,
+        )
 
-            # init generate records
-            (conversation, message) = self._init_generate_records(application_generate_entity, conversation)
+        # init generate records
+        (conversation, message) = self._init_generate_records(application_generate_entity, conversation)
 
-            # init queue manager
-            queue_manager = MessageBasedAppQueueManager(
-                task_id=application_generate_entity.task_id,
-                user_id=application_generate_entity.user_id,
-                invoke_from=application_generate_entity.invoke_from,
-                conversation_id=conversation.id,
-                app_mode=conversation.mode,
-                message_id=message.id,
-            )
+        # init queue manager
+        queue_manager = MessageBasedAppQueueManager(
+            task_id=application_generate_entity.task_id,
+            user_id=application_generate_entity.user_id,
+            invoke_from=application_generate_entity.invoke_from,
+            conversation_id=conversation.id,
+            app_mode=conversation.mode,
+            message_id=message.id,
+        )
 
-            # new thread with request context and contextvars
-            context = contextvars.copy_context()
+        # new thread with request context and contextvars
+        context = contextvars.copy_context()
 
-            worker_thread = threading.Thread(
-                target=self._generate_worker,
-                kwargs={
-                    "flask_app": current_app._get_current_object(),  # type: ignore
-                    "context": context,
-                    "application_generate_entity": application_generate_entity,
-                    "queue_manager": queue_manager,
-                    "conversation_id": conversation.id,
-                    "message_id": message.id,
-                },
-            )
+        worker_thread = threading.Thread(
+            target=self._generate_worker,
+            kwargs={
+                "flask_app": current_app._get_current_object(),  # type: ignore
+                "context": context,
+                "application_generate_entity": application_generate_entity,
+                "queue_manager": queue_manager,
+                "conversation_id": conversation.id,
+                "message_id": message.id,
+            },
+        )
 
-            worker_thread.start()
+        worker_thread.start()
 
-            # return response or stream generator
-            response = self._handle_response(
-                application_generate_entity=application_generate_entity,
-                queue_manager=queue_manager,
-                conversation=conversation,
-                message=message,
-                user=user,
-                stream=streaming,
-            )
-            return AgentChatAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
+        # return response or stream generator
+        response = self._handle_response(
+            application_generate_entity=application_generate_entity,
+            queue_manager=queue_manager,
+            conversation=conversation,
+            message=message,
+            user=user,
+            stream=streaming,
+        )
+        return AgentChatAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
 
     def _generate_worker(
         self,

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

@@ -1,9 +1,6 @@
 import logging
 from typing import cast
 
-from graphon.model_runtime.entities.llm_entities import LLMMode
-from graphon.model_runtime.entities.model_entities import ModelFeature, ModelPropertyKey
-from graphon.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
 from sqlalchemy import select
 
 from core.agent.cot_chat_agent_runner import CotChatAgentRunner
@@ -18,6 +15,9 @@ from core.app.entities.queue_entities import QueueAnnotationReplyEvent
 from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance
 from core.moderation.base import ModerationError
+from dify_graph.model_runtime.entities.llm_entities import LLMMode
+from dify_graph.model_runtime.entities.model_entities import ModelFeature, ModelPropertyKey
+from dify_graph.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
 from extensions.ext_database import db
 from models.model import App, Conversation, Message
 

api/core/app/apps/base_app_generate_response_converter.py

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

api/core/app/apps/base_app_generator.py

@@ -1,89 +1,27 @@
 from collections.abc import Generator, Mapping, Sequence
-from contextlib import AbstractContextManager, nullcontext
 from typing import TYPE_CHECKING, Any, Union, final
 
-from graphon.enums import NodeType
-from graphon.file import File, FileUploadConfig
-from graphon.variables.input_entities import VariableEntityType
 from sqlalchemy.orm import Session
 
-from core.app.apps.draft_variable_saver import (
+from core.app.entities.app_invoke_entities import InvokeFrom
+from dify_graph.enums import NodeType
+from dify_graph.file import File, FileUploadConfig
+from dify_graph.repositories.draft_variable_repository import (
     DraftVariableSaver,
     DraftVariableSaverFactory,
     NoopDraftVariableSaver,
 )
-from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom
-from core.app.file_access import DatabaseFileAccessController, FileAccessScope, bind_file_access_scope
-from extensions.ext_database import db
+from dify_graph.variables.input_entities import VariableEntityType
 from factories import file_factory
 from libs.orjson import orjson_dumps
 from models import Account, EndUser
 from services.workflow_draft_variable_service import DraftVariableSaver as DraftVariableSaverImpl
 
 if TYPE_CHECKING:
-    from graphon.variables.input_entities import VariableEntity
-
-
-@final
-class _DebuggerDraftVariableSaver:
-    """Adapter that binds SQLAlchemy session setup outside the saver port."""
-
-    def __init__(
-        self,
-        *,
-        account: Account,
-        app_id: str,
-        node_id: str,
-        node_type: NodeType,
-        node_execution_id: str,
-        enclosing_node_id: str | None = None,
-    ) -> None:
-        self._account = account
-        self._app_id = app_id
-        self._node_id = node_id
-        self._node_type = node_type
-        self._node_execution_id = node_execution_id
-        self._enclosing_node_id = enclosing_node_id
-
-    def save(self, process_data: Mapping[str, Any] | None, outputs: Mapping[str, Any] | None) -> None:
-        with Session(db.engine) as session, session.begin():
-            DraftVariableSaverImpl(
-                session=session,
-                app_id=self._app_id,
-                node_id=self._node_id,
-                node_type=self._node_type,
-                node_execution_id=self._node_execution_id,
-                enclosing_node_id=self._enclosing_node_id,
-                user=self._account,
-            ).save(process_data, outputs)
+    from dify_graph.variables.input_entities import VariableEntity
 
 
 class BaseAppGenerator:
-    _file_access_controller: DatabaseFileAccessController = DatabaseFileAccessController()
-
-    @staticmethod
-    def _bind_file_access_scope(
-        *,
-        tenant_id: str,
-        user: Account | EndUser,
-        invoke_from: InvokeFrom,
-    ) -> AbstractContextManager[None]:
-        """Bind request-scoped file ownership markers for downstream file lookups."""
-
-        user_id = getattr(user, "id", None)
-        if not isinstance(user_id, str) or not user_id:
-            return nullcontext()
-
-        user_from = UserFrom.ACCOUNT if isinstance(user, Account) else UserFrom.END_USER
-        return bind_file_access_scope(
-            FileAccessScope(
-                tenant_id=tenant_id,
-                user_id=user_id,
-                user_from=user_from,
-                invoke_from=invoke_from,
-            )
-        )
-
     def _prepare_user_inputs(
         self,
         *,
@@ -112,7 +50,6 @@ class BaseAppGenerator:
                     allowed_file_upload_methods=entity_dictionary[k].allowed_file_upload_methods or [],
                 ),
                 strict_type_validation=strict_type_validation,
-                access_controller=self._file_access_controller,
             )
             for k, v in user_inputs.items()
             if isinstance(v, dict) and entity_dictionary[k].type == VariableEntityType.FILE
@@ -127,7 +64,6 @@ class BaseAppGenerator:
                     allowed_file_extensions=entity_dictionary[k].allowed_file_extensions or [],
                     allowed_file_upload_methods=entity_dictionary[k].allowed_file_upload_methods or [],
                 ),
-                access_controller=self._file_access_controller,
             )
             for k, v in user_inputs.items()
             if isinstance(v, list)
@@ -290,30 +226,32 @@ class BaseAppGenerator:
             assert isinstance(account, Account)
 
             def draft_var_saver_factory(
+                session: Session,
                 app_id: str,
                 node_id: str,
                 node_type: NodeType,
                 node_execution_id: str,
                 enclosing_node_id: str | None = None,
             ) -> DraftVariableSaver:
-                return _DebuggerDraftVariableSaver(
-                    account=account,
+                return DraftVariableSaverImpl(
+                    session=session,
                     app_id=app_id,
                     node_id=node_id,
                     node_type=node_type,
                     node_execution_id=node_execution_id,
                     enclosing_node_id=enclosing_node_id,
+                    user=account,
                 )
         else:
 
             def draft_var_saver_factory(
+                session: Session,
                 app_id: str,
                 node_id: str,
                 node_type: NodeType,
                 node_execution_id: str,
                 enclosing_node_id: str | None = None,
             ) -> DraftVariableSaver:
-                _ = app_id, node_id, node_type, node_execution_id, enclosing_node_id
                 return NoopDraftVariableSaver()
 
         return draft_var_saver_factory

api/core/app/apps/base_app_queue_manager.py

@@ -7,7 +7,6 @@ from enum import IntEnum, auto
 from typing import Any
 
 from cachetools import TTLCache, cachedmethod
-from graphon.runtime import GraphRuntimeState
 from redis.exceptions import RedisError
 from sqlalchemy.orm import DeclarativeMeta
 
@@ -21,6 +20,7 @@ from core.app.entities.queue_entities import (
     QueueStopEvent,
     WorkflowQueueMessage,
 )
+from dify_graph.runtime import GraphRuntimeState
 from extensions.ext_redis import redis_client
 
 logger = logging.getLogger(__name__)
@@ -61,30 +61,27 @@ class AppQueueManager(ABC):
         listen_timeout = dify_config.APP_MAX_EXECUTION_TIME
         start_time = time.time()
         last_ping_time: int | float = 0
-        try:
-            while True:
-                try:
-                    message = self._q.get(timeout=1)
-                    if message is None:
-                        break
+        while True:
+            try:
+                message = self._q.get(timeout=1)
+                if message is None:
+                    break
 
-                    yield message
-                except queue.Empty:
-                    continue
-                finally:
-                    elapsed_time = time.time() - start_time
-                    if elapsed_time >= listen_timeout or self._is_stopped():
-                        # publish two messages to make sure the client can receive the stop signal
-                        # and stop listening after the stop signal processed
-                        self.publish(
-                            QueueStopEvent(stopped_by=QueueStopEvent.StopBy.USER_MANUAL), PublishFrom.TASK_PIPELINE
-                        )
+                yield message
+            except queue.Empty:
+                continue
+            finally:
+                elapsed_time = time.time() - start_time
+                if elapsed_time >= listen_timeout or self._is_stopped():
+                    # publish two messages to make sure the client can receive the stop signal
+                    # and stop listening after the stop signal processed
+                    self.publish(
+                        QueueStopEvent(stopped_by=QueueStopEvent.StopBy.USER_MANUAL), PublishFrom.TASK_PIPELINE
+                    )
 
-                    if elapsed_time // 10 > last_ping_time:
-                        self.publish(QueuePingEvent(), PublishFrom.TASK_PIPELINE)
-                        last_ping_time = elapsed_time // 10
-        finally:
-            self._graph_runtime_state = None  # Release reference once consumers finish or close the generator.
+                if elapsed_time // 10 > last_ping_time:
+                    self.publish(QueuePingEvent(), PublishFrom.TASK_PIPELINE)
+                    last_ping_time = elapsed_time // 10
 
     def stop_listen(self):
         """
@@ -93,6 +90,7 @@ class AppQueueManager(ABC):
         """
         self._clear_task_belong_cache()
         self._q.put(None)
+        self._graph_runtime_state = None  # Release reference to allow GC to reclaim memory
 
     def _clear_task_belong_cache(self) -> None:
         """

api/core/app/apps/base_app_runner.py

@@ -5,17 +5,6 @@ from collections.abc import Generator, Mapping, Sequence
 from mimetypes import guess_extension
 from typing import TYPE_CHECKING, Any, Union
 
-from graphon.file import FileTransferMethod, FileType
-from graphon.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk, LLMResultChunkDelta, LLMUsage
-from graphon.model_runtime.entities.message_entities import (
-    AssistantPromptMessage,
-    ImagePromptMessageContent,
-    PromptMessage,
-    TextPromptMessageContent,
-)
-from graphon.model_runtime.entities.model_entities import ModelPropertyKey
-from graphon.model_runtime.errors.invoke import InvokeBadRequestError
-
 from core.app.app_config.entities import ExternalDataVariableEntity, PromptTemplateEntity
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
 from core.app.entities.app_invoke_entities import (
@@ -40,12 +29,22 @@ from core.prompt.advanced_prompt_transform import AdvancedPromptTransform
 from core.prompt.entities.advanced_prompt_entities import ChatModelMessage, CompletionModelPromptTemplate, MemoryConfig
 from core.prompt.simple_prompt_transform import ModelMode, SimplePromptTransform
 from core.tools.tool_file_manager import ToolFileManager
+from dify_graph.file.enums import FileTransferMethod, FileType
+from dify_graph.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk, LLMResultChunkDelta, LLMUsage
+from dify_graph.model_runtime.entities.message_entities import (
+    AssistantPromptMessage,
+    ImagePromptMessageContent,
+    PromptMessage,
+    TextPromptMessageContent,
+)
+from dify_graph.model_runtime.entities.model_entities import ModelPropertyKey
+from dify_graph.model_runtime.errors.invoke import InvokeBadRequestError
 from extensions.ext_database import db
 from models.enums import CreatorUserRole, MessageFileBelongsTo
 from models.model import App, AppMode, Message, MessageAnnotation, MessageFile
 
 if TYPE_CHECKING:
-    from graphon.file import File
+    from dify_graph.file.models import File
 
 _logger = logging.getLogger(__name__)
 

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

@@ -1,4 +1,3 @@
-import contextvars
 import logging
 import threading
 import uuid
@@ -6,7 +5,6 @@ from collections.abc import Generator, Mapping
 from typing import Any, Literal, Union, overload
 
 from flask import Flask, copy_current_request_context, current_app
-from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
 from pydantic import ValidationError
 
 from configs import dify_config
@@ -22,6 +20,7 @@ from core.app.apps.message_based_app_generator import MessageBasedAppGenerator
 from core.app.apps.message_based_app_queue_manager import MessageBasedAppQueueManager
 from core.app.entities.app_invoke_entities import ChatAppGenerateEntity, InvokeFrom
 from core.ops.ops_trace_manager import TraceQueueManager
+from dify_graph.model_runtime.errors.invoke import InvokeAuthorizationError
 from extensions.ext_database import db
 from factories import file_factory
 from models import Account
@@ -121,96 +120,89 @@ class ChatAppGenerator(MessageBasedAppGenerator):
         #
         # For implementation reference, see the `_parse_file` function and
         # `DraftWorkflowNodeRunApi` class which handle this properly.
-        with self._bind_file_access_scope(tenant_id=app_model.tenant_id, user=user, invoke_from=invoke_from):
-            files = args["files"] if args.get("files") else []
-            file_extra_config = FileUploadConfigManager.convert(
-                override_model_config_dict or app_model_config.to_dict()
+        files = args["files"] if args.get("files") else []
+        file_extra_config = FileUploadConfigManager.convert(override_model_config_dict or app_model_config.to_dict())
+        if file_extra_config:
+            file_objs = file_factory.build_from_mappings(
+                mappings=files,
+                tenant_id=app_model.tenant_id,
+                config=file_extra_config,
             )
-            if file_extra_config:
-                file_objs = file_factory.build_from_mappings(
-                    mappings=files,
-                    tenant_id=app_model.tenant_id,
-                    config=file_extra_config,
-                    access_controller=self._file_access_controller,
-                )
-            else:
-                file_objs = []
+        else:
+            file_objs = []
 
-            # convert to app config
-            app_config = ChatAppConfigManager.get_app_config(
-                app_model=app_model,
-                app_model_config=app_model_config,
-                conversation=conversation,
-                override_config_dict=override_model_config_dict,
-            )
+        # convert to app config
+        app_config = ChatAppConfigManager.get_app_config(
+            app_model=app_model,
+            app_model_config=app_model_config,
+            conversation=conversation,
+            override_config_dict=override_model_config_dict,
+        )
 
-            # get tracing instance
-            trace_manager = TraceQueueManager(
-                app_id=app_model.id, user_id=user.id if isinstance(user, Account) else user.session_id
-            )
+        # get tracing instance
+        trace_manager = TraceQueueManager(
+            app_id=app_model.id, user_id=user.id if isinstance(user, Account) else user.session_id
+        )
 
-            # init application generate entity
-            application_generate_entity = ChatAppGenerateEntity(
-                task_id=str(uuid.uuid4()),
-                app_config=app_config,
-                model_conf=ModelConfigConverter.convert(app_config),
-                file_upload_config=file_extra_config,
-                conversation_id=conversation.id if conversation else None,
-                inputs=self._prepare_user_inputs(
-                    user_inputs=inputs, variables=app_config.variables, tenant_id=app_model.tenant_id
-                ),
-                query=query,
-                files=list(file_objs),
-                parent_message_id=args.get("parent_message_id") if invoke_from != InvokeFrom.SERVICE_API else UUID_NIL,
-                user_id=user.id,
-                invoke_from=invoke_from,
-                extras=extras,
-                trace_manager=trace_manager,
-                stream=streaming,
-            )
+        # init application generate entity
+        application_generate_entity = ChatAppGenerateEntity(
+            task_id=str(uuid.uuid4()),
+            app_config=app_config,
+            model_conf=ModelConfigConverter.convert(app_config),
+            file_upload_config=file_extra_config,
+            conversation_id=conversation.id if conversation else None,
+            inputs=self._prepare_user_inputs(
+                user_inputs=inputs, variables=app_config.variables, tenant_id=app_model.tenant_id
+            ),
+            query=query,
+            files=list(file_objs),
+            parent_message_id=args.get("parent_message_id") if invoke_from != InvokeFrom.SERVICE_API else UUID_NIL,
+            user_id=user.id,
+            invoke_from=invoke_from,
+            extras=extras,
+            trace_manager=trace_manager,
+            stream=streaming,
+        )
 
-            # init generate records
-            (conversation, message) = self._init_generate_records(application_generate_entity, conversation)
+        # init generate records
+        (conversation, message) = self._init_generate_records(application_generate_entity, conversation)
 
-            # init queue manager
-            queue_manager = MessageBasedAppQueueManager(
-                task_id=application_generate_entity.task_id,
-                user_id=application_generate_entity.user_id,
-                invoke_from=application_generate_entity.invoke_from,
+        # init queue manager
+        queue_manager = MessageBasedAppQueueManager(
+            task_id=application_generate_entity.task_id,
+            user_id=application_generate_entity.user_id,
+            invoke_from=application_generate_entity.invoke_from,
+            conversation_id=conversation.id,
+            app_mode=conversation.mode,
+            message_id=message.id,
+        )
+
+        # new thread with request context
+        @copy_current_request_context
+        def worker_with_context():
+            return self._generate_worker(
+                flask_app=current_app._get_current_object(),  # type: ignore
+                application_generate_entity=application_generate_entity,
+                queue_manager=queue_manager,
                 conversation_id=conversation.id,
-                app_mode=conversation.mode,
                 message_id=message.id,
             )
 
-            context = contextvars.copy_context()
+        worker_thread = threading.Thread(target=worker_with_context)
 
-            # new thread with request context
-            @copy_current_request_context
-            def worker_with_context():
-                return context.run(
-                    self._generate_worker,
-                    flask_app=current_app._get_current_object(),  # type: ignore
-                    application_generate_entity=application_generate_entity,
-                    queue_manager=queue_manager,
-                    conversation_id=conversation.id,
-                    message_id=message.id,
-                )
+        worker_thread.start()
 
-            worker_thread = threading.Thread(target=worker_with_context)
+        # return response or stream generator
+        response = self._handle_response(
+            application_generate_entity=application_generate_entity,
+            queue_manager=queue_manager,
+            conversation=conversation,
+            message=message,
+            user=user,
+            stream=streaming,
+        )
 
-            worker_thread.start()
-
-            # return response or stream generator
-            response = self._handle_response(
-                application_generate_entity=application_generate_entity,
-                queue_manager=queue_manager,
-                conversation=conversation,
-                message=message,
-                user=user,
-                stream=streaming,
-            )
-
-            return ChatAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
+        return ChatAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
 
     def _generate_worker(
         self,

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

@@ -1,8 +1,6 @@
 import logging
 from typing import cast
 
-from graphon.file import File
-from graphon.model_runtime.entities.message_entities import ImagePromptMessageContent
 from sqlalchemy import select
 
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
@@ -17,6 +15,8 @@ from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance
 from core.moderation.base import ModerationError
 from core.rag.retrieval.dataset_retrieval import DatasetRetrieval
+from dify_graph.file import File
+from dify_graph.model_runtime.entities.message_entities import ImagePromptMessageContent
 from extensions.ext_database import db
 from models.model import App, Conversation, Message
 
@@ -223,6 +223,7 @@ class ChatAppRunner(AppRunner):
             model_parameters=application_generate_entity.model_conf.parameters,
             stop=stop,
             stream=application_generate_entity.stream,
+            user=application_generate_entity.user_id,
         )
 
         # handle invoke result

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

@@ -4,9 +4,7 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING
 
-from graphon.runtime import GraphRuntimeState
-
-from core.workflow.system_variables import SystemVariableKey, get_system_text
+from dify_graph.runtime import GraphRuntimeState
 
 if TYPE_CHECKING:
     from core.app.task_pipeline.based_generate_task_pipeline import BasedGenerateTaskPipeline
@@ -32,10 +30,10 @@ class GraphRuntimeStateSupport:
         return self._resolve_graph_runtime_state(graph_runtime_state)
 
     def _extract_workflow_run_id(self, graph_runtime_state: GraphRuntimeState) -> str:
-        workflow_run_id = get_system_text(graph_runtime_state.variable_pool, SystemVariableKey.WORKFLOW_EXECUTION_ID)
-        if not workflow_run_id:
+        system_variables = graph_runtime_state.variable_pool.system_variables
+        if not system_variables or not system_variables.workflow_execution_id:
             raise ValueError("workflow_execution_id missing from runtime state")
-        return workflow_run_id
+        return str(system_variables.workflow_execution_id)
 
     def _resolve_graph_runtime_state(
         self,

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

@@ -1,4 +1,3 @@
-import json
 import logging
 import time
 from collections.abc import Mapping, Sequence
@@ -6,19 +5,6 @@ from dataclasses import dataclass
 from datetime import datetime
 from typing import Any, NewType, TypedDict, Union
 
-from graphon.entities import WorkflowStartReason
-from graphon.entities.pause_reason import HumanInputRequired
-from graphon.enums import (
-    BuiltinNodeTypes,
-    WorkflowExecutionStatus,
-    WorkflowNodeExecutionMetadataKey,
-    WorkflowNodeExecutionStatus,
-)
-from graphon.file import FILE_MODEL_IDENTITY, File
-from graphon.runtime import GraphRuntimeState
-from graphon.variables.segments import ArrayFileSegment, FileSegment, Segment
-from graphon.variables.variables import Variable
-from graphon.workflow_type_encoder import WorkflowRuntimeTypeConverter
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
@@ -64,9 +50,21 @@ 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.human_input_forms import load_form_tokens_by_form_id
-from core.workflow.system_variables import SystemVariableKey, system_variables_to_mapping
 from core.workflow.workflow_entry import WorkflowEntry
+from dify_graph.entities.pause_reason import HumanInputRequired
+from dify_graph.entities.workflow_start_reason import WorkflowStartReason
+from dify_graph.enums import (
+    BuiltinNodeTypes,
+    SystemVariableKey,
+    WorkflowExecutionStatus,
+    WorkflowNodeExecutionMetadataKey,
+    WorkflowNodeExecutionStatus,
+)
+from dify_graph.file import FILE_MODEL_IDENTITY, File
+from dify_graph.runtime import GraphRuntimeState
+from dify_graph.system_variable import SystemVariable
+from dify_graph.variables.segments import ArrayFileSegment, FileSegment, Segment
+from dify_graph.workflow_type_encoder import WorkflowRuntimeTypeConverter
 from extensions.ext_database import db
 from libs.datetime_utils import naive_utc_now
 from models import Account, EndUser
@@ -113,11 +111,11 @@ class WorkflowResponseConverter:
         *,
         application_generate_entity: Union[AdvancedChatAppGenerateEntity, WorkflowAppGenerateEntity],
         user: Union[Account, EndUser],
-        system_variables: Sequence[Variable],
+        system_variables: SystemVariable,
     ):
         self._application_generate_entity = application_generate_entity
         self._user = user
-        self._system_variables = system_variables_to_mapping(system_variables)
+        self._system_variables = system_variables
         self._workflow_inputs = self._prepare_workflow_inputs()
 
         # Disable truncation for SERVICE_API calls to keep backward compatibility.
@@ -135,7 +133,7 @@ class WorkflowResponseConverter:
     # ------------------------------------------------------------------
     def _prepare_workflow_inputs(self) -> Mapping[str, Any]:
         inputs = dict(self._application_generate_entity.inputs)
-        for field_name, value in self._system_variables.items():
+        for field_name, value in self._system_variables.to_dict().items():
             # TODO(@future-refactor): store system variables separately from user inputs so we don't
             # need to flatten `sys.*` entries into the input payload just for rerun/export tooling.
             if field_name == SystemVariableKey.CONVERSATION_ID:
@@ -320,23 +318,13 @@ class WorkflowResponseConverter:
         pause_reasons = [reason.model_dump(mode="json") for reason in event.reasons]
         human_input_form_ids = [reason.form_id for reason in event.reasons if isinstance(reason, HumanInputRequired)]
         expiration_times_by_form_id: dict[str, datetime] = {}
-        display_in_ui_by_form_id: dict[str, bool] = {}
-        form_token_by_form_id: dict[str, str] = {}
         if human_input_form_ids:
-            stmt = select(
-                HumanInputForm.id,
-                HumanInputForm.expiration_time,
-                HumanInputForm.form_definition,
-            ).where(HumanInputForm.id.in_(human_input_form_ids))
+            stmt = select(HumanInputForm.id, HumanInputForm.expiration_time).where(
+                HumanInputForm.id.in_(human_input_form_ids)
+            )
             with Session(bind=db.engine) as session:
-                for form_id, expiration_time, form_definition in session.execute(stmt):
+                for form_id, expiration_time in session.execute(stmt):
                     expiration_times_by_form_id[str(form_id)] = expiration_time
-                    try:
-                        definition_payload = json.loads(form_definition) if form_definition else {}
-                    except (TypeError, json.JSONDecodeError):
-                        definition_payload = {}
-                    display_in_ui_by_form_id[str(form_id)] = bool(definition_payload.get("display_in_ui"))
-                form_token_by_form_id = load_form_tokens_by_form_id(human_input_form_ids, session=session)
 
         responses: list[StreamResponse] = []
 
@@ -356,8 +344,8 @@ class WorkflowResponseConverter:
                             form_content=reason.form_content,
                             inputs=reason.inputs,
                             actions=reason.actions,
-                            display_in_ui=display_in_ui_by_form_id.get(reason.form_id, False),
-                            form_token=form_token_by_form_id.get(reason.form_id),
+                            display_in_ui=reason.display_in_ui,
+                            form_token=reason.form_token,
                             resolved_default_values=reason.resolved_default_values,
                             expiration_time=int(expiration_time.timestamp()),
                         ),

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

@@ -1,4 +1,3 @@
-import contextvars
 import logging
 import threading
 import uuid
@@ -6,7 +5,6 @@ from collections.abc import Generator, Mapping
 from typing import Any, Literal, Union, overload
 
 from flask import Flask, copy_current_request_context, current_app
-from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
 from pydantic import ValidationError
 from sqlalchemy import select
 
@@ -22,6 +20,7 @@ from core.app.apps.message_based_app_generator import MessageBasedAppGenerator
 from core.app.apps.message_based_app_queue_manager import MessageBasedAppQueueManager
 from core.app.entities.app_invoke_entities import CompletionAppGenerateEntity, InvokeFrom
 from core.ops.ops_trace_manager import TraceQueueManager
+from dify_graph.model_runtime.errors.invoke import InvokeAuthorizationError
 from extensions.ext_database import db
 from factories import file_factory
 from models import Account, App, EndUser, Message
@@ -109,90 +108,83 @@ class CompletionAppGenerator(MessageBasedAppGenerator):
         #
         # For implementation reference, see the `_parse_file` function and
         # `DraftWorkflowNodeRunApi` class which handle this properly.
-        with self._bind_file_access_scope(tenant_id=app_model.tenant_id, user=user, invoke_from=invoke_from):
-            files = args["files"] if args.get("files") else []
-            file_extra_config = FileUploadConfigManager.convert(
-                override_model_config_dict or app_model_config.to_dict()
+        files = args["files"] if args.get("files") else []
+        file_extra_config = FileUploadConfigManager.convert(override_model_config_dict or app_model_config.to_dict())
+        if file_extra_config:
+            file_objs = file_factory.build_from_mappings(
+                mappings=files,
+                tenant_id=app_model.tenant_id,
+                config=file_extra_config,
             )
-            if file_extra_config:
-                file_objs = file_factory.build_from_mappings(
-                    mappings=files,
-                    tenant_id=app_model.tenant_id,
-                    config=file_extra_config,
-                    access_controller=self._file_access_controller,
-                )
-            else:
-                file_objs = []
+        else:
+            file_objs = []
 
-            # convert to app config
-            app_config = CompletionAppConfigManager.get_app_config(
-                app_model=app_model, app_model_config=app_model_config, override_config_dict=override_model_config_dict
-            )
+        # convert to app config
+        app_config = CompletionAppConfigManager.get_app_config(
+            app_model=app_model, app_model_config=app_model_config, override_config_dict=override_model_config_dict
+        )
 
-            # get tracing instance
-            trace_manager = TraceQueueManager(
-                app_id=app_model.id, user_id=user.id if isinstance(user, Account) else user.session_id
-            )
+        # get tracing instance
+        trace_manager = TraceQueueManager(
+            app_id=app_model.id, user_id=user.id if isinstance(user, Account) else user.session_id
+        )
 
-            # init application generate entity
-            application_generate_entity = CompletionAppGenerateEntity(
-                task_id=str(uuid.uuid4()),
-                app_config=app_config,
-                model_conf=ModelConfigConverter.convert(app_config),
-                file_upload_config=file_extra_config,
-                inputs=self._prepare_user_inputs(
-                    user_inputs=inputs, variables=app_config.variables, tenant_id=app_model.tenant_id
-                ),
-                query=query,
-                files=list(file_objs),
-                user_id=user.id,
-                stream=streaming,
-                invoke_from=invoke_from,
-                extras={},
-                trace_manager=trace_manager,
-            )
+        # init application generate entity
+        application_generate_entity = CompletionAppGenerateEntity(
+            task_id=str(uuid.uuid4()),
+            app_config=app_config,
+            model_conf=ModelConfigConverter.convert(app_config),
+            file_upload_config=file_extra_config,
+            inputs=self._prepare_user_inputs(
+                user_inputs=inputs, variables=app_config.variables, tenant_id=app_model.tenant_id
+            ),
+            query=query,
+            files=list(file_objs),
+            user_id=user.id,
+            stream=streaming,
+            invoke_from=invoke_from,
+            extras={},
+            trace_manager=trace_manager,
+        )
 
-            # init generate records
-            (conversation, message) = self._init_generate_records(application_generate_entity)
+        # init generate records
+        (conversation, message) = self._init_generate_records(application_generate_entity)
 
-            # init queue manager
-            queue_manager = MessageBasedAppQueueManager(
-                task_id=application_generate_entity.task_id,
-                user_id=application_generate_entity.user_id,
-                invoke_from=application_generate_entity.invoke_from,
-                conversation_id=conversation.id,
-                app_mode=conversation.mode,
+        # init queue manager
+        queue_manager = MessageBasedAppQueueManager(
+            task_id=application_generate_entity.task_id,
+            user_id=application_generate_entity.user_id,
+            invoke_from=application_generate_entity.invoke_from,
+            conversation_id=conversation.id,
+            app_mode=conversation.mode,
+            message_id=message.id,
+        )
+
+        # new thread with request context
+        @copy_current_request_context
+        def worker_with_context():
+            return self._generate_worker(
+                flask_app=current_app._get_current_object(),  # type: ignore
+                application_generate_entity=application_generate_entity,
+                queue_manager=queue_manager,
                 message_id=message.id,
             )
 
-            context = contextvars.copy_context()
+        worker_thread = threading.Thread(target=worker_with_context)
 
-            # new thread with request context
-            @copy_current_request_context
-            def worker_with_context():
-                return context.run(
-                    self._generate_worker,
-                    flask_app=current_app._get_current_object(),  # type: ignore
-                    application_generate_entity=application_generate_entity,
-                    queue_manager=queue_manager,
-                    message_id=message.id,
-                )
+        worker_thread.start()
 
-            worker_thread = threading.Thread(target=worker_with_context)
+        # return response or stream generator
+        response = self._handle_response(
+            application_generate_entity=application_generate_entity,
+            queue_manager=queue_manager,
+            conversation=conversation,
+            message=message,
+            user=user,
+            stream=streaming,
+        )
 
-            worker_thread.start()
-
-            # return response or stream generator
-            response = self._handle_response(
-                application_generate_entity=application_generate_entity,
-                queue_manager=queue_manager,
-                conversation=conversation,
-                message=message,
-                user=user,
-                stream=streaming,
-            )
-
-            return CompletionAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
+        return CompletionAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
 
     def _generate_worker(
         self,
@@ -288,76 +280,71 @@ class CompletionAppGenerator(MessageBasedAppGenerator):
         model_dict["completion_params"] = completion_params
         override_model_config_dict["model"] = model_dict
 
-        with self._bind_file_access_scope(tenant_id=app_model.tenant_id, user=user, invoke_from=invoke_from):
-            # parse files
-            file_extra_config = FileUploadConfigManager.convert(override_model_config_dict)
-            if file_extra_config:
-                file_objs = file_factory.build_from_mappings(
-                    mappings=message.message_files,
-                    tenant_id=app_model.tenant_id,
-                    config=file_extra_config,
-                    access_controller=self._file_access_controller,
-                )
-            else:
-                file_objs = []
-
-            # convert to app config
-            app_config = CompletionAppConfigManager.get_app_config(
-                app_model=app_model, app_model_config=app_model_config, override_config_dict=override_model_config_dict
+        # parse files
+        file_extra_config = FileUploadConfigManager.convert(override_model_config_dict)
+        if file_extra_config:
+            file_objs = file_factory.build_from_mappings(
+                mappings=message.message_files,
+                tenant_id=app_model.tenant_id,
+                config=file_extra_config,
             )
+        else:
+            file_objs = []
 
-            # init application generate entity
-            application_generate_entity = CompletionAppGenerateEntity(
-                task_id=str(uuid.uuid4()),
-                app_config=app_config,
-                model_conf=ModelConfigConverter.convert(app_config),
-                inputs=message.inputs,
-                query=message.query,
-                files=list(file_objs),
-                user_id=user.id,
-                stream=stream,
-                invoke_from=invoke_from,
-                extras={},
-            )
+        # convert to app config
+        app_config = CompletionAppConfigManager.get_app_config(
+            app_model=app_model, app_model_config=app_model_config, override_config_dict=override_model_config_dict
+        )
 
-            # init generate records
-            (conversation, message) = self._init_generate_records(application_generate_entity)
+        # init application generate entity
+        application_generate_entity = CompletionAppGenerateEntity(
+            task_id=str(uuid.uuid4()),
+            app_config=app_config,
+            model_conf=ModelConfigConverter.convert(app_config),
+            inputs=message.inputs,
+            query=message.query,
+            files=list(file_objs),
+            user_id=user.id,
+            stream=stream,
+            invoke_from=invoke_from,
+            extras={},
+        )
 
-            # init queue manager
-            queue_manager = MessageBasedAppQueueManager(
-                task_id=application_generate_entity.task_id,
-                user_id=application_generate_entity.user_id,
-                invoke_from=application_generate_entity.invoke_from,
-                conversation_id=conversation.id,
-                app_mode=conversation.mode,
+        # init generate records
+        (conversation, message) = self._init_generate_records(application_generate_entity)
+
+        # init queue manager
+        queue_manager = MessageBasedAppQueueManager(
+            task_id=application_generate_entity.task_id,
+            user_id=application_generate_entity.user_id,
+            invoke_from=application_generate_entity.invoke_from,
+            conversation_id=conversation.id,
+            app_mode=conversation.mode,
+            message_id=message.id,
+        )
+
+        # new thread with request context
+        @copy_current_request_context
+        def worker_with_context():
+            return self._generate_worker(
+                flask_app=current_app._get_current_object(),  # type: ignore
+                application_generate_entity=application_generate_entity,
+                queue_manager=queue_manager,
                 message_id=message.id,
             )
 
-            context = contextvars.copy_context()
+        worker_thread = threading.Thread(target=worker_with_context)
 
-            # new thread with request context
-            @copy_current_request_context
-            def worker_with_context():
-                return context.run(
-                    self._generate_worker,
-                    flask_app=current_app._get_current_object(),  # type: ignore
-                    application_generate_entity=application_generate_entity,
-                    queue_manager=queue_manager,
-                    message_id=message.id,
-                )
+        worker_thread.start()
 
-            worker_thread = threading.Thread(target=worker_with_context)
+        # return response or stream generator
+        response = self._handle_response(
+            application_generate_entity=application_generate_entity,
+            queue_manager=queue_manager,
+            conversation=conversation,
+            message=message,
+            user=user,
+            stream=stream,
+        )
 
-            worker_thread.start()
-
-            # return response or stream generator
-            response = self._handle_response(
-                application_generate_entity=application_generate_entity,
-                queue_manager=queue_manager,
-                conversation=conversation,
-                message=message,
-                user=user,
-                stream=stream,
-            )
-
-            return CompletionAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
+        return CompletionAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)

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

@@ -1,8 +1,6 @@
 import logging
 from typing import cast
 
-from graphon.file import File
-from graphon.model_runtime.entities.message_entities import ImagePromptMessageContent
 from sqlalchemy import select
 
 from core.app.apps.base_app_queue_manager import AppQueueManager
@@ -15,6 +13,8 @@ from core.callback_handler.index_tool_callback_handler import DatasetIndexToolCa
 from core.model_manager import ModelInstance
 from core.moderation.base import ModerationError
 from core.rag.retrieval.dataset_retrieval import DatasetRetrieval
+from dify_graph.file import File
+from dify_graph.model_runtime.entities.message_entities import ImagePromptMessageContent
 from extensions.ext_database import db
 from models.model import App, Message
 
@@ -181,6 +181,7 @@ class CompletionAppRunner(AppRunner):
             model_parameters=application_generate_entity.model_conf.parameters,
             stop=stop,
             stream=application_generate_entity.stream,
+            user=application_generate_entity.user_id,
         )
 
         # handle invoke result

api/core/app/apps/message_based_app_generator.py

@@ -28,7 +28,6 @@ from core.app.entities.task_entities import (
 )
 from core.app.task_pipeline.easy_ui_based_generate_task_pipeline import EasyUIBasedGenerateTaskPipeline
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
-from core.workflow.file_reference import resolve_file_record_id
 from extensions.ext_database import db
 from extensions.ext_redis import get_pubsub_broadcast_channel
 from libs.broadcast_channel.channel import Topic
@@ -228,7 +227,7 @@ class MessageBasedAppGenerator(BaseAppGenerator):
                     transfer_method=file.transfer_method,
                     belongs_to=MessageFileBelongsTo.USER,
                     url=file.remote_url,
-                    upload_file_id=resolve_file_record_id(file.reference),
+                    upload_file_id=file.related_id,
                     created_by_role=(CreatorUserRole.ACCOUNT if account_id else CreatorUserRole.END_USER),
                     created_by=account_id or end_user_id or "",
                 )

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

@@ -10,8 +10,6 @@ from collections.abc import Generator, Mapping
 from typing import Any, Literal, Union, cast, overload
 
 from flask import Flask, current_app
-from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
-from graphon.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader
 from pydantic import ValidationError
 from sqlalchemy import select
 from sqlalchemy.orm import Session, sessionmaker
@@ -20,7 +18,6 @@ import contexts
 from configs import dify_config
 from core.app.apps.base_app_generator import BaseAppGenerator
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
-from core.app.apps.draft_variable_saver import DraftVariableSaverFactory
 from core.app.apps.exc import GenerateTaskStoppedError
 from core.app.apps.pipeline.pipeline_config_manager import PipelineConfigManager
 from core.app.apps.pipeline.pipeline_queue_manager import PipelineQueueManager
@@ -37,11 +34,12 @@ from core.datasource.entities.datasource_entities import (
 from core.datasource.online_drive.online_drive_plugin import OnlineDriveDatasourcePlugin
 from core.entities.knowledge_entities import PipelineDataset, PipelineDocument
 from core.rag.index_processor.constant.built_in_field import BuiltInField
-from core.repositories.factory import (
-    DifyCoreRepositoryFactory,
-    WorkflowExecutionRepository,
-    WorkflowNodeExecutionRepository,
-)
+from core.repositories.factory import DifyCoreRepositoryFactory
+from dify_graph.model_runtime.errors.invoke import InvokeAuthorizationError
+from dify_graph.repositories.draft_variable_repository import DraftVariableSaverFactory
+from dify_graph.repositories.workflow_execution_repository import WorkflowExecutionRepository
+from dify_graph.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
+from dify_graph.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader
 from extensions.ext_database import db
 from libs.flask_utils import preserve_flask_contexts
 from models import Account, EndUser, Workflow, WorkflowNodeExecutionTriggeredFrom