[autofix.ci] apply automated fixes

.agents/skills/component-refactoring/SKILL.md

@@ -367,7 +367,7 @@ For each extraction:
   ┌────────────────────────────────────────┐
   │ 1. Extract code                        │
   │ 2. Run: pnpm lint:fix                  │
-  │ 3. Run: pnpm type-check                │
+  │ 3. Run: pnpm type-check:tsgo           │
   │ 4. Run: pnpm test                      │
   │ 5. Test functionality manually         │
   │ 6. PASS? → Next extraction             │

.agents/skills/frontend-testing/references/checklist.md

@@ -127,7 +127,7 @@ For the current file being tested:
 - [ ] Run full directory test: `pnpm test path/to/directory/`
 - [ ] Check coverage report: `pnpm test:coverage`
 - [ ] Run `pnpm lint:fix` on all test files
-- [ ] Run `pnpm type-check`
+- [ ] Run `pnpm type-check:tsgo`
 
 ## Common Issues to Watch
 

.github/workflows/api-tests.yml

@@ -16,7 +16,7 @@ concurrency:
 jobs:
   api-unit:
     name: API Unit Tests
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     env:
       COVERAGE_FILE: coverage-unit
     defaults:
@@ -62,7 +62,7 @@ jobs:
 
   api-integration:
     name: API Integration Tests
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     env:
       COVERAGE_FILE: coverage-integration
       STORAGE_TYPE: opendal
@@ -137,7 +137,7 @@ jobs:
 
   api-coverage:
     name: API Coverage
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     needs:
       - api-unit
       - api-integration

.github/workflows/autofix.yml

@@ -13,7 +13,7 @@ permissions:
 jobs:
   autofix:
     if: github.repository == 'langgenius/dify'
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     steps:
       - name: Complete merge group check
         if: github.event_name == 'merge_group'

.github/workflows/build-push.yml

@@ -26,9 +26,6 @@ jobs:
   build:
     runs-on: ${{ matrix.runs_on }}
     if: github.repository == 'langgenius/dify'
-    permissions:
-      contents: read
-      id-token: write
     strategy:
       matrix:
         include:
@@ -38,28 +35,28 @@ jobs:
             build_context: "{{defaultContext}}:api"
             file: "Dockerfile"
             platform: linux/amd64
-            runs_on: depot-ubuntu-24.04-4
+            runs_on: ubuntu-latest
           - service_name: "build-api-arm64"
             image_name_env: "DIFY_API_IMAGE_NAME"
             artifact_context: "api"
             build_context: "{{defaultContext}}:api"
             file: "Dockerfile"
             platform: linux/arm64
-            runs_on: depot-ubuntu-24.04-4
+            runs_on: ubuntu-24.04-arm
           - service_name: "build-web-amd64"
             image_name_env: "DIFY_WEB_IMAGE_NAME"
             artifact_context: "web"
             build_context: "{{defaultContext}}"
             file: "web/Dockerfile"
             platform: linux/amd64
-            runs_on: depot-ubuntu-24.04-4
+            runs_on: ubuntu-latest
           - service_name: "build-web-arm64"
             image_name_env: "DIFY_WEB_IMAGE_NAME"
             artifact_context: "web"
             build_context: "{{defaultContext}}"
             file: "web/Dockerfile"
             platform: linux/arm64
-            runs_on: depot-ubuntu-24.04-4
+            runs_on: ubuntu-24.04-arm
 
     steps:
       - name: Prepare
@@ -73,8 +70,8 @@ jobs:
           username: ${{ env.DOCKERHUB_USER }}
           password: ${{ env.DOCKERHUB_TOKEN }}
 
-      - name: Set up Depot CLI
-        uses: depot/setup-action@v1
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@4d04d5d9486b7bd6fa91e7baf45bbb4f8b9deedd # v4.0.0
 
       - name: Extract metadata for Docker
         id: meta
@@ -84,15 +81,16 @@ jobs:
 
       - name: Build Docker image
         id: build
-        uses: depot/build-push-action@v1
+        uses: docker/build-push-action@bcafcacb16a39f128d818304e6c9c0c18556b85f # v7.1.0
         with:
-          project: ${{ vars.DEPOT_PROJECT_ID }}
           context: ${{ matrix.build_context }}
           file: ${{ matrix.file }}
           platforms: ${{ matrix.platform }}
           build-args: COMMIT_SHA=${{ fromJSON(steps.meta.outputs.json).labels['org.opencontainers.image.revision'] }}
           labels: ${{ steps.meta.outputs.labels }}
           outputs: type=image,name=${{ env[matrix.image_name_env] }},push-by-digest=true,name-canonical=true,push=true
+          cache-from: type=gha,scope=${{ matrix.service_name }}
+          cache-to: type=gha,mode=max,scope=${{ matrix.service_name }}
 
       - name: Export digest
         env:
@@ -110,33 +108,9 @@ jobs:
           if-no-files-found: error
           retention-days: 1
 
-  fork-build-validate:
-    if: github.repository != 'langgenius/dify'
-    runs-on: ubuntu-24.04
-    strategy:
-      matrix:
-        include:
-          - service_name: "validate-api-amd64"
-            build_context: "{{defaultContext}}:api"
-            file: "Dockerfile"
-          - service_name: "validate-web-amd64"
-            build_context: "{{defaultContext}}"
-            file: "web/Dockerfile"
-    steps:
-      - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@98e3b2c9eab4f4f98a95c0c0a3ea5e5e672fd2a8 # v3.10.0
-
-      - name: Validate Docker image
-        uses: docker/build-push-action@5cd29d66b4a8d8e6f4d5dfe2e9329f0b1d446289 # v6.18.0
-        with:
-          push: false
-          context: ${{ matrix.build_context }}
-          file: ${{ matrix.file }}
-          platforms: linux/amd64
-
   create-manifest:
     needs: build
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     if: github.repository == 'langgenius/dify'
     strategy:
       matrix:

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

@@ -9,7 +9,7 @@ concurrency:
 
 jobs:
   db-migration-test-postgres:
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
 
     steps:
       - name: Checkout code
@@ -59,7 +59,7 @@ jobs:
         run: uv run --directory api flask upgrade-db
 
   db-migration-test-mysql:
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
 
     steps:
       - name: Checkout code
@@ -110,28 +110,6 @@ jobs:
           sed -i 's/DB_PORT=5432/DB_PORT=3306/' .env
           sed -i 's/DB_USERNAME=postgres/DB_USERNAME=root/' .env
 
-      # hoverkraft-tech/compose-action@v2.6.0 only waits for `docker compose up -d`
-      # to return (container processes started); it does not wait on healthcheck
-      # status. mysql:8.0's first-time init takes 15-30s, so without an explicit
-      # wait the migration runs while InnoDB is still initialising and gets
-      # killed with "Lost connection during query". Poll a real SELECT until it
-      # succeeds.
-      - name: Wait for MySQL to accept queries
-        run: |
-          set +e
-          for i in $(seq 1 60); do
-            if docker run --rm --network host mysql:8.0 \
-                mysql -h 127.0.0.1 -P 3306 -uroot -pdifyai123456 \
-                -e 'SELECT 1' >/dev/null 2>&1; then
-              echo "MySQL ready after ${i}s"
-              exit 0
-            fi
-            sleep 1
-          done
-          echo "MySQL not ready after 60s; dumping container logs:"
-          docker compose -f docker/docker-compose.middleware.yaml --profile mysql logs --tail=200 db_mysql
-          exit 1
-
       - name: Run DB Migration
         env:
           DEBUG: true

.github/workflows/deploy-agent-dev.yml

@@ -13,7 +13,7 @@ on:
 
 jobs:
   deploy:
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     if: |
       github.event.workflow_run.conclusion == 'success' &&
       github.event.workflow_run.head_branch == 'deploy/agent-dev'

.github/workflows/deploy-dev.yml

@@ -10,7 +10,7 @@ on:
 
 jobs:
   deploy:
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     if: |
       github.event.workflow_run.conclusion == 'success' &&
       github.event.workflow_run.head_branch == 'deploy/dev'

.github/workflows/deploy-enterprise.yml

@@ -13,7 +13,7 @@ on:
 
 jobs:
   deploy:
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     if: |
       github.event.workflow_run.conclusion == 'success' &&
       github.event.workflow_run.head_branch == 'deploy/enterprise'

.github/workflows/deploy-hitl.yml

@@ -10,7 +10,7 @@ on:
 
 jobs:
   deploy:
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     if: |
       github.event.workflow_run.conclusion == 'success' &&
       github.event.workflow_run.head_branch == 'build/feat/hitl'

.github/workflows/docker-build.yml

@@ -14,69 +14,40 @@ concurrency:
 
 jobs:
   build-docker:
-    if: github.event.pull_request.head.repo.full_name == github.repository
     runs-on: ${{ matrix.runs_on }}
-    permissions:
-      contents: read
-      id-token: write
     strategy:
       matrix:
         include:
           - service_name: "api-amd64"
             platform: linux/amd64
-            runs_on: depot-ubuntu-24.04-4
+            runs_on: ubuntu-latest
             context: "{{defaultContext}}:api"
             file: "Dockerfile"
           - service_name: "api-arm64"
             platform: linux/arm64
-            runs_on: depot-ubuntu-24.04-4
+            runs_on: ubuntu-24.04-arm
             context: "{{defaultContext}}:api"
             file: "Dockerfile"
           - service_name: "web-amd64"
             platform: linux/amd64
-            runs_on: depot-ubuntu-24.04-4
+            runs_on: ubuntu-latest
             context: "{{defaultContext}}"
             file: "web/Dockerfile"
           - service_name: "web-arm64"
             platform: linux/arm64
-            runs_on: depot-ubuntu-24.04-4
-            context: "{{defaultContext}}"
-            file: "web/Dockerfile"
-    steps:
-      - name: Set up Depot CLI
-        uses: depot/setup-action@v1
-
-      - name: Build Docker Image
-        uses: depot/build-push-action@v1
-        with:
-          project: ${{ vars.DEPOT_PROJECT_ID }}
-          push: false
-          context: ${{ matrix.context }}
-          file: ${{ matrix.file }}
-          platforms: ${{ matrix.platform }}
-
-  build-docker-fork:
-    if: github.event.pull_request.head.repo.full_name != github.repository
-    runs-on: ubuntu-24.04
-    permissions:
-      contents: read
-    strategy:
-      matrix:
-        include:
-          - service_name: "api-amd64"
-            context: "{{defaultContext}}:api"
-            file: "Dockerfile"
-          - service_name: "web-amd64"
+            runs_on: ubuntu-24.04-arm
             context: "{{defaultContext}}"
             file: "web/Dockerfile"
     steps:
       - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@98e3b2c9eab4f4f98a95c0c0a3ea5e5e672fd2a8 # v3.10.0
+        uses: docker/setup-buildx-action@4d04d5d9486b7bd6fa91e7baf45bbb4f8b9deedd # v4.0.0
 
       - name: Build Docker Image
-        uses: docker/build-push-action@5cd29d66b4a8d8e6f4d5dfe2e9329f0b1d446289 # v6.18.0
+        uses: docker/build-push-action@bcafcacb16a39f128d818304e6c9c0c18556b85f # v7.1.0
         with:
           push: false
           context: ${{ matrix.context }}
           file: ${{ matrix.file }}
-          platforms: linux/amd64
+          platforms: ${{ matrix.platform }}
+          cache-from: type=gha
+          cache-to: type=gha,mode=max

.github/workflows/labeler.yml

@@ -7,7 +7,7 @@ jobs:
     permissions:
       contents: read
       pull-requests: write
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     steps:
       - uses: actions/labeler@634933edcd8ababfe52f92936142cc22ac488b1b # v6.0.1
         with:

.github/workflows/main-ci.yml

@@ -23,7 +23,7 @@ concurrency:
 jobs:
   pre_job:
     name: Skip Duplicate Checks
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     outputs:
       should_skip: ${{ steps.skip_check.outputs.should_skip || 'false' }}
     steps:
@@ -39,7 +39,7 @@ jobs:
     name: Check Changed Files
     needs: pre_job
     if: needs.pre_job.outputs.should_skip != 'true'
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     outputs:
       api-changed: ${{ steps.changes.outputs.api }}
       e2e-changed: ${{ steps.changes.outputs.e2e }}
@@ -141,7 +141,7 @@ jobs:
       - pre_job
       - check-changes
     if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.api-changed != 'true'
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     steps:
       - name: Report skipped API tests
         run: echo "No API-related changes detected; skipping API tests."
@@ -154,7 +154,7 @@ jobs:
       - check-changes
       - api-tests-run
       - api-tests-skip
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     steps:
       - name: Finalize API Tests status
         env:
@@ -201,7 +201,7 @@ jobs:
       - pre_job
       - check-changes
     if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.web-changed != 'true'
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     steps:
       - name: Report skipped web tests
         run: echo "No web-related changes detected; skipping web tests."
@@ -214,7 +214,7 @@ jobs:
       - check-changes
       - web-tests-run
       - web-tests-skip
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     steps:
       - name: Finalize Web Tests status
         env:
@@ -260,7 +260,7 @@ jobs:
       - pre_job
       - check-changes
     if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.e2e-changed != 'true'
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     steps:
       - name: Report skipped web full-stack e2e
         run: echo "No E2E-related changes detected; skipping web full-stack E2E."
@@ -273,7 +273,7 @@ jobs:
       - check-changes
       - web-e2e-run
       - web-e2e-skip
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     steps:
       - name: Finalize Web Full-Stack E2E status
         env:
@@ -325,7 +325,7 @@ jobs:
       - pre_job
       - check-changes
     if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.vdb-changed != 'true'
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     steps:
       - name: Report skipped VDB tests
         run: echo "No VDB-related changes detected; skipping VDB tests."
@@ -338,7 +338,7 @@ jobs:
       - check-changes
       - vdb-tests-run
       - vdb-tests-skip
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     steps:
       - name: Finalize VDB Tests status
         env:
@@ -384,7 +384,7 @@ jobs:
       - pre_job
       - check-changes
     if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.migration-changed != 'true'
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     steps:
       - name: Report skipped DB migration tests
         run: echo "No migration-related changes detected; skipping DB migration tests."
@@ -397,7 +397,7 @@ jobs:
       - check-changes
       - db-migration-test-run
       - db-migration-test-skip
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     steps:
       - name: Finalize DB Migration Test status
         env:

.github/workflows/pyrefly-diff-comment.yml

@@ -12,7 +12,7 @@ permissions: {}
 jobs:
   comment:
     name: Comment PR with pyrefly diff
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     permissions:
       actions: read
       contents: read

.github/workflows/pyrefly-diff.yml

@@ -10,7 +10,7 @@ permissions:
 
 jobs:
   pyrefly-diff:
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     permissions:
       contents: read
       issues: write

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

@@ -12,7 +12,7 @@ permissions: {}
 jobs:
   comment:
     name: Comment PR with type coverage
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     permissions:
       actions: read
       contents: read

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

@@ -10,7 +10,7 @@ permissions:
 
 jobs:
   pyrefly-type-coverage:
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     permissions:
       contents: read
       issues: write

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

@@ -16,7 +16,7 @@ jobs:
     name: Validate PR title
     permissions:
       pull-requests: read
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     steps:
       - name: Complete merge group check
         if: github.event_name == 'merge_group'

.github/workflows/stale.yml

@@ -12,7 +12,7 @@ on:
 jobs:
   stale:
 
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     permissions:
       issues: write
       pull-requests: write

.github/workflows/style.yml

@@ -15,7 +15,7 @@ permissions:
 jobs:
   python-style:
     name: Python Style
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
 
     steps:
       - name: Checkout code
@@ -57,7 +57,7 @@ jobs:
 
   web-style:
     name: Web Style
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     defaults:
       run:
         working-directory: ./web
@@ -110,8 +110,6 @@ jobs:
       - name: Web tsslint
         if: steps.changed-files.outputs.any_changed == 'true'
         working-directory: ./web
-        env:
-          NODE_OPTIONS: --max-old-space-size=4096
         run: vp run lint:tss
 
       - name: Web type check
@@ -133,7 +131,7 @@ jobs:
 
   superlinter:
     name: SuperLinter
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
 
     steps:
       - name: Checkout code

.github/workflows/tool-test-sdks.yaml

@@ -18,7 +18,7 @@ concurrency:
 jobs:
   build:
     name: unit test for Node.js SDK
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
 
     defaults:
       run:

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

@@ -35,7 +35,7 @@ concurrency:
 jobs:
   translate:
     if: github.repository == 'langgenius/dify'
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     timeout-minutes: 120
 
     steps:
@@ -158,7 +158,7 @@ jobs:
 
       - name: Run Claude Code for Translation Sync
         if: steps.context.outputs.CHANGED_FILES != ''
-        uses: anthropics/claude-code-action@567fe954a4527e81f132d87d1bdbcc94f7737434 # v1.0.107
+        uses: anthropics/claude-code-action@38ec876110f9fbf8b950c79f534430740c3ac009 # v1.0.101
         with:
           anthropic_api_key: ${{ secrets.ANTHROPIC_API_KEY }}
           github_token: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/trigger-i18n-sync.yml

@@ -16,7 +16,7 @@ concurrency:
 jobs:
   trigger:
     if: github.repository == 'langgenius/dify'
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     timeout-minutes: 5
 
     steps:

.github/workflows/vdb-tests-full.yml

@@ -16,7 +16,7 @@ jobs:
   test:
     name: Full VDB Tests
     if: github.repository == 'langgenius/dify'
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     strategy:
       matrix:
         python-version:

.github/workflows/vdb-tests.yml

@@ -13,7 +13,7 @@ concurrency:
 jobs:
   test:
     name: VDB Smoke Tests
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
     strategy:
       matrix:
         python-version:

.github/workflows/web-e2e.yml

@@ -13,7 +13,7 @@ concurrency:
 jobs:
   test:
     name: Web Full-Stack E2E
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: ubuntu-latest
     defaults:
       run:
         shell: bash

.github/workflows/web-tests.yml

@@ -16,7 +16,7 @@ concurrency:
 jobs:
   test:
     name: Web Tests (${{ matrix.shardIndex }}/${{ matrix.shardTotal }})
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: ubuntu-latest
     env:
       VITEST_COVERAGE_SCOPE: app-components
     strategy:
@@ -54,7 +54,7 @@ jobs:
     name: Merge Test Reports
     if: ${{ !cancelled() }}
     needs: [test]
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: ubuntu-latest
     env:
       CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
     defaults:
@@ -92,7 +92,7 @@ jobs:
 
   dify-ui-test:
     name: dify-ui Tests
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: ubuntu-latest
     env:
       CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
     defaults:

AGENTS.md

@@ -30,7 +30,7 @@ The codebase is split into:
 ## Language Style
 
 - **Python**: Keep type hints on functions and attributes, and implement relevant special methods (e.g., `__repr__`, `__str__`). Prefer `TypedDict` over `dict` or `Mapping` for type safety and better code documentation.
-- **TypeScript**: Use the strict config, rely on ESLint (`pnpm lint:fix` preferred) plus `pnpm type-check`, and avoid `any` types.
+- **TypeScript**: Use the strict config, rely on ESLint (`pnpm lint:fix` preferred) plus `pnpm type-check:tsgo`, and avoid `any` types.
 
 ## General Practices
 

README.md

@@ -139,6 +139,19 @@ Star Dify on GitHub and be instantly notified of new releases.
 
 If you need to customize the configuration, please refer to the comments in our [.env.example](docker/.env.example) file and update the corresponding values in your `.env` file. Additionally, you might need to make adjustments to the `docker-compose.yaml` file itself, such as changing image versions, port mappings, or volume mounts, based on your specific deployment environment and requirements. After making any changes, please re-run `docker compose up -d`. You can find the full list of available environment variables [here](https://docs.dify.ai/getting-started/install-self-hosted/environments).
 
+#### Customizing Suggested Questions
+
+You can now customize the "Suggested Questions After Answer" feature to better fit your use case. For example, to generate longer, more technical questions:
+
+```bash
+# In your .env file
+SUGGESTED_QUESTIONS_PROMPT='Please help me predict the five most likely technical follow-up questions a developer would ask. Focus on implementation details, best practices, and architecture considerations. Keep each question between 40-60 characters. Output must be JSON array: ["question1","question2","question3","question4","question5"]'
+SUGGESTED_QUESTIONS_MAX_TOKENS=512
+SUGGESTED_QUESTIONS_TEMPERATURE=0.3
+```
+
+See the [Suggested Questions Configuration Guide](docs/suggested-questions-configuration.md) for detailed examples and usage instructions.
+
 ### Metrics Monitoring with Grafana
 
 Import the dashboard to Grafana, using Dify's PostgreSQL database as data source, to monitor metrics in granularity of apps, tenants, messages, and more.
@@ -147,7 +160,7 @@ Import the dashboard to Grafana, using Dify's PostgreSQL database as data source
 
 ### Deployment with Kubernetes
 
-If you'd like to configure a highly available setup, there are community-contributed [Helm Charts](https://helm.sh/) and YAML files which allow Dify to be deployed on Kubernetes.
+If you'd like to configure a highly-available setup, there are community-contributed [Helm Charts](https://helm.sh/) and YAML files which allow Dify to be deployed on Kubernetes.
 
 - [Helm Chart by @LeoQuote](https://github.com/douban/charts/tree/master/charts/dify)
 - [Helm Chart by @BorisPolonsky](https://github.com/BorisPolonsky/dify-helm)

api/.env.example

@@ -659,11 +659,6 @@ INNER_API_KEY_FOR_PLUGIN=QaHbTe77CtuXmsfyhR7+vRjI/+XbV1AaFy691iy+kGDv2Jvy0/eAh8Y
 MARKETPLACE_ENABLED=true
 MARKETPLACE_API_URL=https://marketplace.dify.ai
 
-# Creators Platform configuration
-CREATORS_PLATFORM_FEATURES_ENABLED=true
-CREATORS_PLATFORM_API_URL=https://creators.dify.ai
-CREATORS_PLATFORM_OAUTH_CLIENT_ID=
-
 # Endpoint configuration
 ENDPOINT_URL_TEMPLATE=http://localhost:5002/e/{hook_id}
 
@@ -714,6 +709,22 @@ SWAGGER_UI_PATH=/swagger-ui.html
 # Set to false to export dataset IDs as plain text for easier cross-environment import
 DSL_EXPORT_ENCRYPT_DATASET_ID=true
 
+# Suggested Questions After Answer Configuration
+# These environment variables allow customization of the suggested questions feature
+#
+# Custom prompt for generating suggested questions (optional)
+# If not set, uses the default prompt that generates 3 questions under 20 characters each
+# Example: "Please help me predict the five most likely technical follow-up questions a developer would ask. Focus on implementation details, best practices, and architecture considerations. Keep each question between 40-60 characters. Output must be JSON array: [\"question1\",\"question2\",\"question3\",\"question4\",\"question5\"]"
+# SUGGESTED_QUESTIONS_PROMPT=
+
+# Maximum number of tokens for suggested questions generation (default: 256)
+# Adjust this value for longer questions or more questions
+# SUGGESTED_QUESTIONS_MAX_TOKENS=256
+
+# Temperature for suggested questions generation (default: 0.0)
+# Higher values (0.5-1.0) produce more creative questions, lower values (0.0-0.3) produce more focused questions
+# SUGGESTED_QUESTIONS_TEMPERATURE=0
+
 # Tenant isolated task queue configuration
 TENANT_ISOLATED_TASK_CONCURRENCY=1
 

api/README.md

@@ -101,11 +101,3 @@ The scripts resolve paths relative to their location, so you can run them from a
    uv run ruff format ./        # Format code
    uv run basedpyright .        # Type checking
    ```
-
-## Generate TS stub
-
-```
-uv run dev/generate_swagger_specs.py --output-dir openapi
-```
-
-use https://jsontotable.org/openapi-to-typescript to convert to typescript

api/commands/plugin.py

@@ -11,7 +11,7 @@ from configs import dify_config
 from core.helper import encrypter
 from core.plugin.entities.plugin_daemon import CredentialType
 from core.plugin.impl.plugin import PluginInstaller
-from core.tools.utils.system_encryption import encrypt_system_params
+from core.tools.utils.system_oauth_encryption import encrypt_system_oauth_params
 from extensions.ext_database import db
 from models import Tenant
 from models.oauth import DatasourceOauthParamConfig, DatasourceProvider
@@ -44,7 +44,7 @@ def setup_system_tool_oauth_client(provider, client_params):
 
         click.echo(click.style(f"Encrypting client params: {client_params}", fg="yellow"))
         click.echo(click.style(f"Using SECRET_KEY: `{dify_config.SECRET_KEY}`", fg="yellow"))
-        oauth_client_params = encrypt_system_params(client_params_dict)
+        oauth_client_params = encrypt_system_oauth_params(client_params_dict)
         click.echo(click.style("Client params encrypted successfully.", fg="green"))
     except Exception as e:
         click.echo(click.style(f"Error parsing client params: {str(e)}", fg="red"))
@@ -94,7 +94,7 @@ def setup_system_trigger_oauth_client(provider, client_params):
 
         click.echo(click.style(f"Encrypting client params: {client_params}", fg="yellow"))
         click.echo(click.style(f"Using SECRET_KEY: `{dify_config.SECRET_KEY}`", fg="yellow"))
-        oauth_client_params = encrypt_system_params(client_params_dict)
+        oauth_client_params = encrypt_system_oauth_params(client_params_dict)
         click.echo(click.style("Client params encrypted successfully.", fg="green"))
     except Exception as e:
         click.echo(click.style(f"Error parsing client params: {str(e)}", fg="red"))

api/configs/feature/__init__.py

@@ -287,27 +287,6 @@ class MarketplaceConfig(BaseSettings):
     )
 
 
-class CreatorsPlatformConfig(BaseSettings):
-    """
-    Configuration for Creators Platform integration
-    """
-
-    CREATORS_PLATFORM_FEATURES_ENABLED: bool = Field(
-        description="Enable or disable Creators Platform features",
-        default=True,
-    )
-
-    CREATORS_PLATFORM_API_URL: HttpUrl = Field(
-        description="Creators Platform API URL",
-        default=HttpUrl("https://creators.dify.ai"),
-    )
-
-    CREATORS_PLATFORM_OAUTH_CLIENT_ID: str = Field(
-        description="OAuth client ID for Creators Platform integration",
-        default="",
-    )
-
-
 class EndpointConfig(BaseSettings):
     """
     Configuration for various application endpoints and URLs
@@ -1400,7 +1379,6 @@ class FeatureConfig(
     AuthConfig,  # Changed from OAuthConfig to AuthConfig
     BillingConfig,
     CodeExecutionSandboxConfig,
-    CreatorsPlatformConfig,
     TriggerConfig,
     AsyncWorkflowConfig,
     PluginConfig,

api/controllers/common/human_input.py

@@ -1,6 +0,0 @@
-from pydantic import BaseModel, JsonValue
-
-
-class HumanInputFormSubmitPayload(BaseModel):
-    inputs: dict[str, JsonValue]
-    action: str

api/controllers/console/app/app.py

@@ -692,32 +692,6 @@ class AppExportApi(Resource):
         return payload.model_dump(mode="json")
 
 
-@console_ns.route("/apps/<uuid:app_id>/publish-to-creators-platform")
-class AppPublishToCreatorsPlatformApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=None)
-    @edit_permission_required
-    def post(self, app_model):
-        """Publish app to Creators Platform"""
-        from configs import dify_config
-        from core.helper.creators import get_redirect_url, upload_dsl
-
-        if not dify_config.CREATORS_PLATFORM_FEATURES_ENABLED:
-            return {"error": "Creators Platform features are not enabled"}, 403
-
-        current_user, _ = current_account_with_tenant()
-
-        dsl_content = AppDslService.export_dsl(app_model=app_model, include_secret=False)
-        dsl_bytes = dsl_content.encode("utf-8")
-
-        claim_code = upload_dsl(dsl_bytes)
-        redirect_url = get_redirect_url(str(current_user.id), claim_code)
-
-        return {"redirect_url": redirect_url}
-
-
 @console_ns.route("/apps/<uuid:app_id>/name")
 class AppNameApi(Resource):
     @console_ns.doc("check_app_name")

api/controllers/console/datasets/hit_testing_base.py

@@ -38,48 +38,6 @@ class HitTestingPayload(BaseModel):
 
 
 class DatasetsHitTestingBase:
-    @staticmethod
-    def _normalize_hit_testing_query(query: Any) -> str:
-        """Return the user-visible query string from legacy and current response shapes."""
-        if isinstance(query, str):
-            return query
-
-        if isinstance(query, dict):
-            content = query.get("content")
-            if isinstance(content, str):
-                return content
-
-        raise ValueError("Invalid hit testing query response")
-
-    @staticmethod
-    def _normalize_hit_testing_records(records: Any) -> list[dict[str, Any]]:
-        """Coerce nullable collection fields into lists before response validation."""
-        if not isinstance(records, list):
-            return []
-
-        normalized_records: list[dict[str, Any]] = []
-        for record in records:
-            if not isinstance(record, dict):
-                continue
-
-            normalized_record = dict(record)
-            segment = normalized_record.get("segment")
-            if isinstance(segment, dict):
-                normalized_segment = dict(segment)
-                if normalized_segment.get("keywords") is None:
-                    normalized_segment["keywords"] = []
-                normalized_record["segment"] = normalized_segment
-
-            if normalized_record.get("child_chunks") is None:
-                normalized_record["child_chunks"] = []
-
-            if normalized_record.get("files") is None:
-                normalized_record["files"] = []
-
-            normalized_records.append(normalized_record)
-
-        return normalized_records
-
     @staticmethod
     def get_and_validate_dataset(dataset_id: str):
         assert isinstance(current_user, Account)
@@ -117,12 +75,7 @@ class DatasetsHitTestingBase:
                 attachment_ids=args.get("attachment_ids"),
                 limit=10,
             )
-            return {
-                "query": DatasetsHitTestingBase._normalize_hit_testing_query(response.get("query")),
-                "records": DatasetsHitTestingBase._normalize_hit_testing_records(
-                    marshal(response.get("records", []), hit_testing_record_fields)
-                ),
-            }
+            return {"query": response["query"], "records": marshal(response["records"], hit_testing_record_fields)}
         except services.errors.index.IndexNotInitializedError:
             raise DatasetNotInitializedError()
         except ProviderTokenNotInitError as ex:

api/controllers/console/human_input_form.py

@@ -8,10 +8,10 @@ from collections.abc import Generator
 
 from flask import Response, jsonify, request
 from flask_restx import Resource
+from pydantic import BaseModel
 from sqlalchemy import select
 from sqlalchemy.orm import Session, sessionmaker
 
-from controllers.common.human_input import HumanInputFormSubmitPayload
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, setup_required
 from controllers.web.error import InvalidArgumentError, NotFoundError
@@ -20,11 +20,11 @@ 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
-from core.workflow.human_input_policy import HumanInputSurface, is_recipient_type_allowed_for_surface
 from extensions.ext_database import db
 from libs.login import current_account_with_tenant, login_required
 from models import App
 from models.enums import CreatorUserRole
+from models.human_input import RecipientType
 from models.model import AppMode
 from models.workflow import WorkflowRun
 from repositories.factory import DifyAPIRepositoryFactory
@@ -34,6 +34,11 @@ from services.workflow_event_snapshot_service import build_workflow_event_stream
 logger = logging.getLogger(__name__)
 
 
+class HumanInputFormSubmitPayload(BaseModel):
+    inputs: dict
+    action: str
+
+
 def _jsonify_form_definition(form: Form) -> Response:
     payload = form.get_definition().model_dump()
     payload["expiration_time"] = int(form.expiration_time.timestamp())
@@ -51,11 +56,6 @@ class ConsoleHumanInputFormApi(Resource):
         if form.tenant_id != current_tenant_id:
             raise NotFoundError("App not found")
 
-    @staticmethod
-    def _ensure_console_recipient_type(form: Form) -> None:
-        if not is_recipient_type_allowed_for_surface(form.recipient_type, HumanInputSurface.CONSOLE):
-            raise NotFoundError("form not found")
-
     @setup_required
     @login_required
     @account_initialization_required
@@ -99,8 +99,10 @@ class ConsoleHumanInputFormApi(Resource):
             raise NotFoundError(f"form not found, token={form_token}")
 
         self._ensure_console_access(form)
-        self._ensure_console_recipient_type(form)
+
         recipient_type = form.recipient_type
+        if recipient_type not in {RecipientType.CONSOLE, RecipientType.BACKSTAGE}:
+            raise NotFoundError(f"form not found, token={form_token}")
         # The type checker is not smart enought to validate the following invariant.
         # So we need to assert it manually.
         assert recipient_type is not None, "recipient_type cannot be None here."

api/controllers/console/tag/tags.py

@@ -37,11 +37,6 @@ class TagBindingRemovePayload(BaseModel):
     type: TagType = Field(description="Tag type")
 
 
-class TagBindingItemDeletePayload(BaseModel):
-    target_id: str = Field(description="Target ID to unbind tag from")
-    type: TagType = Field(description="Tag type")
-
-
 class TagListQueryParam(BaseModel):
     type: Literal["knowledge", "app", ""] = Field("", description="Tag type filter")
     keyword: str | None = Field(None, description="Search keyword")
@@ -75,7 +70,6 @@ register_schema_models(
     TagBasePayload,
     TagBindingPayload,
     TagBindingRemovePayload,
-    TagBindingItemDeletePayload,
     TagListQueryParam,
     TagResponse,
 )
@@ -158,107 +152,41 @@ class TagUpdateDeleteApi(Resource):
         return "", 204
 
 
-def _require_tag_binding_edit_permission() -> None:
-    """
-    Ensure the current account can edit tag bindings.
-
-    Tag binding operations are allowed for users who can edit resources (app/dataset) within the current tenant.
-    """
-    current_user, _ = current_account_with_tenant()
-    # The role of the current user in the ta table must be admin, owner, editor, or dataset_operator
-    if not (current_user.has_edit_permission or current_user.is_dataset_editor):
-        raise Forbidden()
-
-
-def _create_tag_bindings() -> tuple[dict[str, str], int]:
-    _require_tag_binding_edit_permission()
-
-    payload = TagBindingPayload.model_validate(console_ns.payload or {})
-    TagService.save_tag_binding(
-        TagBindingCreatePayload(
-            tag_ids=payload.tag_ids,
-            target_id=payload.target_id,
-            type=payload.type,
-        )
-    )
-    return {"result": "success"}, 200
-
-
-def _remove_tag_binding() -> tuple[dict[str, str], int]:
-    _require_tag_binding_edit_permission()
-
-    payload = TagBindingRemovePayload.model_validate(console_ns.payload or {})
-    TagService.delete_tag_binding(
-        TagBindingDeletePayload(
-            tag_id=payload.tag_id,
-            target_id=payload.target_id,
-            type=payload.type,
-        )
-    )
-    return {"result": "success"}, 200
-
-
-@console_ns.route("/tag-bindings")
-class TagBindingCollectionApi(Resource):
-    """Canonical collection resource for tag binding creation."""
-
-    @console_ns.doc("create_tag_binding")
+@console_ns.route("/tag-bindings/create")
+class TagBindingCreateApi(Resource):
     @console_ns.expect(console_ns.models[TagBindingPayload.__name__])
     @setup_required
     @login_required
     @account_initialization_required
     def post(self):
-        return _create_tag_bindings()
+        current_user, _ = current_account_with_tenant()
+        # The role of the current user in the ta table must be admin, owner, editor, or dataset_operator
+        if not (current_user.has_edit_permission or current_user.is_dataset_editor):
+            raise Forbidden()
 
-
-@console_ns.route("/tag-bindings/<uuid:id>")
-class TagBindingItemApi(Resource):
-    """Canonical item resource for tag binding deletion."""
-
-    @console_ns.doc("delete_tag_binding")
-    @console_ns.doc(params={"id": "Tag ID"})
-    @console_ns.expect(console_ns.models[TagBindingItemDeletePayload.__name__])
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def delete(self, id):
-        _require_tag_binding_edit_permission()
-        payload = TagBindingItemDeletePayload.model_validate(console_ns.payload or {})
-        TagService.delete_tag_binding(
-            TagBindingDeletePayload(
-                tag_id=str(id),
-                target_id=payload.target_id,
-                type=payload.type,
-            )
+        payload = TagBindingPayload.model_validate(console_ns.payload or {})
+        TagService.save_tag_binding(
+            TagBindingCreatePayload(tag_ids=payload.tag_ids, target_id=payload.target_id, type=payload.type)
         )
+
         return {"result": "success"}, 200
 
 
-@console_ns.route("/tag-bindings/create")
-class DeprecatedTagBindingCreateApi(Resource):
-    """Deprecated verb-based alias for tag binding creation."""
-
-    @console_ns.doc("create_tag_binding_deprecated")
-    @console_ns.doc(deprecated=True)
-    @console_ns.doc(description="Deprecated legacy alias. Use POST /tag-bindings instead.")
-    @console_ns.expect(console_ns.models[TagBindingPayload.__name__])
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def post(self):
-        return _create_tag_bindings()
-
-
 @console_ns.route("/tag-bindings/remove")
-class DeprecatedTagBindingRemoveApi(Resource):
-    """Deprecated verb-based alias for tag binding deletion."""
-
-    @console_ns.doc("delete_tag_binding_deprecated")
-    @console_ns.doc(deprecated=True)
-    @console_ns.doc(description="Deprecated legacy alias. Use DELETE /tag-bindings/{id} instead.")
+class TagBindingDeleteApi(Resource):
     @console_ns.expect(console_ns.models[TagBindingRemovePayload.__name__])
     @setup_required
     @login_required
     @account_initialization_required
     def post(self):
-        return _remove_tag_binding()
+        current_user, _ = current_account_with_tenant()
+        # The role of the current user in the ta table must be admin, owner, editor, or dataset_operator
+        if not (current_user.has_edit_permission or current_user.is_dataset_editor):
+            raise Forbidden()
+
+        payload = TagBindingRemovePayload.model_validate(console_ns.payload or {})
+        TagService.delete_tag_binding(
+            TagBindingDeletePayload(tag_id=payload.tag_id, target_id=payload.target_id, type=payload.type)
+        )
+
+        return {"result": "success"}, 200

api/controllers/console/workspace/endpoint.py

@@ -1,11 +1,3 @@
-"""Console workspace endpoint controllers.
-
-This module exposes workspace-scoped plugin endpoint management APIs. The
-canonical write routes follow resource-oriented paths, while the historical
-verb-based aliases stay available as deprecated resources so OpenAPI metadata
-marks only the legacy paths as deprecated.
-"""
-
 from typing import Any
 
 from flask import request
@@ -33,12 +25,7 @@ class EndpointIdPayload(BaseModel):
     endpoint_id: str
 
 
-class EndpointUpdatePayload(BaseModel):
-    settings: dict[str, Any]
-    name: str = Field(min_length=1)
-
-
-class LegacyEndpointUpdatePayload(EndpointIdPayload):
+class EndpointUpdatePayload(EndpointIdPayload):
     settings: dict[str, Any]
     name: str = Field(min_length=1)
 
@@ -89,7 +76,6 @@ register_schema_models(
     EndpointCreatePayload,
     EndpointIdPayload,
     EndpointUpdatePayload,
-    LegacyEndpointUpdatePayload,
     EndpointListQuery,
     EndpointListForPluginQuery,
     EndpointCreateResponse,
@@ -102,60 +88,8 @@ register_schema_models(
 )
 
 
-def _create_endpoint() -> dict[str, bool]:
-    """Create a plugin endpoint for the current workspace."""
-    user, tenant_id = current_account_with_tenant()
-
-    args = EndpointCreatePayload.model_validate(console_ns.payload)
-
-    try:
-        return {
-            "success": EndpointService.create_endpoint(
-                tenant_id=tenant_id,
-                user_id=user.id,
-                plugin_unique_identifier=args.plugin_unique_identifier,
-                name=args.name,
-                settings=args.settings,
-            )
-        }
-    except PluginPermissionDeniedError as e:
-        raise ValueError(e.description) from e
-
-
-def _update_endpoint(endpoint_id: str) -> dict[str, bool]:
-    """Update a plugin endpoint identified by the canonical path parameter."""
-    user, tenant_id = current_account_with_tenant()
-
-    args = EndpointUpdatePayload.model_validate(console_ns.payload)
-
-    return {
-        "success": EndpointService.update_endpoint(
-            tenant_id=tenant_id,
-            user_id=user.id,
-            endpoint_id=endpoint_id,
-            name=args.name,
-            settings=args.settings,
-        )
-    }
-
-
-def _delete_endpoint(endpoint_id: str) -> dict[str, bool]:
-    """Delete a plugin endpoint identified by the canonical path parameter."""
-    user, tenant_id = current_account_with_tenant()
-
-    return {
-        "success": EndpointService.delete_endpoint(
-            tenant_id=tenant_id,
-            user_id=user.id,
-            endpoint_id=endpoint_id,
-        )
-    }
-
-
-@console_ns.route("/workspaces/current/endpoints")
-class EndpointCollectionApi(Resource):
-    """Canonical collection resource for endpoint creation."""
-
+@console_ns.route("/workspaces/current/endpoints/create")
+class EndpointCreateApi(Resource):
     @console_ns.doc("create_endpoint")
     @console_ns.doc(description="Create a new plugin endpoint")
     @console_ns.expect(console_ns.models[EndpointCreatePayload.__name__])
@@ -170,33 +104,22 @@ class EndpointCollectionApi(Resource):
     @is_admin_or_owner_required
     @account_initialization_required
     def post(self):
-        return _create_endpoint()
+        user, tenant_id = current_account_with_tenant()
 
+        args = EndpointCreatePayload.model_validate(console_ns.payload)
 
-@console_ns.route("/workspaces/current/endpoints/create")
-class DeprecatedEndpointCreateApi(Resource):
-    """Deprecated verb-based alias for endpoint creation."""
-
-    @console_ns.doc("create_endpoint_deprecated")
-    @console_ns.doc(deprecated=True)
-    @console_ns.doc(
-        description=(
-            "Deprecated legacy alias for creating a plugin endpoint. Use POST /workspaces/current/endpoints instead."
-        )
-    )
-    @console_ns.expect(console_ns.models[EndpointCreatePayload.__name__])
-    @console_ns.response(
-        200,
-        "Endpoint created successfully",
-        console_ns.models[EndpointCreateResponse.__name__],
-    )
-    @console_ns.response(403, "Admin privileges required")
-    @setup_required
-    @login_required
-    @is_admin_or_owner_required
-    @account_initialization_required
-    def post(self):
-        return _create_endpoint()
+        try:
+            return {
+                "success": EndpointService.create_endpoint(
+                    tenant_id=tenant_id,
+                    user_id=user.id,
+                    plugin_unique_identifier=args.plugin_unique_identifier,
+                    name=args.name,
+                    settings=args.settings,
+                )
+            }
+        except PluginPermissionDeniedError as e:
+            raise ValueError(e.description) from e
 
 
 @console_ns.route("/workspaces/current/endpoints/list")
@@ -267,56 +190,10 @@ class EndpointListForSinglePluginApi(Resource):
         )
 
 
-@console_ns.route("/workspaces/current/endpoints/<string:id>")
-class EndpointItemApi(Resource):
-    """Canonical item resource for endpoint updates and deletion."""
-
+@console_ns.route("/workspaces/current/endpoints/delete")
+class EndpointDeleteApi(Resource):
     @console_ns.doc("delete_endpoint")
     @console_ns.doc(description="Delete a plugin endpoint")
-    @console_ns.doc(params={"id": {"description": "Endpoint ID", "type": "string", "required": True}})
-    @console_ns.response(
-        200,
-        "Endpoint deleted successfully",
-        console_ns.models[EndpointDeleteResponse.__name__],
-    )
-    @console_ns.response(403, "Admin privileges required")
-    @setup_required
-    @login_required
-    @is_admin_or_owner_required
-    @account_initialization_required
-    def delete(self, id: str):
-        return _delete_endpoint(endpoint_id=id)
-
-    @console_ns.doc("update_endpoint")
-    @console_ns.doc(description="Update a plugin endpoint")
-    @console_ns.expect(console_ns.models[EndpointUpdatePayload.__name__])
-    @console_ns.doc(params={"id": {"description": "Endpoint ID", "type": "string", "required": True}})
-    @console_ns.response(
-        200,
-        "Endpoint updated successfully",
-        console_ns.models[EndpointUpdateResponse.__name__],
-    )
-    @console_ns.response(403, "Admin privileges required")
-    @setup_required
-    @login_required
-    @is_admin_or_owner_required
-    @account_initialization_required
-    def patch(self, id: str):
-        return _update_endpoint(endpoint_id=id)
-
-
-@console_ns.route("/workspaces/current/endpoints/delete")
-class DeprecatedEndpointDeleteApi(Resource):
-    """Deprecated verb-based alias for endpoint deletion."""
-
-    @console_ns.doc("delete_endpoint_deprecated")
-    @console_ns.doc(deprecated=True)
-    @console_ns.doc(
-        description=(
-            "Deprecated legacy alias for deleting a plugin endpoint. "
-            "Use DELETE /workspaces/current/endpoints/{id} instead."
-        )
-    )
     @console_ns.expect(console_ns.models[EndpointIdPayload.__name__])
     @console_ns.response(
         200,
@@ -329,23 +206,22 @@ class DeprecatedEndpointDeleteApi(Resource):
     @is_admin_or_owner_required
     @account_initialization_required
     def post(self):
+        user, tenant_id = current_account_with_tenant()
+
         args = EndpointIdPayload.model_validate(console_ns.payload)
-        return _delete_endpoint(endpoint_id=args.endpoint_id)
+
+        return {
+            "success": EndpointService.delete_endpoint(
+                tenant_id=tenant_id, user_id=user.id, endpoint_id=args.endpoint_id
+            )
+        }
 
 
 @console_ns.route("/workspaces/current/endpoints/update")
-class DeprecatedEndpointUpdateApi(Resource):
-    """Deprecated verb-based alias for endpoint updates."""
-
-    @console_ns.doc("update_endpoint_deprecated")
-    @console_ns.doc(deprecated=True)
-    @console_ns.doc(
-        description=(
-            "Deprecated legacy alias for updating a plugin endpoint. "
-            "Use PATCH /workspaces/current/endpoints/{id} instead."
-        )
-    )
-    @console_ns.expect(console_ns.models[LegacyEndpointUpdatePayload.__name__])
+class EndpointUpdateApi(Resource):
+    @console_ns.doc("update_endpoint")
+    @console_ns.doc(description="Update a plugin endpoint")
+    @console_ns.expect(console_ns.models[EndpointUpdatePayload.__name__])
     @console_ns.response(
         200,
         "Endpoint updated successfully",
@@ -357,8 +233,19 @@ class DeprecatedEndpointUpdateApi(Resource):
     @is_admin_or_owner_required
     @account_initialization_required
     def post(self):
-        args = LegacyEndpointUpdatePayload.model_validate(console_ns.payload)
-        return _update_endpoint(endpoint_id=args.endpoint_id)
+        user, tenant_id = current_account_with_tenant()
+
+        args = EndpointUpdatePayload.model_validate(console_ns.payload)
+
+        return {
+            "success": EndpointService.update_endpoint(
+                tenant_id=tenant_id,
+                user_id=user.id,
+                endpoint_id=args.endpoint_id,
+                name=args.name,
+                settings=args.settings,
+            )
+        }
 
 
 @console_ns.route("/workspaces/current/endpoints/enable")

api/controllers/service_api/__init__.py

@@ -23,11 +23,9 @@ from .app import (
     conversation,
     file,
     file_preview,
-    human_input_form,
     message,
     site,
     workflow,
-    workflow_events,
 )
 from .dataset import (
     dataset,
@@ -52,7 +50,6 @@ __all__ = [
     "file",
     "file_preview",
     "hit_testing",
-    "human_input_form",
     "index",
     "message",
     "metadata",
@@ -61,7 +58,6 @@ __all__ = [
     "segment",
     "site",
     "workflow",
-    "workflow_events",
 ]
 
 api.add_namespace(service_api_ns)

api/controllers/service_api/app/human_input_form.py

@@ -1,137 +0,0 @@
-"""
-Service API human input form endpoints.
-
-This module exposes app-token authenticated APIs for fetching and submitting
-paused human input forms in workflow/chatflow runs.
-"""
-
-import json
-import logging
-from datetime import datetime
-
-from flask import Response
-from flask_restx import Resource
-from werkzeug.exceptions import BadRequest, NotFound
-
-from controllers.common.human_input import HumanInputFormSubmitPayload
-from controllers.common.schema import register_schema_models
-from controllers.service_api import service_api_ns
-from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate_app_token
-from core.workflow.human_input_policy import HumanInputSurface, is_recipient_type_allowed_for_surface
-from extensions.ext_database import db
-from models.model import App, EndUser
-from services.human_input_service import Form, FormNotFoundError, HumanInputService
-
-logger = logging.getLogger(__name__)
-
-
-register_schema_models(service_api_ns, HumanInputFormSubmitPayload)
-
-
-def _stringify_default_values(values: dict[str, object]) -> dict[str, str]:
-    result: dict[str, str] = {}
-    for key, value in values.items():
-        if value is None:
-            result[key] = ""
-        elif isinstance(value, (dict, list)):
-            result[key] = json.dumps(value, ensure_ascii=False)
-        else:
-            result[key] = str(value)
-    return result
-
-
-def _to_timestamp(value: datetime) -> int:
-    return int(value.timestamp())
-
-
-def _jsonify_form_definition(form: Form) -> Response:
-    definition_payload = form.get_definition().model_dump()
-    payload = {
-        "form_content": definition_payload["rendered_content"],
-        "inputs": definition_payload["inputs"],
-        "resolved_default_values": _stringify_default_values(definition_payload["default_values"]),
-        "user_actions": definition_payload["user_actions"],
-        "expiration_time": _to_timestamp(form.expiration_time),
-    }
-    return Response(json.dumps(payload, ensure_ascii=False), mimetype="application/json")
-
-
-def _ensure_form_belongs_to_app(form: Form, app_model: App) -> None:
-    if form.app_id != app_model.id or form.tenant_id != app_model.tenant_id:
-        raise NotFound("Form not found")
-
-
-def _ensure_form_is_allowed_for_service_api(form: Form) -> None:
-    # Keep app-token callers scoped to the public web-form surface; internal HITL
-    # routes must continue to flow through console-only authentication.
-    if not is_recipient_type_allowed_for_surface(form.recipient_type, HumanInputSurface.SERVICE_API):
-        raise NotFound("Form not found")
-
-
-@service_api_ns.route("/form/human_input/<string:form_token>")
-class WorkflowHumanInputFormApi(Resource):
-    @service_api_ns.doc("get_human_input_form")
-    @service_api_ns.doc(description="Get a paused human input form by token")
-    @service_api_ns.doc(params={"form_token": "Human input form token"})
-    @service_api_ns.doc(
-        responses={
-            200: "Form retrieved successfully",
-            401: "Unauthorized - invalid API token",
-            404: "Form not found",
-            412: "Form already submitted or expired",
-        }
-    )
-    @validate_app_token
-    def get(self, app_model: App, form_token: str):
-        service = HumanInputService(db.engine)
-        form = service.get_form_by_token(form_token)
-        if form is None:
-            raise NotFound("Form not found")
-
-        _ensure_form_belongs_to_app(form, app_model)
-        _ensure_form_is_allowed_for_service_api(form)
-        service.ensure_form_active(form)
-        return _jsonify_form_definition(form)
-
-    @service_api_ns.expect(service_api_ns.models[HumanInputFormSubmitPayload.__name__])
-    @service_api_ns.doc("submit_human_input_form")
-    @service_api_ns.doc(description="Submit a paused human input form by token")
-    @service_api_ns.doc(params={"form_token": "Human input form token"})
-    @service_api_ns.doc(
-        responses={
-            200: "Form submitted successfully",
-            400: "Bad request - invalid submission data",
-            401: "Unauthorized - invalid API token",
-            404: "Form not found",
-            412: "Form already submitted or expired",
-        }
-    )
-    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))
-    def post(self, app_model: App, end_user: EndUser, form_token: str):
-        payload = HumanInputFormSubmitPayload.model_validate(service_api_ns.payload or {})
-
-        service = HumanInputService(db.engine)
-        form = service.get_form_by_token(form_token)
-        if form is None:
-            raise NotFound("Form not found")
-
-        _ensure_form_belongs_to_app(form, app_model)
-        _ensure_form_is_allowed_for_service_api(form)
-
-        recipient_type = form.recipient_type
-        if recipient_type is None:
-            logger.warning("Recipient type is None for form, form_id=%s", form.id)
-            raise BadRequest("Form recipient type is invalid")
-
-        try:
-            service.submit_form_by_token(
-                recipient_type=recipient_type,
-                form_token=form_token,
-                selected_action_id=payload.action,
-                form_data=payload.inputs,
-                submission_end_user_id=end_user.id,
-            )
-        except FormNotFoundError:
-            raise NotFound("Form not found")
-
-        return {}, 200

api/controllers/service_api/app/workflow_events.py

@@ -1,142 +0,0 @@
-"""
-Service API workflow resume event stream endpoints.
-"""
-
-import json
-from collections.abc import Generator
-
-from flask import Response, request
-from flask_restx import Resource
-from sqlalchemy.orm import sessionmaker
-from werkzeug.exceptions import NotFound
-
-from controllers.service_api import service_api_ns
-from controllers.service_api.app.error import NotWorkflowAppError
-from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate_app_token
-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
-from core.app.entities.task_entities import StreamEvent
-from core.workflow.human_input_policy import HumanInputSurface
-from extensions.ext_database import db
-from models.enums import CreatorUserRole
-from models.model import App, AppMode, EndUser
-from repositories.factory import DifyAPIRepositoryFactory
-from services.workflow_event_snapshot_service import build_workflow_event_stream
-
-
-@service_api_ns.route("/workflow/<string:task_id>/events")
-class WorkflowEventsApi(Resource):
-    """Service API for getting workflow execution events after resume."""
-
-    @service_api_ns.doc("get_workflow_events")
-    @service_api_ns.doc(description="Get workflow execution events stream after resume")
-    @service_api_ns.doc(
-        params={
-            "task_id": "Workflow run ID",
-            "user": "End user identifier (query param)",
-            "include_state_snapshot": (
-                "Whether to replay from persisted state snapshot, "
-                'specify `"true"` to include a status snapshot of executed nodes'
-            ),
-            "continue_on_pause": (
-                "Whether to keep the stream open across workflow_paused events,"
-                'specify `"true"` to keep the stream open for `workflow_paused` events.'
-            ),
-        }
-    )
-    @service_api_ns.doc(
-        responses={
-            200: "SSE event stream",
-            401: "Unauthorized - invalid API token",
-            404: "Workflow run not found",
-        }
-    )
-    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.QUERY, required=True))
-    def get(self, app_model: App, end_user: EndUser, task_id: str):
-        app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in {AppMode.WORKFLOW, AppMode.ADVANCED_CHAT}:
-            raise NotWorkflowAppError()
-
-        session_maker = sessionmaker(db.engine)
-        repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker)
-        workflow_run = repo.get_workflow_run_by_id_and_tenant_id(
-            tenant_id=app_model.tenant_id,
-            run_id=task_id,
-        )
-
-        if workflow_run is None:
-            raise NotFound("Workflow run not found")
-
-        if workflow_run.app_id != app_model.id:
-            raise NotFound("Workflow run not found")
-
-        if workflow_run.created_by_role != CreatorUserRole.END_USER:
-            raise NotFound("Workflow run not found")
-
-        if workflow_run.created_by != end_user.id:
-            raise NotFound("Workflow run not found")
-
-        workflow_run_entity = workflow_run
-
-        if workflow_run_entity.finished_at is not None:
-            response = WorkflowResponseConverter.workflow_run_result_to_finish_response(
-                task_id=workflow_run_entity.id,
-                workflow_run=workflow_run_entity,
-                creator_user=end_user,
-            )
-
-            payload = response.model_dump(mode="json")
-            payload["event"] = response.event.value
-
-            def _generate_finished_events() -> Generator[str, None, None]:
-                yield f"data: {json.dumps(payload)}\n\n"
-
-            event_generator = _generate_finished_events
-        else:
-            msg_generator = MessageGenerator()
-            generator: BaseAppGenerator
-            if app_mode == AppMode.ADVANCED_CHAT:
-                generator = AdvancedChatAppGenerator()
-            elif app_mode == AppMode.WORKFLOW:
-                generator = WorkflowAppGenerator()
-            else:
-                raise NotWorkflowAppError()
-
-            include_state_snapshot = request.args.get("include_state_snapshot", "false").lower() == "true"
-            continue_on_pause = request.args.get("continue_on_pause", "false").lower() == "true"
-            terminal_events: list[StreamEvent] | None = [] if continue_on_pause else None
-
-            def _generate_stream_events():
-                if include_state_snapshot:
-                    return generator.convert_to_event_stream(
-                        build_workflow_event_stream(
-                            app_mode=app_mode,
-                            workflow_run=workflow_run_entity,
-                            tenant_id=app_model.tenant_id,
-                            app_id=app_model.id,
-                            session_maker=session_maker,
-                            human_input_surface=HumanInputSurface.SERVICE_API,
-                            close_on_pause=not continue_on_pause,
-                        )
-                    )
-                return generator.convert_to_event_stream(
-                    msg_generator.retrieve_events(
-                        app_mode,
-                        workflow_run_entity.id,
-                        terminal_events=terminal_events,
-                    ),
-                )
-
-            event_generator = _generate_stream_events
-
-        return Response(
-            event_generator(),
-            mimetype="text/event-stream",
-            headers={
-                "Cache-Control": "no-cache",
-                "Connection": "keep-alive",
-            },
-        )

api/controllers/service_api/dataset/document.py

@@ -1,12 +1,4 @@
-"""Service API endpoints for dataset document management.
-
-The canonical Service API paths use hyphenated route segments. Legacy underscore
-aliases remain registered for backward compatibility, but they must stay marked
-deprecated in generated API docs so clients migrate toward the canonical paths.
-"""
-
 import json
-from collections.abc import Mapping
 from contextlib import ExitStack
 from typing import Self
 from uuid import UUID
@@ -125,137 +117,12 @@ register_schema_models(
 )
 
 
-def _create_document_by_text(tenant_id: str, dataset_id: UUID) -> tuple[Mapping[str, object], int]:
-    """Create a document from text for both canonical and legacy routes."""
-    payload = DocumentTextCreatePayload.model_validate(service_api_ns.payload or {})
-    args = payload.model_dump(exclude_none=True)
-
-    dataset_id_str = str(dataset_id)
-    tenant_id_str = str(tenant_id)
-    dataset = db.session.scalar(
-        select(Dataset).where(Dataset.tenant_id == tenant_id_str, Dataset.id == dataset_id_str).limit(1)
-    )
-
-    if not dataset:
-        raise ValueError("Dataset does not exist.")
-
-    if not dataset.indexing_technique and not args["indexing_technique"]:
-        raise ValueError("indexing_technique is required.")
-
-    embedding_model_provider = payload.embedding_model_provider
-    embedding_model = payload.embedding_model
-    if embedding_model_provider and embedding_model:
-        DatasetService.check_embedding_model_setting(tenant_id_str, embedding_model_provider, embedding_model)
-
-    retrieval_model = payload.retrieval_model
-    if (
-        retrieval_model
-        and retrieval_model.reranking_model
-        and retrieval_model.reranking_model.reranking_provider_name
-        and retrieval_model.reranking_model.reranking_model_name
-    ):
-        DatasetService.check_reranking_model_setting(
-            tenant_id_str,
-            retrieval_model.reranking_model.reranking_provider_name,
-            retrieval_model.reranking_model.reranking_model_name,
-        )
-
-    if not current_user:
-        raise ValueError("current_user is required")
-
-    upload_file = FileService(db.engine).upload_text(
-        text=payload.text, text_name=payload.name, user_id=current_user.id, tenant_id=tenant_id_str
-    )
-    data_source = {
-        "type": "upload_file",
-        "info_list": {"data_source_type": "upload_file", "file_info_list": {"file_ids": [upload_file.id]}},
-    }
-    args["data_source"] = data_source
-    knowledge_config = KnowledgeConfig.model_validate(args)
-    DocumentService.document_create_args_validate(knowledge_config)
-
-    if not current_user:
-        raise ValueError("current_user is required")
-
-    try:
-        documents, batch = DocumentService.save_document_with_dataset_id(
-            dataset=dataset,
-            knowledge_config=knowledge_config,
-            account=current_user,
-            dataset_process_rule=dataset.latest_process_rule if "process_rule" not in args else None,
-            created_from="api",
-        )
-    except ProviderTokenNotInitError as ex:
-        raise ProviderNotInitializeError(ex.description)
-    document = documents[0]
-
-    documents_and_batch_fields = {"document": marshal(document, document_fields), "batch": batch}
-    return documents_and_batch_fields, 200
-
-
-def _update_document_by_text(tenant_id: str, dataset_id: UUID, document_id: UUID) -> tuple[Mapping[str, object], int]:
-    """Update a document from text for both canonical and legacy routes."""
-    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)
-    )
-    args = payload.model_dump(exclude_none=True)
-    if not dataset:
-        raise ValueError("Dataset does not exist.")
-
-    retrieval_model = payload.retrieval_model
-    if (
-        retrieval_model
-        and retrieval_model.reranking_model
-        and retrieval_model.reranking_model.reranking_provider_name
-        and retrieval_model.reranking_model.reranking_model_name
-    ):
-        DatasetService.check_reranking_model_setting(
-            tenant_id,
-            retrieval_model.reranking_model.reranking_provider_name,
-            retrieval_model.reranking_model.reranking_model_name,
-        )
-
-    # indexing_technique is already set in dataset since this is an update
-    args["indexing_technique"] = dataset.indexing_technique
-
-    if args.get("text"):
-        text = args.get("text")
-        name = args.get("name")
-        if not current_user:
-            raise ValueError("current_user is required")
-        upload_file = FileService(db.engine).upload_text(
-            text=str(text), text_name=str(name), user_id=current_user.id, tenant_id=tenant_id
-        )
-        data_source = {
-            "type": "upload_file",
-            "info_list": {"data_source_type": "upload_file", "file_info_list": {"file_ids": [upload_file.id]}},
-        }
-        args["data_source"] = data_source
-
-    args["original_document_id"] = str(document_id)
-    knowledge_config = KnowledgeConfig.model_validate(args)
-    DocumentService.document_create_args_validate(knowledge_config)
-
-    try:
-        documents, batch = DocumentService.save_document_with_dataset_id(
-            dataset=dataset,
-            knowledge_config=knowledge_config,
-            account=current_user,
-            dataset_process_rule=dataset.latest_process_rule if "process_rule" not in args else None,
-            created_from="api",
-        )
-    except ProviderTokenNotInitError as ex:
-        raise ProviderNotInitializeError(ex.description)
-    document = documents[0]
-
-    documents_and_batch_fields = {"document": marshal(document, document_fields), "batch": batch}
-    return documents_and_batch_fields, 200
-
-
-@service_api_ns.route("/datasets/<uuid:dataset_id>/document/create-by-text")
+@service_api_ns.route(
+    "/datasets/<uuid:dataset_id>/document/create_by_text",
+    "/datasets/<uuid:dataset_id>/document/create-by-text",
+)
 class DocumentAddByTextApi(DatasetApiResource):
-    """Resource for the canonical text document creation route."""
+    """Resource for documents."""
 
     @service_api_ns.expect(service_api_ns.models[DocumentTextCreatePayload.__name__])
     @service_api_ns.doc("create_document_by_text")
@@ -271,43 +138,81 @@ class DocumentAddByTextApi(DatasetApiResource):
     @cloud_edition_billing_resource_check("vector_space", "dataset")
     @cloud_edition_billing_resource_check("documents", "dataset")
     @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
-    def post(self, tenant_id: str, dataset_id: UUID):
+    def post(self, tenant_id, dataset_id):
         """Create document by text."""
-        return _create_document_by_text(tenant_id=tenant_id, dataset_id=dataset_id)
+        payload = DocumentTextCreatePayload.model_validate(service_api_ns.payload or {})
+        args = payload.model_dump(exclude_none=True)
 
-
-@service_api_ns.route("/datasets/<uuid:dataset_id>/document/create_by_text")
-class DeprecatedDocumentAddByTextApi(DatasetApiResource):
-    """Deprecated resource alias for text document creation."""
-
-    @service_api_ns.expect(service_api_ns.models[DocumentTextCreatePayload.__name__])
-    @service_api_ns.doc("create_document_by_text_deprecated")
-    @service_api_ns.doc(deprecated=True)
-    @service_api_ns.doc(
-        description=(
-            "Deprecated legacy alias for creating a new document by providing text content. "
-            "Use /datasets/{dataset_id}/document/create-by-text instead."
+        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)
         )
-    )
-    @service_api_ns.doc(params={"dataset_id": "Dataset ID"})
-    @service_api_ns.doc(
-        responses={
-            200: "Document created successfully",
-            401: "Unauthorized - invalid API token",
-            400: "Bad request - invalid parameters",
+
+        if not dataset:
+            raise ValueError("Dataset does not exist.")
+
+        if not dataset.indexing_technique and not args["indexing_technique"]:
+            raise ValueError("indexing_technique is required.")
+
+        embedding_model_provider = payload.embedding_model_provider
+        embedding_model = payload.embedding_model
+        if embedding_model_provider and embedding_model:
+            DatasetService.check_embedding_model_setting(tenant_id, embedding_model_provider, embedding_model)
+
+        retrieval_model = payload.retrieval_model
+        if (
+            retrieval_model
+            and retrieval_model.reranking_model
+            and retrieval_model.reranking_model.reranking_provider_name
+            and retrieval_model.reranking_model.reranking_model_name
+        ):
+            DatasetService.check_reranking_model_setting(
+                tenant_id,
+                retrieval_model.reranking_model.reranking_provider_name,
+                retrieval_model.reranking_model.reranking_model_name,
+            )
+
+        if not current_user:
+            raise ValueError("current_user is required")
+
+        upload_file = FileService(db.engine).upload_text(
+            text=payload.text, text_name=payload.name, user_id=current_user.id, tenant_id=tenant_id
+        )
+        data_source = {
+            "type": "upload_file",
+            "info_list": {"data_source_type": "upload_file", "file_info_list": {"file_ids": [upload_file.id]}},
         }
-    )
-    @cloud_edition_billing_resource_check("vector_space", "dataset")
-    @cloud_edition_billing_resource_check("documents", "dataset")
-    @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
-    def post(self, tenant_id: str, dataset_id: UUID):
-        """Create document by text through the deprecated underscore alias."""
-        return _create_document_by_text(tenant_id=tenant_id, dataset_id=dataset_id)
+        args["data_source"] = data_source
+        knowledge_config = KnowledgeConfig.model_validate(args)
+        # validate args
+        DocumentService.document_create_args_validate(knowledge_config)
+
+        if not current_user:
+            raise ValueError("current_user is required")
+
+        try:
+            documents, batch = DocumentService.save_document_with_dataset_id(
+                dataset=dataset,
+                knowledge_config=knowledge_config,
+                account=current_user,
+                dataset_process_rule=dataset.latest_process_rule if "process_rule" not in args else None,
+                created_from="api",
+            )
+        except ProviderTokenNotInitError as ex:
+            raise ProviderNotInitializeError(ex.description)
+        document = documents[0]
+
+        documents_and_batch_fields = {"document": marshal(document, document_fields), "batch": batch}
+        return documents_and_batch_fields, 200
 
 
-@service_api_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/update-by-text")
+@service_api_ns.route(
+    "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/update_by_text",
+    "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/update-by-text",
+)
 class DocumentUpdateByTextApi(DatasetApiResource):
-    """Resource for the canonical text document update route."""
+    """Resource for update documents."""
 
     @service_api_ns.expect(service_api_ns.models[DocumentTextUpdate.__name__])
     @service_api_ns.doc("update_document_by_text")
@@ -324,35 +229,62 @@ class DocumentUpdateByTextApi(DatasetApiResource):
     @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
     def post(self, tenant_id: str, dataset_id: UUID, document_id: UUID):
         """Update document by text."""
-        return _update_document_by_text(tenant_id=tenant_id, dataset_id=dataset_id, document_id=document_id)
-
-
-@service_api_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/update_by_text")
-class DeprecatedDocumentUpdateByTextApi(DatasetApiResource):
-    """Deprecated resource alias for text document updates."""
-
-    @service_api_ns.expect(service_api_ns.models[DocumentTextUpdate.__name__])
-    @service_api_ns.doc("update_document_by_text_deprecated")
-    @service_api_ns.doc(deprecated=True)
-    @service_api_ns.doc(
-        description=(
-            "Deprecated legacy alias for updating an existing document by providing text content. "
-            "Use /datasets/{dataset_id}/documents/{document_id}/update-by-text instead."
+        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)
         )
-    )
-    @service_api_ns.doc(params={"dataset_id": "Dataset ID", "document_id": "Document ID"})
-    @service_api_ns.doc(
-        responses={
-            200: "Document updated successfully",
-            401: "Unauthorized - invalid API token",
-            404: "Document not found",
-        }
-    )
-    @cloud_edition_billing_resource_check("vector_space", "dataset")
-    @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
-    def post(self, tenant_id: str, dataset_id: UUID, document_id: UUID):
-        """Update document by text through the deprecated underscore alias."""
-        return _update_document_by_text(tenant_id=tenant_id, dataset_id=dataset_id, document_id=document_id)
+        args = payload.model_dump(exclude_none=True)
+        if not dataset:
+            raise ValueError("Dataset does not exist.")
+
+        retrieval_model = payload.retrieval_model
+        if (
+            retrieval_model
+            and retrieval_model.reranking_model
+            and retrieval_model.reranking_model.reranking_provider_name
+            and retrieval_model.reranking_model.reranking_model_name
+        ):
+            DatasetService.check_reranking_model_setting(
+                tenant_id,
+                retrieval_model.reranking_model.reranking_provider_name,
+                retrieval_model.reranking_model.reranking_model_name,
+            )
+
+        # indexing_technique is already set in dataset since this is an update
+        args["indexing_technique"] = dataset.indexing_technique
+
+        if args.get("text"):
+            text = args.get("text")
+            name = args.get("name")
+            if not current_user:
+                raise ValueError("current_user is required")
+            upload_file = FileService(db.engine).upload_text(
+                text=str(text), text_name=str(name), user_id=current_user.id, tenant_id=tenant_id
+            )
+            data_source = {
+                "type": "upload_file",
+                "info_list": {"data_source_type": "upload_file", "file_info_list": {"file_ids": [upload_file.id]}},
+            }
+            args["data_source"] = data_source
+        # validate args
+        args["original_document_id"] = str(document_id)
+        knowledge_config = KnowledgeConfig.model_validate(args)
+        DocumentService.document_create_args_validate(knowledge_config)
+
+        try:
+            documents, batch = DocumentService.save_document_with_dataset_id(
+                dataset=dataset,
+                knowledge_config=knowledge_config,
+                account=current_user,
+                dataset_process_rule=dataset.latest_process_rule if "process_rule" not in args else None,
+                created_from="api",
+            )
+        except ProviderTokenNotInitError as ex:
+            raise ProviderNotInitializeError(ex.description)
+        document = documents[0]
+
+        documents_and_batch_fields = {"document": marshal(document, document_fields), "batch": batch}
+        return documents_and_batch_fields, 200
 
 
 @service_api_ns.route(
@@ -468,98 +400,15 @@ class DocumentAddByFileApi(DatasetApiResource):
         return documents_and_batch_fields, 200
 
 
-def _update_document_by_file(tenant_id: str, dataset_id: UUID, document_id: UUID) -> tuple[Mapping[str, object], int]:
-    """Update a document from an uploaded file for canonical and deprecated routes."""
-    dataset_id_str = str(dataset_id)
-    tenant_id_str = str(tenant_id)
-    dataset = db.session.scalar(
-        select(Dataset).where(Dataset.tenant_id == tenant_id_str, Dataset.id == dataset_id_str).limit(1)
-    )
-
-    if not dataset:
-        raise ValueError("Dataset does not exist.")
-
-    if dataset.provider == "external":
-        raise ValueError("External datasets are not supported.")
-
-    args: dict[str, object] = {}
-    if "data" in request.form:
-        args = json.loads(request.form["data"])
-    if "doc_form" not in args:
-        args["doc_form"] = dataset.chunk_structure or "text_model"
-    if "doc_language" not in args:
-        args["doc_language"] = "English"
-
-    # indexing_technique is already set in dataset since this is an update
-    args["indexing_technique"] = dataset.indexing_technique
-
-    if "file" in request.files:
-        # save file info
-        file = request.files["file"]
-
-        if len(request.files) > 1:
-            raise TooManyFilesError()
-
-        if not file.filename:
-            raise FilenameNotExistsError
-
-        if not current_user:
-            raise ValueError("current_user is required")
-
-        try:
-            upload_file = FileService(db.engine).upload_file(
-                filename=file.filename,
-                content=file.read(),
-                mimetype=file.mimetype,
-                user=current_user,
-                source="datasets",
-            )
-        except services.errors.file.FileTooLargeError as file_too_large_error:
-            raise FileTooLargeError(file_too_large_error.description)
-        except services.errors.file.UnsupportedFileTypeError:
-            raise UnsupportedFileTypeError()
-        data_source = {
-            "type": "upload_file",
-            "info_list": {"data_source_type": "upload_file", "file_info_list": {"file_ids": [upload_file.id]}},
-        }
-        args["data_source"] = data_source
-
-    # validate args
-    args["original_document_id"] = str(document_id)
-
-    knowledge_config = KnowledgeConfig.model_validate(args)
-    DocumentService.document_create_args_validate(knowledge_config)
-
-    try:
-        documents, _ = DocumentService.save_document_with_dataset_id(
-            dataset=dataset,
-            knowledge_config=knowledge_config,
-            account=dataset.created_by_account,
-            dataset_process_rule=dataset.latest_process_rule if "process_rule" not in args else None,
-            created_from="api",
-        )
-    except ProviderTokenNotInitError as ex:
-        raise ProviderNotInitializeError(ex.description)
-    document = documents[0]
-    documents_and_batch_fields = {"document": marshal(document, document_fields), "batch": document.batch}
-    return documents_and_batch_fields, 200
-
-
 @service_api_ns.route(
     "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/update_by_file",
     "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/update-by-file",
 )
-class DeprecatedDocumentUpdateByFileApi(DatasetApiResource):
-    """Deprecated resource aliases for file document updates."""
+class DocumentUpdateByFileApi(DatasetApiResource):
+    """Resource for update documents."""
 
-    @service_api_ns.doc("update_document_by_file_deprecated")
-    @service_api_ns.doc(deprecated=True)
-    @service_api_ns.doc(
-        description=(
-            "Deprecated legacy alias for updating an existing document by uploading a file. "
-            "Use PATCH /datasets/{dataset_id}/documents/{document_id} instead."
-        )
-    )
+    @service_api_ns.doc("update_document_by_file")
+    @service_api_ns.doc(description="Update an existing document by uploading a file")
     @service_api_ns.doc(params={"dataset_id": "Dataset ID", "document_id": "Document ID"})
     @service_api_ns.doc(
         responses={
@@ -570,9 +419,82 @@ class DeprecatedDocumentUpdateByFileApi(DatasetApiResource):
     )
     @cloud_edition_billing_resource_check("vector_space", "dataset")
     @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
-    def post(self, tenant_id: str, dataset_id: UUID, document_id: UUID):
-        """Update document by file through the deprecated file-update aliases."""
-        return _update_document_by_file(tenant_id=tenant_id, dataset_id=dataset_id, document_id=document_id)
+    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)
+        )
+
+        if not dataset:
+            raise ValueError("Dataset does not exist.")
+
+        if dataset.provider == "external":
+            raise ValueError("External datasets are not supported.")
+
+        args = {}
+        if "data" in request.form:
+            args = json.loads(request.form["data"])
+        if "doc_form" not in args:
+            args["doc_form"] = dataset.chunk_structure or "text_model"
+        if "doc_language" not in args:
+            args["doc_language"] = "English"
+
+        # get dataset info
+        dataset_id = str(dataset_id)
+        tenant_id = str(tenant_id)
+
+        # indexing_technique is already set in dataset since this is an update
+        args["indexing_technique"] = dataset.indexing_technique
+
+        if "file" in request.files:
+            # save file info
+            file = request.files["file"]
+
+            if len(request.files) > 1:
+                raise TooManyFilesError()
+
+            if not file.filename:
+                raise FilenameNotExistsError
+
+            if not current_user:
+                raise ValueError("current_user is required")
+
+            try:
+                upload_file = FileService(db.engine).upload_file(
+                    filename=file.filename,
+                    content=file.read(),
+                    mimetype=file.mimetype,
+                    user=current_user,
+                    source="datasets",
+                )
+            except services.errors.file.FileTooLargeError as file_too_large_error:
+                raise FileTooLargeError(file_too_large_error.description)
+            except services.errors.file.UnsupportedFileTypeError:
+                raise UnsupportedFileTypeError()
+            data_source = {
+                "type": "upload_file",
+                "info_list": {"data_source_type": "upload_file", "file_info_list": {"file_ids": [upload_file.id]}},
+            }
+            args["data_source"] = data_source
+        # validate args
+        args["original_document_id"] = str(document_id)
+
+        knowledge_config = KnowledgeConfig.model_validate(args)
+        DocumentService.document_create_args_validate(knowledge_config)
+
+        try:
+            documents, _ = DocumentService.save_document_with_dataset_id(
+                dataset=dataset,
+                knowledge_config=knowledge_config,
+                account=dataset.created_by_account,
+                dataset_process_rule=dataset.latest_process_rule if "process_rule" not in args else None,
+                created_from="api",
+            )
+        except ProviderTokenNotInitError as ex:
+            raise ProviderNotInitializeError(ex.description)
+        document = documents[0]
+        documents_and_batch_fields = {"document": marshal(document, document_fields), "batch": document.batch}
+        return documents_and_batch_fields, 200
 
 
 @service_api_ns.route("/datasets/<uuid:dataset_id>/documents")
@@ -886,22 +808,6 @@ class DocumentApi(DatasetApiResource):
 
         return response
 
-    @service_api_ns.doc("update_document_by_file")
-    @service_api_ns.doc(description="Update an existing document by uploading a file")
-    @service_api_ns.doc(params={"dataset_id": "Dataset ID", "document_id": "Document ID"})
-    @service_api_ns.doc(
-        responses={
-            200: "Document updated successfully",
-            401: "Unauthorized - invalid API token",
-            404: "Document not found",
-        }
-    )
-    @cloud_edition_billing_resource_check("vector_space", "dataset")
-    @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
-    def patch(self, tenant_id: str, dataset_id: UUID, document_id: UUID):
-        """Update document by file on the canonical document resource."""
-        return _update_document_by_file(tenant_id=tenant_id, dataset_id=dataset_id, document_id=document_id)
-
     @service_api_ns.doc("delete_document")
     @service_api_ns.doc(description="Delete a document")
     @service_api_ns.doc(params={"dataset_id": "Dataset ID", "document_id": "Document ID"})

api/controllers/web/human_input_form.py

@@ -9,11 +9,11 @@ from typing import Any, NotRequired, TypedDict
 
 from flask import Response, request
 from flask_restx import Resource
+from pydantic import BaseModel
 from sqlalchemy import select
 from werkzeug.exceptions import Forbidden
 
 from configs import dify_config
-from controllers.common.human_input import HumanInputFormSubmitPayload
 from controllers.web import web_ns
 from controllers.web.error import NotFoundError, WebFormRateLimitExceededError
 from controllers.web.site import serialize_app_site_payload
@@ -26,6 +26,11 @@ from services.human_input_service import Form, FormNotFoundError, HumanInputServ
 logger = logging.getLogger(__name__)
 
 
+class HumanInputFormSubmitPayload(BaseModel):
+    inputs: dict
+    action: str
+
+
 _FORM_SUBMIT_RATE_LIMITER = RateLimiter(
     prefix="web_form_submit_rate_limit",
     max_attempts=dify_config.WEB_FORM_SUBMIT_RATE_LIMIT_MAX_ATTEMPTS,

api/controllers/web/workflow_events.py

@@ -84,6 +84,9 @@ class WorkflowEventsApi(WebApiResource):
 
             def _generate_stream_events():
                 if include_state_snapshot:
+                    # TODO(wylswz): events between shapshot and live tail may be lost.
+                    # TODO(wylswz): previous message chunks are not replayed. In order to support replay, we need
+                    # to figure out a way to deduplicate events between snapshot and stream.
                     return generator.convert_to_event_stream(
                         build_workflow_event_stream(
                             app_mode=app_mode,

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

@@ -1,7 +1,5 @@
 from typing import Any
 
-CUSTOM_FOLLOW_UP_PROMPT_MAX_LENGTH = 1000
-
 
 class SuggestedQuestionsAfterAnswerConfigManager:
     @classmethod
@@ -22,11 +20,7 @@ class SuggestedQuestionsAfterAnswerConfigManager:
     @classmethod
     def validate_and_set_defaults(cls, config: dict[str, Any]) -> tuple[dict[str, Any], list[str]]:
         """
-        Validate and set defaults for suggested questions feature.
-
-        Optional fields:
-        - prompt: custom instruction prompt.
-        - model: provider/model configuration for suggested question generation.
+        Validate and set defaults for suggested questions feature
 
         :param config: app model config args
         """
@@ -45,27 +39,4 @@ class SuggestedQuestionsAfterAnswerConfigManager:
         if not isinstance(config["suggested_questions_after_answer"]["enabled"], bool):
             raise ValueError("enabled in suggested_questions_after_answer must be of boolean type")
 
-        prompt = config["suggested_questions_after_answer"].get("prompt")
-        if prompt is not None and not isinstance(prompt, str):
-            raise ValueError("prompt in suggested_questions_after_answer must be of string type")
-        if isinstance(prompt, str) and len(prompt) > CUSTOM_FOLLOW_UP_PROMPT_MAX_LENGTH:
-            raise ValueError(
-                f"prompt in suggested_questions_after_answer must be less than or equal to "
-                f"{CUSTOM_FOLLOW_UP_PROMPT_MAX_LENGTH} characters"
-            )
-
-        if "model" in config["suggested_questions_after_answer"]:
-            model_config = config["suggested_questions_after_answer"]["model"]
-            if not isinstance(model_config, dict):
-                raise ValueError("model in suggested_questions_after_answer must be of object type")
-
-            if "provider" not in model_config or not isinstance(model_config["provider"], str):
-                raise ValueError("provider in suggested_questions_after_answer.model must be of string type")
-
-            if "name" not in model_config or not isinstance(model_config["name"], str):
-                raise ValueError("name in suggested_questions_after_answer.model must be of string type")
-
-            if "completion_params" in model_config and not isinstance(model_config["completion_params"], dict):
-                raise ValueError("completion_params in suggested_questions_after_answer.model must be of object type")
-
         return config, ["suggested_questions_after_answer"]

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

@@ -34,11 +34,7 @@ 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
 from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, InvokeFrom
-from core.app.entities.task_entities import (
-    AdvancedChatPausedBlockingResponse,
-    ChatbotAppBlockingResponse,
-    ChatbotAppStreamResponse,
-)
+from core.app.entities.task_entities import ChatbotAppBlockingResponse, ChatbotAppStreamResponse
 from core.app.layers.pause_state_persist_layer import PauseStateLayerConfig, PauseStatePersistenceLayer
 from core.helper.trace_id_helper import extract_external_trace_id_from_args
 from core.ops.ops_trace_manager import TraceQueueManager
@@ -659,11 +655,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         user: Account | EndUser,
         draft_var_saver_factory: DraftVariableSaverFactory,
         stream: bool = False,
-    ) -> (
-        ChatbotAppBlockingResponse
-        | AdvancedChatPausedBlockingResponse
-        | Generator[ChatbotAppStreamResponse, None, None]
-    ):
+    ) -> ChatbotAppBlockingResponse | Generator[ChatbotAppStreamResponse, None, None]:
         """
         Handle response.
         :param application_generate_entity: application generate entity

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

@@ -3,7 +3,7 @@ from typing import Any, cast
 
 from core.app.apps.base_app_generate_response_converter import AppGenerateResponseConverter
 from core.app.entities.task_entities import (
-    AdvancedChatPausedBlockingResponse,
+    AppBlockingResponse,
     AppStreamResponse,
     ChatbotAppBlockingResponse,
     ChatbotAppStreamResponse,
@@ -12,40 +12,22 @@ from core.app.entities.task_entities import (
     NodeFinishStreamResponse,
     NodeStartStreamResponse,
     PingStreamResponse,
-    StreamEvent,
 )
 
 
-class AdvancedChatAppGenerateResponseConverter(
-    AppGenerateResponseConverter[ChatbotAppBlockingResponse | AdvancedChatPausedBlockingResponse]
-):
+class AdvancedChatAppGenerateResponseConverter(AppGenerateResponseConverter):
+    _blocking_response_type = ChatbotAppBlockingResponse
+
     @classmethod
-    def convert_blocking_full_response(
-        cls, blocking_response: ChatbotAppBlockingResponse | AdvancedChatPausedBlockingResponse
-    ) -> dict[str, Any]:
+    def convert_blocking_full_response(cls, blocking_response: AppBlockingResponse) -> dict[str, Any]:
         """
         Convert blocking full response.
         :param blocking_response: blocking response
         :return:
         """
-        if isinstance(blocking_response, AdvancedChatPausedBlockingResponse):
-            paused_data = blocking_response.data.model_dump(mode="json")
-            return {
-                "event": StreamEvent.WORKFLOW_PAUSED.value,
-                "task_id": blocking_response.task_id,
-                "id": blocking_response.data.id,
-                "message_id": blocking_response.data.message_id,
-                "conversation_id": blocking_response.data.conversation_id,
-                "mode": blocking_response.data.mode,
-                "answer": blocking_response.data.answer,
-                "metadata": blocking_response.data.metadata,
-                "created_at": blocking_response.data.created_at,
-                "workflow_run_id": blocking_response.data.workflow_run_id,
-                "data": paused_data,
-            }
-
+        blocking_response = cast(ChatbotAppBlockingResponse, blocking_response)
         response = {
-            "event": StreamEvent.MESSAGE.value,
+            "event": "message",
             "task_id": blocking_response.task_id,
             "id": blocking_response.data.id,
             "message_id": blocking_response.data.message_id,
@@ -59,9 +41,7 @@ class AdvancedChatAppGenerateResponseConverter(
         return response
 
     @classmethod
-    def convert_blocking_simple_response(
-        cls, blocking_response: ChatbotAppBlockingResponse | AdvancedChatPausedBlockingResponse
-    ) -> dict[str, Any]:
+    def convert_blocking_simple_response(cls, blocking_response: AppBlockingResponse) -> dict[str, Any]:
         """
         Convert blocking simple response.
         :param blocking_response: blocking response
@@ -70,8 +50,7 @@ class AdvancedChatAppGenerateResponseConverter(
         response = cls.convert_blocking_full_response(blocking_response)
 
         metadata = response.get("metadata", {})
-        if isinstance(metadata, dict):
-            response["metadata"] = cls._get_simple_metadata(metadata)
+        response["metadata"] = cls._get_simple_metadata(metadata)
 
         return response
 

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

@@ -53,18 +53,14 @@ from core.app.entities.queue_entities import (
     WorkflowQueueMessage,
 )
 from core.app.entities.task_entities import (
-    AdvancedChatPausedBlockingResponse,
     ChatbotAppBlockingResponse,
     ChatbotAppStreamResponse,
     ErrorStreamResponse,
-    HumanInputRequiredPauseReasonPayload,
-    HumanInputRequiredResponse,
     MessageAudioEndStreamResponse,
     MessageAudioStreamResponse,
     MessageEndStreamResponse,
     PingStreamResponse,
     StreamResponse,
-    WorkflowPauseStreamResponse,
     WorkflowTaskState,
 )
 from core.app.task_pipeline.based_generate_task_pipeline import BasedGenerateTaskPipeline
@@ -214,13 +210,7 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         if message.status == MessageStatus.PAUSED and message.answer:
             self._task_state.answer = message.answer
 
-    def process(
-        self,
-    ) -> Union[
-        ChatbotAppBlockingResponse,
-        AdvancedChatPausedBlockingResponse,
-        Generator[ChatbotAppStreamResponse, None, None],
-    ]:
+    def process(self) -> Union[ChatbotAppBlockingResponse, Generator[ChatbotAppStreamResponse, None, None]]:
         """
         Process generate task pipeline.
         :return:
@@ -236,39 +226,14 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         else:
             return self._to_blocking_response(generator)
 
-    def _to_blocking_response(
-        self, generator: Generator[StreamResponse, None, None]
-    ) -> Union[ChatbotAppBlockingResponse, AdvancedChatPausedBlockingResponse]:
+    def _to_blocking_response(self, generator: Generator[StreamResponse, None, None]) -> ChatbotAppBlockingResponse:
         """
         Process blocking response.
         :return:
         """
-        human_input_responses: list[HumanInputRequiredResponse] = []
         for stream_response in generator:
             if isinstance(stream_response, ErrorStreamResponse):
                 raise stream_response.err
-            elif isinstance(stream_response, HumanInputRequiredResponse):
-                human_input_responses.append(stream_response)
-            elif isinstance(stream_response, WorkflowPauseStreamResponse):
-                return AdvancedChatPausedBlockingResponse(
-                    task_id=stream_response.task_id,
-                    data=AdvancedChatPausedBlockingResponse.Data(
-                        id=self._message_id,
-                        mode=self._conversation_mode,
-                        conversation_id=self._conversation_id,
-                        message_id=self._message_id,
-                        workflow_run_id=stream_response.data.workflow_run_id,
-                        answer=self._task_state.answer,
-                        metadata=self._message_end_to_stream_response().metadata,
-                        created_at=self._message_created_at,
-                        paused_nodes=stream_response.data.paused_nodes,
-                        reasons=stream_response.data.reasons,
-                        status=stream_response.data.status,
-                        elapsed_time=stream_response.data.elapsed_time,
-                        total_tokens=stream_response.data.total_tokens,
-                        total_steps=stream_response.data.total_steps,
-                    ),
-                )
             elif isinstance(stream_response, MessageEndStreamResponse):
                 extras = {}
                 if stream_response.metadata:
@@ -289,41 +254,8 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
             else:
                 continue
 
-        if human_input_responses:
-            return self._build_paused_blocking_response_from_human_input(human_input_responses)
-
         raise ValueError("queue listening stopped unexpectedly.")
 
-    def _build_paused_blocking_response_from_human_input(
-        self, human_input_responses: list[HumanInputRequiredResponse]
-    ) -> AdvancedChatPausedBlockingResponse:
-        runtime_state = self._resolve_graph_runtime_state()
-        paused_nodes = list(dict.fromkeys(response.data.node_id for response in human_input_responses))
-        reasons = [
-            HumanInputRequiredPauseReasonPayload.from_response_data(response.data).model_dump(mode="json")
-            for response in human_input_responses
-        ]
-
-        return AdvancedChatPausedBlockingResponse(
-            task_id=self._application_generate_entity.task_id,
-            data=AdvancedChatPausedBlockingResponse.Data(
-                id=self._message_id,
-                mode=self._conversation_mode,
-                conversation_id=self._conversation_id,
-                message_id=self._message_id,
-                workflow_run_id=human_input_responses[-1].workflow_run_id,
-                answer=self._task_state.answer,
-                metadata=self._message_end_to_stream_response().metadata,
-                created_at=self._message_created_at,
-                paused_nodes=paused_nodes,
-                reasons=reasons,
-                status=WorkflowExecutionStatus.PAUSED,
-                elapsed_time=time.perf_counter() - self._base_task_pipeline.start_at,
-                total_tokens=runtime_state.total_tokens,
-                total_steps=runtime_state.node_run_steps,
-            ),
-        )
-
     def _to_stream_response(
         self, generator: Generator[StreamResponse, None, None]
     ) -> Generator[ChatbotAppStreamResponse, Any, None]:

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

@@ -1,8 +1,6 @@
 from collections.abc import Generator
 from typing import Any, cast
 
-from pydantic import JsonValue
-
 from core.app.apps.base_app_generate_response_converter import AppGenerateResponseConverter
 from core.app.entities.task_entities import (
     AppStreamResponse,
@@ -14,9 +12,11 @@ from core.app.entities.task_entities import (
 )
 
 
-class AgentChatAppGenerateResponseConverter(AppGenerateResponseConverter[ChatbotAppBlockingResponse]):
+class AgentChatAppGenerateResponseConverter(AppGenerateResponseConverter):
+    _blocking_response_type = ChatbotAppBlockingResponse
+
     @classmethod
-    def convert_blocking_full_response(cls, blocking_response: ChatbotAppBlockingResponse):
+    def convert_blocking_full_response(cls, blocking_response: ChatbotAppBlockingResponse):  # type: ignore[override]
         """
         Convert blocking full response.
         :param blocking_response: blocking response
@@ -37,7 +37,7 @@ class AgentChatAppGenerateResponseConverter(AppGenerateResponseConverter[Chatbot
         return response
 
     @classmethod
-    def convert_blocking_simple_response(cls, blocking_response: ChatbotAppBlockingResponse):
+    def convert_blocking_simple_response(cls, blocking_response: ChatbotAppBlockingResponse):  # type: ignore[override]
         """
         Convert blocking simple response.
         :param blocking_response: blocking response
@@ -70,7 +70,7 @@ class AgentChatAppGenerateResponseConverter(AppGenerateResponseConverter[Chatbot
                 yield "ping"
                 continue
 
-            response_chunk: dict[str, JsonValue] = {
+            response_chunk = {
                 "event": sub_stream_response.event.value,
                 "conversation_id": chunk.conversation_id,
                 "message_id": chunk.message_id,
@@ -101,7 +101,7 @@ class AgentChatAppGenerateResponseConverter(AppGenerateResponseConverter[Chatbot
                 yield "ping"
                 continue
 
-            response_chunk: dict[str, JsonValue] = {
+            response_chunk = {
                 "event": sub_stream_response.event.value,
                 "conversation_id": chunk.conversation_id,
                 "message_id": chunk.message_id,

api/core/app/apps/base_app_generate_response_converter.py

@@ -1,9 +1,7 @@
 import logging
 from abc import ABC, abstractmethod
 from collections.abc import Generator, Mapping
-from typing import Any, Union, cast
-
-from pydantic import JsonValue
+from typing import Any, Union
 
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.app.entities.task_entities import AppBlockingResponse, AppStreamResponse
@@ -13,10 +11,8 @@ from graphon.model_runtime.errors.invoke import InvokeError
 logger = logging.getLogger(__name__)
 
 
-class AppGenerateResponseConverter[TBlockingResponse: AppBlockingResponse](ABC):
-    @classmethod
-    def _cast_blocking_response(cls, response: AppBlockingResponse) -> TBlockingResponse:
-        return cast(TBlockingResponse, response)
+class AppGenerateResponseConverter(ABC):
+    _blocking_response_type: type[AppBlockingResponse]
 
     @classmethod
     def convert(
@@ -24,7 +20,7 @@ class AppGenerateResponseConverter[TBlockingResponse: AppBlockingResponse](ABC):
     ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any], Any, None]:
         if invoke_from in {InvokeFrom.DEBUGGER, InvokeFrom.SERVICE_API}:
             if isinstance(response, AppBlockingResponse):
-                return cls.convert_blocking_full_response(cls._cast_blocking_response(response))
+                return cls.convert_blocking_full_response(response)
             else:
 
                 def _generate_full_response() -> Generator[dict[str, Any] | str, Any, None]:
@@ -33,7 +29,7 @@ class AppGenerateResponseConverter[TBlockingResponse: AppBlockingResponse](ABC):
                 return _generate_full_response()
         else:
             if isinstance(response, AppBlockingResponse):
-                return cls.convert_blocking_simple_response(cls._cast_blocking_response(response))
+                return cls.convert_blocking_simple_response(response)
             else:
 
                 def _generate_simple_response() -> Generator[dict[str, Any] | str, Any, None]:
@@ -43,12 +39,12 @@ class AppGenerateResponseConverter[TBlockingResponse: AppBlockingResponse](ABC):
 
     @classmethod
     @abstractmethod
-    def convert_blocking_full_response(cls, blocking_response: TBlockingResponse) -> dict[str, Any]:
+    def convert_blocking_full_response(cls, blocking_response: AppBlockingResponse) -> dict[str, Any]:
         raise NotImplementedError
 
     @classmethod
     @abstractmethod
-    def convert_blocking_simple_response(cls, blocking_response: TBlockingResponse) -> dict[str, Any]:
+    def convert_blocking_simple_response(cls, blocking_response: AppBlockingResponse) -> dict[str, Any]:
         raise NotImplementedError
 
     @classmethod
@@ -110,13 +106,13 @@ class AppGenerateResponseConverter[TBlockingResponse: AppBlockingResponse](ABC):
         return metadata
 
     @classmethod
-    def _error_to_stream_response(cls, e: Exception) -> dict[str, JsonValue]:
+    def _error_to_stream_response(cls, e: Exception) -> dict[str, Any]:
         """
         Error to stream response.
         :param e: exception
         :return:
         """
-        error_responses: dict[type[Exception], dict[str, JsonValue]] = {
+        error_responses: dict[type[Exception], dict[str, Any]] = {
             ValueError: {"code": "invalid_param", "status": 400},
             ProviderTokenNotInitError: {"code": "provider_not_initialize", "status": 400},
             QuotaExceededError: {
@@ -130,7 +126,7 @@ class AppGenerateResponseConverter[TBlockingResponse: AppBlockingResponse](ABC):
         }
 
         # Determine the response based on the type of exception
-        data: dict[str, JsonValue] | None = None
+        data: dict[str, Any] | None = None
         for k, v in error_responses.items():
             if isinstance(e, k):
                 data = v

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

@@ -1,8 +1,6 @@
 from collections.abc import Generator
 from typing import Any, cast
 
-from pydantic import JsonValue
-
 from core.app.apps.base_app_generate_response_converter import AppGenerateResponseConverter
 from core.app.entities.task_entities import (
     AppStreamResponse,
@@ -14,9 +12,11 @@ from core.app.entities.task_entities import (
 )
 
 
-class ChatAppGenerateResponseConverter(AppGenerateResponseConverter[ChatbotAppBlockingResponse]):
+class ChatAppGenerateResponseConverter(AppGenerateResponseConverter):
+    _blocking_response_type = ChatbotAppBlockingResponse
+
     @classmethod
-    def convert_blocking_full_response(cls, blocking_response: ChatbotAppBlockingResponse):
+    def convert_blocking_full_response(cls, blocking_response: ChatbotAppBlockingResponse):  # type: ignore[override]
         """
         Convert blocking full response.
         :param blocking_response: blocking response
@@ -37,7 +37,7 @@ class ChatAppGenerateResponseConverter(AppGenerateResponseConverter[ChatbotAppBl
         return response
 
     @classmethod
-    def convert_blocking_simple_response(cls, blocking_response: ChatbotAppBlockingResponse):
+    def convert_blocking_simple_response(cls, blocking_response: ChatbotAppBlockingResponse):  # type: ignore[override]
         """
         Convert blocking simple response.
         :param blocking_response: blocking response
@@ -70,7 +70,7 @@ class ChatAppGenerateResponseConverter(AppGenerateResponseConverter[ChatbotAppBl
                 yield "ping"
                 continue
 
-            response_chunk: dict[str, JsonValue] = {
+            response_chunk = {
                 "event": sub_stream_response.event.value,
                 "conversation_id": chunk.conversation_id,
                 "message_id": chunk.message_id,
@@ -101,7 +101,7 @@ class ChatAppGenerateResponseConverter(AppGenerateResponseConverter[ChatbotAppBl
                 yield "ping"
                 continue
 
-            response_chunk: dict[str, JsonValue] = {
+            response_chunk = {
                 "event": sub_stream_response.event.value,
                 "conversation_id": chunk.conversation_id,
                 "message_id": chunk.message_id,

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

@@ -52,7 +52,6 @@ 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.human_input_policy import HumanInputSurface, enrich_human_input_pause_reasons
 from core.workflow.system_variables import SystemVariableKey, system_variables_to_mapping
 from core.workflow.workflow_entry import WorkflowEntry
 from extensions.ext_database import db
@@ -337,26 +336,7 @@ class WorkflowResponseConverter:
                     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,
-                    surface=(
-                        HumanInputSurface.SERVICE_API
-                        if self._application_generate_entity.invoke_from == InvokeFrom.SERVICE_API
-                        else None
-                    ),
-                )
-
-        # Reconnect paths must preserve the same pause-reason contract as live streams;
-        # otherwise clients see schema drift after resume.
-        pause_reasons = enrich_human_input_pause_reasons(
-            pause_reasons,
-            form_tokens_by_form_id=form_token_by_form_id,
-            expiration_times_by_form_id={
-                form_id: int(expiration_time.timestamp())
-                for form_id, expiration_time in expiration_times_by_form_id.items()
-            },
-        )
+                form_token_by_form_id = load_form_tokens_by_form_id(human_input_form_ids, session=session)
 
         responses: list[StreamResponse] = []
 

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

@@ -1,8 +1,6 @@
 from collections.abc import Generator
 from typing import Any, cast
 
-from pydantic import JsonValue
-
 from core.app.apps.base_app_generate_response_converter import AppGenerateResponseConverter
 from core.app.entities.task_entities import (
     AppStreamResponse,
@@ -14,15 +12,17 @@ from core.app.entities.task_entities import (
 )
 
 
-class CompletionAppGenerateResponseConverter(AppGenerateResponseConverter[CompletionAppBlockingResponse]):
+class CompletionAppGenerateResponseConverter(AppGenerateResponseConverter):
+    _blocking_response_type = CompletionAppBlockingResponse
+
     @classmethod
-    def convert_blocking_full_response(cls, blocking_response: CompletionAppBlockingResponse):
+    def convert_blocking_full_response(cls, blocking_response: CompletionAppBlockingResponse):  # type: ignore[override]
         """
         Convert blocking full response.
         :param blocking_response: blocking response
         :return:
         """
-        response: dict[str, Any] = {
+        response = {
             "event": "message",
             "task_id": blocking_response.task_id,
             "id": blocking_response.data.id,
@@ -36,7 +36,7 @@ class CompletionAppGenerateResponseConverter(AppGenerateResponseConverter[Comple
         return response
 
     @classmethod
-    def convert_blocking_simple_response(cls, blocking_response: CompletionAppBlockingResponse):
+    def convert_blocking_simple_response(cls, blocking_response: CompletionAppBlockingResponse):  # type: ignore[override]
         """
         Convert blocking simple response.
         :param blocking_response: blocking response
@@ -69,7 +69,7 @@ class CompletionAppGenerateResponseConverter(AppGenerateResponseConverter[Comple
                 yield "ping"
                 continue
 
-            response_chunk: dict[str, JsonValue] = {
+            response_chunk = {
                 "event": sub_stream_response.event.value,
                 "message_id": chunk.message_id,
                 "created_at": chunk.created_at,
@@ -99,7 +99,7 @@ class CompletionAppGenerateResponseConverter(AppGenerateResponseConverter[Comple
                 yield "ping"
                 continue
 
-            response_chunk: dict[str, JsonValue] = {
+            response_chunk = {
                 "event": sub_stream_response.event.value,
                 "message_id": chunk.message_id,
                 "created_at": chunk.created_at,

api/core/app/apps/message_based_app_generator.py

@@ -311,7 +311,7 @@ class MessageBasedAppGenerator(BaseAppGenerator):
         cls,
         app_mode: AppMode,
         workflow_run_id: str,
-        idle_timeout=300,
+        idle_timeout: float = 300,
         on_subscribe: Callable[[], None] | None = None,
     ) -> Generator[Mapping | str, None, None]:
         topic = cls.get_response_topic(app_mode, workflow_run_id)

api/core/app/apps/message_generator.py

@@ -1,7 +1,6 @@
-from collections.abc import Callable, Generator, Iterable, Mapping
+from collections.abc import Callable, Generator, Mapping
 
 from core.app.apps.streaming_utils import stream_topic_events
-from core.app.entities.task_entities import StreamEvent
 from extensions.ext_redis import get_pubsub_broadcast_channel
 from libs.broadcast_channel.channel import Topic
 from models.model import AppMode
@@ -24,10 +23,9 @@ class MessageGenerator:
         cls,
         app_mode: AppMode,
         workflow_run_id: str,
-        idle_timeout=300,
+        idle_timeout: float = 300,
         ping_interval: float = 10.0,
         on_subscribe: Callable[[], None] | None = None,
-        terminal_events: Iterable[str | StreamEvent] | None = None,
     ) -> Generator[Mapping | str, None, None]:
         topic = cls.get_response_topic(app_mode, workflow_run_id)
         return stream_topic_events(
@@ -35,5 +33,4 @@ class MessageGenerator:
             idle_timeout=idle_timeout,
             ping_interval=ping_interval,
             on_subscribe=on_subscribe,
-            terminal_events=terminal_events,
         )

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

@@ -13,9 +13,11 @@ from core.app.entities.task_entities import (
 )
 
 
-class WorkflowAppGenerateResponseConverter(AppGenerateResponseConverter[WorkflowAppBlockingResponse]):
+class WorkflowAppGenerateResponseConverter(AppGenerateResponseConverter):
+    _blocking_response_type = WorkflowAppBlockingResponse
+
     @classmethod
-    def convert_blocking_full_response(cls, blocking_response: WorkflowAppBlockingResponse) -> dict[str, object]:
+    def convert_blocking_full_response(cls, blocking_response: WorkflowAppBlockingResponse) -> dict[str, Any]:  # type: ignore[override]
         """
         Convert blocking full response.
         :param blocking_response: blocking response
@@ -24,7 +26,7 @@ class WorkflowAppGenerateResponseConverter(AppGenerateResponseConverter[Workflow
         return dict(blocking_response.model_dump())
 
     @classmethod
-    def convert_blocking_simple_response(cls, blocking_response: WorkflowAppBlockingResponse) -> dict[str, object]:
+    def convert_blocking_simple_response(cls, blocking_response: WorkflowAppBlockingResponse) -> dict[str, Any]:  # type: ignore[override]
         """
         Convert blocking simple response.
         :param blocking_response: blocking response

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

@@ -27,11 +27,7 @@ from core.app.apps.workflow.generate_response_converter import WorkflowAppGenera
 from core.app.apps.workflow.generate_task_pipeline import WorkflowAppGenerateTaskPipeline
 from core.app.entities.app_invoke_entities import InvokeFrom, RagPipelineGenerateEntity
 from core.app.entities.rag_pipeline_invoke_entities import RagPipelineInvokeEntity
-from core.app.entities.task_entities import (
-    WorkflowAppBlockingResponse,
-    WorkflowAppPausedBlockingResponse,
-    WorkflowAppStreamResponse,
-)
+from core.app.entities.task_entities import WorkflowAppBlockingResponse, WorkflowAppStreamResponse
 from core.datasource.entities.datasource_entities import (
     DatasourceProviderType,
     OnlineDriveBrowseFilesRequest,
@@ -631,11 +627,7 @@ class PipelineGenerator(BaseAppGenerator):
         user: Account | EndUser,
         draft_var_saver_factory: DraftVariableSaverFactory,
         stream: bool = False,
-    ) -> (
-        WorkflowAppBlockingResponse
-        | WorkflowAppPausedBlockingResponse
-        | Generator[WorkflowAppStreamResponse, None, None]
-    ):
+    ) -> WorkflowAppBlockingResponse | Generator[WorkflowAppStreamResponse, None, None]:
         """
         Handle response.
         :param application_generate_entity: application generate entity

api/core/app/apps/streaming_utils.py

@@ -27,6 +27,9 @@ def stream_topic_events(
     terminal_values = _normalize_terminal_events(terminal_events)
     last_msg_time = time.time()
     last_ping_time = last_msg_time
+    # The application layer intentionally does not use broadcast-channel replay;
+    # callers that need historical events should compose them from persisted state
+    # (see ``build_workflow_event_stream``) and then tail the live stream.
     with topic.subscribe() as sub:
         # on_subscribe fires only after the Redis subscription is active.
         # This is used to gate task start and reduce pub/sub race for the first event.
@@ -59,7 +62,7 @@ def stream_topic_events(
 
 
 def _normalize_terminal_events(terminal_events: Iterable[str | StreamEvent] | None) -> set[str]:
-    if terminal_events is None:
+    if not terminal_events:
         return {StreamEvent.WORKFLOW_FINISHED.value, StreamEvent.WORKFLOW_PAUSED.value}
     values: set[str] = set()
     for item in terminal_events:

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

@@ -25,11 +25,7 @@ from core.app.apps.workflow.app_runner import WorkflowAppRunner
 from core.app.apps.workflow.generate_response_converter import WorkflowAppGenerateResponseConverter
 from core.app.apps.workflow.generate_task_pipeline import WorkflowAppGenerateTaskPipeline
 from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity
-from core.app.entities.task_entities import (
-    WorkflowAppBlockingResponse,
-    WorkflowAppPausedBlockingResponse,
-    WorkflowAppStreamResponse,
-)
+from core.app.entities.task_entities import WorkflowAppBlockingResponse, WorkflowAppStreamResponse
 from core.app.layers.pause_state_persist_layer import PauseStateLayerConfig, PauseStatePersistenceLayer
 from core.db.session_factory import session_factory
 from core.helper.trace_id_helper import extract_external_trace_id_from_args
@@ -616,11 +612,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
         user: Account | EndUser,
         draft_var_saver_factory: DraftVariableSaverFactory,
         stream: bool = False,
-    ) -> (
-        WorkflowAppBlockingResponse
-        | WorkflowAppPausedBlockingResponse
-        | Generator[WorkflowAppStreamResponse, None, None]
-    ):
+    ) -> WorkflowAppBlockingResponse | Generator[WorkflowAppStreamResponse, None, None]:
         """
         Handle response.
         :param application_generate_entity: application generate entity

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

@@ -9,29 +9,24 @@ from core.app.entities.task_entities import (
     NodeStartStreamResponse,
     PingStreamResponse,
     WorkflowAppBlockingResponse,
-    WorkflowAppPausedBlockingResponse,
     WorkflowAppStreamResponse,
 )
 
 
-class WorkflowAppGenerateResponseConverter(
-    AppGenerateResponseConverter[WorkflowAppBlockingResponse | WorkflowAppPausedBlockingResponse]
-):
+class WorkflowAppGenerateResponseConverter(AppGenerateResponseConverter):
+    _blocking_response_type = WorkflowAppBlockingResponse
+
     @classmethod
-    def convert_blocking_full_response(
-        cls, blocking_response: WorkflowAppBlockingResponse | WorkflowAppPausedBlockingResponse
-    ) -> dict[str, Any]:
+    def convert_blocking_full_response(cls, blocking_response: WorkflowAppBlockingResponse):  # type: ignore[override]
         """
         Convert blocking full response.
         :param blocking_response: blocking response
         :return:
         """
-        return dict(blocking_response.model_dump())
+        return blocking_response.model_dump()
 
     @classmethod
-    def convert_blocking_simple_response(
-        cls, blocking_response: WorkflowAppBlockingResponse | WorkflowAppPausedBlockingResponse
-    ) -> dict[str, Any]:
+    def convert_blocking_simple_response(cls, blocking_response: WorkflowAppBlockingResponse):  # type: ignore[override]
         """
         Convert blocking simple response.
         :param blocking_response: blocking response

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

@@ -42,15 +42,12 @@ from core.app.entities.queue_entities import (
 )
 from core.app.entities.task_entities import (
     ErrorStreamResponse,
-    HumanInputRequiredPauseReasonPayload,
-    HumanInputRequiredResponse,
     MessageAudioEndStreamResponse,
     MessageAudioStreamResponse,
     PingStreamResponse,
     StreamResponse,
     TextChunkStreamResponse,
     WorkflowAppBlockingResponse,
-    WorkflowAppPausedBlockingResponse,
     WorkflowAppStreamResponse,
     WorkflowFinishStreamResponse,
     WorkflowPauseStreamResponse,
@@ -121,11 +118,7 @@ class WorkflowAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         )
         self._graph_runtime_state: GraphRuntimeState | None = self._base_task_pipeline.queue_manager.graph_runtime_state
 
-    def process(
-        self,
-    ) -> Union[
-        WorkflowAppBlockingResponse, WorkflowAppPausedBlockingResponse, Generator[WorkflowAppStreamResponse, None, None]
-    ]:
+    def process(self) -> Union[WorkflowAppBlockingResponse, Generator[WorkflowAppStreamResponse, None, None]]:
         """
         Process generate task pipeline.
         :return:
@@ -136,24 +129,19 @@ class WorkflowAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         else:
             return self._to_blocking_response(generator)
 
-    def _to_blocking_response(
-        self, generator: Generator[StreamResponse, None, None]
-    ) -> Union[WorkflowAppBlockingResponse, WorkflowAppPausedBlockingResponse]:
+    def _to_blocking_response(self, generator: Generator[StreamResponse, None, None]) -> WorkflowAppBlockingResponse:
         """
         To blocking response.
         :return:
         """
-        human_input_responses: list[HumanInputRequiredResponse] = []
         for stream_response in generator:
             if isinstance(stream_response, ErrorStreamResponse):
                 raise stream_response.err
-            elif isinstance(stream_response, HumanInputRequiredResponse):
-                human_input_responses.append(stream_response)
             elif isinstance(stream_response, WorkflowPauseStreamResponse):
-                return WorkflowAppPausedBlockingResponse(
+                response = WorkflowAppBlockingResponse(
                     task_id=self._application_generate_entity.task_id,
                     workflow_run_id=stream_response.data.workflow_run_id,
-                    data=WorkflowAppPausedBlockingResponse.Data(
+                    data=WorkflowAppBlockingResponse.Data(
                         id=stream_response.data.workflow_run_id,
                         workflow_id=self._workflow.id,
                         status=stream_response.data.status,
@@ -164,13 +152,12 @@ class WorkflowAppGenerateTaskPipeline(GraphRuntimeStateSupport):
                         total_steps=stream_response.data.total_steps,
                         created_at=stream_response.data.created_at,
                         finished_at=None,
-                        paused_nodes=stream_response.data.paused_nodes,
-                        reasons=stream_response.data.reasons,
                     ),
                 )
 
+                return response
             elif isinstance(stream_response, WorkflowFinishStreamResponse):
-                return WorkflowAppBlockingResponse(
+                response = WorkflowAppBlockingResponse(
                     task_id=self._application_generate_entity.task_id,
                     workflow_run_id=stream_response.data.id,
                     data=WorkflowAppBlockingResponse.Data(
@@ -187,44 +174,12 @@ class WorkflowAppGenerateTaskPipeline(GraphRuntimeStateSupport):
                     ),
                 )
 
+                return response
             else:
                 continue
 
-        if human_input_responses:
-            return self._build_paused_blocking_response_from_human_input(human_input_responses)
-
         raise ValueError("queue listening stopped unexpectedly.")
 
-    def _build_paused_blocking_response_from_human_input(
-        self, human_input_responses: list[HumanInputRequiredResponse]
-    ) -> WorkflowAppPausedBlockingResponse:
-        runtime_state = self._resolve_graph_runtime_state()
-        paused_nodes = list(dict.fromkeys(response.data.node_id for response in human_input_responses))
-        created_at = int(runtime_state.start_at)
-        reasons = [
-            HumanInputRequiredPauseReasonPayload.from_response_data(response.data).model_dump(mode="json")
-            for response in human_input_responses
-        ]
-
-        return WorkflowAppPausedBlockingResponse(
-            task_id=self._application_generate_entity.task_id,
-            workflow_run_id=human_input_responses[-1].workflow_run_id,
-            data=WorkflowAppPausedBlockingResponse.Data(
-                id=human_input_responses[-1].workflow_run_id,
-                workflow_id=self._workflow.id,
-                status=WorkflowExecutionStatus.PAUSED,
-                outputs={},
-                error=None,
-                elapsed_time=time.perf_counter() - self._base_task_pipeline.start_at,
-                total_tokens=runtime_state.total_tokens,
-                total_steps=runtime_state.node_run_steps,
-                created_at=created_at,
-                finished_at=None,
-                paused_nodes=paused_nodes,
-                reasons=reasons,
-            ),
-        )
-
     def _to_stream_response(
         self, generator: Generator[StreamResponse, None, None]
     ) -> Generator[WorkflowAppStreamResponse, None, None]:

api/core/app/apps/workflow_app_runner.py

@@ -410,7 +410,7 @@ class WorkflowBasedAppRunner:
         elif isinstance(event, GraphRunFailedEvent):
             self._publish_event(QueueWorkflowFailedEvent(error=event.error, exceptions_count=event.exceptions_count))
         elif isinstance(event, GraphRunAbortedEvent):
-            self._publish_event(QueueWorkflowFailedEvent(error=event.reason or "Unknown error", exceptions_count=0))
+            self._publish_event(QueueWorkflowPartialSuccessEvent(outputs={}, exceptions_count=0))
         elif isinstance(event, GraphRunPausedEvent):
             runtime_state = workflow_entry.graph_engine.graph_runtime_state
             paused_nodes = runtime_state.get_paused_nodes()

api/core/app/entities/task_entities.py

@@ -1,13 +1,12 @@
 from collections.abc import Mapping, Sequence
 from enum import StrEnum
-from typing import Any, Literal
+from typing import Any
 
-from pydantic import BaseModel, ConfigDict, Field, JsonValue
+from pydantic import BaseModel, ConfigDict, Field
 
 from core.app.entities.agent_strategy import AgentStrategyInfo
 from core.rag.entities import RetrievalSourceMetadata
 from graphon.entities import WorkflowStartReason
-from graphon.entities.pause_reason import PauseReasonType
 from graphon.enums import WorkflowExecutionStatus, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
 from graphon.model_runtime.entities.llm_entities import LLMResult, LLMUsage
 from graphon.nodes.human_input.entities import FormInput, UserAction
@@ -296,40 +295,6 @@ class HumanInputRequiredResponse(StreamResponse):
     data: Data
 
 
-class HumanInputRequiredPauseReasonPayload(BaseModel):
-    """
-    Public pause-reason payload used by blocking responses when only
-    ``human_input_required`` events are available.
-    """
-
-    TYPE: Literal[PauseReasonType.HUMAN_INPUT_REQUIRED] = PauseReasonType.HUMAN_INPUT_REQUIRED
-    form_id: str
-    node_id: str
-    node_title: str
-    form_content: str
-    inputs: Sequence[FormInput] = Field(default_factory=list)
-    actions: Sequence[UserAction] = Field(default_factory=list)
-    display_in_ui: bool = False
-    form_token: str | None = None
-    resolved_default_values: Mapping[str, Any] = Field(default_factory=dict)
-    expiration_time: int
-
-    @classmethod
-    def from_response_data(cls, data: HumanInputRequiredResponse.Data) -> "HumanInputRequiredPauseReasonPayload":
-        return cls(
-            form_id=data.form_id,
-            node_id=data.node_id,
-            node_title=data.node_title,
-            form_content=data.form_content,
-            inputs=data.inputs,
-            actions=data.actions,
-            display_in_ui=data.display_in_ui,
-            form_token=data.form_token,
-            resolved_default_values=data.resolved_default_values,
-            expiration_time=data.expiration_time,
-        )
-
-
 class HumanInputFormFilledResponse(StreamResponse):
     class Data(BaseModel):
         """
@@ -390,7 +355,7 @@ class NodeStartStreamResponse(StreamResponse):
     workflow_run_id: str
     data: Data
 
-    def to_ignore_detail_dict(self) -> dict[str, JsonValue]:
+    def to_ignore_detail_dict(self):
         return {
             "event": self.event.value,
             "task_id": self.task_id,
@@ -447,7 +412,7 @@ class NodeFinishStreamResponse(StreamResponse):
     workflow_run_id: str
     data: Data
 
-    def to_ignore_detail_dict(self) -> dict[str, JsonValue]:
+    def to_ignore_detail_dict(self):
         return {
             "event": self.event.value,
             "task_id": self.task_id,
@@ -809,34 +774,6 @@ class ChatbotAppBlockingResponse(AppBlockingResponse):
     data: Data
 
 
-class AdvancedChatPausedBlockingResponse(AppBlockingResponse):
-    """
-    ChatbotAppPausedBlockingResponse entity
-    """
-
-    class Data(BaseModel):
-        """
-        Data entity
-        """
-
-        id: str
-        mode: str
-        conversation_id: str
-        message_id: str
-        workflow_run_id: str
-        answer: str
-        metadata: Mapping[str, object] = Field(default_factory=dict)
-        created_at: int
-        paused_nodes: Sequence[str] = Field(default_factory=list)
-        reasons: Sequence[Mapping[str, Any]] = Field(default_factory=list[Mapping[str, Any]])
-        status: WorkflowExecutionStatus
-        elapsed_time: float
-        total_tokens: int
-        total_steps: int
-
-    data: Data
-
-
 class CompletionAppBlockingResponse(AppBlockingResponse):
     """
     CompletionAppBlockingResponse entity
@@ -882,33 +819,6 @@ class WorkflowAppBlockingResponse(AppBlockingResponse):
     data: Data
 
 
-class WorkflowAppPausedBlockingResponse(AppBlockingResponse):
-    """
-    WorkflowAppPausedBlockingResponse entity
-    """
-
-    class Data(BaseModel):
-        """
-        Data entity
-        """
-
-        id: str
-        workflow_id: str
-        status: WorkflowExecutionStatus
-        outputs: Mapping[str, Any] | None = None
-        error: str | None = None
-        elapsed_time: float
-        total_tokens: int
-        total_steps: int
-        created_at: int
-        finished_at: int | None
-        paused_nodes: Sequence[str] = Field(default_factory=list)
-        reasons: Sequence[Mapping[str, Any]] = Field(default_factory=list)
-
-    workflow_run_id: str
-    data: Data
-
-
 class AgentLogStreamResponse(StreamResponse):
     """
     AgentLogStreamResponse entity

api/core/app/llm/model_access.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-from copy import deepcopy
 from typing import Any
 
 from core.app.entities.app_invoke_entities import DifyRunContext, ModelConfigWithCredentialsEntity
@@ -15,21 +14,8 @@ from graphon.nodes.llm.protocols import CredentialsProvider
 
 
 class DifyCredentialsProvider:
-    """Resolves and returns LLM credentials for a given provider and model.
-
-    Fetched credentials are stored in :attr:`credentials_cache` and reused for
-    subsequent ``fetch`` calls for the same ``(provider_name, model_name)``.
-    Because of that cache, a single instance can return stale credentials after
-    the tenant or provider configuration changes (e.g. API key rotation).
-
-    Do **not** keep one instance for the lifetime of a process or across
-    unrelated invocations. Create a new provider per request, workflow run, or
-    other bounded scope where up-to-date credentials matter.
-    """
-
     tenant_id: str
     provider_manager: ProviderManager
-    credentials_cache: dict[tuple[str, str], dict[str, Any]]
 
     def __init__(
         self,
@@ -44,12 +30,8 @@ class DifyCredentialsProvider:
                 user_id=run_context.user_id,
             )
         self.provider_manager = provider_manager
-        self.credentials_cache = {}
 
     def fetch(self, provider_name: str, model_name: str) -> dict[str, Any]:
-        if (provider_name, model_name) in self.credentials_cache:
-            return deepcopy(self.credentials_cache[(provider_name, model_name)])
-
         provider_configurations = self.provider_manager.get_configurations(self.tenant_id)
         provider_configuration = provider_configurations.get(provider_name)
         if not provider_configuration:
@@ -64,7 +46,6 @@ class DifyCredentialsProvider:
         if credentials is None:
             raise ProviderTokenNotInitError(f"Model {model_name} credentials is not initialized.")
 
-        self.credentials_cache[(provider_name, model_name)] = deepcopy(credentials)
         return credentials
 
 
@@ -84,8 +65,7 @@ class DifyModelFactory:
                 provider_manager=create_plugin_provider_manager(
                     tenant_id=run_context.tenant_id,
                     user_id=run_context.user_id,
-                ),
-                enable_credentials_cache=True,
+                )
             )
         self.model_manager = model_manager
 
@@ -104,7 +84,7 @@ def build_dify_model_access(run_context: DifyRunContext) -> tuple[CredentialsPro
         tenant_id=run_context.tenant_id,
         user_id=run_context.user_id,
     )
-    model_manager = ModelManager(provider_manager=provider_manager, enable_credentials_cache=True)
+    model_manager = ModelManager(provider_manager=provider_manager)
 
     return (
         DifyCredentialsProvider(run_context=run_context, provider_manager=provider_manager),

api/core/helper/creators.py

@@ -1,41 +0,0 @@
-"""
-Helper module for Creators Platform integration.
-
-Provides functionality to upload DSL files to the Creators Platform
-and generate redirect URLs with OAuth authorization codes.
-"""
-
-import logging
-from urllib.parse import urlencode
-
-import httpx
-from yarl import URL
-
-from configs import dify_config
-
-logger = logging.getLogger(__name__)
-
-creators_platform_api_url = URL(str(dify_config.CREATORS_PLATFORM_API_URL))
-
-
-def upload_dsl(dsl_file_bytes: bytes, filename: str = "template.yaml") -> str:
-    url = str(creators_platform_api_url / "api/v1/templates/anonymous-upload")
-    response = httpx.post(url, files={"file": (filename, dsl_file_bytes)}, timeout=30)
-    response.raise_for_status()
-    data = response.json()
-    claim_code = data.get("data", {}).get("claim_code")
-    if not claim_code:
-        raise ValueError("Creators Platform did not return a valid claim_code")
-    return claim_code
-
-
-def get_redirect_url(user_account_id: str, claim_code: str) -> str:
-    base_url = str(dify_config.CREATORS_PLATFORM_API_URL).rstrip("/")
-    params: dict[str, str] = {"dsl_claim_code": claim_code}
-    client_id = str(dify_config.CREATORS_PLATFORM_OAUTH_CLIENT_ID or "")
-    if client_id:
-        from services.oauth_server import OAuthServerService
-
-        oauth_code = OAuthServerService.sign_oauth_authorization_code(client_id, user_account_id)
-        params["oauth_code"] = oauth_code
-    return f"{base_url}?{urlencode(params)}"

api/core/llm_generator/llm_generator.py

@@ -2,7 +2,7 @@ import json
 import logging
 import re
 from collections.abc import Sequence
-from typing import Any, NotRequired, Protocol, TypedDict, cast
+from typing import Any, Protocol, TypedDict, cast
 
 import json_repair
 from sqlalchemy import select
@@ -18,6 +18,8 @@ from core.llm_generator.prompts import (
     LLM_MODIFY_CODE_SYSTEM,
     LLM_MODIFY_PROMPT_SYSTEM,
     PYTHON_CODE_GENERATOR_PROMPT_TEMPLATE,
+    SUGGESTED_QUESTIONS_MAX_TOKENS,
+    SUGGESTED_QUESTIONS_TEMPERATURE,
     SYSTEM_STRUCTURED_OUTPUT_GENERATE,
     WORKFLOW_RULE_CONFIG_PROMPT_GENERATE_TEMPLATE,
 )
@@ -39,36 +41,6 @@ from models.workflow import Workflow
 logger = logging.getLogger(__name__)
 
 
-class SuggestedQuestionsModelConfig(TypedDict):
-    provider: str
-    name: str
-    completion_params: NotRequired[dict[str, object]]
-
-
-def _normalize_completion_params(completion_params: dict[str, object]) -> tuple[dict[str, object], list[str]]:
-    """
-    Normalize raw completion params into invocation parameters and stop sequences.
-
-    This mirrors the app-model access path by separating ``stop`` from provider
-    parameters before invocation, then drops non-positive token limits because
-    some plugin-backed models reject ``0`` after mapping ``max_tokens`` to their
-    provider-specific output-token field.
-    """
-    normalized_parameters = dict(completion_params)
-    stop_value = normalized_parameters.pop("stop", [])
-    if isinstance(stop_value, list) and all(isinstance(item, str) for item in stop_value):
-        stop = stop_value
-    else:
-        stop = []
-
-    for token_limit_key in ("max_tokens", "max_output_tokens"):
-        token_limit = normalized_parameters.get(token_limit_key)
-        if isinstance(token_limit, int | float) and token_limit <= 0:
-            normalized_parameters.pop(token_limit_key, None)
-
-    return normalized_parameters, stop
-
-
 class WorkflowServiceInterface(Protocol):
     def get_draft_workflow(self, app_model: App, workflow_id: str | None = None) -> Workflow | None:
         pass
@@ -151,15 +123,8 @@ class LLMGenerator:
         return name
 
     @classmethod
-    def generate_suggested_questions_after_answer(
-        cls,
-        tenant_id: str,
-        histories: str,
-        *,
-        instruction_prompt: str | None = None,
-        model_config: object | None = None,
-    ) -> Sequence[str]:
-        output_parser = SuggestedQuestionsAfterAnswerOutputParser(instruction_prompt=instruction_prompt)
+    def generate_suggested_questions_after_answer(cls, tenant_id: str, histories: str) -> Sequence[str]:
+        output_parser = SuggestedQuestionsAfterAnswerOutputParser()
         format_instructions = output_parser.get_format_instructions()
 
         prompt_template = PromptTemplateParser(template="{{histories}}\n{{format_instructions}}\nquestions:\n")
@@ -168,36 +133,10 @@ class LLMGenerator:
 
         try:
             model_manager = ModelManager.for_tenant(tenant_id=tenant_id)
-            configured_model = cast(dict[str, object], model_config) if isinstance(model_config, dict) else {}
-            provider = configured_model.get("provider")
-            model_name = configured_model.get("name")
-            use_configured_model = False
-
-            if isinstance(provider, str) and provider and isinstance(model_name, str) and model_name:
-                try:
-                    model_instance = model_manager.get_model_instance(
-                        tenant_id=tenant_id,
-                        model_type=ModelType.LLM,
-                        provider=provider,
-                        model=model_name,
-                    )
-                    use_configured_model = True
-                except Exception:
-                    logger.warning(
-                        "Failed to use configured suggested-questions model %s/%s, fallback to default model",
-                        provider,
-                        model_name,
-                        exc_info=True,
-                    )
-                    model_instance = model_manager.get_default_model_instance(
-                        tenant_id=tenant_id,
-                        model_type=ModelType.LLM,
-                    )
-            else:
-                model_instance = model_manager.get_default_model_instance(
-                    tenant_id=tenant_id,
-                    model_type=ModelType.LLM,
-                )
+            model_instance = model_manager.get_default_model_instance(
+                tenant_id=tenant_id,
+                model_type=ModelType.LLM,
+            )
         except InvokeAuthorizationError:
             return []
 
@@ -206,29 +145,19 @@ class LLMGenerator:
         questions: Sequence[str] = []
 
         try:
-            configured_completion_params = configured_model.get("completion_params")
-            if use_configured_model and isinstance(configured_completion_params, dict):
-                model_parameters, stop = _normalize_completion_params(configured_completion_params)
-            elif use_configured_model:
-                model_parameters = {}
-                stop = []
-            else:
-                # Default-model generation keeps the built-in suggested-questions tuning.
-                model_parameters = {
-                    "max_tokens": 2560,
-                    "temperature": 0.0,
-                }
-                stop = []
-
             response: LLMResult = model_instance.invoke_llm(
                 prompt_messages=list(prompt_messages),
-                model_parameters=model_parameters,
-                stop=stop,
+                model_parameters={
+                    "max_tokens": SUGGESTED_QUESTIONS_MAX_TOKENS,
+                    "temperature": SUGGESTED_QUESTIONS_TEMPERATURE,
+                },
                 stream=False,
             )
 
             text_content = response.message.get_text_content()
             questions = output_parser.parse(text_content) if text_content else []
+        except InvokeError:
+            questions = []
         except Exception:
             logger.exception("Failed to generate suggested questions after answer")
             questions = []

api/core/llm_generator/output_parser/suggested_questions_after_answer.py

@@ -3,28 +3,17 @@ import logging
 import re
 from collections.abc import Sequence
 
-from core.llm_generator.prompts import DEFAULT_SUGGESTED_QUESTIONS_AFTER_ANSWER_INSTRUCTION_PROMPT
+from core.llm_generator.prompts import SUGGESTED_QUESTIONS_AFTER_ANSWER_INSTRUCTION_PROMPT
 
 logger = logging.getLogger(__name__)
 
 
 class SuggestedQuestionsAfterAnswerOutputParser:
-    def __init__(self, instruction_prompt: str | None = None) -> None:
-        self._instruction_prompt = self._build_instruction_prompt(instruction_prompt)
-
-    @staticmethod
-    def _build_instruction_prompt(instruction_prompt: str | None) -> str:
-        if not instruction_prompt or not instruction_prompt.strip():
-            return DEFAULT_SUGGESTED_QUESTIONS_AFTER_ANSWER_INSTRUCTION_PROMPT
-
-        return f'{instruction_prompt}\nYou must output a JSON array like ["question1", "question2", "question3"].'
-
     def get_format_instructions(self) -> str:
-        return self._instruction_prompt
+        return SUGGESTED_QUESTIONS_AFTER_ANSWER_INSTRUCTION_PROMPT
 
     def parse(self, text: str) -> Sequence[str]:
-        stripped_text = text.strip()
-        action_match = re.search(r"\[.*?\]", stripped_text, re.DOTALL)
+        action_match = re.search(r"\[.*?\]", text.strip(), re.DOTALL)
         questions: list[str] = []
         if action_match is not None:
             try:
@@ -34,6 +23,4 @@ class SuggestedQuestionsAfterAnswerOutputParser:
             else:
                 if isinstance(json_obj, list):
                     questions = [question for question in json_obj if isinstance(question, str)]
-        elif stripped_text:
-            logger.warning("Failed to find suggested questions payload array in text: %r", stripped_text[:200])
         return questions

api/core/llm_generator/prompts.py

@@ -1,4 +1,5 @@
 # Written by YORKI MINAKO🤡, Edited by Xiaoyi, Edited by yasu-oh
+import os
 
 CONVERSATION_TITLE_PROMPT = """You are asked to generate a concise chat title by decomposing the user’s input into two parts: “Intention” and “Subject”.
 
@@ -95,8 +96,8 @@ JAVASCRIPT_CODE_GENERATOR_PROMPT_TEMPLATE = (
 )
 
 
-# Default prompt and model parameters for suggested questions.
-DEFAULT_SUGGESTED_QUESTIONS_AFTER_ANSWER_INSTRUCTION_PROMPT = (
+# Default prompt for suggested questions (can be overridden by environment variable)
+_DEFAULT_SUGGESTED_QUESTIONS_AFTER_ANSWER_PROMPT = (
     "Please help me predict the three most likely questions that human would ask, "
     "and keep each question under 20 characters.\n"
     "MAKE SURE your output is the SAME language as the Assistant's latest response. "
@@ -104,6 +105,15 @@ DEFAULT_SUGGESTED_QUESTIONS_AFTER_ANSWER_INSTRUCTION_PROMPT = (
     '["question1","question2","question3"]\n'
 )
 
+# Environment variable override for suggested questions prompt
+SUGGESTED_QUESTIONS_AFTER_ANSWER_INSTRUCTION_PROMPT = os.getenv(
+    "SUGGESTED_QUESTIONS_PROMPT", _DEFAULT_SUGGESTED_QUESTIONS_AFTER_ANSWER_PROMPT
+)
+
+# Configurable LLM parameters for suggested questions (can be overridden by environment variables)
+SUGGESTED_QUESTIONS_MAX_TOKENS = int(os.getenv("SUGGESTED_QUESTIONS_MAX_TOKENS", "256"))
+SUGGESTED_QUESTIONS_TEMPERATURE = float(os.getenv("SUGGESTED_QUESTIONS_TEMPERATURE", "0"))
+
 GENERATOR_QA_PROMPT = (
     "<Task> The user will send a long text. Generate a Question and Answer pairs only using the knowledge"
     " in the long text. Please think step by step."

api/core/model_manager.py

@@ -1,6 +1,5 @@
 import logging
 from collections.abc import Callable, Generator, Iterable, Mapping, Sequence
-from copy import deepcopy
 from typing import IO, Any, Literal, Optional, ParamSpec, TypeVar, Union, cast, overload
 
 from configs import dify_config
@@ -37,13 +36,11 @@ class ModelInstance:
     Model instance class.
     """
 
-    def __init__(self, provider_model_bundle: ProviderModelBundle, model: str, credentials: dict | None = None) -> None:
+    def __init__(self, provider_model_bundle: ProviderModelBundle, model: str):
         self.provider_model_bundle = provider_model_bundle
         self.model_name = model
         self.provider = provider_model_bundle.configuration.provider.provider
-        if credentials is None:
-            credentials = self._fetch_credentials_from_bundle(provider_model_bundle, model)
-        self.credentials = credentials
+        self.credentials = self._fetch_credentials_from_bundle(provider_model_bundle, model)
         # Runtime LLM invocation fields.
         self.parameters: Mapping[str, Any] = {}
         self.stop: Sequence[str] = ()
@@ -437,30 +434,8 @@ class ModelInstance:
 
 
 class ModelManager:
-    """Resolves :class:`ModelInstance` objects for a tenant and provider.
-
-    When ``enable_credentials_cache`` is ``True``, resolved credentials for each
-    ``(tenant_id, provider, model_type, model)`` are stored in
-    ``_credentials_cache`` and reused. That can return **stale** credentials after
-    API keys or provider settings change, so a manager constructed with
-    ``enable_credentials_cache=True`` should not be kept for the lifetime of a
-    process or shared across unrelated work. Prefer a new manager per request,
-    workflow run, or similar bounded scope.
-
-    The default is ``enable_credentials_cache=False``; in that mode the internal
-    credential cache is not populated, and each ``get_model_instance`` call
-    loads credentials from the current provider configuration.
-    """
-
-    def __init__(
-        self,
-        provider_manager: ProviderManager,
-        *,
-        enable_credentials_cache: bool = False,
-    ) -> None:
+    def __init__(self, provider_manager: ProviderManager):
         self._provider_manager = provider_manager
-        self._credentials_cache: dict[tuple[str, str, str, str], Any] = {}
-        self._enable_credentials_cache = enable_credentials_cache
 
     @classmethod
     def for_tenant(cls, tenant_id: str, user_id: str | None = None) -> "ModelManager":
@@ -488,19 +463,8 @@ class ModelManager:
             tenant_id=tenant_id, provider=provider, model_type=model_type
         )
 
-        cred_cache_key = (tenant_id, provider, model_type.value, model)
-
-        if cred_cache_key in self._credentials_cache:
-            return ModelInstance(
-                provider_model_bundle,
-                model,
-                deepcopy(self._credentials_cache[cred_cache_key]),
-            )
-
-        ret = ModelInstance(provider_model_bundle, model)
-        if self._enable_credentials_cache:
-            self._credentials_cache[cred_cache_key] = deepcopy(ret.credentials)
-        return ret
+        model_instance = ModelInstance(provider_model_bundle, model)
+        return model_instance
 
     def get_default_provider_model_name(self, tenant_id: str, model_type: ModelType) -> tuple[str | None, str | None]:
         """

api/core/rag/datasource/keyword/jieba/jieba.py

@@ -156,8 +156,7 @@ class Jieba(BaseKeyword):
         if dataset_keyword_table:
             keyword_table_dict = dataset_keyword_table.keyword_table_dict
             if keyword_table_dict:
-                data: Any = keyword_table_dict["__data__"]
-                return dict(data["table"])
+                return dict(keyword_table_dict["__data__"]["table"])
         else:
             keyword_data_source_type = dify_config.KEYWORD_DATA_SOURCE_TYPE
             dataset_keyword_table = DatasetKeywordTable(

api/core/rag/datasource/keyword/jieba/jieba_keyword_table_handler.py

@@ -109,7 +109,7 @@ class JiebaKeywordTableHandler:
         """Extract keywords with JIEBA tfidf."""
         keywords = self._tfidf.extract_tags(
             sentence=text,
-            topK=max_keywords_per_chunk or 10,
+            topK=max_keywords_per_chunk,
         )
         # jieba.analyse.extract_tags returns an untyped list when withFlag is False by default.
         keywords = cast(list[str], keywords)

api/core/rag/datasource/retrieval_service.py

@@ -551,7 +551,6 @@ class RetrievalService:
                 child_index_nodes = session.execute(child_chunk_stmt).scalars().all()
 
                 for i in child_index_nodes:
-                    assert i.index_node_id
                     segment_ids.append(i.segment_id)
                     if i.segment_id in child_chunk_map:
                         child_chunk_map[i.segment_id].append(i)

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

@@ -39,58 +39,6 @@ class AbstractVectorFactory(ABC):
         return index_struct_dict
 
 
-class _LazyEmbeddings(Embeddings):
-    """Lazy proxy that defers materializing the real embedding model.
-
-    Constructing the real embeddings (via ``ModelManager.get_model_instance``)
-    transitively calls ``FeatureService.get_features`` → ``BillingService``
-    HTTP GETs (see ``provider_manager.py``). Cleanup paths
-    (``delete_by_ids`` / ``delete`` / ``text_exists``) do not need embeddings
-    at all, so deferring this until an ``embed_*`` method is actually invoked
-    keeps cleanup tasks resilient to transient billing-API failures and avoids
-    leaving stranded ``document_segments`` / ``child_chunks`` whenever billing
-    hiccups.
-
-    Existing callers that perform create / search operations are unaffected:
-    the first ``embed_*`` call materializes the underlying model and the
-    behavior is identical from that point on.
-    """
-
-    def __init__(self, dataset: Dataset):
-        self._dataset = dataset
-        self._real: Embeddings | None = None
-
-    def _ensure(self) -> Embeddings:
-        if self._real is None:
-            model_manager = ModelManager.for_tenant(tenant_id=self._dataset.tenant_id)
-            embedding_model = model_manager.get_model_instance(
-                tenant_id=self._dataset.tenant_id,
-                provider=self._dataset.embedding_model_provider,
-                model_type=ModelType.TEXT_EMBEDDING,
-                model=self._dataset.embedding_model,
-            )
-            self._real = CacheEmbedding(embedding_model)
-        return self._real
-
-    def embed_documents(self, texts: list[str]) -> list[list[float]]:
-        return self._ensure().embed_documents(texts)
-
-    def embed_multimodal_documents(self, multimodel_documents: list[dict[str, Any]]) -> list[list[float]]:
-        return self._ensure().embed_multimodal_documents(multimodel_documents)
-
-    def embed_query(self, text: str) -> list[float]:
-        return self._ensure().embed_query(text)
-
-    def embed_multimodal_query(self, multimodel_document: dict[str, Any]) -> list[float]:
-        return self._ensure().embed_multimodal_query(multimodel_document)
-
-    async def aembed_documents(self, texts: list[str]) -> list[list[float]]:
-        return await self._ensure().aembed_documents(texts)
-
-    async def aembed_query(self, text: str) -> list[float]:
-        return await self._ensure().aembed_query(text)
-
-
 class Vector:
     def __init__(self, dataset: Dataset, attributes: list | None = None):
         if attributes is None:
@@ -112,11 +60,7 @@ class Vector:
                 "original_chunk_id",
             ]
         self._dataset = dataset
-        # Use a lazy proxy so cleanup paths (delete_by_ids / delete / text_exists)
-        # never transitively trigger billing API calls during ``Vector(dataset)``
-        # construction. The real embedding model is materialized only when an
-        # ``embed_*`` method is actually invoked (i.e. create / search paths).
-        self._embeddings: Embeddings = _LazyEmbeddings(dataset)
+        self._embeddings = self._get_embeddings()
         self._attributes = attributes
         self._vector_processor = self._init_vector()
 

api/core/rag/docstore/dataset_docstore.py

@@ -11,7 +11,6 @@ from core.rag.models.document import AttachmentDocument, Document
 from extensions.ext_database import db
 from graphon.model_runtime.entities.model_entities import ModelType
 from models.dataset import ChildChunk, Dataset, DocumentSegment, SegmentAttachmentBinding
-from models.enums import SegmentType
 
 
 class DatasetDocumentStore:
@@ -128,7 +127,6 @@ class DatasetDocumentStore:
                 if save_child:
                     if doc.children:
                         for position, child in enumerate(doc.children, start=1):
-                            assert self._document_id
                             child_segment = ChildChunk(
                                 tenant_id=self._dataset.tenant_id,
                                 dataset_id=self._dataset.id,
@@ -139,7 +137,7 @@ class DatasetDocumentStore:
                                 index_node_hash=child.metadata.get("doc_hash"),
                                 content=child.page_content,
                                 word_count=len(child.page_content),
-                                type=SegmentType.AUTOMATIC,
+                                type="automatic",
                                 created_by=self._user_id,
                             )
                             db.session.add(child_segment)
@@ -165,7 +163,6 @@ class DatasetDocumentStore:
                     )
                     # add new child chunks
                     for position, child in enumerate(doc.children, start=1):
-                        assert self._document_id
                         child_segment = ChildChunk(
                             tenant_id=self._dataset.tenant_id,
                             dataset_id=self._dataset.id,
@@ -176,7 +173,7 @@ class DatasetDocumentStore:
                             index_node_hash=child.metadata.get("doc_hash"),
                             content=child.page_content,
                             word_count=len(child.page_content),
-                            type=SegmentType.AUTOMATIC,
+                            type="automatic",
                             created_by=self._user_id,
                         )
                         db.session.add(child_segment)

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

@@ -31,7 +31,7 @@ class FunctionCallMultiDatasetRouter:
             result: LLMResult = model_instance.invoke_llm(  # pyright: ignore[reportCallIssue, reportArgumentType]
                 prompt_messages=prompt_messages,
                 tools=dataset_tools,
-                stream=False,  # pyright: ignore[reportArgumentType]
+                stream=False,
                 model_parameters={"temperature": 0.2, "top_p": 0.3, "max_tokens": 1500},
             )
             usage = result.usage or LLMUsage.empty_usage()

api/core/tools/utils/system_oauth_encryption.py

@@ -14,23 +14,23 @@ from configs import dify_config
 logger = logging.getLogger(__name__)
 
 
-class EncryptionError(Exception):
-    """Encryption/decryption specific error"""
+class OAuthEncryptionError(Exception):
+    """OAuth encryption/decryption specific error"""
 
     pass
 
 
-class SystemEncrypter:
+class SystemOAuthEncrypter:
     """
-    A simple parameters encrypter using AES-CBC encryption.
+    A simple OAuth parameters encrypter using AES-CBC encryption.
 
-    This class provides methods to encrypt and decrypt parameters
+    This class provides methods to encrypt and decrypt OAuth parameters
     using AES-CBC mode with a key derived from the application's SECRET_KEY.
     """
 
     def __init__(self, secret_key: str | None = None):
         """
-        Initialize the encrypter.
+        Initialize the OAuth encrypter.
 
         Args:
             secret_key: Optional secret key. If not provided, uses dify_config.SECRET_KEY
@@ -43,19 +43,19 @@ class SystemEncrypter:
         # Generate a fixed 256-bit key using SHA-256
         self.key = hashlib.sha256(secret_key.encode()).digest()
 
-    def encrypt_params(self, params: Mapping[str, Any]) -> str:
+    def encrypt_oauth_params(self, oauth_params: Mapping[str, Any]) -> str:
         """
-        Encrypt parameters.
+        Encrypt OAuth parameters.
 
         Args:
-            params: Parameters dictionary, e.g., {"client_id": "xxx", "client_secret": "xxx"}
+            oauth_params: OAuth parameters dictionary, e.g., {"client_id": "xxx", "client_secret": "xxx"}
 
         Returns:
             Base64-encoded encrypted string
 
         Raises:
-            EncryptionError: If encryption fails
-            ValueError: If params is invalid
+            OAuthEncryptionError: If encryption fails
+            ValueError: If oauth_params is invalid
         """
 
         try:
@@ -66,7 +66,7 @@ class SystemEncrypter:
             cipher = AES.new(self.key, AES.MODE_CBC, iv)
 
             # Encrypt data
-            padded_data = pad(TypeAdapter(dict).dump_json(dict(params)), AES.block_size)
+            padded_data = pad(TypeAdapter(dict).dump_json(dict(oauth_params)), AES.block_size)
             encrypted_data = cipher.encrypt(padded_data)
 
             # Combine IV and encrypted data
@@ -76,20 +76,20 @@ class SystemEncrypter:
             return base64.b64encode(combined).decode()
 
         except Exception as e:
-            raise EncryptionError(f"Encryption failed: {str(e)}") from e
+            raise OAuthEncryptionError(f"Encryption failed: {str(e)}") from e
 
-    def decrypt_params(self, encrypted_data: str) -> Mapping[str, Any]:
+    def decrypt_oauth_params(self, encrypted_data: str) -> Mapping[str, Any]:
         """
-        Decrypt parameters.
+        Decrypt OAuth parameters.
 
         Args:
             encrypted_data: Base64-encoded encrypted string
 
         Returns:
-            Decrypted parameters dictionary
+            Decrypted OAuth parameters dictionary
 
         Raises:
-            EncryptionError: If decryption fails
+            OAuthEncryptionError: If decryption fails
             ValueError: If encrypted_data is invalid
         """
         if not isinstance(encrypted_data, str):
@@ -118,70 +118,70 @@ class SystemEncrypter:
             unpadded_data = unpad(decrypted_data, AES.block_size)
 
             # Parse JSON
-            params: Mapping[str, Any] = TypeAdapter(Mapping[str, Any]).validate_json(unpadded_data)
+            oauth_params: Mapping[str, Any] = TypeAdapter(Mapping[str, Any]).validate_json(unpadded_data)
 
-            if not isinstance(params, dict):
+            if not isinstance(oauth_params, dict):
                 raise ValueError("Decrypted data is not a valid dictionary")
 
-            return params
+            return oauth_params
 
         except Exception as e:
-            raise EncryptionError(f"Decryption failed: {str(e)}") from e
+            raise OAuthEncryptionError(f"Decryption failed: {str(e)}") from e
 
 
 # Factory function for creating encrypter instances
-def create_system_encrypter(secret_key: str | None = None) -> SystemEncrypter:
+def create_system_oauth_encrypter(secret_key: str | None = None) -> SystemOAuthEncrypter:
     """
-    Create an encrypter instance.
+    Create an OAuth encrypter instance.
 
     Args:
         secret_key: Optional secret key. If not provided, uses dify_config.SECRET_KEY
 
     Returns:
-        SystemEncrypter instance
+        SystemOAuthEncrypter instance
     """
-    return SystemEncrypter(secret_key=secret_key)
+    return SystemOAuthEncrypter(secret_key=secret_key)
 
 
 # Global encrypter instance (for backward compatibility)
-_encrypter: SystemEncrypter | None = None
+_oauth_encrypter: SystemOAuthEncrypter | None = None
 
 
-def get_system_encrypter() -> SystemEncrypter:
+def get_system_oauth_encrypter() -> SystemOAuthEncrypter:
     """
-    Get the global encrypter instance.
+    Get the global OAuth encrypter instance.
 
     Returns:
-        SystemEncrypter instance
+        SystemOAuthEncrypter instance
     """
-    global _encrypter
-    if _encrypter is None:
-        _encrypter = SystemEncrypter()
-    return _encrypter
+    global _oauth_encrypter
+    if _oauth_encrypter is None:
+        _oauth_encrypter = SystemOAuthEncrypter()
+    return _oauth_encrypter
 
 
 # Convenience functions for backward compatibility
-def encrypt_system_params(params: Mapping[str, Any]) -> str:
+def encrypt_system_oauth_params(oauth_params: Mapping[str, Any]) -> str:
     """
-    Encrypt parameters using the global encrypter.
+    Encrypt OAuth parameters using the global encrypter.
 
     Args:
-        params: Parameters dictionary
+        oauth_params: OAuth parameters dictionary
 
     Returns:
         Base64-encoded encrypted string
     """
-    return get_system_encrypter().encrypt_params(params)
+    return get_system_oauth_encrypter().encrypt_oauth_params(oauth_params)
 
 
-def decrypt_system_params(encrypted_data: str) -> Mapping[str, Any]:
+def decrypt_system_oauth_params(encrypted_data: str) -> Mapping[str, Any]:
     """
-    Decrypt parameters using the global encrypter.
+    Decrypt OAuth parameters using the global encrypter.
 
     Args:
         encrypted_data: Base64-encoded encrypted string
 
     Returns:
-        Decrypted parameters dictionary
+        Decrypted OAuth parameters dictionary
     """
-    return get_system_encrypter().decrypt_params(encrypted_data)
+    return get_system_oauth_encrypter().decrypt_oauth_params(encrypted_data)

api/core/workflow/human_input_forms.py

@@ -12,16 +12,20 @@ from collections.abc import Sequence
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
-from core.workflow.human_input_policy import HumanInputSurface, get_preferred_form_token
 from extensions.ext_database import db
 from models.human_input import HumanInputFormRecipient, RecipientType
 
+_FORM_TOKEN_PRIORITY = {
+    RecipientType.BACKSTAGE: 0,
+    RecipientType.CONSOLE: 1,
+    RecipientType.STANDALONE_WEB_APP: 2,
+}
+
 
 def load_form_tokens_by_form_id(
     form_ids: Sequence[str],
     *,
     session: Session | None = None,
-    surface: HumanInputSurface | None = None,
 ) -> dict[str, str]:
     """Load the preferred access token for each human input form."""
     unique_form_ids = list(dict.fromkeys(form_ids))
@@ -29,43 +33,23 @@ def load_form_tokens_by_form_id(
         return {}
 
     if session is not None:
-        return _load_form_tokens_by_form_id(session, unique_form_ids, surface=surface)
+        return _load_form_tokens_by_form_id(session, unique_form_ids)
 
     with Session(bind=db.engine, expire_on_commit=False) as new_session:
-        return _load_form_tokens_by_form_id(new_session, unique_form_ids, surface=surface)
+        return _load_form_tokens_by_form_id(new_session, unique_form_ids)
 
 
-def _load_form_tokens_by_form_id(
-    session: Session,
-    form_ids: Sequence[str],
-    *,
-    surface: HumanInputSurface | None = None,
-) -> dict[str, str]:
-    recipients_by_form_id: dict[str, list[tuple[RecipientType, str]]] = {}
+def _load_form_tokens_by_form_id(session: Session, form_ids: Sequence[str]) -> dict[str, str]:
+    tokens_by_form_id: dict[str, tuple[int, str]] = {}
     stmt = select(HumanInputFormRecipient).where(HumanInputFormRecipient.form_id.in_(form_ids))
     for recipient in session.scalars(stmt):
-        if not recipient.access_token:
+        priority = _FORM_TOKEN_PRIORITY.get(recipient.recipient_type)
+        if priority is None or not recipient.access_token:
             continue
-        recipients_by_form_id.setdefault(recipient.form_id, []).append(
-            (recipient.recipient_type, recipient.access_token)
-        )
 
-    tokens_by_form_id: dict[str, str] = {}
-    for form_id, recipients in recipients_by_form_id.items():
-        token = _get_surface_form_token(recipients, surface=surface)
-        if token is not None:
-            tokens_by_form_id[form_id] = token
-    return tokens_by_form_id
+        candidate = (priority, recipient.access_token)
+        current = tokens_by_form_id.get(recipient.form_id)
+        if current is None or candidate[0] < current[0]:
+            tokens_by_form_id[recipient.form_id] = candidate
 
-
-def _get_surface_form_token(
-    recipients: Sequence[tuple[RecipientType, str]],
-    *,
-    surface: HumanInputSurface | None,
-) -> str | None:
-    if surface == HumanInputSurface.SERVICE_API:
-        for recipient_type, token in recipients:
-            if recipient_type == RecipientType.STANDALONE_WEB_APP and token:
-                return token
-
-    return get_preferred_form_token(recipients)
+    return {form_id: token for form_id, (_, token) in tokens_by_form_id.items()}

api/core/workflow/human_input_policy.py

@@ -1,73 +0,0 @@
-from __future__ import annotations
-
-from collections.abc import Mapping, Sequence
-from enum import StrEnum
-from typing import Any
-
-from graphon.entities.pause_reason import PauseReasonType
-from models.human_input import RecipientType
-
-
-class HumanInputSurface(StrEnum):
-    SERVICE_API = "service_api"
-    CONSOLE = "console"
-
-
-# Service API is intentionally narrower than other surfaces: app-token callers
-# should only be able to act on end-user web forms, not internal console flows.
-_ALLOWED_RECIPIENT_TYPES_BY_SURFACE: dict[HumanInputSurface, frozenset[RecipientType]] = {
-    HumanInputSurface.SERVICE_API: frozenset({RecipientType.STANDALONE_WEB_APP}),
-    HumanInputSurface.CONSOLE: frozenset({RecipientType.CONSOLE, RecipientType.BACKSTAGE}),
-}
-
-# A single HITL form can have multiple recipient records; this shared priority
-# keeps every API surface consistent about which resume token to expose.
-_RECIPIENT_TOKEN_PRIORITY: dict[RecipientType, int] = {
-    RecipientType.BACKSTAGE: 0,
-    RecipientType.CONSOLE: 1,
-    RecipientType.STANDALONE_WEB_APP: 2,
-}
-
-
-def is_recipient_type_allowed_for_surface(
-    recipient_type: RecipientType | None,
-    surface: HumanInputSurface,
-) -> bool:
-    if recipient_type is None:
-        return False
-    return recipient_type in _ALLOWED_RECIPIENT_TYPES_BY_SURFACE[surface]
-
-
-def get_preferred_form_token(
-    recipients: Sequence[tuple[RecipientType, str]],
-) -> str | None:
-    chosen_token: str | None = None
-    chosen_priority: int | None = None
-    for recipient_type, token in recipients:
-        priority = _RECIPIENT_TOKEN_PRIORITY.get(recipient_type)
-        if priority is None or not token:
-            continue
-        if chosen_priority is None or priority < chosen_priority:
-            chosen_priority = priority
-            chosen_token = token
-    return chosen_token
-
-
-def enrich_human_input_pause_reasons(
-    reasons: Sequence[Mapping[str, Any]],
-    *,
-    form_tokens_by_form_id: Mapping[str, str],
-    expiration_times_by_form_id: Mapping[str, int],
-) -> list[dict[str, Any]]:
-    enriched: list[dict[str, Any]] = []
-    for reason in reasons:
-        updated = dict(reason)
-        if updated.get("TYPE") == PauseReasonType.HUMAN_INPUT_REQUIRED:
-            form_id = updated.get("form_id")
-            if isinstance(form_id, str):
-                updated["form_token"] = form_tokens_by_form_id.get(form_id)
-                expiration_time = expiration_times_by_form_id.get(form_id)
-                if expiration_time is not None:
-                    updated["expiration_time"] = expiration_time
-        enriched.append(updated)
-    return enriched

api/dev/generate_swagger_specs.py

@@ -1,172 +0,0 @@
-"""Generate Flask-RESTX Swagger 2.0 specs without booting the full backend.
-
-This helper intentionally avoids `app_factory.create_app()`. The normal backend
-startup eagerly initializes database, Redis, Celery, and storage extensions,
-which is unnecessary when the goal is only to serialize the Flask-RESTX
-`/swagger.json` documents.
-"""
-
-from __future__ import annotations
-
-import argparse
-import json
-import logging
-import os
-import sys
-from dataclasses import dataclass
-from pathlib import Path
-
-from flask import Flask
-from flask_restx.swagger import Swagger
-
-logger = logging.getLogger(__name__)
-
-API_ROOT = Path(__file__).resolve().parents[1]
-if str(API_ROOT) not in sys.path:
-    sys.path.insert(0, str(API_ROOT))
-
-
-@dataclass(frozen=True)
-class SpecTarget:
-    route: str
-    filename: str
-
-
-SPEC_TARGETS: tuple[SpecTarget, ...] = (
-    SpecTarget(route="/console/api/swagger.json", filename="console-swagger.json"),
-    SpecTarget(route="/api/swagger.json", filename="web-swagger.json"),
-    SpecTarget(route="/v1/swagger.json", filename="service-swagger.json"),
-)
-
-_ORIGINAL_REGISTER_MODEL = Swagger.register_model
-_ORIGINAL_REGISTER_FIELD = Swagger.register_field
-
-
-def _apply_runtime_defaults() -> None:
-    """Force the small config surface required for Swagger generation."""
-
-    os.environ.setdefault("SECRET_KEY", "spec-export")
-    os.environ.setdefault("STORAGE_TYPE", "local")
-    os.environ.setdefault("STORAGE_LOCAL_PATH", "/tmp/dify-storage")
-    os.environ.setdefault("SWAGGER_UI_ENABLED", "true")
-
-    from configs import dify_config
-
-    dify_config.SECRET_KEY = os.environ["SECRET_KEY"]
-    dify_config.STORAGE_TYPE = "local"
-    dify_config.STORAGE_LOCAL_PATH = os.environ["STORAGE_LOCAL_PATH"]
-    dify_config.SWAGGER_UI_ENABLED = os.environ["SWAGGER_UI_ENABLED"].lower() == "true"
-
-
-def _patch_swagger_for_inline_nested_dicts() -> None:
-    """Teach Flask-RESTX Swagger generation to tolerate inline nested field maps.
-
-    Some existing controllers use `fields.Nested({...})` with a raw field mapping
-    instead of a named `api.model(...)`. Flask-RESTX crashes on those anonymous
-    dicts during schema registration, so this helper upgrades them into temporary
-    named models at export time.
-    """
-
-    if getattr(Swagger, "_dify_inline_nested_dict_patch", False):
-        return
-
-    def get_or_create_inline_model(self: Swagger, nested_fields: dict[object, object]) -> object:
-        anonymous_models = getattr(self, "_anonymous_inline_models", None)
-        if anonymous_models is None:
-            anonymous_models = {}
-            self._anonymous_inline_models = anonymous_models
-
-        anonymous_name = anonymous_models.get(id(nested_fields))
-        if anonymous_name is None:
-            anonymous_name = f"_AnonymousInlineModel{len(anonymous_models) + 1}"
-            anonymous_models[id(nested_fields)] = anonymous_name
-            self.api.model(anonymous_name, nested_fields)
-
-        return self.api.models[anonymous_name]
-
-    def register_model_with_inline_dict_support(self: Swagger, model: object) -> dict[str, str]:
-        if isinstance(model, dict):
-            model = get_or_create_inline_model(self, model)
-
-        return _ORIGINAL_REGISTER_MODEL(self, model)
-
-    def register_field_with_inline_dict_support(self: Swagger, field: object) -> None:
-        nested = getattr(field, "nested", None)
-        if isinstance(nested, dict):
-            field.model = get_or_create_inline_model(self, nested)  # type: ignore
-
-        _ORIGINAL_REGISTER_FIELD(self, field)
-
-    Swagger.register_model = register_model_with_inline_dict_support
-    Swagger.register_field = register_field_with_inline_dict_support
-    Swagger._dify_inline_nested_dict_patch = True
-
-
-def create_spec_app() -> Flask:
-    """Build a minimal Flask app that only mounts the Swagger-producing blueprints."""
-
-    _apply_runtime_defaults()
-    _patch_swagger_for_inline_nested_dicts()
-
-    app = Flask(__name__)
-
-    from controllers.console import bp as console_bp
-    from controllers.service_api import bp as service_api_bp
-    from controllers.web import bp as web_bp
-
-    app.register_blueprint(console_bp)
-    app.register_blueprint(web_bp)
-    app.register_blueprint(service_api_bp)
-
-    return app
-
-
-def generate_specs(output_dir: Path) -> list[Path]:
-    """Write all Swagger specs to `output_dir` and return the written paths."""
-
-    output_dir.mkdir(parents=True, exist_ok=True)
-
-    app = create_spec_app()
-    client = app.test_client()
-
-    written_paths: list[Path] = []
-    for target in SPEC_TARGETS:
-        response = client.get(target.route)
-        if response.status_code != 200:
-            raise RuntimeError(f"failed to fetch {target.route}: {response.status_code}")
-
-        payload = response.get_json()
-        if not isinstance(payload, dict):
-            raise RuntimeError(f"unexpected response payload for {target.route}")
-
-        output_path = output_dir / target.filename
-        output_path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
-        written_paths.append(output_path)
-
-    return written_paths
-
-
-def parse_args() -> argparse.Namespace:
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument(
-        "-o",
-        "--output-dir",
-        type=Path,
-        default=Path("openapi"),
-        help="Directory where the Swagger JSON files will be written.",
-    )
-    return parser.parse_args()
-
-
-def main() -> int:
-    args = parse_args()
-    written_paths = generate_specs(args.output_dir)
-
-    for path in written_paths:
-        logger.debug(path)
-
-    return 0
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

api/enums/quota_type.py

@@ -1,17 +1,56 @@
+import logging
+from dataclasses import dataclass
 from enum import StrEnum, auto
 
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class QuotaCharge:
+    """
+    Result of a quota consumption operation.
+
+    Attributes:
+        success: Whether the quota charge succeeded
+        charge_id: UUID for refund, or None if failed/disabled
+    """
+
+    success: bool
+    charge_id: str | None
+    _quota_type: "QuotaType"
+
+    def refund(self) -> None:
+        """
+        Refund this quota charge.
+
+        Safe to call even if charge failed or was disabled.
+        This method guarantees no exceptions will be raised.
+        """
+        if self.charge_id:
+            self._quota_type.refund(self.charge_id)
+            logger.info("Refunded quota for %s with charge_id: %s", self._quota_type.value, self.charge_id)
+
 
 class QuotaType(StrEnum):
     """
     Supported quota types for tenant feature usage.
+
+    Add additional types here whenever new billable features become available.
     """
 
+    # Trigger execution quota
     TRIGGER = auto()
+
+    # Workflow execution quota
     WORKFLOW = auto()
+
     UNLIMITED = auto()
 
     @property
     def billing_key(self) -> str:
+        """
+        Get the billing key for the feature.
+        """
         match self:
             case QuotaType.TRIGGER:
                 return "trigger_event"
@@ -19,3 +58,152 @@ class QuotaType(StrEnum):
                 return "api_rate_limit"
             case _:
                 raise ValueError(f"Invalid quota type: {self}")
+
+    def consume(self, tenant_id: str, amount: int = 1) -> QuotaCharge:
+        """
+        Consume quota for the feature.
+
+        Args:
+            tenant_id: The tenant identifier
+            amount: Amount to consume (default: 1)
+
+        Returns:
+            QuotaCharge with success status and charge_id for refund
+
+        Raises:
+            QuotaExceededError: When quota is insufficient
+        """
+        from configs import dify_config
+        from services.billing_service import BillingService
+        from services.errors.app import QuotaExceededError
+
+        if not dify_config.BILLING_ENABLED:
+            logger.debug("Billing disabled, allowing request for %s", tenant_id)
+            return QuotaCharge(success=True, charge_id=None, _quota_type=self)
+
+        logger.info("Consuming %d %s quota for tenant %s", amount, self.value, tenant_id)
+
+        if amount <= 0:
+            raise ValueError("Amount to consume must be greater than 0")
+
+        try:
+            response = BillingService.update_tenant_feature_plan_usage(tenant_id, self.billing_key, delta=amount)
+
+            if response.get("result") != "success":
+                logger.warning(
+                    "Failed to consume quota for %s, feature %s details: %s",
+                    tenant_id,
+                    self.value,
+                    response.get("detail"),
+                )
+                raise QuotaExceededError(feature=self.value, tenant_id=tenant_id, required=amount)
+
+            charge_id = response.get("history_id")
+            logger.debug(
+                "Successfully consumed %d %s quota for tenant %s, charge_id: %s",
+                amount,
+                self.value,
+                tenant_id,
+                charge_id,
+            )
+            return QuotaCharge(success=True, charge_id=charge_id, _quota_type=self)
+
+        except QuotaExceededError:
+            raise
+        except Exception:
+            # fail-safe: allow request on billing errors
+            logger.exception("Failed to consume quota for %s, feature %s", tenant_id, self.value)
+            return unlimited()
+
+    def check(self, tenant_id: str, amount: int = 1) -> bool:
+        """
+        Check if tenant has sufficient quota without consuming.
+
+        Args:
+            tenant_id: The tenant identifier
+            amount: Amount to check (default: 1)
+
+        Returns:
+            True if quota is sufficient, False otherwise
+        """
+        from configs import dify_config
+
+        if not dify_config.BILLING_ENABLED:
+            return True
+
+        if amount <= 0:
+            raise ValueError("Amount to check must be greater than 0")
+
+        try:
+            remaining = self.get_remaining(tenant_id)
+            return remaining >= amount if remaining != -1 else True
+        except Exception:
+            logger.exception("Failed to check quota for %s, feature %s", tenant_id, self.value)
+            # fail-safe: allow request on billing errors
+            return True
+
+    def refund(self, charge_id: str) -> None:
+        """
+        Refund quota using charge_id from consume().
+
+        This method guarantees no exceptions will be raised.
+        All errors are logged but silently handled.
+
+        Args:
+            charge_id: The UUID returned from consume()
+        """
+        try:
+            from configs import dify_config
+            from services.billing_service import BillingService
+
+            if not dify_config.BILLING_ENABLED:
+                return
+
+            if not charge_id:
+                logger.warning("Cannot refund: charge_id is empty")
+                return
+
+            logger.info("Refunding %s quota with charge_id: %s", self.value, charge_id)
+
+            response = BillingService.refund_tenant_feature_plan_usage(charge_id)
+            if response.get("result") == "success":
+                logger.debug("Successfully refunded %s quota, charge_id: %s", self.value, charge_id)
+            else:
+                logger.warning("Refund failed for charge_id: %s", charge_id)
+
+        except Exception:
+            # Catch ALL exceptions - refund must never fail
+            logger.exception("Failed to refund quota for charge_id: %s", charge_id)
+            # Don't raise - refund is best-effort and must be silent
+
+    def get_remaining(self, tenant_id: str) -> int:
+        """
+        Get remaining quota for the tenant.
+
+        Args:
+            tenant_id: The tenant identifier
+
+        Returns:
+            Remaining quota amount
+        """
+        from services.billing_service import BillingService
+
+        try:
+            usage_info = BillingService.get_tenant_feature_plan_usage(tenant_id, self.billing_key)
+            # Assuming the API returns a dict with 'remaining' or 'limit' and 'used'
+            if isinstance(usage_info, dict):
+                return usage_info.get("remaining", 0)
+            # If it returns a simple number, treat it as remaining
+            return int(usage_info) if usage_info else 0
+        except Exception:
+            logger.exception("Failed to get remaining quota for %s, feature %s", tenant_id, self.value)
+            return -1
+
+
+def unlimited() -> QuotaCharge:
+    """
+    Return a quota charge for unlimited quota.
+
+    This is useful for features that are not subject to quota limits, such as the UNLIMITED quota type.
+    """
+    return QuotaCharge(success=True, charge_id=None, _quota_type=QuotaType.UNLIMITED)

api/fields/message_fields.py

@@ -58,6 +58,7 @@ class MessageListItem(ResponseModel):
     message_files: list[MessageFile]
     status: str
     error: str | None = None
+    workflow_run_id: str | None = None
     extra_contents: list[ExecutionExtraContentDomainModel]
 
     @field_validator("inputs", mode="before")

api/libs/broadcast_channel/redis/streams_channel.py

@@ -76,7 +76,6 @@ class _StreamsSubscription(Subscription):
         # reading and writing the _listener / `_closed` attribute.
         self._lock = threading.Lock()
         self._closed: bool = False
-        # self._closed = threading.Event()
         self._listener: threading.Thread | None = None
 
     def _listen(self) -> None:

api/models/comment.py

@@ -8,7 +8,7 @@ from sqlalchemy import Index, func
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from .account import Account
-from .base import Base, gen_uuidv7_string
+from .base import Base
 from .engine import db
 from .types import StringUUID
 
@@ -42,7 +42,7 @@ class WorkflowComment(Base):
         Index("workflow_comments_created_at_idx", "created_at"),
     )
 
-    id: Mapped[str] = mapped_column(StringUUID, default=gen_uuidv7_string)
+    id: Mapped[str] = mapped_column(StringUUID, server_default=sa.text("uuidv7()"))
     tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
     app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
     position_x: Mapped[float] = mapped_column(sa.Float)
@@ -149,7 +149,7 @@ class WorkflowCommentReply(Base):
         Index("comment_replies_created_at_idx", "created_at"),
     )
 
-    id: Mapped[str] = mapped_column(StringUUID, default=gen_uuidv7_string)
+    id: Mapped[str] = mapped_column(StringUUID, server_default=sa.text("uuidv7()"))
     comment_id: Mapped[str] = mapped_column(
         StringUUID, sa.ForeignKey("workflow_comments.id", ondelete="CASCADE"), nullable=False
     )
@@ -194,7 +194,7 @@ class WorkflowCommentMention(Base):
         Index("comment_mentions_user_idx", "mentioned_user_id"),
     )
 
-    id: Mapped[str] = mapped_column(StringUUID, default=gen_uuidv7_string)
+    id: Mapped[str] = mapped_column(StringUUID, server_default=sa.text("uuidv7()"))
     comment_id: Mapped[str] = mapped_column(
         StringUUID, sa.ForeignKey("workflow_comments.id", ondelete="CASCADE"), nullable=False
     )

api/models/dataset.py

@@ -1036,7 +1036,7 @@ class DocumentSegment(Base):
         return attachment_list
 
 
-class ChildChunk(TypeBase):
+class ChildChunk(Base):
     __tablename__ = "child_chunks"
     __table_args__ = (
         sa.PrimaryKeyConstraint("id", name="child_chunk_pkey"),
@@ -1046,42 +1046,29 @@ class ChildChunk(TypeBase):
     )
 
     # initial fields
-    id: Mapped[str] = mapped_column(StringUUID, nullable=False, default_factory=lambda: str(uuid4()), init=False)
-    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
-    dataset_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
-    document_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
-    segment_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    id = mapped_column(StringUUID, nullable=False, default=lambda: str(uuid4()))
+    tenant_id = mapped_column(StringUUID, nullable=False)
+    dataset_id = mapped_column(StringUUID, nullable=False)
+    document_id = mapped_column(StringUUID, nullable=False)
+    segment_id = mapped_column(StringUUID, nullable=False)
     position: Mapped[int] = mapped_column(sa.Integer, nullable=False)
-    content: Mapped[str] = mapped_column(LongText, nullable=False)
+    content = mapped_column(LongText, nullable=False)
     word_count: Mapped[int] = mapped_column(sa.Integer, nullable=False)
     # indexing fields
-    created_by: Mapped[str] = mapped_column(StringUUID, nullable=False)
-    created_at: Mapped[datetime] = mapped_column(
-        DateTime, nullable=False, server_default=sa.func.current_timestamp(), init=False
-    )
-    updated_by: Mapped[str | None] = mapped_column(StringUUID, nullable=True, init=False)
-    updated_at: Mapped[datetime] = mapped_column(
-        DateTime,
-        nullable=False,
-        server_default=sa.func.current_timestamp(),
-        onupdate=func.current_timestamp(),
-        init=False,
-    )
-    indexing_at: Mapped[datetime | None] = mapped_column(
-        DateTime, nullable=True, insert_default=None, server_default=None, init=False
-    )
-    completed_at: Mapped[datetime | None] = mapped_column(
-        DateTime, nullable=True, insert_default=None, server_default=None, init=False
-    )
-    index_node_id: Mapped[str | None] = mapped_column(String(255), nullable=True, default=None)
-    index_node_hash: Mapped[str | None] = mapped_column(String(255), nullable=True, default=None)
+    index_node_id = mapped_column(String(255), nullable=True)
+    index_node_hash = mapped_column(String(255), nullable=True)
     type: Mapped[SegmentType] = mapped_column(
-        EnumText(SegmentType, length=255),
-        nullable=False,
-        server_default=sa.text("'automatic'"),
-        default=SegmentType.AUTOMATIC,
+        EnumText(SegmentType, length=255), nullable=False, server_default=sa.text("'automatic'")
     )
-    error: Mapped[str | None] = mapped_column(LongText, nullable=True, init=False)
+    created_by = mapped_column(StringUUID, nullable=False)
+    created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, server_default=sa.func.current_timestamp())
+    updated_by = mapped_column(StringUUID, nullable=True)
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime, nullable=False, server_default=sa.func.current_timestamp(), onupdate=func.current_timestamp()
+    )
+    indexing_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    completed_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    error = mapped_column(LongText, nullable=True)
 
     @property
     def dataset(self):

api/models/model.py

@@ -91,19 +91,6 @@ class EnabledConfig(TypedDict):
     enabled: bool
 
 
-class SuggestedQuestionsAfterAnswerModelConfig(TypedDict):
-    provider: str
-    name: str
-    mode: NotRequired[str]
-    completion_params: NotRequired[dict[str, Any]]
-
-
-class SuggestedQuestionsAfterAnswerConfig(TypedDict):
-    enabled: bool
-    model: NotRequired[SuggestedQuestionsAfterAnswerModelConfig]
-    prompt: NotRequired[str]
-
-
 class EmbeddingModelInfo(TypedDict):
     embedding_provider_name: str
     embedding_model_name: str
@@ -233,7 +220,7 @@ class ModelConfig(TypedDict):
 class AppModelConfigDict(TypedDict):
     opening_statement: str | None
     suggested_questions: list[str]
-    suggested_questions_after_answer: SuggestedQuestionsAfterAnswerConfig
+    suggested_questions_after_answer: EnabledConfig
     speech_to_text: EnabledConfig
     text_to_speech: EnabledConfig
     retriever_resource: EnabledConfig
@@ -693,13 +680,8 @@ class AppModelConfig(TypeBase):
         return cast(EnabledConfig, json.loads(value) if value else {"enabled": default_enabled})
 
     @property
-    def suggested_questions_after_answer_dict(self) -> SuggestedQuestionsAfterAnswerConfig:
-        return cast(
-            SuggestedQuestionsAfterAnswerConfig,
-            json.loads(self.suggested_questions_after_answer)
-            if self.suggested_questions_after_answer
-            else {"enabled": False},
-        )
+    def suggested_questions_after_answer_dict(self) -> EnabledConfig:
+        return self._get_enabled_config(self.suggested_questions_after_answer)
 
     @property
     def speech_to_text_dict(self) -> EnabledConfig:
@@ -1867,18 +1849,15 @@ class MessageAnnotation(TypeBase):
     )
 
     id: Mapped[str] = mapped_column(
-        StringUUID,
-        insert_default=lambda: str(uuid4()),
-        default_factory=lambda: str(uuid4()),
-        init=False,
+        StringUUID, insert_default=lambda: str(uuid4()), default_factory=lambda: str(uuid4()), init=False
     )
     app_id: Mapped[str] = mapped_column(StringUUID)
     question: Mapped[str] = mapped_column(LongText, nullable=False)
     content: Mapped[str] = mapped_column(LongText, nullable=False)
-    hit_count: Mapped[int] = mapped_column(sa.Integer, nullable=False, server_default=sa.text("0"), init=False)
     account_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
     conversation_id: Mapped[str | None] = mapped_column(StringUUID, sa.ForeignKey("conversations.id"), default=None)
     message_id: Mapped[str | None] = mapped_column(StringUUID, default=None)
+    hit_count: Mapped[int] = mapped_column(sa.Integer, nullable=False, server_default=sa.text("0"), default=0)
     created_at: Mapped[datetime] = mapped_column(
         sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
     )
@@ -2182,7 +2161,7 @@ class ApiToken(Base):  # bug: this uses setattr so idk the field.
             return result
 
 
-class UploadFile(TypeBase):
+class UploadFile(Base):
     __tablename__ = "upload_files"
     __table_args__ = (
         sa.PrimaryKeyConstraint("id", name="upload_file_pkey"),
@@ -2190,12 +2169,9 @@ class UploadFile(TypeBase):
     )
 
     # NOTE: The `id` field is generated within the application to minimize extra roundtrips
-    # (especially when generating `source_url`) and keep model metadata portable across databases.
-    id: Mapped[str] = mapped_column(
-        StringUUID,
-        init=False,
-        default_factory=lambda: str(uuid4()),
-    )
+    # (especially when generating `source_url`).
+    # The `server_default` serves as a fallback mechanism.
+    id: Mapped[str] = mapped_column(StringUUID, default=lambda: str(uuid4()))
     tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
     storage_type: Mapped[StorageType] = mapped_column(EnumText(StorageType, length=255), nullable=False)
     key: Mapped[str] = mapped_column(String(255), nullable=False)
@@ -2203,6 +2179,16 @@ class UploadFile(TypeBase):
     size: Mapped[int] = mapped_column(sa.Integer, nullable=False)
     extension: Mapped[str] = mapped_column(String(255), nullable=False)
     mime_type: Mapped[str] = mapped_column(String(255), nullable=True)
+
+    # The `created_by_role` field indicates whether the file was created by an `Account` or an `EndUser`.
+    # Its value is derived from the `CreatorUserRole` enumeration.
+    created_by_role: Mapped[CreatorUserRole] = mapped_column(
+        EnumText(CreatorUserRole, length=255),
+        nullable=False,
+        server_default=sa.text("'account'"),
+        default=CreatorUserRole.ACCOUNT,
+    )
+
     # The `created_by` field stores the ID of the entity that created this upload file.
     #
     # If `created_by_role` is `ACCOUNT`, it corresponds to `Account.id`.
@@ -2221,18 +2207,10 @@ class UploadFile(TypeBase):
     # `used` may indicate whether the file has been utilized by another service.
     used: Mapped[bool] = mapped_column(sa.Boolean, nullable=False, server_default=sa.text("false"))
 
-    # The `created_by_role` field indicates whether the file was created by an `Account` or an `EndUser`.
-    # Its value is derived from the `CreatorUserRole` enumeration.
-    created_by_role: Mapped[CreatorUserRole] = mapped_column(
-        EnumText(CreatorUserRole, length=255),
-        nullable=False,
-        server_default=sa.text("'account'"),
-        default=CreatorUserRole.ACCOUNT,
-    )
     # `used_by` may indicate the ID of the user who utilized this file.
-    used_by: Mapped[str | None] = mapped_column(StringUUID, nullable=True, default=None)
-    used_at: Mapped[datetime | None] = mapped_column(sa.DateTime, nullable=True, default=None)
-    hash: Mapped[str | None] = mapped_column(String(255), nullable=True, default=None)
+    used_by: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
+    used_at: Mapped[datetime | None] = mapped_column(sa.DateTime, nullable=True)
+    hash: Mapped[str | None] = mapped_column(String(255), nullable=True)
     source_url: Mapped[str] = mapped_column(LongText, default="")
 
     def __init__(

api/models/workflow.py

@@ -50,7 +50,7 @@ from libs.uuid_utils import uuidv7
 from ._workflow_exc import NodeNotFoundError, WorkflowDataError
 
 if TYPE_CHECKING:
-    from .model import AppMode
+    from .model import AppMode, UploadFile
 
 
 from constants import DEFAULT_FILE_NUMBER_LIMITS, HIDDEN_VALUE
@@ -63,10 +63,6 @@ from .account import Account
 from .base import Base, DefaultFieldsDCMixin, TypeBase
 from .engine import db
 from .enums import CreatorUserRole, DraftVariableType, ExecutionOffLoadType, WorkflowRunTriggeredFrom
-
-# UploadFile uses TypeBase while workflow execution offload models use Base, so relationships
-# must target the class object directly instead of relying on string lookup across registries.
-from .model import UploadFile
 from .types import EnumText, LongText, StringUUID
 from .utils.file_input_compat import (
     build_file_from_mapping_without_lookup,
@@ -1100,6 +1096,8 @@ class WorkflowNodeExecutionModel(Base):  # This model is expected to have `offlo
 
     @staticmethod
     def _load_full_content(session: orm.Session, file_id: str, storage: Storage):
+        from .model import UploadFile
+
         stmt = sa.select(UploadFile).where(UploadFile.id == file_id)
         file = session.scalars(stmt).first()
         assert file is not None, f"UploadFile with id {file_id} should exist but not"
@@ -1193,11 +1191,10 @@ class WorkflowNodeExecutionOffload(Base):
     )
 
     file: Mapped[Optional["UploadFile"]] = orm.relationship(
-        UploadFile,
         foreign_keys=[file_id],
         lazy="raise",
         uselist=False,
-        primaryjoin=lambda: orm.foreign(WorkflowNodeExecutionOffload.file_id) == UploadFile.id,
+        primaryjoin="WorkflowNodeExecutionOffload.file_id == UploadFile.id",
     )
 
 
@@ -1971,11 +1968,10 @@ class WorkflowDraftVariableFile(Base):
 
     # Relationship to UploadFile
     upload_file: Mapped["UploadFile"] = orm.relationship(
-        UploadFile,
         foreign_keys=[upload_file_id],
         lazy="raise",
         uselist=False,
-        primaryjoin=lambda: orm.foreign(WorkflowDraftVariableFile.upload_file_id) == UploadFile.id,
+        primaryjoin="WorkflowDraftVariableFile.upload_file_id == UploadFile.id",
     )
 
 

api/providers/trace/trace-aliyun/tests/unit_tests/aliyun_trace/data_exporter/test_traceclient.py

@@ -225,10 +225,8 @@ class TestSpanBuilder:
         span = builder.build_span(span_data)
         assert isinstance(span, ReadableSpan)
         assert span.name == "test-span"
-        assert span.context is not None
         assert span.context.trace_id == 123
         assert span.context.span_id == 456
-        assert span.parent is not None
         assert span.parent.span_id == 789
         assert span.resource == resource
         assert span.attributes == {"attr1": "val1"}

api/providers/trace/trace-aliyun/tests/unit_tests/aliyun_trace/entities/test_aliyun_trace_entity.py

@@ -64,13 +64,12 @@ class TestSpanData:
 
     def test_span_data_missing_required_fields(self):
         with pytest.raises(ValidationError):
-            SpanData.model_validate(
-                {
-                    "trace_id": 123,
-                    "name": "test_span",
-                    "start_time": 1000,
-                    "end_time": 2000,
-                }
+            SpanData(
+                trace_id=123,
+                # span_id missing
+                name="test_span",
+                start_time=1000,
+                end_time=2000,
             )
 
     def test_span_data_arbitrary_types_allowed(self):

api/providers/trace/trace-aliyun/tests/unit_tests/aliyun_trace/test_aliyun_trace.py

@@ -2,14 +2,12 @@ from __future__ import annotations
 
 from datetime import UTC, datetime
 from types import SimpleNamespace
-from typing import cast
 from unittest.mock import MagicMock
 
 import dify_trace_aliyun.aliyun_trace as aliyun_trace_module
 import pytest
 from dify_trace_aliyun.aliyun_trace import AliyunDataTrace
 from dify_trace_aliyun.config import AliyunConfig
-from dify_trace_aliyun.entities.aliyun_trace_entity import SpanData, TraceMetadata
 from dify_trace_aliyun.entities.semconv import (
     GEN_AI_COMPLETION,
     GEN_AI_INPUT_MESSAGE,
@@ -46,7 +44,7 @@ class RecordingTraceClient:
         self.endpoint = endpoint
         self.added_spans: list[object] = []
 
-    def add_span(self, span: object) -> None:
+    def add_span(self, span) -> None:
         self.added_spans.append(span)
 
     def api_check(self) -> bool:
@@ -65,35 +63,11 @@ def _make_link(trace_id: int = 1, span_id: int = 2) -> Link:
         trace_id=trace_id,
         span_id=span_id,
         is_remote=False,
-        trace_flags=TraceFlags(TraceFlags.SAMPLED),
+        trace_flags=TraceFlags.SAMPLED,
     )
     return Link(context)
 
 
-def _make_trace_metadata(
-    trace_id: int = 1,
-    workflow_span_id: int = 2,
-    session_id: str = "s",
-    user_id: str = "u",
-    links: list[Link] | None = None,
-) -> TraceMetadata:
-    return TraceMetadata(
-        trace_id=trace_id,
-        workflow_span_id=workflow_span_id,
-        session_id=session_id,
-        user_id=user_id,
-        links=[] if links is None else links,
-    )
-
-
-def _recording_trace_client(trace_instance: AliyunDataTrace) -> RecordingTraceClient:
-    return cast(RecordingTraceClient, trace_instance.trace_client)
-
-
-def _recorded_span_data(trace_instance: AliyunDataTrace) -> list[SpanData]:
-    return cast(list[SpanData], _recording_trace_client(trace_instance).added_spans)
-
-
 def _make_workflow_trace_info(**overrides) -> WorkflowTraceInfo:
     defaults = {
         "workflow_id": "workflow-id",
@@ -289,20 +263,20 @@ def test_workflow_trace_adds_workflow_and_node_spans(trace_instance: AliyunDataT
     trace_instance.workflow_trace(trace_info)
 
     add_workflow_span.assert_called_once()
-    passed_trace_metadata = cast(TraceMetadata, add_workflow_span.call_args.args[1])
+    passed_trace_metadata = add_workflow_span.call_args.args[1]
     assert passed_trace_metadata.trace_id == 111
     assert passed_trace_metadata.workflow_span_id == 222
     assert passed_trace_metadata.session_id == "c"
     assert passed_trace_metadata.user_id == "u"
     assert passed_trace_metadata.links == []
 
-    assert _recording_trace_client(trace_instance).added_spans == ["span-1", "span-2"]
+    assert trace_instance.trace_client.added_spans == ["span-1", "span-2"]
 
 
 def test_message_trace_returns_early_if_no_message_data(trace_instance: AliyunDataTrace):
     trace_info = _make_message_trace_info(message_data=None)
     trace_instance.message_trace(trace_info)
-    assert _recording_trace_client(trace_instance).added_spans == []
+    assert trace_instance.trace_client.added_spans == []
 
 
 def test_message_trace_creates_message_and_llm_spans(trace_instance: AliyunDataTrace, monkeypatch: pytest.MonkeyPatch):
@@ -328,9 +302,8 @@ def test_message_trace_creates_message_and_llm_spans(trace_instance: AliyunDataT
     )
     trace_instance.message_trace(trace_info)
 
-    spans = _recorded_span_data(trace_instance)
-    assert len(spans) == 2
-    message_span, llm_span = spans
+    assert len(trace_instance.trace_client.added_spans) == 2
+    message_span, llm_span = trace_instance.trace_client.added_spans
 
     assert message_span.name == "message"
     assert message_span.trace_id == 10
@@ -351,7 +324,7 @@ def test_message_trace_creates_message_and_llm_spans(trace_instance: AliyunDataT
 def test_dataset_retrieval_trace_returns_early_if_no_message_data(trace_instance: AliyunDataTrace):
     trace_info = _make_dataset_retrieval_trace_info(message_data=None)
     trace_instance.dataset_retrieval_trace(trace_info)
-    assert _recording_trace_client(trace_instance).added_spans == []
+    assert trace_instance.trace_client.added_spans == []
 
 
 def test_dataset_retrieval_trace_creates_span(trace_instance: AliyunDataTrace, monkeypatch: pytest.MonkeyPatch):
@@ -365,9 +338,8 @@ def test_dataset_retrieval_trace_creates_span(trace_instance: AliyunDataTrace, m
     monkeypatch.setattr(aliyun_trace_module, "extract_retrieval_documents", lambda _: [{"doc": "d"}])
 
     trace_instance.dataset_retrieval_trace(_make_dataset_retrieval_trace_info(inputs="query"))
-    spans = _recorded_span_data(trace_instance)
-    assert len(spans) == 1
-    span = spans[0]
+    assert len(trace_instance.trace_client.added_spans) == 1
+    span = trace_instance.trace_client.added_spans[0]
     assert span.name == "dataset_retrieval"
     assert span.attributes[RETRIEVAL_QUERY] == "query"
     assert span.attributes[RETRIEVAL_DOCUMENT] == '[{"doc": "d"}]'
@@ -376,7 +348,7 @@ def test_dataset_retrieval_trace_creates_span(trace_instance: AliyunDataTrace, m
 def test_tool_trace_returns_early_if_no_message_data(trace_instance: AliyunDataTrace):
     trace_info = _make_tool_trace_info(message_data=None)
     trace_instance.tool_trace(trace_info)
-    assert _recording_trace_client(trace_instance).added_spans == []
+    assert trace_instance.trace_client.added_spans == []
 
 
 def test_tool_trace_creates_span(trace_instance: AliyunDataTrace, monkeypatch: pytest.MonkeyPatch):
@@ -399,9 +371,8 @@ def test_tool_trace_creates_span(trace_instance: AliyunDataTrace, monkeypatch: p
         )
     )
 
-    spans = _recorded_span_data(trace_instance)
-    assert len(spans) == 1
-    span = spans[0]
+    assert len(trace_instance.trace_client.added_spans) == 1
+    span = trace_instance.trace_client.added_spans[0]
     assert span.name == "my-tool"
     assert span.status == status
     assert span.attributes[TOOL_NAME] == "my-tool"
@@ -438,7 +409,7 @@ def test_get_workflow_node_executions_builds_repo_and_fetches(
 def test_build_workflow_node_span_routes_llm_type(trace_instance: AliyunDataTrace, monkeypatch: pytest.MonkeyPatch):
     node_execution = MagicMock(spec=WorkflowNodeExecution)
     trace_info = _make_workflow_trace_info()
-    trace_metadata = _make_trace_metadata()
+    trace_metadata = MagicMock()
 
     monkeypatch.setattr(trace_instance, "build_workflow_llm_span", MagicMock(return_value="llm"))
 
@@ -451,7 +422,7 @@ def test_build_workflow_node_span_routes_knowledge_retrieval_type(
 ):
     node_execution = MagicMock(spec=WorkflowNodeExecution)
     trace_info = _make_workflow_trace_info()
-    trace_metadata = _make_trace_metadata()
+    trace_metadata = MagicMock()
 
     monkeypatch.setattr(trace_instance, "build_workflow_retrieval_span", MagicMock(return_value="retrieval"))
 
@@ -462,7 +433,7 @@ def test_build_workflow_node_span_routes_knowledge_retrieval_type(
 def test_build_workflow_node_span_routes_tool_type(trace_instance: AliyunDataTrace, monkeypatch: pytest.MonkeyPatch):
     node_execution = MagicMock(spec=WorkflowNodeExecution)
     trace_info = _make_workflow_trace_info()
-    trace_metadata = _make_trace_metadata()
+    trace_metadata = MagicMock()
 
     monkeypatch.setattr(trace_instance, "build_workflow_tool_span", MagicMock(return_value="tool"))
 
@@ -473,7 +444,7 @@ def test_build_workflow_node_span_routes_tool_type(trace_instance: AliyunDataTra
 def test_build_workflow_node_span_routes_code_type(trace_instance: AliyunDataTrace, monkeypatch: pytest.MonkeyPatch):
     node_execution = MagicMock(spec=WorkflowNodeExecution)
     trace_info = _make_workflow_trace_info()
-    trace_metadata = _make_trace_metadata()
+    trace_metadata = MagicMock()
 
     monkeypatch.setattr(trace_instance, "build_workflow_task_span", MagicMock(return_value="task"))
 
@@ -486,7 +457,7 @@ def test_build_workflow_node_span_handles_errors(
 ):
     node_execution = MagicMock(spec=WorkflowNodeExecution)
     trace_info = _make_workflow_trace_info()
-    trace_metadata = _make_trace_metadata()
+    trace_metadata = MagicMock()
 
     monkeypatch.setattr(trace_instance, "build_workflow_task_span", MagicMock(side_effect=RuntimeError("boom")))
     node_execution.node_type = BuiltinNodeTypes.CODE
@@ -501,7 +472,7 @@ def test_build_workflow_task_span(trace_instance: AliyunDataTrace, monkeypatch:
     status = Status(StatusCode.OK)
     monkeypatch.setattr(aliyun_trace_module, "get_workflow_node_status", lambda _: status)
 
-    trace_metadata = _make_trace_metadata()
+    trace_metadata = SimpleNamespace(trace_id=1, workflow_span_id=2, session_id="s", user_id="u", links=[])
     node_execution = MagicMock(spec=WorkflowNodeExecution)
     node_execution.id = "node-id"
     node_execution.title = "title"
@@ -523,7 +494,7 @@ def test_build_workflow_tool_span(trace_instance: AliyunDataTrace, monkeypatch:
     status = Status(StatusCode.OK)
     monkeypatch.setattr(aliyun_trace_module, "get_workflow_node_status", lambda _: status)
 
-    trace_metadata = _make_trace_metadata(links=[_make_link()])
+    trace_metadata = SimpleNamespace(trace_id=1, workflow_span_id=2, session_id="s", user_id="u", links=[_make_link()])
     node_execution = MagicMock(spec=WorkflowNodeExecution)
     node_execution.id = "node-id"
     node_execution.title = "my-tool"
@@ -556,7 +527,7 @@ def test_build_workflow_retrieval_span(trace_instance: AliyunDataTrace, monkeypa
         aliyun_trace_module, "format_retrieval_documents", lambda docs: [{"formatted": True}] if docs else []
     )
 
-    trace_metadata = _make_trace_metadata()
+    trace_metadata = SimpleNamespace(trace_id=1, workflow_span_id=2, session_id="s", user_id="u", links=[])
     node_execution = MagicMock(spec=WorkflowNodeExecution)
     node_execution.id = "node-id"
     node_execution.title = "retrieval"
@@ -585,7 +556,7 @@ def test_build_workflow_llm_span(trace_instance: AliyunDataTrace, monkeypatch: p
     monkeypatch.setattr(aliyun_trace_module, "format_input_messages", lambda _: "in")
     monkeypatch.setattr(aliyun_trace_module, "format_output_messages", lambda _: "out")
 
-    trace_metadata = _make_trace_metadata()
+    trace_metadata = SimpleNamespace(trace_id=1, workflow_span_id=2, session_id="s", user_id="u", links=[])
     node_execution = MagicMock(spec=WorkflowNodeExecution)
     node_execution.id = "node-id"
     node_execution.title = "llm"
@@ -623,7 +594,7 @@ def test_add_workflow_span(trace_instance: AliyunDataTrace, monkeypatch: pytest.
     status = Status(StatusCode.OK)
     monkeypatch.setattr(aliyun_trace_module, "create_status_from_error", lambda _: status)
 
-    trace_metadata = _make_trace_metadata()
+    trace_metadata = SimpleNamespace(trace_id=1, workflow_span_id=2, session_id="s", user_id="u", links=[])
 
     # CASE 1: With message_id
     trace_info = _make_workflow_trace_info(
@@ -631,11 +602,9 @@ def test_add_workflow_span(trace_instance: AliyunDataTrace, monkeypatch: pytest.
     )
     trace_instance.add_workflow_span(trace_info, trace_metadata)
 
-    client = _recording_trace_client(trace_instance)
-    spans = _recorded_span_data(trace_instance)
-    assert len(spans) == 2
-    message_span = spans[0]
-    workflow_span = spans[1]
+    assert len(trace_instance.trace_client.added_spans) == 2
+    message_span = trace_instance.trace_client.added_spans[0]
+    workflow_span = trace_instance.trace_client.added_spans[1]
 
     assert message_span.name == "message"
     assert message_span.span_kind == SpanKind.SERVER
@@ -645,14 +614,13 @@ def test_add_workflow_span(trace_instance: AliyunDataTrace, monkeypatch: pytest.
     assert workflow_span.span_kind == SpanKind.INTERNAL
     assert workflow_span.parent_span_id == 20
 
-    client.added_spans.clear()
+    trace_instance.trace_client.added_spans.clear()
 
     # CASE 2: Without message_id
     trace_info_no_msg = _make_workflow_trace_info(message_id=None)
     trace_instance.add_workflow_span(trace_info_no_msg, trace_metadata)
-    spans = _recorded_span_data(trace_instance)
-    assert len(spans) == 1
-    span = spans[0]
+    assert len(trace_instance.trace_client.added_spans) == 1
+    span = trace_instance.trace_client.added_spans[0]
     assert span.name == "workflow"
     assert span.span_kind == SpanKind.SERVER
     assert span.parent_span_id is None
@@ -673,8 +641,7 @@ def test_suggested_question_trace(trace_instance: AliyunDataTrace, monkeypatch:
     trace_info = _make_suggested_question_trace_info(suggested_question=["how?"])
     trace_instance.suggested_question_trace(trace_info)
 
-    spans = _recorded_span_data(trace_instance)
-    assert len(spans) == 1
-    span = spans[0]
+    assert len(trace_instance.trace_client.added_spans) == 1
+    span = trace_instance.trace_client.added_spans[0]
     assert span.name == "suggested_question"
     assert span.attributes[GEN_AI_COMPLETION] == '["how?"]'

api/providers/trace/trace-aliyun/tests/unit_tests/aliyun_trace/test_aliyun_trace_utils.py

@@ -1,6 +1,4 @@
 import json
-from collections.abc import Mapping
-from typing import Any, cast
 from unittest.mock import MagicMock
 
 from dify_trace_aliyun.entities.semconv import (
@@ -172,7 +170,7 @@ def test_create_common_span_attributes():
 
 def test_format_retrieval_documents():
     # Not a list
-    assert format_retrieval_documents(cast(list[object], "not a list")) == []
+    assert format_retrieval_documents("not a list") == []
 
     # Valid list
     docs = [
@@ -213,7 +211,7 @@ def test_format_retrieval_documents():
 
 def test_format_input_messages():
     # Not a dict
-    assert format_input_messages(cast(Mapping[str, Any], None)) == serialize_json_data([])
+    assert format_input_messages(None) == serialize_json_data([])
 
     # No prompts
     assert format_input_messages({}) == serialize_json_data([])
@@ -246,7 +244,7 @@ def test_format_input_messages():
 
 def test_format_output_messages():
     # Not a dict
-    assert format_output_messages(cast(Mapping[str, Any], None)) == serialize_json_data([])
+    assert format_output_messages(None) == serialize_json_data([])
 
     # No text
     assert format_output_messages({"finish_reason": "stop"}) == serialize_json_data([])

api/providers/trace/trace-aliyun/tests/unit_tests/test_config_entity.py

@@ -25,13 +25,13 @@ class TestAliyunConfig:
     def test_missing_required_fields(self):
         """Test that required fields are enforced"""
         with pytest.raises(ValidationError):
-            AliyunConfig.model_validate({})
+            AliyunConfig()
 
         with pytest.raises(ValidationError):
-            AliyunConfig.model_validate({"license_key": "test_license"})
+            AliyunConfig(license_key="test_license")
 
         with pytest.raises(ValidationError):
-            AliyunConfig.model_validate({"endpoint": "https://tracing-analysis-dc-hz.aliyuncs.com"})
+            AliyunConfig(endpoint="https://tracing-analysis-dc-hz.aliyuncs.com")
 
     def test_app_name_validation_empty(self):
         """Test app_name validation with empty value"""

api/providers/trace/trace-arize-phoenix/tests/unit_tests/arize_phoenix_trace/test_arize_phoenix_trace.py

@@ -1,5 +1,4 @@
 from datetime import UTC, datetime, timedelta
-from typing import cast
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -130,7 +129,7 @@ def test_set_span_status():
             return "SilentErrorRepr"
 
     span.reset_mock()
-    set_span_status(span, cast(Exception | str | None, SilentError()))
+    set_span_status(span, SilentError())
     assert span.add_event.call_args[1]["attributes"][OTELSpanAttributes.EXCEPTION_MESSAGE] == "SilentErrorRepr"
 
 

api/providers/trace/trace-langfuse/tests/unit_tests/test_config_entity.py

@@ -28,13 +28,13 @@ class TestLangfuseConfig:
     def test_missing_required_fields(self):
         """Test that required fields are enforced"""
         with pytest.raises(ValidationError):
-            LangfuseConfig.model_validate({})
+            LangfuseConfig()
 
         with pytest.raises(ValidationError):
-            LangfuseConfig.model_validate({"public_key": "public"})
+            LangfuseConfig(public_key="public")
 
         with pytest.raises(ValidationError):
-            LangfuseConfig.model_validate({"secret_key": "secret"})
+            LangfuseConfig(secret_key="secret")
 
     def test_host_validation_empty(self):
         """Test host validation with empty value"""

api/providers/trace/trace-langfuse/tests/unit_tests/test_langfuse_trace.py

@@ -2,7 +2,6 @@
 
 from datetime import datetime, timedelta
 from types import SimpleNamespace
-from typing import cast
 from unittest.mock import MagicMock, patch
 
 from dify_trace_langfuse.config import LangfuseConfig
@@ -135,4 +134,4 @@ class TestLangFuseDataTraceCompletionStartTime:
 
         assert trace._get_completion_start_time(start_time, None) is None
         assert trace._get_completion_start_time(start_time, -1) is None
-        assert trace._get_completion_start_time(start_time, cast(float | int | None, "invalid")) is None
+        assert trace._get_completion_start_time(start_time, "invalid") is None

api/providers/trace/trace-langsmith/tests/unit_tests/test_config_entity.py

@@ -21,13 +21,13 @@ class TestLangSmithConfig:
     def test_missing_required_fields(self):
         """Test that required fields are enforced"""
         with pytest.raises(ValidationError):
-            LangSmithConfig.model_validate({})
+            LangSmithConfig()
 
         with pytest.raises(ValidationError):
-            LangSmithConfig.model_validate({"api_key": "key"})
+            LangSmithConfig(api_key="key")
 
         with pytest.raises(ValidationError):
-            LangSmithConfig.model_validate({"project": "project"})
+            LangSmithConfig(project="project")
 
     def test_endpoint_validation_https_only(self):
         """Test endpoint validation only allows HTTPS"""

api/providers/trace/trace-mlflow/tests/unit_tests/mlflow_trace/test_mlflow_trace.py

@@ -599,6 +599,7 @@ class TestMessageTrace:
         span = MagicMock()
         mock_tracing["start"].return_value = span
         mock_tracing["set"].return_value = "token"
+        mock_db.session.query.return_value.where.return_value.first.return_value = None
 
         trace_instance.message_trace(_make_message_trace_info())
         mock_tracing["start"].assert_called_once()
@@ -608,6 +609,7 @@ class TestMessageTrace:
         span = MagicMock()
         mock_tracing["start"].return_value = span
         mock_tracing["set"].return_value = "token"
+        mock_db.session.query.return_value.where.return_value.first.return_value = None
 
         trace_info = _make_message_trace_info(error="something broke")
         trace_instance.message_trace(trace_info)
@@ -618,6 +620,7 @@ class TestMessageTrace:
         span = MagicMock()
         mock_tracing["start"].return_value = span
         mock_tracing["set"].return_value = "token"
+        mock_db.session.query.return_value.where.return_value.first.return_value = None
         monkeypatch.setenv("FILES_URL", "http://files.test")
 
         file_data = SimpleNamespace(url="path/to/file.png")
@@ -635,6 +638,7 @@ class TestMessageTrace:
         span = MagicMock()
         mock_tracing["start"].return_value = span
         mock_tracing["set"].return_value = "token"
+        mock_db.session.query.return_value.where.return_value.first.return_value = None
 
         trace_info = _make_message_trace_info(file_list=None, message_file_data=None)
         trace_instance.message_trace(trace_info)
@@ -647,6 +651,7 @@ class TestMessageTrace:
 
         end_user = MagicMock()
         end_user.session_id = "session-xyz"
+        mock_db.session.query.return_value.where.return_value.first.return_value = end_user
 
         trace_info = _make_message_trace_info(
             metadata={"from_end_user_id": "eu-1", "conversation_id": "c1"},
@@ -659,6 +664,7 @@ class TestMessageTrace:
         span = MagicMock()
         mock_tracing["start"].return_value = span
         mock_tracing["set"].return_value = "token"
+        mock_db.session.query.return_value.where.return_value.first.return_value = None
 
         trace_info = _make_message_trace_info(
             metadata={"from_account_id": "acc-1"},

api/providers/trace/trace-opik/tests/unit_tests/test_opik_trace.py

@@ -12,7 +12,6 @@ from __future__ import annotations
 
 import uuid
 from datetime import datetime
-from typing import cast
 from unittest.mock import MagicMock, patch
 
 from dify_trace_opik.opik_trace import OpikDataTrace, _seed_to_uuid4, prepare_opik_uuid
@@ -70,14 +69,6 @@ def _make_opik_trace_instance() -> OpikDataTrace:
     return instance
 
 
-def _add_trace_mock(instance: OpikDataTrace) -> MagicMock:
-    return cast(MagicMock, instance.add_trace)
-
-
-def _add_span_mock(instance: OpikDataTrace) -> MagicMock:
-    return cast(MagicMock, instance.add_span)
-
-
 # ---------------------------------------------------------------------------
 # _seed_to_uuid4
 # ---------------------------------------------------------------------------
@@ -164,21 +155,21 @@ class TestWorkflowTraceWithoutMessageId:
     def test_root_span_is_created(self):
         trace_info = _make_workflow_trace_info(message_id=None)
         instance = self._run(trace_info)
-        assert _add_span_mock(instance).called
+        assert instance.add_span.called
 
     def test_root_span_id_matches_expected(self):
         trace_info = _make_workflow_trace_info(message_id=None)
         instance = self._run(trace_info)
 
         expected = self._expected_root_span_id(trace_info)
-        root_span_kwargs = _add_span_mock(instance).call_args_list[0][0][0]
+        root_span_kwargs = instance.add_span.call_args_list[0][0][0]
         assert root_span_kwargs["id"] == expected
 
     def test_root_span_has_no_parent(self):
         trace_info = _make_workflow_trace_info(message_id=None)
         instance = self._run(trace_info)
 
-        root_span_kwargs = _add_span_mock(instance).call_args_list[0][0][0]
+        root_span_kwargs = instance.add_span.call_args_list[0][0][0]
         assert root_span_kwargs["parent_span_id"] is None
 
     def test_trace_name_is_workflow_trace(self):
@@ -186,21 +177,21 @@ class TestWorkflowTraceWithoutMessageId:
         trace_info = _make_workflow_trace_info(message_id=None)
         instance = self._run(trace_info)
 
-        trace_kwargs = _add_trace_mock(instance).call_args_list[0][0][0]
+        trace_kwargs = instance.add_trace.call_args_list[0][0][0]
         assert trace_kwargs["name"] == TraceTaskName.WORKFLOW_TRACE
 
     def test_root_span_name_is_workflow_trace(self):
         trace_info = _make_workflow_trace_info(message_id=None)
         instance = self._run(trace_info)
 
-        root_span_kwargs = _add_span_mock(instance).call_args_list[0][0][0]
+        root_span_kwargs = instance.add_span.call_args_list[0][0][0]
         assert root_span_kwargs["name"] == TraceTaskName.WORKFLOW_TRACE
 
     def test_root_span_has_workflow_tag(self):
         trace_info = _make_workflow_trace_info(message_id=None)
         instance = self._run(trace_info)
 
-        root_span_kwargs = _add_span_mock(instance).call_args_list[0][0][0]
+        root_span_kwargs = instance.add_span.call_args_list[0][0][0]
         assert "workflow" in root_span_kwargs["tags"]
 
     def test_node_execution_spans_are_parented_to_root(self):
@@ -223,9 +214,8 @@ class TestWorkflowTraceWithoutMessageId:
         instance = self._run(trace_info, node_executions=[node_exec])
 
         # call_args_list[0] = root span, [1] = node execution span
-        add_span = _add_span_mock(instance)
-        assert add_span.call_count == 2
-        node_span_kwargs = add_span.call_args_list[1][0][0]
+        assert instance.add_span.call_count == 2
+        node_span_kwargs = instance.add_span.call_args_list[1][0][0]
         assert node_span_kwargs["parent_span_id"] == expected_root_span_id
 
     def test_node_span_not_parented_to_workflow_app_log_id(self):
@@ -250,7 +240,7 @@ class TestWorkflowTraceWithoutMessageId:
         instance = self._run(trace_info, node_executions=[node_exec])
 
         old_parent_id = prepare_opik_uuid(trace_info.start_time, trace_info.workflow_app_log_id)
-        node_span_kwargs = _add_span_mock(instance).call_args_list[1][0][0]
+        node_span_kwargs = instance.add_span.call_args_list[1][0][0]
         assert node_span_kwargs["parent_span_id"] != old_parent_id
 
     def test_root_span_id_differs_from_trace_id(self):
@@ -293,7 +283,7 @@ class TestWorkflowTraceWithMessageId:
         trace_info = _make_workflow_trace_info(message_id=self._MESSAGE_ID)
         instance = self._run(trace_info)
 
-        trace_kwargs = _add_trace_mock(instance).call_args_list[0][0][0]
+        trace_kwargs = instance.add_trace.call_args_list[0][0][0]
         assert trace_kwargs["name"] == TraceTaskName.MESSAGE_TRACE
 
     def test_root_span_uses_workflow_run_id_directly(self):
@@ -302,7 +292,7 @@ class TestWorkflowTraceWithMessageId:
         instance = self._run(trace_info)
 
         expected_root_span_id = prepare_opik_uuid(trace_info.start_time, trace_info.workflow_run_id)
-        root_span_kwargs = _add_span_mock(instance).call_args_list[0][0][0]
+        root_span_kwargs = instance.add_span.call_args_list[0][0][0]
         assert root_span_kwargs["id"] == expected_root_span_id
 
     def test_root_span_id_differs_from_no_message_id_case(self):
@@ -336,5 +326,5 @@ class TestWorkflowTraceWithMessageId:
 
         instance = self._run(trace_info, node_executions=[node_exec])
 
-        node_span_kwargs = _add_span_mock(instance).call_args_list[1][0][0]
+        node_span_kwargs = instance.add_span.call_args_list[1][0][0]
         assert node_span_kwargs["parent_span_id"] == expected_root_span_id

api/providers/trace/trace-tencent/tests/unit_tests/tencent_trace/test_client.py

@@ -5,7 +5,6 @@ from __future__ import annotations
 import sys
 import types
 from types import SimpleNamespace
-from typing import Any, TypedDict, cast
 from unittest.mock import MagicMock
 
 import pytest
@@ -13,7 +12,7 @@ from dify_trace_tencent import client as client_module
 from dify_trace_tencent.client import TencentTraceClient, _get_opentelemetry_sdk_version
 from dify_trace_tencent.entities.tencent_trace_entity import SpanData
 from opentelemetry.sdk.trace import Event
-from opentelemetry.trace import SpanContext, Status, StatusCode, TraceFlags
+from opentelemetry.trace import Status, StatusCode
 
 metric_reader_instances: list[DummyMetricReader] = []
 meter_provider_instances: list[DummyMeterProvider] = []
@@ -81,16 +80,6 @@ class DummyJsonMetricExporterNoTemporality:
         self.kwargs = kwargs
 
 
-class PatchedCoreComponents(TypedDict):
-    span_exporter: MagicMock
-    span_processor: MagicMock
-    tracer: MagicMock
-    span: MagicMock
-    tracer_provider: MagicMock
-    logger: MagicMock
-    trace_api: Any
-
-
 def _add_stub_modules(monkeypatch: pytest.MonkeyPatch) -> None:
     """Drop fake metric modules into sys.modules so the client imports resolve."""
 
@@ -129,7 +118,7 @@ def stub_metric_modules(monkeypatch: pytest.MonkeyPatch) -> None:
 
 
 @pytest.fixture(autouse=True)
-def patch_core_components(monkeypatch: pytest.MonkeyPatch) -> PatchedCoreComponents:
+def patch_core_components(monkeypatch: pytest.MonkeyPatch) -> dict[str, object]:
     span_exporter = MagicMock(name="span_exporter")
     monkeypatch.setattr(client_module, "OTLPSpanExporter", MagicMock(return_value=span_exporter))
 
@@ -179,15 +168,6 @@ def patch_core_components(monkeypatch: pytest.MonkeyPatch) -> PatchedCoreCompone
     }
 
 
-def _make_span_context(trace_id: int = 1, span_id: int = 2) -> SpanContext:
-    return SpanContext(
-        trace_id=trace_id,
-        span_id=span_id,
-        is_remote=False,
-        trace_flags=TraceFlags(TraceFlags.SAMPLED),
-    )
-
-
 def _build_client() -> TencentTraceClient:
     return TencentTraceClient(
         service_name="service",
@@ -228,7 +208,7 @@ def test_resolve_grpc_target_parsable_variants(endpoint: str, expected: tuple[st
 
 
 def test_resolve_grpc_target_handles_errors() -> None:
-    assert TencentTraceClient._resolve_grpc_target(cast(str, 123)) == ("localhost:4317", True, "localhost", 4317)
+    assert TencentTraceClient._resolve_grpc_target(123) == ("localhost:4317", True, "localhost", 4317)
 
 
 @pytest.mark.parametrize(
@@ -268,7 +248,7 @@ def test_record_methods_skip_when_histogram_missing() -> None:
     client.record_trace_duration(0.5)
 
 
-def test_record_llm_duration_handles_exceptions(patch_core_components: PatchedCoreComponents) -> None:
+def test_record_llm_duration_handles_exceptions(patch_core_components: dict[str, object]) -> None:
     client = _build_client()
     client.hist_llm_duration = MagicMock(name="hist_llm_duration")
     client.hist_llm_duration.record.side_effect = RuntimeError("boom")
@@ -278,11 +258,10 @@ def test_record_llm_duration_handles_exceptions(patch_core_components: PatchedCo
     logger.debug.assert_called()
 
 
-def test_create_and_export_span_sets_attributes(patch_core_components: PatchedCoreComponents) -> None:
+def test_create_and_export_span_sets_attributes(patch_core_components: dict[str, object]) -> None:
     client = _build_client()
     span = patch_core_components["span"]
-    ctx = _make_span_context(span_id=2)
-    span.get_span_context.return_value = ctx
+    span.get_span_context.return_value = "ctx"
 
     data = SpanData(
         trace_id=1,
@@ -301,15 +280,14 @@ def test_create_and_export_span_sets_attributes(patch_core_components: PatchedCo
     span.add_event.assert_called_once()
     span.set_status.assert_called_once()
     span.end.assert_called_once_with(end_time=20)
-    assert client.span_contexts[2] == ctx
+    assert client.span_contexts[2] == "ctx"
 
 
-def test_create_and_export_span_uses_parent_context(patch_core_components: PatchedCoreComponents) -> None:
+def test_create_and_export_span_uses_parent_context(patch_core_components: dict[str, object]) -> None:
     client = _build_client()
-    existing_context = _make_span_context(span_id=10)
-    client.span_contexts[10] = existing_context
+    client.span_contexts[10] = "existing"
     span = patch_core_components["span"]
-    span.get_span_context.return_value = _make_span_context(span_id=11)
+    span.get_span_context.return_value = "child"
 
     data = SpanData(
         trace_id=1,
@@ -324,14 +302,14 @@ def test_create_and_export_span_uses_parent_context(patch_core_components: Patch
 
     client._create_and_export_span(data)
     trace_api = patch_core_components["trace_api"]
-    trace_api.NonRecordingSpan.assert_called_once_with(existing_context)
+    trace_api.NonRecordingSpan.assert_called_once_with("existing")
     trace_api.set_span_in_context.assert_called_once()
 
 
-def test_create_and_export_span_exception_logs_error(patch_core_components: PatchedCoreComponents) -> None:
+def test_create_and_export_span_exception_logs_error(patch_core_components: dict[str, object]) -> None:
     client = _build_client()
     span = patch_core_components["span"]
-    span.get_span_context.return_value = _make_span_context(span_id=2)
+    span.get_span_context.return_value = "ctx"
     client.tracer.start_span.side_effect = RuntimeError("boom")
 
     client._create_and_export_span(
@@ -407,7 +385,7 @@ def test_get_project_url() -> None:
     assert client.get_project_url() == "https://console.cloud.tencent.com/apm"
 
 
-def test_shutdown_flushes_all_components(patch_core_components: PatchedCoreComponents) -> None:
+def test_shutdown_flushes_all_components(patch_core_components: dict[str, object]) -> None:
     client = _build_client()
     span_processor = patch_core_components["span_processor"]
     tracer_provider = patch_core_components["tracer_provider"]
@@ -423,11 +401,10 @@ def test_shutdown_flushes_all_components(patch_core_components: PatchedCoreCompo
     metric_reader.shutdown.assert_called_once()
 
 
-def test_shutdown_logs_when_meter_provider_fails(patch_core_components: PatchedCoreComponents) -> None:
+def test_shutdown_logs_when_meter_provider_fails(patch_core_components: dict[str, object]) -> None:
     client = _build_client()
     meter_provider = meter_provider_instances[-1]
     meter_provider.shutdown.side_effect = RuntimeError("boom")
-    assert client.metric_reader is not None
     client.metric_reader.shutdown.side_effect = RuntimeError("boom")
 
     client.shutdown()
@@ -456,7 +433,7 @@ def test_metrics_initialization_failure_sets_histogram_attributes(monkeypatch: p
     assert client.metric_reader is None
 
 
-def test_add_span_logs_exception(monkeypatch: pytest.MonkeyPatch, patch_core_components: PatchedCoreComponents) -> None:
+def test_add_span_logs_exception(monkeypatch: pytest.MonkeyPatch, patch_core_components: dict[str, object]) -> None:
     client = _build_client()
     monkeypatch.setattr(client, "_create_and_export_span", MagicMock(side_effect=RuntimeError("boom")))
 
@@ -477,10 +454,10 @@ def test_add_span_logs_exception(monkeypatch: pytest.MonkeyPatch, patch_core_com
     logger.exception.assert_called_once()
 
 
-def test_create_and_export_span_converts_attribute_types(patch_core_components: PatchedCoreComponents) -> None:
+def test_create_and_export_span_converts_attribute_types(patch_core_components: dict[str, object]) -> None:
     client = _build_client()
     span = patch_core_components["span"]
-    span.get_span_context.return_value = _make_span_context(span_id=2)
+    span.get_span_context.return_value = "ctx"
 
     data = SpanData.model_construct(
         trace_id=1,
@@ -508,7 +485,7 @@ def test_record_llm_duration_converts_attributes() -> None:
     hist_mock = MagicMock(name="hist_llm_duration")
     client.hist_llm_duration = hist_mock
 
-    client.record_llm_duration(0.3, cast(dict[str, str], {"foo": object(), "bar": 2}))
+    client.record_llm_duration(0.3, {"foo": object(), "bar": 2})
     _, attrs = hist_mock.record.call_args.args
     assert isinstance(attrs["foo"], str)
     assert attrs["bar"] == 2
@@ -519,7 +496,7 @@ def test_record_trace_duration_converts_attributes() -> None:
     hist_mock = MagicMock(name="hist_trace_duration")
     client.hist_trace_duration = hist_mock
 
-    client.record_trace_duration(1.0, cast(dict[str, str], {"meta": object(), "ok": True}))
+    client.record_trace_duration(1.0, {"meta": object(), "ok": True})
     _, attrs = hist_mock.record.call_args.args
     assert isinstance(attrs["meta"], str)
     assert attrs["ok"] is True
@@ -535,7 +512,7 @@ def test_record_trace_duration_converts_attributes() -> None:
     ],
 )
 def test_record_methods_handle_exceptions(
-    method: str, attr_name: str, args: tuple[object, ...], patch_core_components: PatchedCoreComponents
+    method: str, attr_name: str, args: tuple[object, ...], patch_core_components: dict[str, object]
 ) -> None:
     client = _build_client()
     hist_mock = MagicMock(name=attr_name)
@@ -550,38 +527,35 @@ def test_record_methods_handle_exceptions(
 def test_metrics_initializes_grpc_metric_exporter() -> None:
     client = _build_client()
     metric_reader = metric_reader_instances[-1]
-    exporter = cast(DummyGrpcMetricExporter, metric_reader.exporter)
 
-    assert isinstance(exporter, DummyGrpcMetricExporter)
+    assert isinstance(metric_reader.exporter, DummyGrpcMetricExporter)
     assert metric_reader.export_interval_millis == client.metrics_export_interval_sec * 1000
-    assert exporter.kwargs["endpoint"] == "trace.example.com:4317"
-    assert exporter.kwargs["insecure"] is False
-    assert cast(dict[str, dict[str, str]], exporter.kwargs)["headers"]["authorization"] == "Bearer token"
+    assert metric_reader.exporter.kwargs["endpoint"] == "trace.example.com:4317"
+    assert metric_reader.exporter.kwargs["insecure"] is False
+    assert metric_reader.exporter.kwargs["headers"]["authorization"] == "Bearer token"
 
 
 def test_metrics_initializes_http_protobuf_metric_exporter(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setenv("OTEL_EXPORTER_OTLP_PROTOCOL", "http/protobuf")
     client = _build_client()
     metric_reader = metric_reader_instances[-1]
-    exporter = cast(DummyHttpMetricExporter, metric_reader.exporter)
 
-    assert isinstance(exporter, DummyHttpMetricExporter)
+    assert isinstance(metric_reader.exporter, DummyHttpMetricExporter)
     assert metric_reader.export_interval_millis == client.metrics_export_interval_sec * 1000
-    assert exporter.kwargs["endpoint"] == client.endpoint
-    assert cast(dict[str, dict[str, str]], exporter.kwargs)["headers"]["authorization"] == "Bearer token"
+    assert metric_reader.exporter.kwargs["endpoint"] == client.endpoint
+    assert metric_reader.exporter.kwargs["headers"]["authorization"] == "Bearer token"
 
 
 def test_metrics_initializes_http_json_metric_exporter(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setenv("OTEL_EXPORTER_OTLP_PROTOCOL", "http/json")
     client = _build_client()
     metric_reader = metric_reader_instances[-1]
-    exporter = cast(DummyJsonMetricExporter, metric_reader.exporter)
 
-    assert isinstance(exporter, DummyJsonMetricExporter)
+    assert isinstance(metric_reader.exporter, DummyJsonMetricExporter)
     assert metric_reader.export_interval_millis == client.metrics_export_interval_sec * 1000
-    assert exporter.kwargs["endpoint"] == client.endpoint
-    assert cast(dict[str, dict[str, str]], exporter.kwargs)["headers"]["authorization"] == "Bearer token"
-    assert "preferred_temporality" in exporter.kwargs
+    assert metric_reader.exporter.kwargs["endpoint"] == client.endpoint
+    assert metric_reader.exporter.kwargs["headers"]["authorization"] == "Bearer token"
+    assert "preferred_temporality" in metric_reader.exporter.kwargs
 
 
 def test_metrics_http_json_metric_exporter_falls_back_without_temporality(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -590,10 +564,9 @@ def test_metrics_http_json_metric_exporter_falls_back_without_temporality(monkey
     monkeypatch.setattr(exporter_module, "OTLPMetricExporter", DummyJsonMetricExporterNoTemporality)
     _ = _build_client()
     metric_reader = metric_reader_instances[-1]
-    exporter = cast(DummyJsonMetricExporterNoTemporality, metric_reader.exporter)
 
-    assert isinstance(exporter, DummyJsonMetricExporterNoTemporality)
-    assert "preferred_temporality" not in exporter.kwargs
+    assert isinstance(metric_reader.exporter, DummyJsonMetricExporterNoTemporality)
+    assert "preferred_temporality" not in metric_reader.exporter.kwargs
 
 
 def test_metrics_http_json_uses_http_fallback_when_no_json_exporter(monkeypatch: pytest.MonkeyPatch) -> None:

api/providers/trace/trace-weave/tests/unit_tests/test_config_entity.py

@@ -31,13 +31,13 @@ class TestWeaveConfig:
     def test_missing_required_fields(self):
         """Test that required fields are enforced"""
         with pytest.raises(ValidationError):
-            WeaveConfig.model_validate({})
+            WeaveConfig()
 
         with pytest.raises(ValidationError):
-            WeaveConfig.model_validate({"api_key": "key"})
+            WeaveConfig(api_key="key")
 
         with pytest.raises(ValidationError):
-            WeaveConfig.model_validate({"project": "project"})
+            WeaveConfig(project="project")
 
     def test_endpoint_validation_https_only(self):
         """Test endpoint validation only allows HTTPS"""