chore(deps): bump google-api-python-client in /api

Bumps [google-api-python-client](https://github.com/googleapis/google-api-python-client) from 2.90.0 to 2.189.0.
- [Release notes](https://github.com/googleapis/google-api-python-client/releases)
- [Commits](https://github.com/googleapis/google-api-python-client/compare/v2.90.0...v2.189.0)

---
updated-dependencies:
- dependency-name: google-api-python-client
  dependency-version: 2.189.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>

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

@@ -480,4 +480,4 @@ const useButtonState = () => {
 ### Related Skills
 
 - `frontend-testing` - For testing refactored components
-- `web/testing/testing.md` - Testing specification
+- `web/docs/test.md` - Testing specification

.agents/skills/frontend-testing/SKILL.md

@@ -7,7 +7,7 @@ description: Generate Vitest + React Testing Library tests for Dify frontend com
 
 This skill enables Claude to generate high-quality, comprehensive frontend tests for the Dify project following established conventions and best practices.
 
-> **⚠️ Authoritative Source**: This skill is derived from `web/testing/testing.md`. Use Vitest mock/timer APIs (`vi.*`).
+> **⚠️ Authoritative Source**: This skill is derived from `web/docs/test.md`. Use Vitest mock/timer APIs (`vi.*`).
 
 ## When to Apply This Skill
 
@@ -309,7 +309,7 @@ For more detailed information, refer to:
 
 ### Primary Specification (MUST follow)
 
-- **`web/testing/testing.md`** - The canonical testing specification. This skill is derived from this document.
+- **`web/docs/test.md`** - The canonical testing specification. This skill is derived from this document.
 
 ### Reference Examples in Codebase
 

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

@@ -4,7 +4,7 @@ This guide defines the workflow for generating tests, especially for complex com
 
 ## Scope Clarification
 
-This guide addresses **multi-file workflow** (how to process multiple test files). For coverage requirements within a single test file, see `web/testing/testing.md` § Coverage Goals.
+This guide addresses **multi-file workflow** (how to process multiple test files). For coverage requirements within a single test file, see `web/docs/test.md` § Coverage Goals.
 
 | Scope | Rule |
 |-------|------|

.codex/skills/component-refactoring

@@ -1 +0,0 @@
-../../.agents/skills/component-refactoring

.codex/skills/frontend-code-review

@@ -1 +0,0 @@
-../../.agents/skills/frontend-code-review

.codex/skills/frontend-testing

@@ -1 +0,0 @@
-../../.agents/skills/frontend-testing

.codex/skills/orpc-contract-first

@@ -1 +0,0 @@
-../../.agents/skills/orpc-contract-first

.github/CODEOWNERS

@@ -9,6 +9,9 @@
 # CODEOWNERS file
 /.github/CODEOWNERS @laipz8200 @crazywoola
 
+# Agents
+/.agents/skills/ @hyoban
+
 # Docs
 /docs/ @crazywoola
 
@@ -21,6 +24,10 @@
 /api/services/tools/mcp_tools_manage_service.py @Nov1c444
 /api/controllers/mcp/ @Nov1c444
 /api/controllers/console/app/mcp_server.py @Nov1c444
+
+# Backend - Tests
+/api/tests/ @laipz8200 @QuantumGhost
+
 /api/tests/**/*mcp* @Nov1c444
 
 # Backend - Workflow - Engine (Core graph execution engine)
@@ -231,6 +238,9 @@
 # Frontend - Base Components
 /web/app/components/base/ @iamjoel @zxhlyh
 
+# Frontend - Base Components Tests
+/web/app/components/base/**/*.spec.tsx @hyoban @CodingOnStar
+
 # Frontend - Utils and Hooks
 /web/utils/classnames.ts @iamjoel @zxhlyh
 /web/utils/time.ts @iamjoel @zxhlyh

.github/workflows/api-tests.yml

@@ -72,6 +72,7 @@ jobs:
           OPENDAL_FS_ROOT: /tmp/dify-storage
         run: |
           uv run --project api pytest \
+            -n auto \
             --timeout "${PYTEST_TIMEOUT:-180}" \
             api/tests/integration_tests/workflow \
             api/tests/integration_tests/tools \

.github/workflows/autofix.yml

@@ -79,29 +79,6 @@ jobs:
           find . -name "*.py" -type f -exec sed -i.bak -E 's/"([^"]+)" \| None/Optional["\1"]/g; s/'"'"'([^'"'"']+)'"'"' \| None/Optional['"'"'\1'"'"']/g' {} \;
           find . -name "*.py.bak" -type f -delete
 
-      - name: Install pnpm
-        uses: pnpm/action-setup@v4
-        with:
-          package_json_file: web/package.json
-          run_install: false
-
-      - name: Setup Node.js
-        uses: actions/setup-node@v6
-        with:
-          node-version: 24
-          cache: pnpm
-          cache-dependency-path: ./web/pnpm-lock.yaml
-
-      - name: Install web dependencies
-        run: |
-          cd web
-          pnpm install --frozen-lockfile
-
-      - name: ESLint autofix
-        run: |
-          cd web
-          pnpm lint:fix || true
-
       # mdformat breaks YAML front matter in markdown files. Add --exclude for directories containing YAML front matter.
       - name: mdformat
         run: |

.github/workflows/deploy-hitl.yml

@@ -4,8 +4,7 @@ on:
   workflow_run:
     workflows: ["Build and Push API & Web"]
     branches:
-      - "feat/hitl-frontend"
-      - "feat/hitl-backend"
+      - "build/feat/hitl"
     types:
       - completed
 
@@ -14,10 +13,7 @@ jobs:
     runs-on: ubuntu-latest
     if: |
       github.event.workflow_run.conclusion == 'success' &&
-      (
-        github.event.workflow_run.head_branch == 'feat/hitl-frontend' ||
-        github.event.workflow_run.head_branch == 'feat/hitl-backend'
-      )
+      github.event.workflow_run.head_branch == 'build/feat/hitl'
     steps:
       - name: Deploy to server
         uses: appleboy/ssh-action@v1

.github/workflows/style.yml

@@ -47,13 +47,9 @@ jobs:
         if: steps.changed-files.outputs.any_changed == 'true'
         run: uv run --directory api --dev lint-imports
 
-      - name: Run Basedpyright Checks
+      - name: Run Type Checks
         if: steps.changed-files.outputs.any_changed == 'true'
-        run: dev/basedpyright-check
-
-      - name: Run Mypy Type Checks
-        if: steps.changed-files.outputs.any_changed == 'true'
-        run: uv --directory api run mypy --exclude-gitignore --exclude 'tests/' --exclude 'migrations/' --check-untyped-defs --disable-error-code=import-untyped .
+        run: make type-check
 
       - name: Dotenv check
         if: steps.changed-files.outputs.any_changed == 'true'

.github/workflows/web-tests.yml

@@ -39,7 +39,7 @@ jobs:
         run: pnpm install --frozen-lockfile
 
       - name: Run tests
-        run: pnpm test:coverage
+        run: pnpm test:ci
 
       - name: Coverage Summary
         if: always()

AGENTS.md

@@ -7,7 +7,7 @@ Dify is an open-source platform for developing LLM applications with an intuitiv
 The codebase is split into:
 
 - **Backend API** (`/api`): Python Flask application organized with Domain-Driven Design
-- **Frontend Web** (`/web`): Next.js 15 application using TypeScript and React 19
+- **Frontend Web** (`/web`): Next.js application using TypeScript and React
 - **Docker deployment** (`/docker`): Containerized deployment configurations
 
 ## Backend Workflow
@@ -18,36 +18,7 @@ The codebase is split into:
 
 ## Frontend Workflow
 
-```bash
-cd web
-pnpm lint:fix
-pnpm type-check:tsgo
-pnpm test
-```
-
-### Frontend Linting
-
-ESLint is used for frontend code quality. Available commands:
-
-```bash
-# Lint all files (report only)
-pnpm lint
-
-# Lint and auto-fix issues
-pnpm lint:fix
-
-# Lint specific files or directories
-pnpm lint:fix app/components/base/button/
-pnpm lint:fix app/components/base/button/index.tsx
-
-# Lint quietly (errors only, no warnings)
-pnpm lint:quiet
-
-# Check code complexity
-pnpm lint:complexity
-```
-
-**Important**: Always run `pnpm lint:fix` before committing. The pre-commit hook runs `lint-staged` which only lints staged files.
+- Read `web/AGENTS.md` for details
 
 ## Testing & Quality Practices
 

CONTRIBUTING.md

@@ -77,7 +77,7 @@ How we prioritize:
 
 For setting up the frontend service, please refer to our comprehensive [guide](https://github.com/langgenius/dify/blob/main/web/README.md) in the `web/README.md` file. This document provides detailed instructions to help you set up the frontend environment properly.
 
-**Testing**: All React components must have comprehensive test coverage. See [web/testing/testing.md](https://github.com/langgenius/dify/blob/main/web/testing/testing.md) for the canonical frontend testing guidelines and follow every requirement described there.
+**Testing**: All React components must have comprehensive test coverage. See [web/docs/test.md](https://github.com/langgenius/dify/blob/main/web/docs/test.md) for the canonical frontend testing guidelines and follow every requirement described there.
 
 #### Backend
 

Makefile

@@ -68,9 +68,11 @@ lint:
 	@echo "✅ Linting complete"
 
 type-check:
-	@echo "📝 Running type check with basedpyright..."
-	@uv run --directory api --dev basedpyright
-	@echo "✅ Type check complete"
+	@echo "📝 Running type checks (basedpyright + mypy + ty)..."
+	@./dev/basedpyright-check $(PATH_TO_CHECK)
+	@uv --directory api run mypy --exclude-gitignore --exclude 'tests/' --exclude 'migrations/' --check-untyped-defs --disable-error-code=import-untyped .
+	@cd api && uv run ty check
+	@echo "✅ Type checks complete"
 
 test:
 	@echo "🧪 Running backend unit tests..."
@@ -78,7 +80,7 @@ test:
 		echo "Target: $(TARGET_TESTS)"; \
 		uv run --project api --dev pytest $(TARGET_TESTS); \
 	else \
-		uv run --project api --dev dev/pytest/pytest_unit_tests.sh; \
+		PYTEST_XDIST_ARGS="-n auto" uv run --project api --dev dev/pytest/pytest_unit_tests.sh; \
 	fi
 	@echo "✅ Tests complete"
 
@@ -130,7 +132,7 @@ help:
 	@echo "  make format         - Format code with ruff"
 	@echo "  make check          - Check code with ruff"
 	@echo "  make lint           - Format, fix, and lint code (ruff, imports, dotenv)"
-	@echo "  make type-check     - Run type checking with basedpyright"
+	@echo "  make type-check     - Run type checks (basedpyright, mypy, ty)"
 	@echo "  make test           - Run backend unit tests (or TARGET_TESTS=./api/tests/<target_tests>)"
 	@echo ""
 	@echo "Docker Build Targets:"

api/.env.example

@@ -617,6 +617,7 @@ PLUGIN_DAEMON_URL=http://127.0.0.1:5002
 PLUGIN_REMOTE_INSTALL_PORT=5003
 PLUGIN_REMOTE_INSTALL_HOST=localhost
 PLUGIN_MAX_PACKAGE_SIZE=15728640
+PLUGIN_MODEL_SCHEMA_CACHE_TTL=3600
 INNER_API_KEY_FOR_PLUGIN=QaHbTe77CtuXmsfyhR7+vRjI/+XbV1AaFy691iy+kGDv2Jvy0/eAh8Y1
 
 # Marketplace configuration
@@ -716,4 +717,3 @@ SANDBOX_EXPIRED_RECORDS_CLEAN_GRACEFUL_PERIOD=21
 SANDBOX_EXPIRED_RECORDS_CLEAN_BATCH_SIZE=1000
 SANDBOX_EXPIRED_RECORDS_RETENTION_DAYS=30
 SANDBOX_EXPIRED_RECORDS_CLEAN_TASK_LOCK_TTL=90000
-

api/.importlinter

@@ -136,7 +136,6 @@ ignore_imports =
     core.workflow.nodes.llm.llm_utils -> models.provider
     core.workflow.nodes.llm.llm_utils -> services.credit_pool_service
     core.workflow.nodes.llm.node -> core.tools.signature
-    core.workflow.nodes.template_transform.template_transform_node -> configs
     core.workflow.nodes.tool.tool_node -> core.callback_handler.workflow_tool_callback_handler
     core.workflow.nodes.tool.tool_node -> core.tools.tool_engine
     core.workflow.nodes.tool.tool_node -> core.tools.tool_manager

api/.ruff.toml

@@ -53,6 +53,7 @@ select = [
     "S301", # suspicious-pickle-usage, disallow use of `pickle` and its wrappers.
     "S302", # suspicious-marshal-usage, disallow use of `marshal` module
     "S311", # suspicious-non-cryptographic-random-usage,
+    "TID",   # flake8-tidy-imports
 
 ]
 
@@ -88,6 +89,7 @@ ignore = [
     "SIM113",  # enumerate-for-loop
     "SIM117",  # multiple-with-statements
     "SIM210",  # if-expr-with-true-false
+    "TID252",  # allow relative imports from parent modules
 ]
 
 [lint.per-file-ignores]
@@ -109,10 +111,20 @@ ignore = [
     "S110", # allow ignoring exceptions in tests code (currently)
 
 ]
+"controllers/console/explore/trial.py" = ["TID251"]
+"controllers/console/human_input_form.py" = ["TID251"]
+"controllers/web/human_input_form.py" = ["TID251"]
 
 [lint.pyflakes]
 allowed-unused-imports = [
-    "_pytest.monkeypatch",
     "tests.integration_tests",
     "tests.unit_tests",
 ]
+
+[lint.flake8-tidy-imports]
+
+[lint.flake8-tidy-imports.banned-api."flask_restx.reqparse"]
+msg = "Use Pydantic payload/query models instead of reqparse."
+
+[lint.flake8-tidy-imports.banned-api."flask_restx.reqparse.RequestParser"]
+msg = "Use Pydantic payload/query models instead of reqparse."

api/README.md

@@ -122,7 +122,7 @@ These commands assume you start from the repository root.
 
    ```bash
    cd api
-   uv run celery -A app.celery worker -P threads -c 2 --loglevel INFO -Q dataset,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor,retention
+   uv run celery -A app.celery worker -P threads -c 2 --loglevel INFO -Q api_token,dataset,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor,retention
    ```
 
 1. Optional: start Celery Beat (scheduled tasks, in a new terminal).

api/app.py

@@ -1,4 +1,12 @@
+from __future__ import annotations
+
 import sys
+from typing import TYPE_CHECKING, cast
+
+if TYPE_CHECKING:
+    from celery import Celery
+
+    celery: Celery
 
 
 def is_db_command() -> bool:
@@ -23,7 +31,7 @@ else:
     from app_factory import create_app
 
     app = create_app()
-    celery = app.extensions["celery"]
+    celery = cast("Celery", app.extensions["celery"])
 
 if __name__ == "__main__":
     app.run(host="0.0.0.0", port=5001)

api/app_factory.py

@@ -149,7 +149,7 @@ def initialize_extensions(app: DifyApp):
             logger.info("Loaded %s (%s ms)", short_name, round((end_time - start_time) * 1000, 2))
 
 
-def create_migrations_app():
+def create_migrations_app() -> DifyApp:
     app = create_flask_app_with_configs()
     from extensions import ext_database, ext_migrate
 

api/commands.py

@@ -739,8 +739,10 @@ def upgrade_db():
 
             click.echo(click.style("Database migration successful!", fg="green"))
 
-        except Exception:
+        except Exception as e:
             logger.exception("Failed to execute database migration")
+            click.echo(click.style(f"Database migration failed: {e}", fg="red"))
+            raise SystemExit(1)
         finally:
             lock.release()
     else:
@@ -1450,54 +1452,58 @@ def clear_orphaned_file_records(force: bool):
         all_ids_in_tables = []
         for ids_table in ids_tables:
             query = ""
-            if ids_table["type"] == "uuid":
-                click.echo(
-                    click.style(
-                        f"- Listing file ids in column {ids_table['column']} in table {ids_table['table']}", fg="white"
+            match ids_table["type"]:
+                case "uuid":
+                    click.echo(
+                        click.style(
+                            f"- Listing file ids in column {ids_table['column']} in table {ids_table['table']}",
+                            fg="white",
+                        )
                     )
-                )
-                query = (
-                    f"SELECT {ids_table['column']} FROM {ids_table['table']} WHERE {ids_table['column']} IS NOT NULL"
-                )
-                with db.engine.begin() as conn:
-                    rs = conn.execute(sa.text(query))
-                for i in rs:
-                    all_ids_in_tables.append({"table": ids_table["table"], "id": str(i[0])})
-            elif ids_table["type"] == "text":
-                click.echo(
-                    click.style(
-                        f"- Listing file-id-like strings in column {ids_table['column']} in table {ids_table['table']}",
-                        fg="white",
+                    c = ids_table["column"]
+                    query = f"SELECT {c} FROM {ids_table['table']} WHERE {c} IS NOT NULL"
+                    with db.engine.begin() as conn:
+                        rs = conn.execute(sa.text(query))
+                    for i in rs:
+                        all_ids_in_tables.append({"table": ids_table["table"], "id": str(i[0])})
+                case "text":
+                    t = ids_table["table"]
+                    click.echo(
+                        click.style(
+                            f"- Listing file-id-like strings in column {ids_table['column']} in table {t}",
+                            fg="white",
+                        )
                     )
-                )
-                query = (
-                    f"SELECT regexp_matches({ids_table['column']}, '{guid_regexp}', 'g') AS extracted_id "
-                    f"FROM {ids_table['table']}"
-                )
-                with db.engine.begin() as conn:
-                    rs = conn.execute(sa.text(query))
-                for i in rs:
-                    for j in i[0]:
-                        all_ids_in_tables.append({"table": ids_table["table"], "id": j})
-            elif ids_table["type"] == "json":
-                click.echo(
-                    click.style(
-                        (
-                            f"- Listing file-id-like JSON string in column {ids_table['column']} "
-                            f"in table {ids_table['table']}"
-                        ),
-                        fg="white",
+                    query = (
+                        f"SELECT regexp_matches({ids_table['column']}, '{guid_regexp}', 'g') AS extracted_id "
+                        f"FROM {ids_table['table']}"
                     )
-                )
-                query = (
-                    f"SELECT regexp_matches({ids_table['column']}::text, '{guid_regexp}', 'g') AS extracted_id "
-                    f"FROM {ids_table['table']}"
-                )
-                with db.engine.begin() as conn:
-                    rs = conn.execute(sa.text(query))
-                for i in rs:
-                    for j in i[0]:
-                        all_ids_in_tables.append({"table": ids_table["table"], "id": j})
+                    with db.engine.begin() as conn:
+                        rs = conn.execute(sa.text(query))
+                    for i in rs:
+                        for j in i[0]:
+                            all_ids_in_tables.append({"table": ids_table["table"], "id": j})
+                case "json":
+                    click.echo(
+                        click.style(
+                            (
+                                f"- Listing file-id-like JSON string in column {ids_table['column']} "
+                                f"in table {ids_table['table']}"
+                            ),
+                            fg="white",
+                        )
+                    )
+                    query = (
+                        f"SELECT regexp_matches({ids_table['column']}::text, '{guid_regexp}', 'g') AS extracted_id "
+                        f"FROM {ids_table['table']}"
+                    )
+                    with db.engine.begin() as conn:
+                        rs = conn.execute(sa.text(query))
+                    for i in rs:
+                        for j in i[0]:
+                            all_ids_in_tables.append({"table": ids_table["table"], "id": j})
+                case _:
+                    pass
         click.echo(click.style(f"Found {len(all_ids_in_tables)} file ids in tables.", fg="white"))
 
     except Exception as e:
@@ -1737,59 +1743,18 @@ def file_usage(
                 if src_filter != src:
                     continue
 
-        if ids_table["type"] == "uuid":
-            # Direct UUID match
-            query = (
-                f"SELECT {ids_table['pk_column']}, {ids_table['column']} "
-                f"FROM {ids_table['table']} WHERE {ids_table['column']} IS NOT NULL"
-            )
-            with db.engine.begin() as conn:
-                rs = conn.execute(sa.text(query))
-                for row in rs:
-                    record_id = str(row[0])
-                    ref_file_id = str(row[1])
-                    if ref_file_id not in file_key_map:
-                        continue
-                    storage_key = file_key_map[ref_file_id]
-
-                    # Apply filters
-                    if file_id and ref_file_id != file_id:
-                        continue
-                    if key and not storage_key.endswith(key):
-                        continue
-
-                    # Only collect items within the requested page range
-                    if offset <= total_count < offset + limit:
-                        paginated_usages.append(
-                            {
-                                "src": f"{ids_table['table']}.{ids_table['column']}",
-                                "record_id": record_id,
-                                "file_id": ref_file_id,
-                                "key": storage_key,
-                            }
-                        )
-                    total_count += 1
-
-        elif ids_table["type"] in ("text", "json"):
-            # Extract UUIDs from text/json content
-            column_cast = f"{ids_table['column']}::text" if ids_table["type"] == "json" else ids_table["column"]
-            query = (
-                f"SELECT {ids_table['pk_column']}, {column_cast} "
-                f"FROM {ids_table['table']} WHERE {ids_table['column']} IS NOT NULL"
-            )
-            with db.engine.begin() as conn:
-                rs = conn.execute(sa.text(query))
-                for row in rs:
-                    record_id = str(row[0])
-                    content = str(row[1])
-
-                    # Find all UUIDs in the content
-                    import re
-
-                    uuid_pattern = re.compile(guid_regexp, re.IGNORECASE)
-                    matches = uuid_pattern.findall(content)
-
-                    for ref_file_id in matches:
+        match ids_table["type"]:
+            case "uuid":
+                # Direct UUID match
+                query = (
+                    f"SELECT {ids_table['pk_column']}, {ids_table['column']} "
+                    f"FROM {ids_table['table']} WHERE {ids_table['column']} IS NOT NULL"
+                )
+                with db.engine.begin() as conn:
+                    rs = conn.execute(sa.text(query))
+                    for row in rs:
+                        record_id = str(row[0])
+                        ref_file_id = str(row[1])
                         if ref_file_id not in file_key_map:
                             continue
                         storage_key = file_key_map[ref_file_id]
@@ -1812,6 +1777,50 @@ def file_usage(
                             )
                         total_count += 1
 
+            case "text" | "json":
+                # Extract UUIDs from text/json content
+                column_cast = f"{ids_table['column']}::text" if ids_table["type"] == "json" else ids_table["column"]
+                query = (
+                    f"SELECT {ids_table['pk_column']}, {column_cast} "
+                    f"FROM {ids_table['table']} WHERE {ids_table['column']} IS NOT NULL"
+                )
+                with db.engine.begin() as conn:
+                    rs = conn.execute(sa.text(query))
+                    for row in rs:
+                        record_id = str(row[0])
+                        content = str(row[1])
+
+                        # Find all UUIDs in the content
+                        import re
+
+                        uuid_pattern = re.compile(guid_regexp, re.IGNORECASE)
+                        matches = uuid_pattern.findall(content)
+
+                        for ref_file_id in matches:
+                            if ref_file_id not in file_key_map:
+                                continue
+                            storage_key = file_key_map[ref_file_id]
+
+                            # Apply filters
+                            if file_id and ref_file_id != file_id:
+                                continue
+                            if key and not storage_key.endswith(key):
+                                continue
+
+                            # Only collect items within the requested page range
+                            if offset <= total_count < offset + limit:
+                                paginated_usages.append(
+                                    {
+                                        "src": f"{ids_table['table']}.{ids_table['column']}",
+                                        "record_id": record_id,
+                                        "file_id": ref_file_id,
+                                        "key": storage_key,
+                                    }
+                                )
+                            total_count += 1
+            case _:
+                pass
+
     # Output results
     if output_json:
         result = {

api/configs/feature/__init__.py

@@ -245,7 +245,7 @@ class PluginConfig(BaseSettings):
 
     PLUGIN_MODEL_SCHEMA_CACHE_TTL: PositiveInt = Field(
         description="TTL in seconds for caching plugin model schemas in Redis",
-        default=24 * 60 * 60,
+        default=60 * 60,
     )
 
 
@@ -1155,6 +1155,16 @@ class CeleryScheduleTasksConfig(BaseSettings):
         default=0,
     )
 
+    # API token last_used_at batch update
+    ENABLE_API_TOKEN_LAST_USED_UPDATE_TASK: bool = Field(
+        description="Enable periodic batch update of API token last_used_at timestamps",
+        default=True,
+    )
+    API_TOKEN_LAST_USED_UPDATE_INTERVAL: int = Field(
+        description="Interval in minutes for batch updating API token last_used_at (default 30)",
+        default=30,
+    )
+
     # Trigger provider refresh (simple version)
     ENABLE_TRIGGER_PROVIDER_REFRESH_TASK: bool = Field(
         description="Enable trigger provider refresh poller",

api/controllers/console/__init__.py

@@ -115,12 +115,6 @@ from .explore import (
     trial,
 )
 
-# Import evaluation controllers
-from .evaluation import evaluation
-
-# Import snippet controllers
-from .snippets import snippet_workflow
-
 # Import tag controllers
 from .tag import tags
 
@@ -134,7 +128,6 @@ from .workspace import (
     model_providers,
     models,
     plugin,
-    snippets,
     tool_providers,
     trigger_providers,
     workspace,
@@ -172,7 +165,6 @@ __all__ = [
     "datasource_content_preview",
     "email_register",
     "endpoint",
-    "evaluation",
     "extension",
     "external",
     "feature",
@@ -205,8 +197,6 @@ __all__ = [
     "saved_message",
     "setup",
     "site",
-    "snippet_workflow",
-    "snippets",
     "spec",
     "statistic",
     "tags",

api/controllers/console/apikey.py

@@ -10,6 +10,7 @@ from libs.helper import TimestampField
 from libs.login import current_account_with_tenant, login_required
 from models.dataset import Dataset
 from models.model import ApiToken, App
+from services.api_token_service import ApiTokenCache
 
 from . import console_ns
 from .wraps import account_initialization_required, edit_permission_required, setup_required
@@ -131,6 +132,11 @@ class BaseApiKeyResource(Resource):
         if key is None:
             flask_restx.abort(HTTPStatus.NOT_FOUND, message="API key not found")
 
+        # Invalidate cache before deleting from database
+        # Type assertion: key is guaranteed to be non-None here because abort() raises
+        assert key is not None  # nosec - for type checker only
+        ApiTokenCache.delete(key.token, key.type)
+
         db.session.query(ApiToken).where(ApiToken.id == api_key_id).delete()
         db.session.commit()
 

api/controllers/console/app/annotation.py

@@ -1,10 +1,11 @@
 from typing import Any, Literal
 
 from flask import abort, make_response, request
-from flask_restx import Resource, fields, marshal, marshal_with
-from pydantic import BaseModel, Field, field_validator
+from flask_restx import Resource
+from pydantic import BaseModel, Field, TypeAdapter, field_validator
 
 from controllers.common.errors import NoFileUploadedError, TooManyFilesError
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import (
     account_initialization_required,
@@ -16,9 +17,11 @@ from controllers.console.wraps import (
 )
 from extensions.ext_redis import redis_client
 from fields.annotation_fields import (
-    annotation_fields,
-    annotation_hit_history_fields,
-    build_annotation_model,
+    Annotation,
+    AnnotationExportList,
+    AnnotationHitHistory,
+    AnnotationHitHistoryList,
+    AnnotationList,
 )
 from libs.helper import uuid_value
 from libs.login import login_required
@@ -89,6 +92,14 @@ reg(CreateAnnotationPayload)
 reg(UpdateAnnotationPayload)
 reg(AnnotationReplyStatusQuery)
 reg(AnnotationFilePayload)
+register_schema_models(
+    console_ns,
+    Annotation,
+    AnnotationList,
+    AnnotationExportList,
+    AnnotationHitHistory,
+    AnnotationHitHistoryList,
+)
 
 
 @console_ns.route("/apps/<uuid:app_id>/annotation-reply/<string:action>")
@@ -107,10 +118,11 @@ class AnnotationReplyActionApi(Resource):
     def post(self, app_id, action: Literal["enable", "disable"]):
         app_id = str(app_id)
         args = AnnotationReplyPayload.model_validate(console_ns.payload)
-        if action == "enable":
-            result = AppAnnotationService.enable_app_annotation(args.model_dump(), app_id)
-        elif action == "disable":
-            result = AppAnnotationService.disable_app_annotation(app_id)
+        match action:
+            case "enable":
+                result = AppAnnotationService.enable_app_annotation(args.model_dump(), app_id)
+            case "disable":
+                result = AppAnnotationService.disable_app_annotation(app_id)
         return result, 200
 
 
@@ -201,33 +213,33 @@ class AnnotationApi(Resource):
 
         app_id = str(app_id)
         annotation_list, total = AppAnnotationService.get_annotation_list_by_app_id(app_id, page, limit, keyword)
-        response = {
-            "data": marshal(annotation_list, annotation_fields),
-            "has_more": len(annotation_list) == limit,
-            "limit": limit,
-            "total": total,
-            "page": page,
-        }
-        return response, 200
+        annotation_models = TypeAdapter(list[Annotation]).validate_python(annotation_list, from_attributes=True)
+        response = AnnotationList(
+            data=annotation_models,
+            has_more=len(annotation_list) == limit,
+            limit=limit,
+            total=total,
+            page=page,
+        )
+        return response.model_dump(mode="json"), 200
 
     @console_ns.doc("create_annotation")
     @console_ns.doc(description="Create a new annotation for an app")
     @console_ns.doc(params={"app_id": "Application ID"})
     @console_ns.expect(console_ns.models[CreateAnnotationPayload.__name__])
-    @console_ns.response(201, "Annotation created successfully", build_annotation_model(console_ns))
+    @console_ns.response(201, "Annotation created successfully", console_ns.models[Annotation.__name__])
     @console_ns.response(403, "Insufficient permissions")
     @setup_required
     @login_required
     @account_initialization_required
     @cloud_edition_billing_resource_check("annotation")
-    @marshal_with(annotation_fields)
     @edit_permission_required
     def post(self, app_id):
         app_id = str(app_id)
         args = CreateAnnotationPayload.model_validate(console_ns.payload)
         data = args.model_dump(exclude_none=True)
         annotation = AppAnnotationService.up_insert_app_annotation_from_message(data, app_id)
-        return annotation
+        return Annotation.model_validate(annotation, from_attributes=True).model_dump(mode="json")
 
     @setup_required
     @login_required
@@ -264,7 +276,7 @@ class AnnotationExportApi(Resource):
     @console_ns.response(
         200,
         "Annotations exported successfully",
-        console_ns.model("AnnotationList", {"data": fields.List(fields.Nested(build_annotation_model(console_ns)))}),
+        console_ns.models[AnnotationExportList.__name__],
     )
     @console_ns.response(403, "Insufficient permissions")
     @setup_required
@@ -274,7 +286,8 @@ class AnnotationExportApi(Resource):
     def get(self, app_id):
         app_id = str(app_id)
         annotation_list = AppAnnotationService.export_annotation_list_by_app_id(app_id)
-        response_data = {"data": marshal(annotation_list, annotation_fields)}
+        annotation_models = TypeAdapter(list[Annotation]).validate_python(annotation_list, from_attributes=True)
+        response_data = AnnotationExportList(data=annotation_models).model_dump(mode="json")
 
         # Create response with secure headers for CSV export
         response = make_response(response_data, 200)
@@ -289,7 +302,7 @@ class AnnotationUpdateDeleteApi(Resource):
     @console_ns.doc("update_delete_annotation")
     @console_ns.doc(description="Update or delete an annotation")
     @console_ns.doc(params={"app_id": "Application ID", "annotation_id": "Annotation ID"})
-    @console_ns.response(200, "Annotation updated successfully", build_annotation_model(console_ns))
+    @console_ns.response(200, "Annotation updated successfully", console_ns.models[Annotation.__name__])
     @console_ns.response(204, "Annotation deleted successfully")
     @console_ns.response(403, "Insufficient permissions")
     @console_ns.expect(console_ns.models[UpdateAnnotationPayload.__name__])
@@ -298,7 +311,6 @@ class AnnotationUpdateDeleteApi(Resource):
     @account_initialization_required
     @cloud_edition_billing_resource_check("annotation")
     @edit_permission_required
-    @marshal_with(annotation_fields)
     def post(self, app_id, annotation_id):
         app_id = str(app_id)
         annotation_id = str(annotation_id)
@@ -306,7 +318,7 @@ class AnnotationUpdateDeleteApi(Resource):
         annotation = AppAnnotationService.update_app_annotation_directly(
             args.model_dump(exclude_none=True), app_id, annotation_id
         )
-        return annotation
+        return Annotation.model_validate(annotation, from_attributes=True).model_dump(mode="json")
 
     @setup_required
     @login_required
@@ -414,14 +426,7 @@ class AnnotationHitHistoryListApi(Resource):
     @console_ns.response(
         200,
         "Hit histories retrieved successfully",
-        console_ns.model(
-            "AnnotationHitHistoryList",
-            {
-                "data": fields.List(
-                    fields.Nested(console_ns.model("AnnotationHitHistoryItem", annotation_hit_history_fields))
-                )
-            },
-        ),
+        console_ns.models[AnnotationHitHistoryList.__name__],
     )
     @console_ns.response(403, "Insufficient permissions")
     @setup_required
@@ -436,11 +441,14 @@ class AnnotationHitHistoryListApi(Resource):
         annotation_hit_history_list, total = AppAnnotationService.get_annotation_hit_histories(
             app_id, annotation_id, page, limit
         )
-        response = {
-            "data": marshal(annotation_hit_history_list, annotation_hit_history_fields),
-            "has_more": len(annotation_hit_history_list) == limit,
-            "limit": limit,
-            "total": total,
-            "page": page,
-        }
-        return response
+        history_models = TypeAdapter(list[AnnotationHitHistory]).validate_python(
+            annotation_hit_history_list, from_attributes=True
+        )
+        response = AnnotationHitHistoryList(
+            data=history_models,
+            has_more=len(annotation_hit_history_list) == limit,
+            limit=limit,
+            total=total,
+            page=page,
+        )
+        return response.model_dump(mode="json")

api/controllers/console/app/app.py

@@ -1,3 +1,4 @@
+import logging
 import uuid
 from datetime import datetime
 from typing import Any, Literal, TypeAlias
@@ -54,6 +55,8 @@ ALLOW_CREATE_APP_MODES = ["chat", "agent-chat", "advanced-chat", "workflow", "co
 
 register_enum_models(console_ns, IconType)
 
+_logger = logging.getLogger(__name__)
+
 
 class AppListQuery(BaseModel):
     page: int = Field(default=1, ge=1, le=99999, description="Page number (1-99999)")
@@ -499,6 +502,7 @@ class AppListApi(Resource):
                     select(Workflow).where(
                         Workflow.version == Workflow.VERSION_DRAFT,
                         Workflow.app_id.in_(workflow_capable_app_ids),
+                        Workflow.tenant_id == current_tenant_id,
                     )
                 )
                 .scalars()
@@ -510,12 +514,14 @@ class AppListApi(Resource):
                 NodeType.TRIGGER_PLUGIN,
             }
             for workflow in draft_workflows:
+                node_id = None
                 try:
-                    for _, node_data in workflow.walk_nodes():
+                    for node_id, node_data in workflow.walk_nodes():
                         if node_data.get("type") in trigger_node_types:
                             draft_trigger_app_ids.add(str(workflow.app_id))
                             break
                 except Exception:
+                    _logger.exception("error while walking nodes, workflow_id=%s, node_id=%s", workflow.id, node_id)
                     continue
 
         for app in app_pagination.items:

api/controllers/console/app/audio.py

@@ -6,6 +6,7 @@ from pydantic import BaseModel, Field
 from werkzeug.exceptions import InternalServerError
 
 import services
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.error import (
     AppUnavailableError,
@@ -33,7 +34,6 @@ from services.errors.audio import (
 )
 
 logger = logging.getLogger(__name__)
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
 
 class TextToSpeechPayload(BaseModel):
@@ -47,13 +47,11 @@ class TextToSpeechVoiceQuery(BaseModel):
     language: str = Field(..., description="Language code")
 
 
-console_ns.schema_model(
-    TextToSpeechPayload.__name__, TextToSpeechPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
-)
-console_ns.schema_model(
-    TextToSpeechVoiceQuery.__name__,
-    TextToSpeechVoiceQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
+class AudioTranscriptResponse(BaseModel):
+    text: str = Field(description="Transcribed text from audio")
+
+
+register_schema_models(console_ns, AudioTranscriptResponse, TextToSpeechPayload, TextToSpeechVoiceQuery)
 
 
 @console_ns.route("/apps/<uuid:app_id>/audio-to-text")
@@ -64,7 +62,7 @@ class ChatMessageAudioApi(Resource):
     @console_ns.response(
         200,
         "Audio transcription successful",
-        console_ns.model("AudioTranscriptResponse", {"text": fields.String(description="Transcribed text from audio")}),
+        console_ns.models[AudioTranscriptResponse.__name__],
     )
     @console_ns.response(400, "Bad request - No audio uploaded or unsupported type")
     @console_ns.response(413, "Audio file too large")

api/controllers/console/app/conversation.py

@@ -508,16 +508,19 @@ class ChatConversationApi(Resource):
                 case "created_at" | "-created_at" | _:
                     query = query.where(Conversation.created_at <= end_datetime_utc)
 
-        if args.annotation_status == "annotated":
-            query = query.options(joinedload(Conversation.message_annotations)).join(  # type: ignore
-                MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id
-            )
-        elif args.annotation_status == "not_annotated":
-            query = (
-                query.outerjoin(MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id)
-                .group_by(Conversation.id)
-                .having(func.count(MessageAnnotation.id) == 0)
-            )
+        match args.annotation_status:
+            case "annotated":
+                query = query.options(joinedload(Conversation.message_annotations)).join(  # type: ignore
+                    MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id
+                )
+            case "not_annotated":
+                query = (
+                    query.outerjoin(MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id)
+                    .group_by(Conversation.id)
+                    .having(func.count(MessageAnnotation.id) == 0)
+                )
+            case "all":
+                pass
 
         if app_model.mode == AppMode.ADVANCED_CHAT:
             query = query.where(Conversation.invoke_from != InvokeFrom.DEBUGGER)

api/controllers/console/app/generator.py

@@ -1,5 +1,4 @@
 from collections.abc import Sequence
-from typing import Any
 
 from flask_restx import Resource
 from pydantic import BaseModel, Field
@@ -12,10 +11,12 @@ from controllers.console.app.error import (
     ProviderQuotaExceededError,
 )
 from controllers.console.wraps import account_initialization_required, setup_required
+from core.app.app_config.entities import ModelConfig
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
 from core.helper.code_executor.code_node_provider import CodeNodeProvider
 from core.helper.code_executor.javascript.javascript_code_provider import JavascriptCodeProvider
 from core.helper.code_executor.python3.python3_code_provider import Python3CodeProvider
+from core.llm_generator.entities import RuleCodeGeneratePayload, RuleGeneratePayload, RuleStructuredOutputPayload
 from core.llm_generator.llm_generator import LLMGenerator
 from core.model_runtime.errors.invoke import InvokeError
 from extensions.ext_database import db
@@ -26,28 +27,13 @@ from services.workflow_service import WorkflowService
 DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
 
-class RuleGeneratePayload(BaseModel):
-    instruction: str = Field(..., description="Rule generation instruction")
-    model_config_data: dict[str, Any] = Field(..., alias="model_config", description="Model configuration")
-    no_variable: bool = Field(default=False, description="Whether to exclude variables")
-
-
-class RuleCodeGeneratePayload(RuleGeneratePayload):
-    code_language: str = Field(default="javascript", description="Programming language for code generation")
-
-
-class RuleStructuredOutputPayload(BaseModel):
-    instruction: str = Field(..., description="Structured output generation instruction")
-    model_config_data: dict[str, Any] = Field(..., alias="model_config", description="Model configuration")
-
-
 class InstructionGeneratePayload(BaseModel):
     flow_id: str = Field(..., description="Workflow/Flow ID")
     node_id: str = Field(default="", description="Node ID for workflow context")
     current: str = Field(default="", description="Current instruction text")
     language: str = Field(default="javascript", description="Programming language (javascript/python)")
     instruction: str = Field(..., description="Instruction for generation")
-    model_config_data: dict[str, Any] = Field(..., alias="model_config", description="Model configuration")
+    model_config_data: ModelConfig = Field(..., alias="model_config", description="Model configuration")
     ideal_output: str = Field(default="", description="Expected ideal output")
 
 
@@ -64,6 +50,7 @@ reg(RuleCodeGeneratePayload)
 reg(RuleStructuredOutputPayload)
 reg(InstructionGeneratePayload)
 reg(InstructionTemplatePayload)
+reg(ModelConfig)
 
 
 @console_ns.route("/rule-generate")
@@ -82,12 +69,7 @@ class RuleGenerateApi(Resource):
         _, current_tenant_id = current_account_with_tenant()
 
         try:
-            rules = LLMGenerator.generate_rule_config(
-                tenant_id=current_tenant_id,
-                instruction=args.instruction,
-                model_config=args.model_config_data,
-                no_variable=args.no_variable,
-            )
+            rules = LLMGenerator.generate_rule_config(tenant_id=current_tenant_id, args=args)
         except ProviderTokenNotInitError as ex:
             raise ProviderNotInitializeError(ex.description)
         except QuotaExceededError:
@@ -118,9 +100,7 @@ class RuleCodeGenerateApi(Resource):
         try:
             code_result = LLMGenerator.generate_code(
                 tenant_id=current_tenant_id,
-                instruction=args.instruction,
-                model_config=args.model_config_data,
-                code_language=args.code_language,
+                args=args,
             )
         except ProviderTokenNotInitError as ex:
             raise ProviderNotInitializeError(ex.description)
@@ -152,8 +132,7 @@ class RuleStructuredOutputGenerateApi(Resource):
         try:
             structured_output = LLMGenerator.generate_structured_output(
                 tenant_id=current_tenant_id,
-                instruction=args.instruction,
-                model_config=args.model_config_data,
+                args=args,
             )
         except ProviderTokenNotInitError as ex:
             raise ProviderNotInitializeError(ex.description)
@@ -204,23 +183,29 @@ class InstructionGenerateApi(Resource):
                     case "llm":
                         return LLMGenerator.generate_rule_config(
                             current_tenant_id,
-                            instruction=args.instruction,
-                            model_config=args.model_config_data,
-                            no_variable=True,
+                            args=RuleGeneratePayload(
+                                instruction=args.instruction,
+                                model_config=args.model_config_data,
+                                no_variable=True,
+                            ),
                         )
                     case "agent":
                         return LLMGenerator.generate_rule_config(
                             current_tenant_id,
-                            instruction=args.instruction,
-                            model_config=args.model_config_data,
-                            no_variable=True,
+                            args=RuleGeneratePayload(
+                                instruction=args.instruction,
+                                model_config=args.model_config_data,
+                                no_variable=True,
+                            ),
                         )
                     case "code":
                         return LLMGenerator.generate_code(
                             tenant_id=current_tenant_id,
-                            instruction=args.instruction,
-                            model_config=args.model_config_data,
-                            code_language=args.language,
+                            args=RuleCodeGeneratePayload(
+                                instruction=args.instruction,
+                                model_config=args.model_config_data,
+                                code_language=args.language,
+                            ),
                         )
                     case _:
                         return {"error": f"invalid node type: {node_type}"}

api/controllers/console/app/message.py

@@ -7,6 +7,7 @@ from pydantic import BaseModel, Field, field_validator
 from sqlalchemy import exists, select
 from werkzeug.exceptions import InternalServerError, NotFound
 
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.error import (
     CompletionRequestError,
@@ -35,7 +36,6 @@ from services.errors.message import MessageNotExistsError, SuggestedQuestionsAft
 from services.message_service import MessageService
 
 logger = logging.getLogger(__name__)
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
 
 class ChatMessagesQuery(BaseModel):
@@ -90,13 +90,22 @@ class FeedbackExportQuery(BaseModel):
         raise ValueError("has_comment must be a boolean value")
 
 
-def reg(cls: type[BaseModel]):
-    console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
+class AnnotationCountResponse(BaseModel):
+    count: int = Field(description="Number of annotations")
 
 
-reg(ChatMessagesQuery)
-reg(MessageFeedbackPayload)
-reg(FeedbackExportQuery)
+class SuggestedQuestionsResponse(BaseModel):
+    data: list[str] = Field(description="Suggested question")
+
+
+register_schema_models(
+    console_ns,
+    ChatMessagesQuery,
+    MessageFeedbackPayload,
+    FeedbackExportQuery,
+    AnnotationCountResponse,
+    SuggestedQuestionsResponse,
+)
 
 # Register models for flask_restx to avoid dict type issues in Swagger
 # Register in dependency order: base models first, then dependent models
@@ -231,7 +240,7 @@ class ChatMessageListApi(Resource):
     @marshal_with(message_infinite_scroll_pagination_model)
     @edit_permission_required
     def get(self, app_model):
-        args = ChatMessagesQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = ChatMessagesQuery.model_validate(request.args.to_dict())
 
         conversation = (
             db.session.query(Conversation)
@@ -356,7 +365,7 @@ class MessageAnnotationCountApi(Resource):
     @console_ns.response(
         200,
         "Annotation count retrieved successfully",
-        console_ns.model("AnnotationCountResponse", {"count": fields.Integer(description="Number of annotations")}),
+        console_ns.models[AnnotationCountResponse.__name__],
     )
     @get_app_model
     @setup_required
@@ -376,9 +385,7 @@ class MessageSuggestedQuestionApi(Resource):
     @console_ns.response(
         200,
         "Suggested questions retrieved successfully",
-        console_ns.model(
-            "SuggestedQuestionsResponse", {"data": fields.List(fields.String(description="Suggested question"))}
-        ),
+        console_ns.models[SuggestedQuestionsResponse.__name__],
     )
     @console_ns.response(404, "Message or conversation not found")
     @setup_required
@@ -428,7 +435,7 @@ class MessageFeedbackExportApi(Resource):
     @login_required
     @account_initialization_required
     def get(self, app_model):
-        args = FeedbackExportQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = FeedbackExportQuery.model_validate(request.args.to_dict())
 
         # Import the service function
         from services.feedback_service import FeedbackService

api/controllers/console/auth/data_source_oauth.py

@@ -2,9 +2,11 @@ import logging
 
 import httpx
 from flask import current_app, redirect, request
-from flask_restx import Resource, fields
+from flask_restx import Resource
+from pydantic import BaseModel, Field
 
 from configs import dify_config
+from controllers.common.schema import register_schema_models
 from libs.login import login_required
 from libs.oauth_data_source import NotionOAuth
 
@@ -14,6 +16,26 @@ from ..wraps import account_initialization_required, is_admin_or_owner_required,
 logger = logging.getLogger(__name__)
 
 
+class OAuthDataSourceResponse(BaseModel):
+    data: str = Field(description="Authorization URL or 'internal' for internal setup")
+
+
+class OAuthDataSourceBindingResponse(BaseModel):
+    result: str = Field(description="Operation result")
+
+
+class OAuthDataSourceSyncResponse(BaseModel):
+    result: str = Field(description="Operation result")
+
+
+register_schema_models(
+    console_ns,
+    OAuthDataSourceResponse,
+    OAuthDataSourceBindingResponse,
+    OAuthDataSourceSyncResponse,
+)
+
+
 def get_oauth_providers():
     with current_app.app_context():
         notion_oauth = NotionOAuth(
@@ -34,10 +56,7 @@ class OAuthDataSource(Resource):
     @console_ns.response(
         200,
         "Authorization URL or internal setup success",
-        console_ns.model(
-            "OAuthDataSourceResponse",
-            {"data": fields.Raw(description="Authorization URL or 'internal' for internal setup")},
-        ),
+        console_ns.models[OAuthDataSourceResponse.__name__],
     )
     @console_ns.response(400, "Invalid provider")
     @console_ns.response(403, "Admin privileges required")
@@ -101,7 +120,7 @@ class OAuthDataSourceBinding(Resource):
     @console_ns.response(
         200,
         "Data source binding success",
-        console_ns.model("OAuthDataSourceBindingResponse", {"result": fields.String(description="Operation result")}),
+        console_ns.models[OAuthDataSourceBindingResponse.__name__],
     )
     @console_ns.response(400, "Invalid provider or code")
     def get(self, provider: str):
@@ -133,7 +152,7 @@ class OAuthDataSourceSync(Resource):
     @console_ns.response(
         200,
         "Data source sync success",
-        console_ns.model("OAuthDataSourceSyncResponse", {"result": fields.String(description="Operation result")}),
+        console_ns.models[OAuthDataSourceSyncResponse.__name__],
     )
     @console_ns.response(400, "Invalid provider or sync failed")
     @setup_required

api/controllers/console/auth/forgot_password.py

@@ -2,10 +2,11 @@ import base64
 import secrets
 
 from flask import request
-from flask_restx import Resource, fields
+from flask_restx import Resource
 from pydantic import BaseModel, Field, field_validator
 from sqlalchemy.orm import Session
 
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.auth.error import (
     EmailCodeError,
@@ -48,8 +49,31 @@ class ForgotPasswordResetPayload(BaseModel):
         return valid_password(value)
 
 
-for model in (ForgotPasswordSendPayload, ForgotPasswordCheckPayload, ForgotPasswordResetPayload):
-    console_ns.schema_model(model.__name__, model.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
+class ForgotPasswordEmailResponse(BaseModel):
+    result: str = Field(description="Operation result")
+    data: str | None = Field(default=None, description="Reset token")
+    code: str | None = Field(default=None, description="Error code if account not found")
+
+
+class ForgotPasswordCheckResponse(BaseModel):
+    is_valid: bool = Field(description="Whether code is valid")
+    email: EmailStr = Field(description="Email address")
+    token: str = Field(description="New reset token")
+
+
+class ForgotPasswordResetResponse(BaseModel):
+    result: str = Field(description="Operation result")
+
+
+register_schema_models(
+    console_ns,
+    ForgotPasswordSendPayload,
+    ForgotPasswordCheckPayload,
+    ForgotPasswordResetPayload,
+    ForgotPasswordEmailResponse,
+    ForgotPasswordCheckResponse,
+    ForgotPasswordResetResponse,
+)
 
 
 @console_ns.route("/forgot-password")
@@ -60,14 +84,7 @@ class ForgotPasswordSendEmailApi(Resource):
     @console_ns.response(
         200,
         "Email sent successfully",
-        console_ns.model(
-            "ForgotPasswordEmailResponse",
-            {
-                "result": fields.String(description="Operation result"),
-                "data": fields.String(description="Reset token"),
-                "code": fields.String(description="Error code if account not found"),
-            },
-        ),
+        console_ns.models[ForgotPasswordEmailResponse.__name__],
     )
     @console_ns.response(400, "Invalid email or rate limit exceeded")
     @setup_required
@@ -106,14 +123,7 @@ class ForgotPasswordCheckApi(Resource):
     @console_ns.response(
         200,
         "Code verified successfully",
-        console_ns.model(
-            "ForgotPasswordCheckResponse",
-            {
-                "is_valid": fields.Boolean(description="Whether code is valid"),
-                "email": fields.String(description="Email address"),
-                "token": fields.String(description="New reset token"),
-            },
-        ),
+        console_ns.models[ForgotPasswordCheckResponse.__name__],
     )
     @console_ns.response(400, "Invalid code or token")
     @setup_required
@@ -163,7 +173,7 @@ class ForgotPasswordResetApi(Resource):
     @console_ns.response(
         200,
         "Password reset successfully",
-        console_ns.model("ForgotPasswordResetResponse", {"result": fields.String(description="Operation result")}),
+        console_ns.models[ForgotPasswordResetResponse.__name__],
     )
     @console_ns.response(400, "Invalid token or password mismatch")
     @setup_required

api/controllers/console/auth/oauth_server.py

@@ -155,43 +155,43 @@ class OAuthServerUserTokenApi(Resource):
             grant_type = OAuthGrantType(payload.grant_type)
         except ValueError:
             raise BadRequest("invalid grant_type")
+        match grant_type:
+            case OAuthGrantType.AUTHORIZATION_CODE:
+                if not payload.code:
+                    raise BadRequest("code is required")
 
-        if grant_type == OAuthGrantType.AUTHORIZATION_CODE:
-            if not payload.code:
-                raise BadRequest("code is required")
+                if payload.client_secret != oauth_provider_app.client_secret:
+                    raise BadRequest("client_secret is invalid")
 
-            if payload.client_secret != oauth_provider_app.client_secret:
-                raise BadRequest("client_secret is invalid")
+                if payload.redirect_uri not in oauth_provider_app.redirect_uris:
+                    raise BadRequest("redirect_uri is invalid")
 
-            if payload.redirect_uri not in oauth_provider_app.redirect_uris:
-                raise BadRequest("redirect_uri is invalid")
+                access_token, refresh_token = OAuthServerService.sign_oauth_access_token(
+                    grant_type, code=payload.code, client_id=oauth_provider_app.client_id
+                )
+                return jsonable_encoder(
+                    {
+                        "access_token": access_token,
+                        "token_type": "Bearer",
+                        "expires_in": OAUTH_ACCESS_TOKEN_EXPIRES_IN,
+                        "refresh_token": refresh_token,
+                    }
+                )
+            case OAuthGrantType.REFRESH_TOKEN:
+                if not payload.refresh_token:
+                    raise BadRequest("refresh_token is required")
 
-            access_token, refresh_token = OAuthServerService.sign_oauth_access_token(
-                grant_type, code=payload.code, client_id=oauth_provider_app.client_id
-            )
-            return jsonable_encoder(
-                {
-                    "access_token": access_token,
-                    "token_type": "Bearer",
-                    "expires_in": OAUTH_ACCESS_TOKEN_EXPIRES_IN,
-                    "refresh_token": refresh_token,
-                }
-            )
-        elif grant_type == OAuthGrantType.REFRESH_TOKEN:
-            if not payload.refresh_token:
-                raise BadRequest("refresh_token is required")
-
-            access_token, refresh_token = OAuthServerService.sign_oauth_access_token(
-                grant_type, refresh_token=payload.refresh_token, client_id=oauth_provider_app.client_id
-            )
-            return jsonable_encoder(
-                {
-                    "access_token": access_token,
-                    "token_type": "Bearer",
-                    "expires_in": OAUTH_ACCESS_TOKEN_EXPIRES_IN,
-                    "refresh_token": refresh_token,
-                }
-            )
+                access_token, refresh_token = OAuthServerService.sign_oauth_access_token(
+                    grant_type, refresh_token=payload.refresh_token, client_id=oauth_provider_app.client_id
+                )
+                return jsonable_encoder(
+                    {
+                        "access_token": access_token,
+                        "token_type": "Bearer",
+                        "expires_in": OAUTH_ACCESS_TOKEN_EXPIRES_IN,
+                        "refresh_token": refresh_token,
+                    }
+                )
 
 
 @console_ns.route("/oauth/provider/account")

api/controllers/console/datasets/data_source.py

@@ -1,6 +1,6 @@
 import json
 from collections.abc import Generator
-from typing import Any, cast
+from typing import Any, Literal, cast
 
 from flask import request
 from flask_restx import Resource, fields, marshal_with
@@ -157,9 +157,8 @@ class DataSourceApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    def patch(self, binding_id, action):
+    def patch(self, binding_id, action: Literal["enable", "disable"]):
         binding_id = str(binding_id)
-        action = str(action)
         with Session(db.engine) as session:
             data_source_binding = session.execute(
                 select(DataSourceOauthBinding).filter_by(id=binding_id)
@@ -167,23 +166,24 @@ class DataSourceApi(Resource):
         if data_source_binding is None:
             raise NotFound("Data source binding not found.")
         # enable binding
-        if action == "enable":
-            if data_source_binding.disabled:
-                data_source_binding.disabled = False
-                data_source_binding.updated_at = naive_utc_now()
-                db.session.add(data_source_binding)
-                db.session.commit()
-            else:
-                raise ValueError("Data source is not disabled.")
-        # disable binding
-        if action == "disable":
-            if not data_source_binding.disabled:
-                data_source_binding.disabled = True
-                data_source_binding.updated_at = naive_utc_now()
-                db.session.add(data_source_binding)
-                db.session.commit()
-            else:
-                raise ValueError("Data source is disabled.")
+        match action:
+            case "enable":
+                if data_source_binding.disabled:
+                    data_source_binding.disabled = False
+                    data_source_binding.updated_at = naive_utc_now()
+                    db.session.add(data_source_binding)
+                    db.session.commit()
+                else:
+                    raise ValueError("Data source is not disabled.")
+            # disable binding
+            case "disable":
+                if not data_source_binding.disabled:
+                    data_source_binding.disabled = True
+                    data_source_binding.updated_at = naive_utc_now()
+                    db.session.add(data_source_binding)
+                    db.session.commit()
+                else:
+                    raise ValueError("Data source is disabled.")
         return {"result": "success"}, 200
 
 

api/controllers/console/datasets/datasets.py

@@ -55,6 +55,7 @@ from libs.login import current_account_with_tenant, login_required
 from models import ApiToken, Dataset, Document, DocumentSegment, UploadFile
 from models.dataset import DatasetPermissionEnum
 from models.provider_ids import ModelProviderID
+from services.api_token_service import ApiTokenCache
 from services.dataset_service import DatasetPermissionService, DatasetService, DocumentService
 
 # Register models for flask_restx to avoid dict type issues in Swagger
@@ -820,6 +821,11 @@ class DatasetApiDeleteApi(Resource):
         if key is None:
             console_ns.abort(404, message="API key not found")
 
+        # Invalidate cache before deleting from database
+        # Type assertion: key is guaranteed to be non-None here because abort() raises
+        assert key is not None  # nosec - for type checker only
+        ApiTokenCache.delete(key.token, key.type)
+
         db.session.query(ApiToken).where(ApiToken.id == api_key_id).delete()
         db.session.commit()
 

api/controllers/console/datasets/datasets_document.py

@@ -576,63 +576,62 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
             if document.indexing_status in {"completed", "error"}:
                 raise DocumentAlreadyFinishedError()
             data_source_info = document.data_source_info_dict
+            match document.data_source_type:
+                case "upload_file":
+                    if not data_source_info:
+                        continue
+                    file_id = data_source_info["upload_file_id"]
+                    file_detail = (
+                        db.session.query(UploadFile)
+                        .where(UploadFile.tenant_id == current_tenant_id, UploadFile.id == file_id)
+                        .first()
+                    )
 
-            if document.data_source_type == "upload_file":
-                if not data_source_info:
-                    continue
-                file_id = data_source_info["upload_file_id"]
-                file_detail = (
-                    db.session.query(UploadFile)
-                    .where(UploadFile.tenant_id == current_tenant_id, UploadFile.id == file_id)
-                    .first()
-                )
+                    if file_detail is None:
+                        raise NotFound("File not found.")
 
-                if file_detail is None:
-                    raise NotFound("File not found.")
+                    extract_setting = ExtractSetting(
+                        datasource_type=DatasourceType.FILE, upload_file=file_detail, document_model=document.doc_form
+                    )
+                    extract_settings.append(extract_setting)
+                case "notion_import":
+                    if not data_source_info:
+                        continue
+                    extract_setting = ExtractSetting(
+                        datasource_type=DatasourceType.NOTION,
+                        notion_info=NotionInfo.model_validate(
+                            {
+                                "credential_id": data_source_info.get("credential_id"),
+                                "notion_workspace_id": data_source_info["notion_workspace_id"],
+                                "notion_obj_id": data_source_info["notion_page_id"],
+                                "notion_page_type": data_source_info["type"],
+                                "tenant_id": current_tenant_id,
+                            }
+                        ),
+                        document_model=document.doc_form,
+                    )
+                    extract_settings.append(extract_setting)
+                case "website_crawl":
+                    if not data_source_info:
+                        continue
+                    extract_setting = ExtractSetting(
+                        datasource_type=DatasourceType.WEBSITE,
+                        website_info=WebsiteInfo.model_validate(
+                            {
+                                "provider": data_source_info["provider"],
+                                "job_id": data_source_info["job_id"],
+                                "url": data_source_info["url"],
+                                "tenant_id": current_tenant_id,
+                                "mode": data_source_info["mode"],
+                                "only_main_content": data_source_info["only_main_content"],
+                            }
+                        ),
+                        document_model=document.doc_form,
+                    )
+                    extract_settings.append(extract_setting)
 
-                extract_setting = ExtractSetting(
-                    datasource_type=DatasourceType.FILE, upload_file=file_detail, document_model=document.doc_form
-                )
-                extract_settings.append(extract_setting)
-
-            elif document.data_source_type == "notion_import":
-                if not data_source_info:
-                    continue
-                extract_setting = ExtractSetting(
-                    datasource_type=DatasourceType.NOTION,
-                    notion_info=NotionInfo.model_validate(
-                        {
-                            "credential_id": data_source_info.get("credential_id"),
-                            "notion_workspace_id": data_source_info["notion_workspace_id"],
-                            "notion_obj_id": data_source_info["notion_page_id"],
-                            "notion_page_type": data_source_info["type"],
-                            "tenant_id": current_tenant_id,
-                        }
-                    ),
-                    document_model=document.doc_form,
-                )
-                extract_settings.append(extract_setting)
-            elif document.data_source_type == "website_crawl":
-                if not data_source_info:
-                    continue
-                extract_setting = ExtractSetting(
-                    datasource_type=DatasourceType.WEBSITE,
-                    website_info=WebsiteInfo.model_validate(
-                        {
-                            "provider": data_source_info["provider"],
-                            "job_id": data_source_info["job_id"],
-                            "url": data_source_info["url"],
-                            "tenant_id": current_tenant_id,
-                            "mode": data_source_info["mode"],
-                            "only_main_content": data_source_info["only_main_content"],
-                        }
-                    ),
-                    document_model=document.doc_form,
-                )
-                extract_settings.append(extract_setting)
-
-            else:
-                raise ValueError("Data source type not support")
+                case _:
+                    raise ValueError("Data source type not support")
             indexing_runner = IndexingRunner()
             try:
                 response = indexing_runner.indexing_estimate(
@@ -954,23 +953,24 @@ class DocumentProcessingApi(DocumentResource):
         if not current_user.is_dataset_editor:
             raise Forbidden()
 
-        if action == "pause":
-            if document.indexing_status != "indexing":
-                raise InvalidActionError("Document not in indexing state.")
+        match action:
+            case "pause":
+                if document.indexing_status != "indexing":
+                    raise InvalidActionError("Document not in indexing state.")
 
-            document.paused_by = current_user.id
-            document.paused_at = naive_utc_now()
-            document.is_paused = True
-            db.session.commit()
+                document.paused_by = current_user.id
+                document.paused_at = naive_utc_now()
+                document.is_paused = True
+                db.session.commit()
 
-        elif action == "resume":
-            if document.indexing_status not in {"paused", "error"}:
-                raise InvalidActionError("Document not in paused or error state.")
+            case "resume":
+                if document.indexing_status not in {"paused", "error"}:
+                    raise InvalidActionError("Document not in paused or error state.")
 
-            document.paused_by = None
-            document.paused_at = None
-            document.is_paused = False
-            db.session.commit()
+                document.paused_by = None
+                document.paused_at = None
+                document.is_paused = False
+                db.session.commit()
 
         return {"result": "success"}, 200
 
@@ -1339,6 +1339,18 @@ class DocumentGenerateSummaryApi(Resource):
             missing_ids = set(document_list) - found_ids
             raise NotFound(f"Some documents not found: {list(missing_ids)}")
 
+        # Update need_summary to True for documents that don't have it set
+        # This handles the case where documents were created when summary_index_setting was disabled
+        documents_to_update = [doc for doc in documents if not doc.need_summary and doc.doc_form != "qa_model"]
+
+        if documents_to_update:
+            document_ids_to_update = [str(doc.id) for doc in documents_to_update]
+            DocumentService.update_documents_need_summary(
+                dataset_id=dataset_id,
+                document_ids=document_ids_to_update,
+                need_summary=True,
+            )
+
         # Dispatch async tasks for each document
         for document in documents:
             # Skip qa_model documents as they don't generate summaries

api/controllers/console/datasets/metadata.py

@@ -126,10 +126,11 @@ class DatasetMetadataBuiltInFieldActionApi(Resource):
             raise NotFound("Dataset not found.")
         DatasetService.check_dataset_permission(dataset, current_user)
 
-        if action == "enable":
-            MetadataService.enable_built_in_field(dataset)
-        elif action == "disable":
-            MetadataService.disable_built_in_field(dataset)
+        match action:
+            case "enable":
+                MetadataService.enable_built_in_field(dataset)
+            case "disable":
+                MetadataService.disable_built_in_field(dataset)
         return {"result": "success"}, 200
 
 

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

@@ -1,10 +1,9 @@
 import json
 import logging
 from typing import Any, Literal, cast
-from uuid import UUID
 
 from flask import abort, request
-from flask_restx import Resource, marshal_with, reqparse  # type: ignore
+from flask_restx import Resource, marshal_with  # type: ignore
 from pydantic import BaseModel, Field
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
@@ -38,7 +37,7 @@ from core.model_runtime.utils.encoders import jsonable_encoder
 from extensions.ext_database import db
 from factories import variable_factory
 from libs import helper
-from libs.helper import TimestampField
+from libs.helper import TimestampField, UUIDStrOrEmpty
 from libs.login import current_account_with_tenant, current_user, login_required
 from models import Account
 from models.dataset import Pipeline
@@ -110,7 +109,7 @@ class NodeIdQuery(BaseModel):
 
 
 class WorkflowRunQuery(BaseModel):
-    last_id: UUID | None = None
+    last_id: UUIDStrOrEmpty | None = None
     limit: int = Field(default=20, ge=1, le=100)
 
 
@@ -121,6 +120,10 @@ class DatasourceVariablesPayload(BaseModel):
     start_node_title: str
 
 
+class RagPipelineRecommendedPluginQuery(BaseModel):
+    type: str = "all"
+
+
 register_schema_models(
     console_ns,
     DraftWorkflowSyncPayload,
@@ -135,6 +138,7 @@ register_schema_models(
     NodeIdQuery,
     WorkflowRunQuery,
     DatasourceVariablesPayload,
+    RagPipelineRecommendedPluginQuery,
 )
 
 
@@ -975,11 +979,8 @@ class RagPipelineRecommendedPluginApi(Resource):
     @login_required
     @account_initialization_required
     def get(self):
-        parser = reqparse.RequestParser()
-        parser.add_argument("type", type=str, location="args", required=False, default="all")
-        args = parser.parse_args()
-        type = args["type"]
+        query = RagPipelineRecommendedPluginQuery.model_validate(request.args.to_dict())
 
         rag_pipeline_service = RagPipelineService()
-        recommended_plugins = rag_pipeline_service.get_recommended_plugins(type)
+        recommended_plugins = rag_pipeline_service.get_recommended_plugins(query.type)
         return recommended_plugins

api/controllers/console/evaluation/__init__.py

@@ -1 +0,0 @@
-# Evaluation controller module

api/controllers/console/evaluation/evaluation.py

@@ -1,288 +0,0 @@
-import logging
-from collections.abc import Callable
-from functools import wraps
-from typing import ParamSpec, TypeVar, Union
-
-from flask import request
-from flask_restx import Resource, fields
-from pydantic import BaseModel, Field
-from werkzeug.exceptions import NotFound
-
-from controllers.common.schema import register_schema_models
-from controllers.console import console_ns
-from controllers.console.wraps import (
-    account_initialization_required,
-    edit_permission_required,
-    setup_required,
-)
-from extensions.ext_database import db
-from libs.helper import TimestampField
-from libs.login import current_account_with_tenant, login_required
-from models import App
-from models.snippet import CustomizedSnippet
-
-logger = logging.getLogger(__name__)
-
-P = ParamSpec("P")
-R = TypeVar("R")
-
-# Valid evaluation target types
-EVALUATE_TARGET_TYPES = {"app", "snippets"}
-
-
-class VersionQuery(BaseModel):
-    """Query parameters for version endpoint."""
-
-    version: str
-
-
-register_schema_models(
-    console_ns,
-    VersionQuery,
-)
-
-
-# Response field definitions
-file_info_fields = {
-    "id": fields.String,
-    "name": fields.String,
-}
-
-evaluation_log_fields = {
-    "created_at": TimestampField,
-    "created_by": fields.String,
-    "test_file": fields.Nested(
-        console_ns.model(
-            "EvaluationTestFile",
-            file_info_fields,
-        )
-    ),
-    "result_file": fields.Nested(
-        console_ns.model(
-            "EvaluationResultFile",
-            file_info_fields,
-        ),
-        allow_null=True,
-    ),
-    "version": fields.String,
-}
-
-evaluation_log_list_model = console_ns.model(
-    "EvaluationLogList",
-    {
-        "data": fields.List(fields.Nested(console_ns.model("EvaluationLog", evaluation_log_fields))),
-    },
-)
-
-customized_matrix_fields = {
-    "evaluation_workflow_id": fields.String,
-    "input_fields": fields.Raw,
-    "output_fields": fields.Raw,
-}
-
-condition_fields = {
-    "name": fields.List(fields.String),
-    "comparison_operator": fields.String,
-    "value": fields.String,
-}
-
-judgement_conditions_fields = {
-    "logical_operator": fields.String,
-    "conditions": fields.List(fields.Nested(console_ns.model("EvaluationCondition", condition_fields))),
-}
-
-evaluation_detail_fields = {
-    "evaluation_model": fields.String,
-    "evaluation_model_provider": fields.String,
-    "customized_matrix": fields.Nested(
-        console_ns.model("EvaluationCustomizedMatrix", customized_matrix_fields),
-        allow_null=True,
-    ),
-    "judgement_conditions": fields.Nested(
-        console_ns.model("EvaluationJudgementConditions", judgement_conditions_fields),
-        allow_null=True,
-    ),
-}
-
-evaluation_detail_model = console_ns.model("EvaluationDetail", evaluation_detail_fields)
-
-
-def get_evaluation_target(view_func: Callable[P, R]):
-    """
-    Decorator to resolve polymorphic evaluation target (app or snippet).
-
-    Validates the target_type parameter and fetches the corresponding
-    model (App or CustomizedSnippet) with tenant isolation.
-    """
-
-    @wraps(view_func)
-    def decorated_view(*args: P.args, **kwargs: P.kwargs):
-        target_type = kwargs.get("evaluate_target_type")
-        target_id = kwargs.get("evaluate_target_id")
-
-        if target_type not in EVALUATE_TARGET_TYPES:
-            raise NotFound(f"Invalid evaluation target type: {target_type}")
-
-        _, current_tenant_id = current_account_with_tenant()
-
-        target_id = str(target_id)
-
-        # Remove path parameters
-        del kwargs["evaluate_target_type"]
-        del kwargs["evaluate_target_id"]
-
-        target: Union[App, CustomizedSnippet] | None = None
-
-        if target_type == "app":
-            target = (
-                db.session.query(App).where(App.id == target_id, App.tenant_id == current_tenant_id).first()
-            )
-        elif target_type == "snippets":
-            target = (
-                db.session.query(CustomizedSnippet)
-                .where(CustomizedSnippet.id == target_id, CustomizedSnippet.tenant_id == current_tenant_id)
-                .first()
-            )
-
-        if not target:
-            raise NotFound(f"{str(target_type)} not found")
-
-        kwargs["target"] = target
-        kwargs["target_type"] = target_type
-
-        return view_func(*args, **kwargs)
-
-    return decorated_view
-
-
-@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/dataset-template/download")
-class EvaluationDatasetTemplateDownloadApi(Resource):
-    @console_ns.doc("download_evaluation_dataset_template")
-    @console_ns.response(200, "Template download URL generated successfully")
-    @console_ns.response(404, "Target not found")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_evaluation_target
-    @edit_permission_required
-    def post(self, target: Union[App, CustomizedSnippet], target_type: str):
-        """
-        Download evaluation dataset template.
-
-        Generates a download URL for the evaluation dataset template
-        based on the target type (app or snippets).
-        """
-        # TODO: Implement actual template generation logic
-        # This is a placeholder implementation
-        return {
-            "download_url": f"/api/evaluation/{target_type}/{target.id}/template.csv",
-        }
-
-
-@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation")
-class EvaluationDetailApi(Resource):
-    @console_ns.doc("get_evaluation_detail")
-    @console_ns.response(200, "Evaluation details retrieved successfully", evaluation_detail_model)
-    @console_ns.response(404, "Target not found")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_evaluation_target
-    def get(self, target: Union[App, CustomizedSnippet], target_type: str):
-        """
-        Get evaluation details for the target.
-
-        Returns evaluation configuration including model settings,
-        customized matrix, and judgement conditions.
-        """
-        # TODO: Implement actual evaluation detail retrieval
-        # This is a placeholder implementation
-        return {
-            "evaluation_model": None,
-            "evaluation_model_provider": None,
-            "customized_matrix": None,
-            "judgement_conditions": None,
-        }
-
-
-@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation/logs")
-class EvaluationLogsApi(Resource):
-    @console_ns.doc("get_evaluation_logs")
-    @console_ns.response(200, "Evaluation logs retrieved successfully", evaluation_log_list_model)
-    @console_ns.response(404, "Target not found")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_evaluation_target
-    def get(self, target: Union[App, CustomizedSnippet], target_type: str):
-        """
-        Get offline evaluation logs for the target.
-
-        Returns a list of evaluation runs with test files,
-        result files, and version information.
-        """
-        # TODO: Implement actual evaluation logs retrieval
-        # This is a placeholder implementation
-        return {
-            "data": [],
-        }
-
-
-@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation/files/<uuid:file_id>")
-class EvaluationFileDownloadApi(Resource):
-    @console_ns.doc("download_evaluation_file")
-    @console_ns.response(200, "File download URL generated successfully")
-    @console_ns.response(404, "Target or file not found")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_evaluation_target
-    def get(self, target: Union[App, CustomizedSnippet], target_type: str, file_id: str):
-        """
-        Download evaluation test file or result file.
-
-        Returns file information and download URL for the specified file.
-        """
-        file_id = str(file_id)
-
-        # TODO: Implement actual file download logic
-        # This is a placeholder implementation
-        return {
-            "created_at": None,
-            "created_by": None,
-            "test_file": None,
-            "result_file": None,
-            "version": None,
-        }
-
-
-@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation/version")
-class EvaluationVersionApi(Resource):
-    @console_ns.doc("get_evaluation_version_detail")
-    @console_ns.expect(console_ns.models.get(VersionQuery.__name__))
-    @console_ns.response(200, "Version details retrieved successfully")
-    @console_ns.response(404, "Target or version not found")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_evaluation_target
-    def get(self, target: Union[App, CustomizedSnippet], target_type: str):
-        """
-        Get evaluation target version details.
-
-        Returns the workflow graph for the specified version.
-        """
-        version = request.args.get("version")
-
-        if not version:
-            return {"message": "version parameter is required"}, 400
-
-        # TODO: Implement actual version detail retrieval
-        # For now, return the current graph if available
-        graph = {}
-        if target_type == "snippets" and isinstance(target, CustomizedSnippet):
-            graph = target.graph_dict
-
-        return {
-            "graph": graph,
-        }

api/controllers/console/explore/trial.py

@@ -1,8 +1,9 @@
 import logging
-from typing import Any, cast
+from typing import Any, Literal, cast
 
 from flask import request
-from flask_restx import Resource, fields, marshal, marshal_with, reqparse
+from flask_restx import Resource, fields, marshal, marshal_with
+from pydantic import BaseModel
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
 
 import services
@@ -51,7 +52,7 @@ from fields.app_fields import (
     tag_fields,
 )
 from fields.dataset_fields import dataset_fields
-from fields.member_fields import build_simple_account_model
+from fields.member_fields import simple_account_fields
 from fields.workflow_fields import (
     conversation_variable_fields,
     pipeline_variable_fields,
@@ -103,7 +104,7 @@ app_detail_fields_with_site_copy["tags"] = fields.List(fields.Nested(tag_model))
 app_detail_fields_with_site_copy["site"] = fields.Nested(site_model)
 app_detail_with_site_model = get_or_create_model("TrialAppDetailWithSite", app_detail_fields_with_site_copy)
 
-simple_account_model = build_simple_account_model(console_ns)
+simple_account_model = get_or_create_model("SimpleAccount", simple_account_fields)
 conversation_variable_model = get_or_create_model("TrialConversationVariable", conversation_variable_fields)
 pipeline_variable_model = get_or_create_model("TrialPipelineVariable", pipeline_variable_fields)
 
@@ -117,7 +118,56 @@ workflow_fields_copy["rag_pipeline_variables"] = fields.List(fields.Nested(pipel
 workflow_model = get_or_create_model("TrialWorkflow", workflow_fields_copy)
 
 
+# Pydantic models for request validation
+DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
+
+
+class WorkflowRunRequest(BaseModel):
+    inputs: dict
+    files: list | None = None
+
+
+class ChatRequest(BaseModel):
+    inputs: dict
+    query: str
+    files: list | None = None
+    conversation_id: str | None = None
+    parent_message_id: str | None = None
+    retriever_from: str = "explore_app"
+
+
+class TextToSpeechRequest(BaseModel):
+    message_id: str | None = None
+    voice: str | None = None
+    text: str | None = None
+    streaming: bool | None = None
+
+
+class CompletionRequest(BaseModel):
+    inputs: dict
+    query: str = ""
+    files: list | None = None
+    response_mode: Literal["blocking", "streaming"] | None = None
+    retriever_from: str = "explore_app"
+
+
+# Register schemas for Swagger documentation
+console_ns.schema_model(
+    WorkflowRunRequest.__name__, WorkflowRunRequest.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
+console_ns.schema_model(
+    ChatRequest.__name__, ChatRequest.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
+console_ns.schema_model(
+    TextToSpeechRequest.__name__, TextToSpeechRequest.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
+console_ns.schema_model(
+    CompletionRequest.__name__, CompletionRequest.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
+
+
 class TrialAppWorkflowRunApi(TrialAppResource):
+    @console_ns.expect(console_ns.models[WorkflowRunRequest.__name__])
     def post(self, trial_app):
         """
         Run workflow
@@ -129,10 +179,8 @@ class TrialAppWorkflowRunApi(TrialAppResource):
         if app_mode != AppMode.WORKFLOW:
             raise NotWorkflowAppError()
 
-        parser = reqparse.RequestParser()
-        parser.add_argument("inputs", type=dict, required=True, nullable=False, location="json")
-        parser.add_argument("files", type=list, required=False, location="json")
-        args = parser.parse_args()
+        request_data = WorkflowRunRequest.model_validate(console_ns.payload)
+        args = request_data.model_dump()
         assert current_user is not None
         try:
             app_id = app_model.id
@@ -183,6 +231,7 @@ class TrialAppWorkflowTaskStopApi(TrialAppResource):
 
 
 class TrialChatApi(TrialAppResource):
+    @console_ns.expect(console_ns.models[ChatRequest.__name__])
     @trial_feature_enable
     def post(self, trial_app):
         app_model = trial_app
@@ -190,14 +239,14 @@ class TrialChatApi(TrialAppResource):
         if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
-        parser = reqparse.RequestParser()
-        parser.add_argument("inputs", type=dict, required=True, location="json")
-        parser.add_argument("query", type=str, required=True, location="json")
-        parser.add_argument("files", type=list, required=False, location="json")
-        parser.add_argument("conversation_id", type=uuid_value, location="json")
-        parser.add_argument("parent_message_id", type=uuid_value, required=False, location="json")
-        parser.add_argument("retriever_from", type=str, required=False, default="explore_app", location="json")
-        args = parser.parse_args()
+        request_data = ChatRequest.model_validate(console_ns.payload)
+        args = request_data.model_dump()
+
+        # Validate UUID values if provided
+        if args.get("conversation_id"):
+            args["conversation_id"] = uuid_value(args["conversation_id"])
+        if args.get("parent_message_id"):
+            args["parent_message_id"] = uuid_value(args["parent_message_id"])
 
         args["auto_generate_name"] = False
 
@@ -320,20 +369,16 @@ class TrialChatAudioApi(TrialAppResource):
 
 
 class TrialChatTextApi(TrialAppResource):
+    @console_ns.expect(console_ns.models[TextToSpeechRequest.__name__])
     @trial_feature_enable
     def post(self, trial_app):
         app_model = trial_app
         try:
-            parser = reqparse.RequestParser()
-            parser.add_argument("message_id", type=str, required=False, location="json")
-            parser.add_argument("voice", type=str, location="json")
-            parser.add_argument("text", type=str, location="json")
-            parser.add_argument("streaming", type=bool, location="json")
-            args = parser.parse_args()
+            request_data = TextToSpeechRequest.model_validate(console_ns.payload)
 
-            message_id = args.get("message_id", None)
-            text = args.get("text", None)
-            voice = args.get("voice", None)
+            message_id = request_data.message_id
+            text = request_data.text
+            voice = request_data.voice
             if not isinstance(current_user, Account):
                 raise ValueError("current_user must be an Account instance")
 
@@ -371,19 +416,15 @@ class TrialChatTextApi(TrialAppResource):
 
 
 class TrialCompletionApi(TrialAppResource):
+    @console_ns.expect(console_ns.models[CompletionRequest.__name__])
     @trial_feature_enable
     def post(self, trial_app):
         app_model = trial_app
         if app_model.mode != "completion":
             raise NotCompletionAppError()
 
-        parser = reqparse.RequestParser()
-        parser.add_argument("inputs", type=dict, required=True, location="json")
-        parser.add_argument("query", type=str, location="json", default="")
-        parser.add_argument("files", type=list, required=False, location="json")
-        parser.add_argument("response_mode", type=str, choices=["blocking", "streaming"], location="json")
-        parser.add_argument("retriever_from", type=str, required=False, default="explore_app", location="json")
-        args = parser.parse_args()
+        request_data = CompletionRequest.model_validate(console_ns.payload)
+        args = request_data.model_dump()
 
         streaming = args["response_mode"] == "streaming"
         args["auto_generate_name"] = False

api/controllers/console/init_validate.py

@@ -1,87 +1,74 @@
 import os
+from typing import Literal
 
 from flask import session
-from flask_restx import Resource, fields
 from pydantic import BaseModel, Field
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from configs import dify_config
+from controllers.fastopenapi import console_router
 from extensions.ext_database import db
 from models.model import DifySetup
 from services.account_service import TenantService
 
-from . import console_ns
 from .error import AlreadySetupError, InitValidateFailedError
 from .wraps import only_edition_self_hosted
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
 
 class InitValidatePayload(BaseModel):
-    password: str = Field(..., max_length=30)
+    password: str = Field(..., max_length=30, description="Initialization password")
 
 
-console_ns.schema_model(
-    InitValidatePayload.__name__,
-    InitValidatePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+class InitStatusResponse(BaseModel):
+    status: Literal["finished", "not_started"] = Field(..., description="Initialization status")
+
+
+class InitValidateResponse(BaseModel):
+    result: str = Field(description="Operation result", examples=["success"])
+
+
+@console_router.get(
+    "/init",
+    response_model=InitStatusResponse,
+    tags=["console"],
 )
+def get_init_status() -> InitStatusResponse:
+    """Get initialization validation status."""
+    init_status = get_init_validate_status()
+    if init_status:
+        return InitStatusResponse(status="finished")
+    return InitStatusResponse(status="not_started")
 
 
-@console_ns.route("/init")
-class InitValidateAPI(Resource):
-    @console_ns.doc("get_init_status")
-    @console_ns.doc(description="Get initialization validation status")
-    @console_ns.response(
-        200,
-        "Success",
-        model=console_ns.model(
-            "InitStatusResponse",
-            {"status": fields.String(description="Initialization status", enum=["finished", "not_started"])},
-        ),
-    )
-    def get(self):
-        """Get initialization validation status"""
-        init_status = get_init_validate_status()
-        if init_status:
-            return {"status": "finished"}
-        return {"status": "not_started"}
+@console_router.post(
+    "/init",
+    response_model=InitValidateResponse,
+    tags=["console"],
+    status_code=201,
+)
+@only_edition_self_hosted
+def validate_init_password(payload: InitValidatePayload) -> InitValidateResponse:
+    """Validate initialization password."""
+    tenant_count = TenantService.get_tenant_count()
+    if tenant_count > 0:
+        raise AlreadySetupError()
 
-    @console_ns.doc("validate_init_password")
-    @console_ns.doc(description="Validate initialization password for self-hosted edition")
-    @console_ns.expect(console_ns.models[InitValidatePayload.__name__])
-    @console_ns.response(
-        201,
-        "Success",
-        model=console_ns.model("InitValidateResponse", {"result": fields.String(description="Operation result")}),
-    )
-    @console_ns.response(400, "Already setup or validation failed")
-    @only_edition_self_hosted
-    def post(self):
-        """Validate initialization password"""
-        # is tenant created
-        tenant_count = TenantService.get_tenant_count()
-        if tenant_count > 0:
-            raise AlreadySetupError()
+    if payload.password != os.environ.get("INIT_PASSWORD"):
+        session["is_init_validated"] = False
+        raise InitValidateFailedError()
 
-        payload = InitValidatePayload.model_validate(console_ns.payload)
-        input_password = payload.password
-
-        if input_password != os.environ.get("INIT_PASSWORD"):
-            session["is_init_validated"] = False
-            raise InitValidateFailedError()
-
-        session["is_init_validated"] = True
-        return {"result": "success"}, 201
+    session["is_init_validated"] = True
+    return InitValidateResponse(result="success")
 
 
-def get_init_validate_status():
+def get_init_validate_status() -> bool:
     if dify_config.EDITION == "SELF_HOSTED":
         if os.environ.get("INIT_PASSWORD"):
             if session.get("is_init_validated"):
                 return True
 
             with Session(db.engine) as db_session:
-                return db_session.execute(select(DifySetup)).scalar_one_or_none()
+                return db_session.execute(select(DifySetup)).scalar_one_or_none() is not None
 
     return True

api/controllers/console/remote_files.py

@@ -1,7 +1,6 @@
 import urllib.parse
 
 import httpx
-from flask_restx import Resource
 from pydantic import BaseModel, Field
 
 import services
@@ -11,7 +10,7 @@ from controllers.common.errors import (
     RemoteFileUploadError,
     UnsupportedFileTypeError,
 )
-from controllers.common.schema import register_schema_models
+from controllers.fastopenapi import console_router
 from core.file import helpers as file_helpers
 from core.helper import ssrf_proxy
 from extensions.ext_database import db
@@ -19,84 +18,74 @@ from fields.file_fields import FileWithSignedUrl, RemoteFileInfo
 from libs.login import current_account_with_tenant
 from services.file_service import FileService
 
-from . import console_ns
-
-register_schema_models(console_ns, RemoteFileInfo, FileWithSignedUrl)
-
-
-@console_ns.route("/remote-files/<path:url>")
-class RemoteFileInfoApi(Resource):
-    @console_ns.response(200, "Remote file info", console_ns.models[RemoteFileInfo.__name__])
-    def get(self, url):
-        decoded_url = urllib.parse.unquote(url)
-        resp = ssrf_proxy.head(decoded_url)
-        if resp.status_code != httpx.codes.OK:
-            # failed back to get method
-            resp = ssrf_proxy.get(decoded_url, timeout=3)
-        resp.raise_for_status()
-        info = RemoteFileInfo(
-            file_type=resp.headers.get("Content-Type", "application/octet-stream"),
-            file_length=int(resp.headers.get("Content-Length", 0)),
-        )
-        return info.model_dump(mode="json")
-
 
 class RemoteFileUploadPayload(BaseModel):
     url: str = Field(..., description="URL to fetch")
 
 
-console_ns.schema_model(
-    RemoteFileUploadPayload.__name__,
-    RemoteFileUploadPayload.model_json_schema(ref_template="#/definitions/{model}"),
+@console_router.get(
+    "/remote-files/<path:url>",
+    response_model=RemoteFileInfo,
+    tags=["console"],
 )
+def get_remote_file_info(url: str) -> RemoteFileInfo:
+    decoded_url = urllib.parse.unquote(url)
+    resp = ssrf_proxy.head(decoded_url)
+    if resp.status_code != httpx.codes.OK:
+        resp = ssrf_proxy.get(decoded_url, timeout=3)
+    resp.raise_for_status()
+    return RemoteFileInfo(
+        file_type=resp.headers.get("Content-Type", "application/octet-stream"),
+        file_length=int(resp.headers.get("Content-Length", 0)),
+    )
 
 
-@console_ns.route("/remote-files/upload")
-class RemoteFileUploadApi(Resource):
-    @console_ns.expect(console_ns.models[RemoteFileUploadPayload.__name__])
-    @console_ns.response(201, "Remote file uploaded", console_ns.models[FileWithSignedUrl.__name__])
-    def post(self):
-        args = RemoteFileUploadPayload.model_validate(console_ns.payload)
-        url = args.url
+@console_router.post(
+    "/remote-files/upload",
+    response_model=FileWithSignedUrl,
+    tags=["console"],
+    status_code=201,
+)
+def upload_remote_file(payload: RemoteFileUploadPayload) -> FileWithSignedUrl:
+    url = payload.url
 
-        try:
-            resp = ssrf_proxy.head(url=url)
-            if resp.status_code != httpx.codes.OK:
-                resp = ssrf_proxy.get(url=url, timeout=3, follow_redirects=True)
-            if resp.status_code != httpx.codes.OK:
-                raise RemoteFileUploadError(f"Failed to fetch file from {url}: {resp.text}")
-        except httpx.RequestError as e:
-            raise RemoteFileUploadError(f"Failed to fetch file from {url}: {str(e)}")
+    try:
+        resp = ssrf_proxy.head(url=url)
+        if resp.status_code != httpx.codes.OK:
+            resp = ssrf_proxy.get(url=url, timeout=3, follow_redirects=True)
+        if resp.status_code != httpx.codes.OK:
+            raise RemoteFileUploadError(f"Failed to fetch file from {url}: {resp.text}")
+    except httpx.RequestError as e:
+        raise RemoteFileUploadError(f"Failed to fetch file from {url}: {str(e)}")
 
-        file_info = helpers.guess_file_info_from_response(resp)
+    file_info = helpers.guess_file_info_from_response(resp)
 
-        if not FileService.is_file_size_within_limit(extension=file_info.extension, file_size=file_info.size):
-            raise FileTooLargeError
+    if not FileService.is_file_size_within_limit(extension=file_info.extension, file_size=file_info.size):
+        raise FileTooLargeError
 
-        content = resp.content if resp.request.method == "GET" else ssrf_proxy.get(url).content
+    content = resp.content if resp.request.method == "GET" else ssrf_proxy.get(url).content
 
-        try:
-            user, _ = current_account_with_tenant()
-            upload_file = FileService(db.engine).upload_file(
-                filename=file_info.filename,
-                content=content,
-                mimetype=file_info.mimetype,
-                user=user,
-                source_url=url,
-            )
-        except services.errors.file.FileTooLargeError as file_too_large_error:
-            raise FileTooLargeError(file_too_large_error.description)
-        except services.errors.file.UnsupportedFileTypeError:
-            raise UnsupportedFileTypeError()
-
-        payload = FileWithSignedUrl(
-            id=upload_file.id,
-            name=upload_file.name,
-            size=upload_file.size,
-            extension=upload_file.extension,
-            url=file_helpers.get_signed_file_url(upload_file_id=upload_file.id),
-            mime_type=upload_file.mime_type,
-            created_by=upload_file.created_by,
-            created_at=int(upload_file.created_at.timestamp()),
+    try:
+        user, _ = current_account_with_tenant()
+        upload_file = FileService(db.engine).upload_file(
+            filename=file_info.filename,
+            content=content,
+            mimetype=file_info.mimetype,
+            user=user,
+            source_url=url,
         )
-        return payload.model_dump(mode="json"), 201
+    except services.errors.file.FileTooLargeError as file_too_large_error:
+        raise FileTooLargeError(file_too_large_error.description)
+    except services.errors.file.UnsupportedFileTypeError:
+        raise UnsupportedFileTypeError()
+
+    return FileWithSignedUrl(
+        id=upload_file.id,
+        name=upload_file.name,
+        size=upload_file.size,
+        extension=upload_file.extension,
+        url=file_helpers.get_signed_file_url(upload_file_id=upload_file.id),
+        mime_type=upload_file.mime_type,
+        created_by=upload_file.created_by,
+        created_at=int(upload_file.created_at.timestamp()),
+    )

api/controllers/console/snippets/payloads.py

@@ -1,75 +0,0 @@
-from typing import Any, Literal
-
-from pydantic import BaseModel, Field
-
-
-class SnippetListQuery(BaseModel):
-    """Query parameters for listing snippets."""
-
-    page: int = Field(default=1, ge=1, le=99999)
-    limit: int = Field(default=20, ge=1, le=100)
-    keyword: str | None = None
-
-
-class IconInfo(BaseModel):
-    """Icon information model."""
-
-    icon: str | None = None
-    icon_type: Literal["emoji", "image"] | None = None
-    icon_background: str | None = None
-    icon_url: str | None = None
-
-
-class InputFieldDefinition(BaseModel):
-    """Input field definition for snippet parameters."""
-
-    default: str | None = None
-    hint: bool | None = None
-    label: str | None = None
-    max_length: int | None = None
-    options: list[str] | None = None
-    placeholder: str | None = None
-    required: bool | None = None
-    type: str | None = None  # e.g., "text-input"
-
-
-class CreateSnippetPayload(BaseModel):
-    """Payload for creating a new snippet."""
-
-    name: str = Field(..., min_length=1, max_length=255)
-    description: str | None = Field(default=None, max_length=2000)
-    type: Literal["node", "group"] = "node"
-    icon_info: IconInfo | None = None
-    graph: dict[str, Any] | None = None
-    input_fields: list[InputFieldDefinition] | None = Field(default_factory=list)
-
-
-class UpdateSnippetPayload(BaseModel):
-    """Payload for updating a snippet."""
-
-    name: str | None = Field(default=None, min_length=1, max_length=255)
-    description: str | None = Field(default=None, max_length=2000)
-    icon_info: IconInfo | None = None
-
-
-class SnippetDraftSyncPayload(BaseModel):
-    """Payload for syncing snippet draft workflow."""
-
-    graph: dict[str, Any]
-    hash: str | None = None
-    environment_variables: list[dict[str, Any]] | None = None
-    conversation_variables: list[dict[str, Any]] | None = None
-    input_variables: list[dict[str, Any]] | None = None
-
-
-class WorkflowRunQuery(BaseModel):
-    """Query parameters for workflow runs."""
-
-    last_id: str | None = None
-    limit: int = Field(default=20, ge=1, le=100)
-
-
-class PublishWorkflowPayload(BaseModel):
-    """Payload for publishing snippet workflow."""
-
-    knowledge_base_setting: dict[str, Any] | None = None

api/controllers/console/snippets/snippet_workflow.py

@@ -1,306 +0,0 @@
-import logging
-from collections.abc import Callable
-from functools import wraps
-from typing import ParamSpec, TypeVar
-
-from flask import request
-from flask_restx import Resource, marshal_with
-from sqlalchemy.orm import Session
-from werkzeug.exceptions import NotFound
-
-from controllers.common.schema import register_schema_models
-from controllers.console import console_ns
-from controllers.console.app.error import DraftWorkflowNotExist, DraftWorkflowNotSync
-from controllers.console.app.workflow import workflow_model
-from controllers.console.app.workflow_run import (
-    workflow_run_detail_model,
-    workflow_run_node_execution_list_model,
-    workflow_run_pagination_model,
-)
-from controllers.console.snippets.payloads import (
-    PublishWorkflowPayload,
-    SnippetDraftSyncPayload,
-    WorkflowRunQuery,
-)
-from controllers.console.wraps import (
-    account_initialization_required,
-    edit_permission_required,
-    setup_required,
-)
-from extensions.ext_database import db
-from factories import variable_factory
-from libs.helper import TimestampField
-from libs.login import current_account_with_tenant, login_required
-from models.snippet import CustomizedSnippet
-from services.errors.app import WorkflowHashNotEqualError
-from services.snippet_service import SnippetService
-
-logger = logging.getLogger(__name__)
-
-P = ParamSpec("P")
-R = TypeVar("R")
-
-# Register Pydantic models with Swagger
-register_schema_models(
-    console_ns,
-    SnippetDraftSyncPayload,
-    WorkflowRunQuery,
-    PublishWorkflowPayload,
-)
-
-
-class SnippetNotFoundError(Exception):
-    """Snippet not found error."""
-
-    pass
-
-
-def get_snippet(view_func: Callable[P, R]):
-    """Decorator to fetch and validate snippet access."""
-
-    @wraps(view_func)
-    def decorated_view(*args: P.args, **kwargs: P.kwargs):
-        if not kwargs.get("snippet_id"):
-            raise ValueError("missing snippet_id in path parameters")
-
-        _, current_tenant_id = current_account_with_tenant()
-
-        snippet_id = str(kwargs.get("snippet_id"))
-        del kwargs["snippet_id"]
-
-        snippet = SnippetService.get_snippet_by_id(
-            snippet_id=snippet_id,
-            tenant_id=current_tenant_id,
-        )
-
-        if not snippet:
-            raise NotFound("Snippet not found")
-
-        kwargs["snippet"] = snippet
-
-        return view_func(*args, **kwargs)
-
-    return decorated_view
-
-
-@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft")
-class SnippetDraftWorkflowApi(Resource):
-    @console_ns.doc("get_snippet_draft_workflow")
-    @console_ns.response(200, "Draft workflow retrieved successfully", workflow_model)
-    @console_ns.response(404, "Snippet or draft workflow not found")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_snippet
-    @edit_permission_required
-    @marshal_with(workflow_model)
-    def get(self, snippet: CustomizedSnippet):
-        """Get draft workflow for snippet."""
-        snippet_service = SnippetService()
-        workflow = snippet_service.get_draft_workflow(snippet=snippet)
-
-        if not workflow:
-            raise DraftWorkflowNotExist()
-
-        return workflow
-
-    @console_ns.doc("sync_snippet_draft_workflow")
-    @console_ns.expect(console_ns.models.get(SnippetDraftSyncPayload.__name__))
-    @console_ns.response(200, "Draft workflow synced successfully")
-    @console_ns.response(400, "Hash mismatch")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_snippet
-    @edit_permission_required
-    def post(self, snippet: CustomizedSnippet):
-        """Sync draft workflow for snippet."""
-        current_user, _ = current_account_with_tenant()
-
-        payload = SnippetDraftSyncPayload.model_validate(console_ns.payload or {})
-
-        try:
-            environment_variables_list = payload.environment_variables or []
-            environment_variables = [
-                variable_factory.build_environment_variable_from_mapping(obj) for obj in environment_variables_list
-            ]
-            conversation_variables_list = payload.conversation_variables or []
-            conversation_variables = [
-                variable_factory.build_conversation_variable_from_mapping(obj) for obj in conversation_variables_list
-            ]
-            snippet_service = SnippetService()
-            workflow = snippet_service.sync_draft_workflow(
-                snippet=snippet,
-                graph=payload.graph,
-                unique_hash=payload.hash,
-                account=current_user,
-                environment_variables=environment_variables,
-                conversation_variables=conversation_variables,
-                input_variables=payload.input_variables,
-            )
-        except WorkflowHashNotEqualError:
-            raise DraftWorkflowNotSync()
-
-        return {
-            "result": "success",
-            "hash": workflow.unique_hash,
-            "updated_at": TimestampField().format(workflow.updated_at or workflow.created_at),
-        }
-
-
-@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/config")
-class SnippetDraftConfigApi(Resource):
-    @console_ns.doc("get_snippet_draft_config")
-    @console_ns.response(200, "Draft config retrieved successfully")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_snippet
-    @edit_permission_required
-    def get(self, snippet: CustomizedSnippet):
-        """Get snippet draft workflow configuration limits."""
-        return {
-            "parallel_depth_limit": 3,
-        }
-
-
-@console_ns.route("/snippets/<uuid:snippet_id>/workflows/publish")
-class SnippetPublishedWorkflowApi(Resource):
-    @console_ns.doc("get_snippet_published_workflow")
-    @console_ns.response(200, "Published workflow retrieved successfully", workflow_model)
-    @console_ns.response(404, "Snippet not found")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_snippet
-    @edit_permission_required
-    @marshal_with(workflow_model)
-    def get(self, snippet: CustomizedSnippet):
-        """Get published workflow for snippet."""
-        if not snippet.is_published:
-            return None
-
-        snippet_service = SnippetService()
-        workflow = snippet_service.get_published_workflow(snippet=snippet)
-
-        return workflow
-
-    @console_ns.doc("publish_snippet_workflow")
-    @console_ns.expect(console_ns.models.get(PublishWorkflowPayload.__name__))
-    @console_ns.response(200, "Workflow published successfully")
-    @console_ns.response(400, "No draft workflow found")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_snippet
-    @edit_permission_required
-    def post(self, snippet: CustomizedSnippet):
-        """Publish snippet workflow."""
-        current_user, _ = current_account_with_tenant()
-        snippet_service = SnippetService()
-
-        with Session(db.engine) as session:
-            snippet = session.merge(snippet)
-            try:
-                workflow = snippet_service.publish_workflow(
-                    session=session,
-                    snippet=snippet,
-                    account=current_user,
-                )
-                workflow_created_at = TimestampField().format(workflow.created_at)
-                session.commit()
-            except ValueError as e:
-                return {"message": str(e)}, 400
-
-        return {
-            "result": "success",
-            "created_at": workflow_created_at,
-        }
-
-
-@console_ns.route("/snippets/<uuid:snippet_id>/workflows/default-workflow-block-configs")
-class SnippetDefaultBlockConfigsApi(Resource):
-    @console_ns.doc("get_snippet_default_block_configs")
-    @console_ns.response(200, "Default block configs retrieved successfully")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_snippet
-    @edit_permission_required
-    def get(self, snippet: CustomizedSnippet):
-        """Get default block configurations for snippet workflow."""
-        snippet_service = SnippetService()
-        return snippet_service.get_default_block_configs()
-
-
-@console_ns.route("/snippets/<uuid:snippet_id>/workflow-runs")
-class SnippetWorkflowRunsApi(Resource):
-    @console_ns.doc("list_snippet_workflow_runs")
-    @console_ns.response(200, "Workflow runs retrieved successfully", workflow_run_pagination_model)
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_snippet
-    @marshal_with(workflow_run_pagination_model)
-    def get(self, snippet: CustomizedSnippet):
-        """List workflow runs for snippet."""
-        query = WorkflowRunQuery.model_validate(
-            {
-                "last_id": request.args.get("last_id"),
-                "limit": request.args.get("limit", type=int, default=20),
-            }
-        )
-        args = {
-            "last_id": query.last_id,
-            "limit": query.limit,
-        }
-
-        snippet_service = SnippetService()
-        result = snippet_service.get_snippet_workflow_runs(snippet=snippet, args=args)
-
-        return result
-
-
-@console_ns.route("/snippets/<uuid:snippet_id>/workflow-runs/<uuid:run_id>")
-class SnippetWorkflowRunDetailApi(Resource):
-    @console_ns.doc("get_snippet_workflow_run_detail")
-    @console_ns.response(200, "Workflow run detail retrieved successfully", workflow_run_detail_model)
-    @console_ns.response(404, "Workflow run not found")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_snippet
-    @marshal_with(workflow_run_detail_model)
-    def get(self, snippet: CustomizedSnippet, run_id):
-        """Get workflow run detail for snippet."""
-        run_id = str(run_id)
-
-        snippet_service = SnippetService()
-        workflow_run = snippet_service.get_snippet_workflow_run(snippet=snippet, run_id=run_id)
-
-        if not workflow_run:
-            raise NotFound("Workflow run not found")
-
-        return workflow_run
-
-
-@console_ns.route("/snippets/<uuid:snippet_id>/workflow-runs/<uuid:run_id>/node-executions")
-class SnippetWorkflowRunNodeExecutionsApi(Resource):
-    @console_ns.doc("list_snippet_workflow_run_node_executions")
-    @console_ns.response(200, "Node executions retrieved successfully", workflow_run_node_execution_list_model)
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_snippet
-    @marshal_with(workflow_run_node_execution_list_model)
-    def get(self, snippet: CustomizedSnippet, run_id):
-        """List node executions for a workflow run."""
-        run_id = str(run_id)
-
-        snippet_service = SnippetService()
-        node_executions = snippet_service.get_snippet_workflow_run_node_executions(
-            snippet=snippet,
-            run_id=run_id,
-        )
-
-        return {"data": node_executions}

api/controllers/console/tag/tags.py

@@ -1,17 +1,27 @@
 from typing import Literal
 
 from flask import request
-from flask_restx import Resource, marshal_with
+from flask_restx import Namespace, Resource, fields, marshal_with
 from pydantic import BaseModel, Field
 from werkzeug.exceptions import Forbidden
 
 from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
-from fields.tag_fields import dataset_tag_fields
 from libs.login import current_account_with_tenant, login_required
 from services.tag_service import TagService
 
+dataset_tag_fields = {
+    "id": fields.String,
+    "name": fields.String,
+    "type": fields.String,
+    "binding_count": fields.String,
+}
+
+
+def build_dataset_tag_fields(api_or_ns: Namespace):
+    return api_or_ns.model("DataSetTag", dataset_tag_fields)
+
 
 class TagBasePayload(BaseModel):
     name: str = Field(description="Tag name", min_length=1, max_length=50)
@@ -110,7 +120,7 @@ class TagUpdateDeleteApi(Resource):
 
         TagService.delete_tag(tag_id)
 
-        return 204
+        return "", 204
 
 
 @console_ns.route("/tag-bindings/create")

api/controllers/console/workspace/account.py

@@ -12,6 +12,7 @@ from sqlalchemy.orm import Session
 
 from configs import dify_config
 from constants.languages import supported_language
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.auth.error import (
     EmailAlreadyInUseError,
@@ -37,7 +38,7 @@ from controllers.console.wraps import (
     setup_required,
 )
 from extensions.ext_database import db
-from fields.member_fields import account_fields
+from fields.member_fields import Account as AccountResponse
 from libs.datetime_utils import naive_utc_now
 from libs.helper import EmailStr, TimestampField, extract_remote_ip, timezone
 from libs.login import current_account_with_tenant, login_required
@@ -170,6 +171,12 @@ reg(ChangeEmailSendPayload)
 reg(ChangeEmailValidityPayload)
 reg(ChangeEmailResetPayload)
 reg(CheckEmailUniquePayload)
+register_schema_models(console_ns, AccountResponse)
+
+
+def _serialize_account(account) -> dict:
+    return AccountResponse.model_validate(account, from_attributes=True).model_dump(mode="json")
+
 
 integrate_fields = {
     "provider": fields.String,
@@ -236,11 +243,11 @@ class AccountProfileApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @marshal_with(account_fields)
+    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
     @enterprise_license_required
     def get(self):
         current_user, _ = current_account_with_tenant()
-        return current_user
+        return _serialize_account(current_user)
 
 
 @console_ns.route("/account/name")
@@ -249,14 +256,14 @@ class AccountNameApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @marshal_with(account_fields)
+    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
     def post(self):
         current_user, _ = current_account_with_tenant()
         payload = console_ns.payload or {}
         args = AccountNamePayload.model_validate(payload)
         updated_account = AccountService.update_account(current_user, name=args.name)
 
-        return updated_account
+        return _serialize_account(updated_account)
 
 
 @console_ns.route("/account/avatar")
@@ -265,7 +272,7 @@ class AccountAvatarApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @marshal_with(account_fields)
+    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
     def post(self):
         current_user, _ = current_account_with_tenant()
         payload = console_ns.payload or {}
@@ -273,7 +280,7 @@ class AccountAvatarApi(Resource):
 
         updated_account = AccountService.update_account(current_user, avatar=args.avatar)
 
-        return updated_account
+        return _serialize_account(updated_account)
 
 
 @console_ns.route("/account/interface-language")
@@ -282,7 +289,7 @@ class AccountInterfaceLanguageApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @marshal_with(account_fields)
+    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
     def post(self):
         current_user, _ = current_account_with_tenant()
         payload = console_ns.payload or {}
@@ -290,7 +297,7 @@ class AccountInterfaceLanguageApi(Resource):
 
         updated_account = AccountService.update_account(current_user, interface_language=args.interface_language)
 
-        return updated_account
+        return _serialize_account(updated_account)
 
 
 @console_ns.route("/account/interface-theme")
@@ -299,7 +306,7 @@ class AccountInterfaceThemeApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @marshal_with(account_fields)
+    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
     def post(self):
         current_user, _ = current_account_with_tenant()
         payload = console_ns.payload or {}
@@ -307,7 +314,7 @@ class AccountInterfaceThemeApi(Resource):
 
         updated_account = AccountService.update_account(current_user, interface_theme=args.interface_theme)
 
-        return updated_account
+        return _serialize_account(updated_account)
 
 
 @console_ns.route("/account/timezone")
@@ -316,7 +323,7 @@ class AccountTimezoneApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @marshal_with(account_fields)
+    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
     def post(self):
         current_user, _ = current_account_with_tenant()
         payload = console_ns.payload or {}
@@ -324,7 +331,7 @@ class AccountTimezoneApi(Resource):
 
         updated_account = AccountService.update_account(current_user, timezone=args.timezone)
 
-        return updated_account
+        return _serialize_account(updated_account)
 
 
 @console_ns.route("/account/password")
@@ -333,7 +340,7 @@ class AccountPasswordApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @marshal_with(account_fields)
+    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
     def post(self):
         current_user, _ = current_account_with_tenant()
         payload = console_ns.payload or {}
@@ -344,7 +351,7 @@ class AccountPasswordApi(Resource):
         except ServiceCurrentPasswordIncorrectError:
             raise CurrentPasswordIncorrectError()
 
-        return {"result": "success"}
+        return _serialize_account(current_user)
 
 
 @console_ns.route("/account/integrates")
@@ -620,7 +627,7 @@ class ChangeEmailResetApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @marshal_with(account_fields)
+    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
     def post(self):
         payload = console_ns.payload or {}
         args = ChangeEmailResetPayload.model_validate(payload)
@@ -649,7 +656,7 @@ class ChangeEmailResetApi(Resource):
             email=normalized_new_email,
         )
 
-        return updated_account
+        return _serialize_account(updated_account)
 
 
 @console_ns.route("/account/change-email/check-email-unique")

api/controllers/console/workspace/endpoint.py

@@ -1,9 +1,10 @@
 from typing import Any
 
 from flask import request
-from flask_restx import Resource, fields
+from flask_restx import Resource
 from pydantic import BaseModel, Field
 
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, is_admin_or_owner_required, setup_required
 from core.model_runtime.utils.encoders import jsonable_encoder
@@ -38,15 +39,53 @@ class EndpointListForPluginQuery(EndpointListQuery):
     plugin_id: str
 
 
+class EndpointCreateResponse(BaseModel):
+    success: bool = Field(description="Operation success")
+
+
+class EndpointListResponse(BaseModel):
+    endpoints: list[dict[str, Any]] = Field(description="Endpoint information")
+
+
+class PluginEndpointListResponse(BaseModel):
+    endpoints: list[dict[str, Any]] = Field(description="Endpoint information")
+
+
+class EndpointDeleteResponse(BaseModel):
+    success: bool = Field(description="Operation success")
+
+
+class EndpointUpdateResponse(BaseModel):
+    success: bool = Field(description="Operation success")
+
+
+class EndpointEnableResponse(BaseModel):
+    success: bool = Field(description="Operation success")
+
+
+class EndpointDisableResponse(BaseModel):
+    success: bool = Field(description="Operation success")
+
+
 def reg(cls: type[BaseModel]):
     console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
 
 
-reg(EndpointCreatePayload)
-reg(EndpointIdPayload)
-reg(EndpointUpdatePayload)
-reg(EndpointListQuery)
-reg(EndpointListForPluginQuery)
+register_schema_models(
+    console_ns,
+    EndpointCreatePayload,
+    EndpointIdPayload,
+    EndpointUpdatePayload,
+    EndpointListQuery,
+    EndpointListForPluginQuery,
+    EndpointCreateResponse,
+    EndpointListResponse,
+    PluginEndpointListResponse,
+    EndpointDeleteResponse,
+    EndpointUpdateResponse,
+    EndpointEnableResponse,
+    EndpointDisableResponse,
+)
 
 
 @console_ns.route("/workspaces/current/endpoints/create")
@@ -57,7 +96,7 @@ class EndpointCreateApi(Resource):
     @console_ns.response(
         200,
         "Endpoint created successfully",
-        console_ns.model("EndpointCreateResponse", {"success": fields.Boolean(description="Operation success")}),
+        console_ns.models[EndpointCreateResponse.__name__],
     )
     @console_ns.response(403, "Admin privileges required")
     @setup_required
@@ -91,9 +130,7 @@ class EndpointListApi(Resource):
     @console_ns.response(
         200,
         "Success",
-        console_ns.model(
-            "EndpointListResponse", {"endpoints": fields.List(fields.Raw(description="Endpoint information"))}
-        ),
+        console_ns.models[EndpointListResponse.__name__],
     )
     @setup_required
     @login_required
@@ -126,9 +163,7 @@ class EndpointListForSinglePluginApi(Resource):
     @console_ns.response(
         200,
         "Success",
-        console_ns.model(
-            "PluginEndpointListResponse", {"endpoints": fields.List(fields.Raw(description="Endpoint information"))}
-        ),
+        console_ns.models[PluginEndpointListResponse.__name__],
     )
     @setup_required
     @login_required
@@ -163,7 +198,7 @@ class EndpointDeleteApi(Resource):
     @console_ns.response(
         200,
         "Endpoint deleted successfully",
-        console_ns.model("EndpointDeleteResponse", {"success": fields.Boolean(description="Operation success")}),
+        console_ns.models[EndpointDeleteResponse.__name__],
     )
     @console_ns.response(403, "Admin privileges required")
     @setup_required
@@ -190,7 +225,7 @@ class EndpointUpdateApi(Resource):
     @console_ns.response(
         200,
         "Endpoint updated successfully",
-        console_ns.model("EndpointUpdateResponse", {"success": fields.Boolean(description="Operation success")}),
+        console_ns.models[EndpointUpdateResponse.__name__],
     )
     @console_ns.response(403, "Admin privileges required")
     @setup_required
@@ -221,7 +256,7 @@ class EndpointEnableApi(Resource):
     @console_ns.response(
         200,
         "Endpoint enabled successfully",
-        console_ns.model("EndpointEnableResponse", {"success": fields.Boolean(description="Operation success")}),
+        console_ns.models[EndpointEnableResponse.__name__],
     )
     @console_ns.response(403, "Admin privileges required")
     @setup_required
@@ -248,7 +283,7 @@ class EndpointDisableApi(Resource):
     @console_ns.response(
         200,
         "Endpoint disabled successfully",
-        console_ns.model("EndpointDisableResponse", {"success": fields.Boolean(description="Operation success")}),
+        console_ns.models[EndpointDisableResponse.__name__],
     )
     @console_ns.response(403, "Admin privileges required")
     @setup_required

api/controllers/console/workspace/members.py

@@ -1,12 +1,12 @@
 from urllib import parse
 
 from flask import abort, request
-from flask_restx import Resource, fields, marshal_with
-from pydantic import BaseModel, Field
+from flask_restx import Resource
+from pydantic import BaseModel, Field, TypeAdapter
 
 import services
 from configs import dify_config
-from controllers.common.schema import get_or_create_model, register_enum_models
+from controllers.common.schema import register_enum_models, register_schema_models
 from controllers.console import console_ns
 from controllers.console.auth.error import (
     CannotTransferOwnerToSelfError,
@@ -25,7 +25,7 @@ from controllers.console.wraps import (
     setup_required,
 )
 from extensions.ext_database import db
-from fields.member_fields import account_with_role_fields, account_with_role_list_fields
+from fields.member_fields import AccountWithRole, AccountWithRoleList
 from libs.helper import extract_remote_ip
 from libs.login import current_account_with_tenant, login_required
 from models.account import Account, TenantAccountRole
@@ -69,12 +69,7 @@ reg(OwnerTransferEmailPayload)
 reg(OwnerTransferCheckPayload)
 reg(OwnerTransferPayload)
 register_enum_models(console_ns, TenantAccountRole)
-
-account_with_role_model = get_or_create_model("AccountWithRole", account_with_role_fields)
-
-account_with_role_list_fields_copy = account_with_role_list_fields.copy()
-account_with_role_list_fields_copy["accounts"] = fields.List(fields.Nested(account_with_role_model))
-account_with_role_list_model = get_or_create_model("AccountWithRoleList", account_with_role_list_fields_copy)
+register_schema_models(console_ns, AccountWithRole, AccountWithRoleList)
 
 
 @console_ns.route("/workspaces/current/members")
@@ -84,13 +79,15 @@ class MemberListApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @marshal_with(account_with_role_list_model)
+    @console_ns.response(200, "Success", console_ns.models[AccountWithRoleList.__name__])
     def get(self):
         current_user, _ = current_account_with_tenant()
         if not current_user.current_tenant:
             raise ValueError("No current tenant")
         members = TenantService.get_tenant_members(current_user.current_tenant)
-        return {"result": "success", "accounts": members}, 200
+        member_models = TypeAdapter(list[AccountWithRole]).validate_python(members, from_attributes=True)
+        response = AccountWithRoleList(accounts=member_models)
+        return response.model_dump(mode="json"), 200
 
 
 @console_ns.route("/workspaces/current/members/invite-email")
@@ -235,13 +232,15 @@ class DatasetOperatorMemberListApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @marshal_with(account_with_role_list_model)
+    @console_ns.response(200, "Success", console_ns.models[AccountWithRoleList.__name__])
     def get(self):
         current_user, _ = current_account_with_tenant()
         if not current_user.current_tenant:
             raise ValueError("No current tenant")
         members = TenantService.get_dataset_operator_members(current_user.current_tenant)
-        return {"result": "success", "accounts": members}, 200
+        member_models = TypeAdapter(list[AccountWithRole]).validate_python(members, from_attributes=True)
+        response = AccountWithRoleList(accounts=member_models)
+        return response.model_dump(mode="json"), 200
 
 
 @console_ns.route("/workspaces/current/members/send-owner-transfer-confirm-email")

api/controllers/console/workspace/snippets.py

@@ -1,202 +0,0 @@
-import logging
-
-from flask import request
-from flask_restx import Resource, marshal, marshal_with
-from sqlalchemy.orm import Session
-from werkzeug.exceptions import NotFound
-
-from controllers.common.schema import register_schema_models
-from controllers.console import console_ns
-from controllers.console.snippets.payloads import (
-    CreateSnippetPayload,
-    SnippetListQuery,
-    UpdateSnippetPayload,
-)
-from controllers.console.wraps import (
-    account_initialization_required,
-    edit_permission_required,
-    setup_required,
-)
-from extensions.ext_database import db
-from fields.snippet_fields import snippet_fields, snippet_list_fields, snippet_pagination_fields
-from libs.login import current_account_with_tenant, login_required
-from models.snippet import SnippetType
-from services.snippet_service import SnippetService
-
-logger = logging.getLogger(__name__)
-
-# Register Pydantic models with Swagger
-register_schema_models(
-    console_ns,
-    SnippetListQuery,
-    CreateSnippetPayload,
-    UpdateSnippetPayload,
-)
-
-# Create namespace models for marshaling
-snippet_model = console_ns.model("Snippet", snippet_fields)
-snippet_list_model = console_ns.model("SnippetList", snippet_list_fields)
-snippet_pagination_model = console_ns.model("SnippetPagination", snippet_pagination_fields)
-
-
-@console_ns.route("/workspaces/current/customized-snippets")
-class CustomizedSnippetsApi(Resource):
-    @console_ns.doc("list_customized_snippets")
-    @console_ns.expect(console_ns.models.get(SnippetListQuery.__name__))
-    @console_ns.response(200, "Snippets retrieved successfully", snippet_pagination_model)
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def get(self):
-        """List customized snippets with pagination and search."""
-        _, current_tenant_id = current_account_with_tenant()
-
-        query_params = request.args.to_dict()
-        query = SnippetListQuery.model_validate(query_params)
-
-        snippets, total, has_more = SnippetService.get_snippets(
-            tenant_id=current_tenant_id,
-            page=query.page,
-            limit=query.limit,
-            keyword=query.keyword,
-        )
-
-        return {
-            "data": marshal(snippets, snippet_list_fields),
-            "page": query.page,
-            "limit": query.limit,
-            "total": total,
-            "has_more": has_more,
-        }, 200
-
-    @console_ns.doc("create_customized_snippet")
-    @console_ns.expect(console_ns.models.get(CreateSnippetPayload.__name__))
-    @console_ns.response(201, "Snippet created successfully", snippet_model)
-    @console_ns.response(400, "Invalid request or name already exists")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @edit_permission_required
-    def post(self):
-        """Create a new customized snippet."""
-        current_user, current_tenant_id = current_account_with_tenant()
-
-        payload = CreateSnippetPayload.model_validate(console_ns.payload or {})
-
-        try:
-            snippet_type = SnippetType(payload.type)
-        except ValueError:
-            snippet_type = SnippetType.NODE
-
-        try:
-            snippet = SnippetService.create_snippet(
-                tenant_id=current_tenant_id,
-                name=payload.name,
-                description=payload.description,
-                snippet_type=snippet_type,
-                icon_info=payload.icon_info.model_dump() if payload.icon_info else None,
-                graph=payload.graph,
-                input_fields=[f.model_dump() for f in payload.input_fields] if payload.input_fields else None,
-                account=current_user,
-            )
-        except ValueError as e:
-            return {"message": str(e)}, 400
-
-        return marshal(snippet, snippet_fields), 201
-
-
-@console_ns.route("/workspaces/current/customized-snippets/<uuid:snippet_id>")
-class CustomizedSnippetDetailApi(Resource):
-    @console_ns.doc("get_customized_snippet")
-    @console_ns.response(200, "Snippet retrieved successfully", snippet_model)
-    @console_ns.response(404, "Snippet not found")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def get(self, snippet_id: str):
-        """Get customized snippet details."""
-        _, current_tenant_id = current_account_with_tenant()
-
-        snippet = SnippetService.get_snippet_by_id(
-            snippet_id=str(snippet_id),
-            tenant_id=current_tenant_id,
-        )
-
-        if not snippet:
-            raise NotFound("Snippet not found")
-
-        return marshal(snippet, snippet_fields), 200
-
-    @console_ns.doc("update_customized_snippet")
-    @console_ns.expect(console_ns.models.get(UpdateSnippetPayload.__name__))
-    @console_ns.response(200, "Snippet updated successfully", snippet_model)
-    @console_ns.response(400, "Invalid request or name already exists")
-    @console_ns.response(404, "Snippet not found")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @edit_permission_required
-    def patch(self, snippet_id: str):
-        """Update customized snippet."""
-        current_user, current_tenant_id = current_account_with_tenant()
-
-        snippet = SnippetService.get_snippet_by_id(
-            snippet_id=str(snippet_id),
-            tenant_id=current_tenant_id,
-        )
-
-        if not snippet:
-            raise NotFound("Snippet not found")
-
-        payload = UpdateSnippetPayload.model_validate(console_ns.payload or {})
-        update_data = payload.model_dump(exclude_unset=True)
-
-        if "icon_info" in update_data and update_data["icon_info"] is not None:
-            update_data["icon_info"] = payload.icon_info.model_dump() if payload.icon_info else None
-
-        if not update_data:
-            return {"message": "No valid fields to update"}, 400
-
-        try:
-            with Session(db.engine, expire_on_commit=False) as session:
-                snippet = session.merge(snippet)
-                snippet = SnippetService.update_snippet(
-                    session=session,
-                    snippet=snippet,
-                    account_id=current_user.id,
-                    data=update_data,
-                )
-                session.commit()
-        except ValueError as e:
-            return {"message": str(e)}, 400
-
-        return marshal(snippet, snippet_fields), 200
-
-    @console_ns.doc("delete_customized_snippet")
-    @console_ns.response(204, "Snippet deleted successfully")
-    @console_ns.response(404, "Snippet not found")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @edit_permission_required
-    def delete(self, snippet_id: str):
-        """Delete customized snippet."""
-        _, current_tenant_id = current_account_with_tenant()
-
-        snippet = SnippetService.get_snippet_by_id(
-            snippet_id=str(snippet_id),
-            tenant_id=current_tenant_id,
-        )
-
-        if not snippet:
-            raise NotFound("Snippet not found")
-
-        with Session(db.engine) as session:
-            snippet = session.merge(snippet)
-            SnippetService.delete_snippet(
-                session=session,
-                snippet=snippet,
-            )
-            session.commit()
-
-        return "", 204

api/controllers/service_api/app/annotation.py

@@ -1,16 +1,16 @@
 from typing import Literal
 
 from flask import request
-from flask_restx import Namespace, Resource, fields
+from flask_restx import Resource
 from flask_restx.api import HTTPStatus
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, TypeAdapter
 
 from controllers.common.schema import register_schema_models
 from controllers.console.wraps import edit_permission_required
 from controllers.service_api import service_api_ns
 from controllers.service_api.wraps import validate_app_token
 from extensions.ext_redis import redis_client
-from fields.annotation_fields import annotation_fields, build_annotation_model
+from fields.annotation_fields import Annotation, AnnotationList
 from models.model import App
 from services.annotation_service import AppAnnotationService
 
@@ -26,7 +26,9 @@ class AnnotationReplyActionPayload(BaseModel):
     embedding_model_name: str = Field(description="Embedding model name")
 
 
-register_schema_models(service_api_ns, AnnotationCreatePayload, AnnotationReplyActionPayload)
+register_schema_models(
+    service_api_ns, AnnotationCreatePayload, AnnotationReplyActionPayload, Annotation, AnnotationList
+)
 
 
 @service_api_ns.route("/apps/annotation-reply/<string:action>")
@@ -45,10 +47,11 @@ class AnnotationReplyActionApi(Resource):
     def post(self, app_model: App, action: Literal["enable", "disable"]):
         """Enable or disable annotation reply feature."""
         args = AnnotationReplyActionPayload.model_validate(service_api_ns.payload or {}).model_dump()
-        if action == "enable":
-            result = AppAnnotationService.enable_app_annotation(args, app_model.id)
-        elif action == "disable":
-            result = AppAnnotationService.disable_app_annotation(app_model.id)
+        match action:
+            case "enable":
+                result = AppAnnotationService.enable_app_annotation(args, app_model.id)
+            case "disable":
+                result = AppAnnotationService.disable_app_annotation(app_model.id)
         return result, 200
 
 
@@ -82,23 +85,6 @@ class AnnotationReplyActionStatusApi(Resource):
         return {"job_id": job_id, "job_status": job_status, "error_msg": error_msg}, 200
 
 
-# Define annotation list response model
-annotation_list_fields = {
-    "data": fields.List(fields.Nested(annotation_fields)),
-    "has_more": fields.Boolean,
-    "limit": fields.Integer,
-    "total": fields.Integer,
-    "page": fields.Integer,
-}
-
-
-def build_annotation_list_model(api_or_ns: Namespace):
-    """Build the annotation list model for the API or Namespace."""
-    copied_annotation_list_fields = annotation_list_fields.copy()
-    copied_annotation_list_fields["data"] = fields.List(fields.Nested(build_annotation_model(api_or_ns)))
-    return api_or_ns.model("AnnotationList", copied_annotation_list_fields)
-
-
 @service_api_ns.route("/apps/annotations")
 class AnnotationListApi(Resource):
     @service_api_ns.doc("list_annotations")
@@ -109,8 +95,12 @@ class AnnotationListApi(Resource):
             401: "Unauthorized - invalid API token",
         }
     )
+    @service_api_ns.response(
+        200,
+        "Annotations retrieved successfully",
+        service_api_ns.models[AnnotationList.__name__],
+    )
     @validate_app_token
-    @service_api_ns.marshal_with(build_annotation_list_model(service_api_ns))
     def get(self, app_model: App):
         """List annotations for the application."""
         page = request.args.get("page", default=1, type=int)
@@ -118,13 +108,15 @@ class AnnotationListApi(Resource):
         keyword = request.args.get("keyword", default="", type=str)
 
         annotation_list, total = AppAnnotationService.get_annotation_list_by_app_id(app_model.id, page, limit, keyword)
-        return {
-            "data": annotation_list,
-            "has_more": len(annotation_list) == limit,
-            "limit": limit,
-            "total": total,
-            "page": page,
-        }
+        annotation_models = TypeAdapter(list[Annotation]).validate_python(annotation_list, from_attributes=True)
+        response = AnnotationList(
+            data=annotation_models,
+            has_more=len(annotation_list) == limit,
+            limit=limit,
+            total=total,
+            page=page,
+        )
+        return response.model_dump(mode="json")
 
     @service_api_ns.expect(service_api_ns.models[AnnotationCreatePayload.__name__])
     @service_api_ns.doc("create_annotation")
@@ -135,13 +127,18 @@ class AnnotationListApi(Resource):
             401: "Unauthorized - invalid API token",
         }
     )
+    @service_api_ns.response(
+        HTTPStatus.CREATED,
+        "Annotation created successfully",
+        service_api_ns.models[Annotation.__name__],
+    )
     @validate_app_token
-    @service_api_ns.marshal_with(build_annotation_model(service_api_ns), code=HTTPStatus.CREATED)
     def post(self, app_model: App):
         """Create a new annotation."""
         args = AnnotationCreatePayload.model_validate(service_api_ns.payload or {}).model_dump()
         annotation = AppAnnotationService.insert_app_annotation_directly(args, app_model.id)
-        return annotation, 201
+        response = Annotation.model_validate(annotation, from_attributes=True)
+        return response.model_dump(mode="json"), HTTPStatus.CREATED
 
 
 @service_api_ns.route("/apps/annotations/<uuid:annotation_id>")
@@ -158,14 +155,19 @@ class AnnotationUpdateDeleteApi(Resource):
             404: "Annotation not found",
         }
     )
+    @service_api_ns.response(
+        200,
+        "Annotation updated successfully",
+        service_api_ns.models[Annotation.__name__],
+    )
     @validate_app_token
     @edit_permission_required
-    @service_api_ns.marshal_with(build_annotation_model(service_api_ns))
     def put(self, app_model: App, annotation_id: str):
         """Update an existing annotation."""
         args = AnnotationCreatePayload.model_validate(service_api_ns.payload or {}).model_dump()
         annotation = AppAnnotationService.update_app_annotation_directly(args, app_model.id, annotation_id)
-        return annotation
+        response = Annotation.model_validate(annotation, from_attributes=True)
+        return response.model_dump(mode="json")
 
     @service_api_ns.doc("delete_annotation")
     @service_api_ns.doc(description="Delete an annotation")

api/controllers/service_api/app/completion.py

@@ -30,6 +30,7 @@ from core.errors.error import (
 from core.helper.trace_id_helper import get_external_trace_id
 from core.model_runtime.errors.invoke import InvokeError
 from libs import helper
+from libs.helper import UUIDStrOrEmpty
 from models.model import App, AppMode, EndUser
 from services.app_generate_service import AppGenerateService
 from services.app_task_service import AppTaskService
@@ -52,7 +53,7 @@ class ChatRequestPayload(BaseModel):
     query: str
     files: list[dict[str, Any]] | None = None
     response_mode: Literal["blocking", "streaming"] | None = None
-    conversation_id: str | None = Field(default=None, description="Conversation UUID")
+    conversation_id: UUIDStrOrEmpty | None = Field(default=None, description="Conversation UUID")
     retriever_from: str = Field(default="dev")
     auto_generate_name: bool = Field(default=True, description="Auto generate conversation name")
     workflow_id: str | None = Field(default=None, description="Workflow ID for advanced chat")

api/controllers/service_api/app/conversation.py

@@ -1,5 +1,4 @@
 from typing import Any, Literal
-from uuid import UUID
 
 from flask import request
 from flask_restx import Resource
@@ -23,12 +22,13 @@ from fields.conversation_variable_fields import (
     build_conversation_variable_infinite_scroll_pagination_model,
     build_conversation_variable_model,
 )
+from libs.helper import UUIDStrOrEmpty
 from models.model import App, AppMode, EndUser
 from services.conversation_service import ConversationService
 
 
 class ConversationListQuery(BaseModel):
-    last_id: UUID | None = Field(default=None, description="Last conversation ID for pagination")
+    last_id: UUIDStrOrEmpty | None = Field(default=None, description="Last conversation ID for pagination")
     limit: int = Field(default=20, ge=1, le=100, description="Number of conversations to return")
     sort_by: Literal["created_at", "-created_at", "updated_at", "-updated_at"] = Field(
         default="-updated_at", description="Sort order for conversations"
@@ -48,7 +48,7 @@ class ConversationRenamePayload(BaseModel):
 
 
 class ConversationVariablesQuery(BaseModel):
-    last_id: UUID | None = Field(default=None, description="Last variable ID for pagination")
+    last_id: UUIDStrOrEmpty | None = Field(default=None, description="Last variable ID for pagination")
     limit: int = Field(default=20, ge=1, le=100, description="Number of variables to return")
     variable_name: str | None = Field(
         default=None, description="Filter variables by name", min_length=1, max_length=255

api/controllers/service_api/app/message.py

@@ -1,6 +1,5 @@
 import logging
 from typing import Literal
-from uuid import UUID
 
 from flask import request
 from flask_restx import Resource
@@ -15,6 +14,7 @@ from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate
 from core.app.entities.app_invoke_entities import InvokeFrom
 from fields.conversation_fields import ResultResponse
 from fields.message_fields import MessageInfiniteScrollPagination, MessageListItem
+from libs.helper import UUIDStrOrEmpty
 from models.model import App, AppMode, EndUser
 from services.errors.message import (
     FirstMessageNotExistsError,
@@ -27,8 +27,8 @@ logger = logging.getLogger(__name__)
 
 
 class MessageListQuery(BaseModel):
-    conversation_id: UUID
-    first_id: UUID | None = None
+    conversation_id: UUIDStrOrEmpty
+    first_id: UUIDStrOrEmpty | None = None
     limit: int = Field(default=20, ge=1, le=100, description="Number of messages to return")
 
 

api/controllers/service_api/dataset/dataset.py

@@ -17,7 +17,7 @@ from controllers.service_api.wraps import (
 from core.model_runtime.entities.model_entities import ModelType
 from core.provider_manager import ProviderManager
 from fields.dataset_fields import dataset_detail_fields
-from fields.tag_fields import build_dataset_tag_fields
+from fields.tag_fields import DataSetTag
 from libs.login import current_user
 from models.account import Account
 from models.dataset import DatasetPermissionEnum
@@ -114,6 +114,7 @@ register_schema_models(
     TagBindingPayload,
     TagUnbindingPayload,
     DatasetListQuery,
+    DataSetTag,
 )
 
 
@@ -395,7 +396,7 @@ class DatasetApi(DatasetApiResource):
         try:
             if DatasetService.delete_dataset(dataset_id_str, current_user):
                 DatasetPermissionService.clear_partial_member_list(dataset_id_str)
-                return 204
+                return "", 204
             else:
                 raise NotFound("Dataset not found.")
         except services.errors.dataset.DatasetInUseError:
@@ -480,15 +481,14 @@ class DatasetTagsApi(DatasetApiResource):
             401: "Unauthorized - invalid API token",
         }
     )
-    @service_api_ns.marshal_with(build_dataset_tag_fields(service_api_ns))
     def get(self, _):
         """Get all knowledge type tags."""
         assert isinstance(current_user, Account)
         cid = current_user.current_tenant_id
         assert cid is not None
         tags = TagService.get_tags("knowledge", cid)
-
-        return tags, 200
+        tag_models = TypeAdapter(list[DataSetTag]).validate_python(tags, from_attributes=True)
+        return [tag.model_dump(mode="json") for tag in tag_models], 200
 
     @service_api_ns.expect(service_api_ns.models[TagCreatePayload.__name__])
     @service_api_ns.doc("create_dataset_tag")
@@ -500,7 +500,6 @@ class DatasetTagsApi(DatasetApiResource):
             403: "Forbidden - insufficient permissions",
         }
     )
-    @service_api_ns.marshal_with(build_dataset_tag_fields(service_api_ns))
     def post(self, _):
         """Add a knowledge type tag."""
         assert isinstance(current_user, Account)
@@ -510,7 +509,9 @@ class DatasetTagsApi(DatasetApiResource):
         payload = TagCreatePayload.model_validate(service_api_ns.payload or {})
         tag = TagService.save_tags({"name": payload.name, "type": "knowledge"})
 
-        response = {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": 0}
+        response = DataSetTag.model_validate(
+            {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": 0}
+        ).model_dump(mode="json")
         return response, 200
 
     @service_api_ns.expect(service_api_ns.models[TagUpdatePayload.__name__])
@@ -523,7 +524,6 @@ class DatasetTagsApi(DatasetApiResource):
             403: "Forbidden - insufficient permissions",
         }
     )
-    @service_api_ns.marshal_with(build_dataset_tag_fields(service_api_ns))
     def patch(self, _):
         assert isinstance(current_user, Account)
         if not (current_user.has_edit_permission or current_user.is_dataset_editor):
@@ -536,8 +536,9 @@ class DatasetTagsApi(DatasetApiResource):
 
         binding_count = TagService.get_tag_binding_count(tag_id)
 
-        response = {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": binding_count}
-
+        response = DataSetTag.model_validate(
+            {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": binding_count}
+        ).model_dump(mode="json")
         return response, 200
 
     @service_api_ns.expect(service_api_ns.models[TagDeletePayload.__name__])
@@ -556,7 +557,7 @@ class DatasetTagsApi(DatasetApiResource):
         payload = TagDeletePayload.model_validate(service_api_ns.payload or {})
         TagService.delete_tag(payload.tag_id)
 
-        return 204
+        return "", 204
 
 
 @service_api_ns.route("/datasets/tags/binding")
@@ -580,7 +581,7 @@ class DatasetTagBindingApi(DatasetApiResource):
         payload = TagBindingPayload.model_validate(service_api_ns.payload or {})
         TagService.save_tag_binding({"tag_ids": payload.tag_ids, "target_id": payload.target_id, "type": "knowledge"})
 
-        return 204
+        return "", 204
 
 
 @service_api_ns.route("/datasets/tags/unbinding")
@@ -604,7 +605,7 @@ class DatasetTagUnbindingApi(DatasetApiResource):
         payload = TagUnbindingPayload.model_validate(service_api_ns.payload or {})
         TagService.delete_tag_binding({"tag_id": payload.tag_id, "target_id": payload.target_id, "type": "knowledge"})
 
-        return 204
+        return "", 204
 
 
 @service_api_ns.route("/datasets/<uuid:dataset_id>/tags")

api/controllers/service_api/dataset/document.py

@@ -746,4 +746,4 @@ class DocumentApi(DatasetApiResource):
         except services.errors.document.DocumentIndexingError:
             raise DocumentIndexingError("Cannot delete document during indexing.")
 
-        return 204
+        return "", 204

api/controllers/service_api/dataset/hit_testing.py

@@ -1,7 +1,10 @@
-from controllers.console.datasets.hit_testing_base import DatasetsHitTestingBase
+from controllers.common.schema import register_schema_model
+from controllers.console.datasets.hit_testing_base import DatasetsHitTestingBase, HitTestingPayload
 from controllers.service_api import service_api_ns
 from controllers.service_api.wraps import DatasetApiResource, cloud_edition_billing_rate_limit_check
 
+register_schema_model(service_api_ns, HitTestingPayload)
+
 
 @service_api_ns.route("/datasets/<uuid:dataset_id>/hit-testing", "/datasets/<uuid:dataset_id>/retrieve")
 class HitTestingApi(DatasetApiResource, DatasetsHitTestingBase):
@@ -15,6 +18,7 @@ class HitTestingApi(DatasetApiResource, DatasetsHitTestingBase):
             404: "Dataset not found",
         }
     )
+    @service_api_ns.expect(service_api_ns.models[HitTestingPayload.__name__])
     @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
     def post(self, tenant_id, dataset_id):
         """Perform hit testing on a dataset.

api/controllers/service_api/dataset/metadata.py

@@ -128,7 +128,7 @@ class DatasetMetadataServiceApi(DatasetApiResource):
         DatasetService.check_dataset_permission(dataset, current_user)
 
         MetadataService.delete_metadata(dataset_id_str, metadata_id_str)
-        return 204
+        return "", 204
 
 
 @service_api_ns.route("/datasets/<uuid:dataset_id>/metadata/built-in")
@@ -168,10 +168,11 @@ class DatasetMetadataBuiltInFieldActionServiceApi(DatasetApiResource):
             raise NotFound("Dataset not found.")
         DatasetService.check_dataset_permission(dataset, current_user)
 
-        if action == "enable":
-            MetadataService.enable_built_in_field(dataset)
-        elif action == "disable":
-            MetadataService.disable_built_in_field(dataset)
+        match action:
+            case "enable":
+                MetadataService.enable_built_in_field(dataset)
+            case "disable":
+                MetadataService.disable_built_in_field(dataset)
         return {"result": "success"}, 200
 
 

api/controllers/service_api/dataset/segment.py

@@ -233,7 +233,7 @@ class DatasetSegmentApi(DatasetApiResource):
         if not segment:
             raise NotFound("Segment not found.")
         SegmentService.delete_segment(segment, document, dataset)
-        return 204
+        return "", 204
 
     @service_api_ns.expect(service_api_ns.models[SegmentUpdatePayload.__name__])
     @service_api_ns.doc("update_segment")
@@ -499,7 +499,7 @@ class DatasetChildChunkApi(DatasetApiResource):
         except ChildChunkDeleteIndexServiceError as e:
             raise ChildChunkDeleteIndexError(str(e))
 
-        return 204
+        return "", 204
 
     @service_api_ns.expect(service_api_ns.models[ChildChunkUpdatePayload.__name__])
     @service_api_ns.doc("update_child_chunk")

api/controllers/service_api/wraps.py

@@ -1,27 +1,24 @@
 import logging
 import time
 from collections.abc import Callable
-from datetime import timedelta
 from enum import StrEnum, auto
 from functools import wraps
-from typing import Concatenate, ParamSpec, TypeVar
+from typing import Concatenate, ParamSpec, TypeVar, cast
 
 from flask import current_app, request
 from flask_login import user_logged_in
 from flask_restx import Resource
 from pydantic import BaseModel
-from sqlalchemy import select, update
-from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden, NotFound, Unauthorized
 
 from enums.cloud_plan import CloudPlan
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
-from libs.datetime_utils import naive_utc_now
 from libs.login import current_user
 from models import Account, Tenant, TenantAccountJoin, TenantStatus
 from models.dataset import Dataset, RateLimitLog
 from models.model import ApiToken, App
+from services.api_token_service import ApiTokenCache, fetch_token_with_single_flight, record_token_usage
 from services.end_user_service import EndUserService
 from services.feature_service import FeatureService
 
@@ -73,14 +70,14 @@ def validate_app_token(view: Callable[P, R] | None = None, *, fetch_user_arg: Fe
 
             # If caller needs end-user context, attach EndUser to current_user
             if fetch_user_arg:
-                if fetch_user_arg.fetch_from == WhereisUserArg.QUERY:
-                    user_id = request.args.get("user")
-                elif fetch_user_arg.fetch_from == WhereisUserArg.JSON:
-                    user_id = request.get_json().get("user")
-                elif fetch_user_arg.fetch_from == WhereisUserArg.FORM:
-                    user_id = request.form.get("user")
-                else:
-                    user_id = None
+                user_id = None
+                match fetch_user_arg.fetch_from:
+                    case WhereisUserArg.QUERY:
+                        user_id = request.args.get("user")
+                    case WhereisUserArg.JSON:
+                        user_id = request.get_json().get("user")
+                    case WhereisUserArg.FORM:
+                        user_id = request.form.get("user")
 
                 if not user_id and fetch_user_arg.required:
                     raise ValueError("Arg user must be provided.")
@@ -296,7 +293,14 @@ def validate_dataset_token(view: Callable[Concatenate[T, P], R] | None = None):
 
 def validate_and_get_api_token(scope: str | None = None):
     """
-    Validate and get API token.
+    Validate and get API token with Redis caching.
+
+    This function uses a two-tier approach:
+    1. First checks Redis cache for the token
+    2. If not cached, queries database and caches the result
+
+    The last_used_at field is updated asynchronously via Celery task
+    to avoid blocking the request.
     """
     auth_header = request.headers.get("Authorization")
     if auth_header is None or " " not in auth_header:
@@ -308,29 +312,18 @@ def validate_and_get_api_token(scope: str | None = None):
     if auth_scheme != "bearer":
         raise Unauthorized("Authorization scheme must be 'Bearer'")
 
-    current_time = naive_utc_now()
-    cutoff_time = current_time - timedelta(minutes=1)
-    with Session(db.engine, expire_on_commit=False) as session:
-        update_stmt = (
-            update(ApiToken)
-            .where(
-                ApiToken.token == auth_token,
-                (ApiToken.last_used_at.is_(None) | (ApiToken.last_used_at < cutoff_time)),
-                ApiToken.type == scope,
-            )
-            .values(last_used_at=current_time)
-        )
-        stmt = select(ApiToken).where(ApiToken.token == auth_token, ApiToken.type == scope)
-        result = session.execute(update_stmt)
-        api_token = session.scalar(stmt)
+    # Try to get token from cache first
+    # Returns a CachedApiToken (plain Python object), not a SQLAlchemy model
+    cached_token = ApiTokenCache.get(auth_token, scope)
+    if cached_token is not None:
+        logger.debug("Token validation served from cache for scope: %s", scope)
+        # Record usage in Redis for later batch update (no Celery task per request)
+        record_token_usage(auth_token, scope)
+        return cast(ApiToken, cached_token)
 
-        if hasattr(result, "rowcount") and result.rowcount > 0:
-            session.commit()
-
-        if not api_token:
-            raise Unauthorized("Access token is invalid")
-
-    return api_token
+    # Cache miss - use Redis lock for single-flight mode
+    # This ensures only one request queries DB for the same token concurrently
+    return fetch_token_with_single_flight(auth_token, scope)
 
 
 class DatasetApiResource(Resource):

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

@@ -14,16 +14,17 @@ class AgentConfigManager:
             agent_dict = config.get("agent_mode", {})
             agent_strategy = agent_dict.get("strategy", "cot")
 
-            if agent_strategy == "function_call":
-                strategy = AgentEntity.Strategy.FUNCTION_CALLING
-            elif agent_strategy in {"cot", "react"}:
-                strategy = AgentEntity.Strategy.CHAIN_OF_THOUGHT
-            else:
-                # old configs, try to detect default strategy
-                if config["model"]["provider"] == "openai":
+            match agent_strategy:
+                case "function_call":
                     strategy = AgentEntity.Strategy.FUNCTION_CALLING
-                else:
+                case "cot" | "react":
                     strategy = AgentEntity.Strategy.CHAIN_OF_THOUGHT
+                case _:
+                    # old configs, try to detect default strategy
+                    if config["model"]["provider"] == "openai":
+                        strategy = AgentEntity.Strategy.FUNCTION_CALLING
+                    else:
+                        strategy = AgentEntity.Strategy.CHAIN_OF_THOUGHT
 
             agent_tools = []
             for tool in agent_dict.get("tools", []):

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

@@ -250,7 +250,7 @@ class WorkflowResponseConverter:
             data=WorkflowFinishStreamResponse.Data(
                 id=run_id,
                 workflow_id=workflow_id,
-                status=status.value,
+                status=status,
                 outputs=encoded_outputs,
                 error=error,
                 elapsed_time=elapsed_time,
@@ -340,13 +340,13 @@ class WorkflowResponseConverter:
         metadata = self._merge_metadata(event.execution_metadata, snapshot)
 
         if isinstance(event, QueueNodeSucceededEvent):
-            status = WorkflowNodeExecutionStatus.SUCCEEDED.value
+            status = WorkflowNodeExecutionStatus.SUCCEEDED
             error_message = event.error
         elif isinstance(event, QueueNodeFailedEvent):
-            status = WorkflowNodeExecutionStatus.FAILED.value
+            status = WorkflowNodeExecutionStatus.FAILED
             error_message = event.error
         else:
-            status = WorkflowNodeExecutionStatus.EXCEPTION.value
+            status = WorkflowNodeExecutionStatus.EXCEPTION
             error_message = event.error
 
         return NodeFinishStreamResponse(
@@ -413,7 +413,7 @@ class WorkflowResponseConverter:
                 process_data_truncated=process_data_truncated,
                 outputs=outputs,
                 outputs_truncated=outputs_truncated,
-                status=WorkflowNodeExecutionStatus.RETRY.value,
+                status=WorkflowNodeExecutionStatus.RETRY,
                 error=event.error,
                 elapsed_time=elapsed_time,
                 execution_metadata=metadata,

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

@@ -120,7 +120,7 @@ class PipelineGenerator(BaseAppGenerator):
                 raise ValueError("Pipeline dataset is required")
         inputs: Mapping[str, Any] = args["inputs"]
         start_node_id: str = args["start_node_id"]
-        datasource_type: str = args["datasource_type"]
+        datasource_type = DatasourceProviderType(args["datasource_type"])
         datasource_info_list: list[Mapping[str, Any]] = self._format_datasource_info_list(
             datasource_type, args["datasource_info_list"], pipeline, workflow, start_node_id, user
         )
@@ -660,7 +660,7 @@ class PipelineGenerator(BaseAppGenerator):
         tenant_id: str,
         dataset_id: str,
         built_in_field_enabled: bool,
-        datasource_type: str,
+        datasource_type: DatasourceProviderType,
         datasource_info: Mapping[str, Any],
         created_from: str,
         position: int,
@@ -668,17 +668,17 @@ class PipelineGenerator(BaseAppGenerator):
         batch: str,
         document_form: str,
     ):
-        if datasource_type == "local_file":
-            name = datasource_info.get("name", "untitled")
-        elif datasource_type == "online_document":
-            name = datasource_info.get("page", {}).get("page_name", "untitled")
-        elif datasource_type == "website_crawl":
-            name = datasource_info.get("title", "untitled")
-        elif datasource_type == "online_drive":
-            name = datasource_info.get("name", "untitled")
-        else:
-            raise ValueError(f"Unsupported datasource type: {datasource_type}")
-
+        match datasource_type:
+            case DatasourceProviderType.LOCAL_FILE:
+                name = datasource_info.get("name", "untitled")
+            case DatasourceProviderType.ONLINE_DOCUMENT:
+                name = datasource_info.get("page", {}).get("page_name", "untitled")
+            case DatasourceProviderType.WEBSITE_CRAWL:
+                name = datasource_info.get("title", "untitled")
+            case DatasourceProviderType.ONLINE_DRIVE:
+                name = datasource_info.get("name", "untitled")
+            case _:
+                raise ValueError(f"Unsupported datasource type: {datasource_type}")
         document = Document(
             tenant_id=tenant_id,
             dataset_id=dataset_id,
@@ -706,7 +706,7 @@ class PipelineGenerator(BaseAppGenerator):
 
     def _format_datasource_info_list(
         self,
-        datasource_type: str,
+        datasource_type: DatasourceProviderType,
         datasource_info_list: list[Mapping[str, Any]],
         pipeline: Pipeline,
         workflow: Workflow,
@@ -716,7 +716,7 @@ class PipelineGenerator(BaseAppGenerator):
         """
         Format datasource info list.
         """
-        if datasource_type == "online_drive":
+        if datasource_type == DatasourceProviderType.ONLINE_DRIVE:
             all_files: list[Mapping[str, Any]] = []
             datasource_node_data = None
             datasource_nodes = workflow.graph_dict.get("nodes", [])

api/core/app/entities/task_entities.py

@@ -7,7 +7,7 @@ from pydantic import BaseModel, ConfigDict, Field
 from core.model_runtime.entities.llm_entities import LLMResult, LLMUsage
 from core.rag.entities.citation_metadata import RetrievalSourceMetadata
 from core.workflow.entities import AgentNodeStrategyInit
-from core.workflow.enums import WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
+from core.workflow.enums import WorkflowExecutionStatus, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
 
 
 class AnnotationReplyAccount(BaseModel):
@@ -223,7 +223,7 @@ class WorkflowFinishStreamResponse(StreamResponse):
 
         id: str
         workflow_id: str
-        status: str
+        status: WorkflowExecutionStatus
         outputs: Mapping[str, Any] | None = None
         error: str | None = None
         elapsed_time: float
@@ -311,7 +311,7 @@ class NodeFinishStreamResponse(StreamResponse):
         process_data_truncated: bool = False
         outputs: Mapping[str, Any] | None = None
         outputs_truncated: bool = True
-        status: str
+        status: WorkflowNodeExecutionStatus
         error: str | None = None
         elapsed_time: float
         execution_metadata: Mapping[WorkflowNodeExecutionMetadataKey, Any] | None = None
@@ -375,7 +375,7 @@ class NodeRetryStreamResponse(StreamResponse):
         process_data_truncated: bool = False
         outputs: Mapping[str, Any] | None = None
         outputs_truncated: bool = False
-        status: str
+        status: WorkflowNodeExecutionStatus
         error: str | None = None
         elapsed_time: float
         execution_metadata: Mapping[WorkflowNodeExecutionMetadataKey, Any] | None = None
@@ -719,7 +719,7 @@ class WorkflowAppBlockingResponse(AppBlockingResponse):
 
         id: str
         workflow_id: str
-        status: str
+        status: WorkflowExecutionStatus
         outputs: Mapping[str, Any] | None = None
         error: str | None = None
         elapsed_time: float

api/core/app/workflow/node_factory.py

@@ -4,13 +4,14 @@ from typing import TYPE_CHECKING, final
 from typing_extensions import override
 
 from configs import dify_config
-from core.file import file_manager
-from core.helper import ssrf_proxy
+from core.file.file_manager import file_manager
 from core.helper.code_executor.code_executor import CodeExecutor
 from core.helper.code_executor.code_node_provider import CodeNodeProvider
+from core.helper.ssrf_proxy import ssrf_proxy
 from core.tools.tool_file_manager import ToolFileManager
+from core.workflow.entities.graph_config import NodeConfigDict
 from core.workflow.enums import NodeType
-from core.workflow.graph import NodeFactory
+from core.workflow.graph.graph import NodeFactory
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.code.code_node import CodeNode
 from core.workflow.nodes.code.limits import CodeNodeLimits
@@ -22,7 +23,6 @@ from core.workflow.nodes.template_transform.template_renderer import (
     Jinja2TemplateRenderer,
 )
 from core.workflow.nodes.template_transform.template_transform_node import TemplateTransformNode
-from libs.typing import is_str, is_str_dict
 
 if TYPE_CHECKING:
     from core.workflow.entities import GraphInitParams
@@ -47,9 +47,10 @@ class DifyNodeFactory(NodeFactory):
         code_providers: Sequence[type[CodeNodeProvider]] | None = None,
         code_limits: CodeNodeLimits | None = None,
         template_renderer: Jinja2TemplateRenderer | None = None,
-        http_request_http_client: HttpClientProtocol = ssrf_proxy,
+        template_transform_max_output_length: int | None = None,
+        http_request_http_client: HttpClientProtocol | None = None,
         http_request_tool_file_manager_factory: Callable[[], ToolFileManager] = ToolFileManager,
-        http_request_file_manager: FileManagerProtocol = file_manager,
+        http_request_file_manager: FileManagerProtocol | None = None,
     ) -> None:
         self.graph_init_params = graph_init_params
         self.graph_runtime_state = graph_runtime_state
@@ -68,12 +69,15 @@ class DifyNodeFactory(NodeFactory):
             max_object_array_length=dify_config.CODE_MAX_OBJECT_ARRAY_LENGTH,
         )
         self._template_renderer = template_renderer or CodeExecutorJinja2TemplateRenderer()
-        self._http_request_http_client = http_request_http_client
+        self._template_transform_max_output_length = (
+            template_transform_max_output_length or dify_config.TEMPLATE_TRANSFORM_MAX_LENGTH
+        )
+        self._http_request_http_client = http_request_http_client or ssrf_proxy
         self._http_request_tool_file_manager_factory = http_request_tool_file_manager_factory
-        self._http_request_file_manager = http_request_file_manager
+        self._http_request_file_manager = http_request_file_manager or file_manager
 
     @override
-    def create_node(self, node_config: dict[str, object]) -> Node:
+    def create_node(self, node_config: NodeConfigDict) -> Node:
         """
         Create a Node instance from node configuration data using the traditional mapping.
 
@@ -82,23 +86,14 @@ class DifyNodeFactory(NodeFactory):
         :raises ValueError: if node type is unknown or configuration is invalid
         """
         # Get node_id from config
-        node_id = node_config.get("id")
-        if not is_str(node_id):
-            raise ValueError("Node config missing id")
+        node_id = node_config["id"]
 
         # Get node type from config
-        node_data = node_config.get("data", {})
-        if not is_str_dict(node_data):
-            raise ValueError(f"Node {node_id} missing data information")
-
-        node_type_str = node_data.get("type")
-        if not is_str(node_type_str):
-            raise ValueError(f"Node {node_id} missing or invalid type information")
-
+        node_data = node_config["data"]
         try:
-            node_type = NodeType(node_type_str)
+            node_type = NodeType(node_data["type"])
         except ValueError:
-            raise ValueError(f"Unknown node type: {node_type_str}")
+            raise ValueError(f"Unknown node type: {node_data['type']}")
 
         # Get node class
         node_mapping = NODE_TYPE_CLASSES_MAPPING.get(node_type)
@@ -131,6 +126,7 @@ class DifyNodeFactory(NodeFactory):
                 graph_init_params=self.graph_init_params,
                 graph_runtime_state=self.graph_runtime_state,
                 template_renderer=self._template_renderer,
+                max_output_length=self._template_transform_max_output_length,
             )
 
         if node_type == NodeType.HTTP_REQUEST:

api/core/file/file_manager.py

@@ -168,3 +168,18 @@ def _to_url(f: File, /):
         return sign_tool_file(tool_file_id=f.related_id, extension=f.extension)
     else:
         raise ValueError(f"Unsupported transfer method: {f.transfer_method}")
+
+
+class FileManager:
+    """
+    Adapter exposing file manager helpers behind FileManagerProtocol.
+
+    This is intentionally a thin wrapper over the existing module-level functions so callers can inject it
+    where a protocol-typed file manager is expected.
+    """
+
+    def download(self, f: File, /) -> bytes:
+        return download(f)
+
+
+file_manager = FileManager()

api/core/helper/code_executor/code_node_provider.py

@@ -47,15 +47,16 @@ class CodeNodeProvider(BaseModel, ABC):
 
     @classmethod
     def get_default_config(cls) -> DefaultConfig:
-        return {
-            "type": "code",
-            "config": {
-                "variables": [
-                    {"variable": "arg1", "value_selector": []},
-                    {"variable": "arg2", "value_selector": []},
-                ],
-                "code_language": cls.get_language(),
-                "code": cls.get_default_code(),
-                "outputs": {"result": {"type": "string", "children": None}},
-            },
+        variables: list[VariableConfig] = [
+            {"variable": "arg1", "value_selector": []},
+            {"variable": "arg2", "value_selector": []},
+        ]
+        outputs: dict[str, OutputConfig] = {"result": {"type": "string", "children": None}}
+
+        config: CodeConfig = {
+            "variables": variables,
+            "code_language": cls.get_language(),
+            "code": cls.get_default_code(),
+            "outputs": outputs,
         }
+        return {"type": "code", "config": config}

api/core/helper/marketplace.py

@@ -6,7 +6,8 @@ from yarl import URL
 
 from configs import dify_config
 from core.helper.download import download_with_size_limit
-from core.plugin.entities.marketplace import MarketplacePluginDeclaration
+from core.plugin.entities.marketplace import MarketplacePluginDeclaration, MarketplacePluginSnapshot
+from extensions.ext_redis import redis_client
 
 marketplace_api_url = URL(str(dify_config.MARKETPLACE_API_URL))
 logger = logging.getLogger(__name__)
@@ -43,28 +44,37 @@ def batch_fetch_plugin_by_ids(plugin_ids: list[str]) -> list[dict]:
     return data.get("data", {}).get("plugins", [])
 
 
-def batch_fetch_plugin_manifests_ignore_deserialization_error(
-    plugin_ids: list[str],
-) -> Sequence[MarketplacePluginDeclaration]:
-    if len(plugin_ids) == 0:
-        return []
-
-    url = str(marketplace_api_url / "api/v1/plugins/batch")
-    response = httpx.post(url, json={"plugin_ids": plugin_ids}, headers={"X-Dify-Version": dify_config.project.version})
-    response.raise_for_status()
-    result: list[MarketplacePluginDeclaration] = []
-    for plugin in response.json()["data"]["plugins"]:
-        try:
-            result.append(MarketplacePluginDeclaration.model_validate(plugin))
-        except Exception:
-            logger.exception(
-                "Failed to deserialize marketplace plugin manifest for %s", plugin.get("plugin_id", "unknown")
-            )
-
-    return result
-
-
 def record_install_plugin_event(plugin_unique_identifier: str):
     url = str(marketplace_api_url / "api/v1/stats/plugins/install_count")
     response = httpx.post(url, json={"unique_identifier": plugin_unique_identifier})
     response.raise_for_status()
+
+
+def fetch_global_plugin_manifest(cache_key_prefix: str, cache_ttl: int) -> None:
+    """
+    Fetch all plugin manifests from marketplace and cache them in Redis.
+    This should be called once per check cycle to populate the instance-level cache.
+
+    Args:
+        cache_key_prefix: Redis key prefix for caching plugin manifests
+        cache_ttl: Cache TTL in seconds
+
+    Raises:
+        httpx.HTTPError: If the HTTP request fails
+        Exception: If any other error occurs during fetching or caching
+    """
+    url = str(marketplace_api_url / "api/v1/dist/plugins/manifest.json")
+    response = httpx.get(url, headers={"X-Dify-Version": dify_config.project.version}, timeout=30)
+    response.raise_for_status()
+
+    raw_json = response.json()
+    plugins_data = raw_json.get("plugins", [])
+
+    # Parse and cache all plugin snapshots
+    for plugin_data in plugins_data:
+        plugin_snapshot = MarketplacePluginSnapshot.model_validate(plugin_data)
+        redis_client.setex(
+            name=f"{cache_key_prefix}{plugin_snapshot.plugin_id}",
+            time=cache_ttl,
+            value=plugin_snapshot.model_dump_json(),
+        )

api/core/helper/ssrf_proxy.py

@@ -230,3 +230,41 @@ def delete(url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any)
 
 def head(url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
     return make_request("HEAD", url, max_retries=max_retries, **kwargs)
+
+
+class SSRFProxy:
+    """
+    Adapter exposing SSRF-protected HTTP helpers behind HttpClientProtocol.
+
+    This is intentionally a thin wrapper over the existing module-level functions so callers can inject it
+    where a protocol-typed HTTP client is expected.
+    """
+
+    @property
+    def max_retries_exceeded_error(self) -> type[Exception]:
+        return max_retries_exceeded_error
+
+    @property
+    def request_error(self) -> type[Exception]:
+        return request_error
+
+    def get(self, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
+        return get(url=url, max_retries=max_retries, **kwargs)
+
+    def head(self, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
+        return head(url=url, max_retries=max_retries, **kwargs)
+
+    def post(self, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
+        return post(url=url, max_retries=max_retries, **kwargs)
+
+    def put(self, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
+        return put(url=url, max_retries=max_retries, **kwargs)
+
+    def delete(self, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
+        return delete(url=url, max_retries=max_retries, **kwargs)
+
+    def patch(self, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
+        return patch(url=url, max_retries=max_retries, **kwargs)
+
+
+ssrf_proxy = SSRFProxy()

api/core/indexing_runner.py

@@ -369,77 +369,78 @@ class IndexingRunner:
         # Generate summary preview
         summary_index_setting = tmp_processing_rule.get("summary_index_setting")
         if summary_index_setting and summary_index_setting.get("enable") and preview_texts:
-            preview_texts = index_processor.generate_summary_preview(tenant_id, preview_texts, summary_index_setting)
+            preview_texts = index_processor.generate_summary_preview(
+                tenant_id, preview_texts, summary_index_setting, doc_language
+            )
 
         return IndexingEstimate(total_segments=total_segments, preview=preview_texts)
 
     def _extract(
         self, index_processor: BaseIndexProcessor, dataset_document: DatasetDocument, process_rule: dict
     ) -> list[Document]:
-        # load file
-        if dataset_document.data_source_type not in {"upload_file", "notion_import", "website_crawl"}:
-            return []
-
         data_source_info = dataset_document.data_source_info_dict
         text_docs = []
-        if dataset_document.data_source_type == "upload_file":
-            if not data_source_info or "upload_file_id" not in data_source_info:
-                raise ValueError("no upload file found")
-            stmt = select(UploadFile).where(UploadFile.id == data_source_info["upload_file_id"])
-            file_detail = db.session.scalars(stmt).one_or_none()
+        match dataset_document.data_source_type:
+            case "upload_file":
+                if not data_source_info or "upload_file_id" not in data_source_info:
+                    raise ValueError("no upload file found")
+                stmt = select(UploadFile).where(UploadFile.id == data_source_info["upload_file_id"])
+                file_detail = db.session.scalars(stmt).one_or_none()
 
-            if file_detail:
+                if file_detail:
+                    extract_setting = ExtractSetting(
+                        datasource_type=DatasourceType.FILE,
+                        upload_file=file_detail,
+                        document_model=dataset_document.doc_form,
+                    )
+                    text_docs = index_processor.extract(extract_setting, process_rule_mode=process_rule["mode"])
+            case "notion_import":
+                if (
+                    not data_source_info
+                    or "notion_workspace_id" not in data_source_info
+                    or "notion_page_id" not in data_source_info
+                ):
+                    raise ValueError("no notion import info found")
                 extract_setting = ExtractSetting(
-                    datasource_type=DatasourceType.FILE,
-                    upload_file=file_detail,
+                    datasource_type=DatasourceType.NOTION,
+                    notion_info=NotionInfo.model_validate(
+                        {
+                            "credential_id": data_source_info.get("credential_id"),
+                            "notion_workspace_id": data_source_info["notion_workspace_id"],
+                            "notion_obj_id": data_source_info["notion_page_id"],
+                            "notion_page_type": data_source_info["type"],
+                            "document": dataset_document,
+                            "tenant_id": dataset_document.tenant_id,
+                        }
+                    ),
                     document_model=dataset_document.doc_form,
                 )
                 text_docs = index_processor.extract(extract_setting, process_rule_mode=process_rule["mode"])
-        elif dataset_document.data_source_type == "notion_import":
-            if (
-                not data_source_info
-                or "notion_workspace_id" not in data_source_info
-                or "notion_page_id" not in data_source_info
-            ):
-                raise ValueError("no notion import info found")
-            extract_setting = ExtractSetting(
-                datasource_type=DatasourceType.NOTION,
-                notion_info=NotionInfo.model_validate(
-                    {
-                        "credential_id": data_source_info.get("credential_id"),
-                        "notion_workspace_id": data_source_info["notion_workspace_id"],
-                        "notion_obj_id": data_source_info["notion_page_id"],
-                        "notion_page_type": data_source_info["type"],
-                        "document": dataset_document,
-                        "tenant_id": dataset_document.tenant_id,
-                    }
-                ),
-                document_model=dataset_document.doc_form,
-            )
-            text_docs = index_processor.extract(extract_setting, process_rule_mode=process_rule["mode"])
-        elif dataset_document.data_source_type == "website_crawl":
-            if (
-                not data_source_info
-                or "provider" not in data_source_info
-                or "url" not in data_source_info
-                or "job_id" not in data_source_info
-            ):
-                raise ValueError("no website import info found")
-            extract_setting = ExtractSetting(
-                datasource_type=DatasourceType.WEBSITE,
-                website_info=WebsiteInfo.model_validate(
-                    {
-                        "provider": data_source_info["provider"],
-                        "job_id": data_source_info["job_id"],
-                        "tenant_id": dataset_document.tenant_id,
-                        "url": data_source_info["url"],
-                        "mode": data_source_info["mode"],
-                        "only_main_content": data_source_info["only_main_content"],
-                    }
-                ),
-                document_model=dataset_document.doc_form,
-            )
-            text_docs = index_processor.extract(extract_setting, process_rule_mode=process_rule["mode"])
+            case "website_crawl":
+                if (
+                    not data_source_info
+                    or "provider" not in data_source_info
+                    or "url" not in data_source_info
+                    or "job_id" not in data_source_info
+                ):
+                    raise ValueError("no website import info found")
+                extract_setting = ExtractSetting(
+                    datasource_type=DatasourceType.WEBSITE,
+                    website_info=WebsiteInfo.model_validate(
+                        {
+                            "provider": data_source_info["provider"],
+                            "job_id": data_source_info["job_id"],
+                            "tenant_id": dataset_document.tenant_id,
+                            "url": data_source_info["url"],
+                            "mode": data_source_info["mode"],
+                            "only_main_content": data_source_info["only_main_content"],
+                        }
+                    ),
+                    document_model=dataset_document.doc_form,
+                )
+                text_docs = index_processor.extract(extract_setting, process_rule_mode=process_rule["mode"])
+            case _:
+                return []
         # update document status to splitting
         self._update_document_index_status(
             document_id=dataset_document.id,

api/core/llm_generator/entities.py

@@ -0,0 +1,20 @@
+"""Shared payload models for LLM generator helpers and controllers."""
+
+from pydantic import BaseModel, Field
+
+from core.app.app_config.entities import ModelConfig
+
+
+class RuleGeneratePayload(BaseModel):
+    instruction: str = Field(..., description="Rule generation instruction")
+    model_config_data: ModelConfig = Field(..., alias="model_config", description="Model configuration")
+    no_variable: bool = Field(default=False, description="Whether to exclude variables")
+
+
+class RuleCodeGeneratePayload(RuleGeneratePayload):
+    code_language: str = Field(default="javascript", description="Programming language for code generation")
+
+
+class RuleStructuredOutputPayload(BaseModel):
+    instruction: str = Field(..., description="Structured output generation instruction")
+    model_config_data: ModelConfig = Field(..., alias="model_config", description="Model configuration")

api/core/llm_generator/llm_generator.py

@@ -6,6 +6,8 @@ from typing import Protocol, cast
 
 import json_repair
 
+from core.app.app_config.entities import ModelConfig
+from core.llm_generator.entities import RuleCodeGeneratePayload, RuleGeneratePayload, RuleStructuredOutputPayload
 from core.llm_generator.output_parser.rule_config_generator import RuleConfigGeneratorOutputParser
 from core.llm_generator.output_parser.suggested_questions_after_answer import SuggestedQuestionsAfterAnswerOutputParser
 from core.llm_generator.prompts import (
@@ -151,19 +153,19 @@ class LLMGenerator:
         return questions
 
     @classmethod
-    def generate_rule_config(cls, tenant_id: str, instruction: str, model_config: dict, no_variable: bool):
+    def generate_rule_config(cls, tenant_id: str, args: RuleGeneratePayload):
         output_parser = RuleConfigGeneratorOutputParser()
 
         error = ""
         error_step = ""
         rule_config = {"prompt": "", "variables": [], "opening_statement": "", "error": ""}
-        model_parameters = model_config.get("completion_params", {})
-        if no_variable:
+        model_parameters = args.model_config_data.completion_params
+        if args.no_variable:
             prompt_template = PromptTemplateParser(WORKFLOW_RULE_CONFIG_PROMPT_GENERATE_TEMPLATE)
 
             prompt_generate = prompt_template.format(
                 inputs={
-                    "TASK_DESCRIPTION": instruction,
+                    "TASK_DESCRIPTION": args.instruction,
                 },
                 remove_template_variables=False,
             )
@@ -175,8 +177,8 @@ class LLMGenerator:
             model_instance = model_manager.get_model_instance(
                 tenant_id=tenant_id,
                 model_type=ModelType.LLM,
-                provider=model_config.get("provider", ""),
-                model=model_config.get("name", ""),
+                provider=args.model_config_data.provider,
+                model=args.model_config_data.name,
             )
 
             try:
@@ -190,7 +192,7 @@ class LLMGenerator:
                 error = str(e)
                 error_step = "generate rule config"
             except Exception as e:
-                logger.exception("Failed to generate rule config, model: %s", model_config.get("name"))
+                logger.exception("Failed to generate rule config, model: %s", args.model_config_data.name)
                 rule_config["error"] = str(e)
 
             rule_config["error"] = f"Failed to {error_step}. Error: {error}" if error else ""
@@ -209,7 +211,7 @@ class LLMGenerator:
         # format the prompt_generate_prompt
         prompt_generate_prompt = prompt_template.format(
             inputs={
-                "TASK_DESCRIPTION": instruction,
+                "TASK_DESCRIPTION": args.instruction,
             },
             remove_template_variables=False,
         )
@@ -220,8 +222,8 @@ class LLMGenerator:
         model_instance = model_manager.get_model_instance(
             tenant_id=tenant_id,
             model_type=ModelType.LLM,
-            provider=model_config.get("provider", ""),
-            model=model_config.get("name", ""),
+            provider=args.model_config_data.provider,
+            model=args.model_config_data.name,
         )
 
         try:
@@ -250,7 +252,7 @@ class LLMGenerator:
             # the second step to generate the task_parameter and task_statement
             statement_generate_prompt = statement_template.format(
                 inputs={
-                    "TASK_DESCRIPTION": instruction,
+                    "TASK_DESCRIPTION": args.instruction,
                     "INPUT_TEXT": prompt_content.message.get_text_content(),
                 },
                 remove_template_variables=False,
@@ -276,7 +278,7 @@ class LLMGenerator:
                 error_step = "generate conversation opener"
 
         except Exception as e:
-            logger.exception("Failed to generate rule config, model: %s", model_config.get("name"))
+            logger.exception("Failed to generate rule config, model: %s", args.model_config_data.name)
             rule_config["error"] = str(e)
 
         rule_config["error"] = f"Failed to {error_step}. Error: {error}" if error else ""
@@ -284,16 +286,20 @@ class LLMGenerator:
         return rule_config
 
     @classmethod
-    def generate_code(cls, tenant_id: str, instruction: str, model_config: dict, code_language: str = "javascript"):
-        if code_language == "python":
+    def generate_code(
+        cls,
+        tenant_id: str,
+        args: RuleCodeGeneratePayload,
+    ):
+        if args.code_language == "python":
             prompt_template = PromptTemplateParser(PYTHON_CODE_GENERATOR_PROMPT_TEMPLATE)
         else:
             prompt_template = PromptTemplateParser(JAVASCRIPT_CODE_GENERATOR_PROMPT_TEMPLATE)
 
         prompt = prompt_template.format(
             inputs={
-                "INSTRUCTION": instruction,
-                "CODE_LANGUAGE": code_language,
+                "INSTRUCTION": args.instruction,
+                "CODE_LANGUAGE": args.code_language,
             },
             remove_template_variables=False,
         )
@@ -302,28 +308,28 @@ class LLMGenerator:
         model_instance = model_manager.get_model_instance(
             tenant_id=tenant_id,
             model_type=ModelType.LLM,
-            provider=model_config.get("provider", ""),
-            model=model_config.get("name", ""),
+            provider=args.model_config_data.provider,
+            model=args.model_config_data.name,
         )
 
         prompt_messages = [UserPromptMessage(content=prompt)]
-        model_parameters = model_config.get("completion_params", {})
+        model_parameters = args.model_config_data.completion_params
         try:
             response: LLMResult = model_instance.invoke_llm(
                 prompt_messages=list(prompt_messages), model_parameters=model_parameters, stream=False
             )
 
             generated_code = response.message.get_text_content()
-            return {"code": generated_code, "language": code_language, "error": ""}
+            return {"code": generated_code, "language": args.code_language, "error": ""}
 
         except InvokeError as e:
             error = str(e)
-            return {"code": "", "language": code_language, "error": f"Failed to generate code. Error: {error}"}
+            return {"code": "", "language": args.code_language, "error": f"Failed to generate code. Error: {error}"}
         except Exception as e:
             logger.exception(
-                "Failed to invoke LLM model, model: %s, language: %s", model_config.get("name"), code_language
+                "Failed to invoke LLM model, model: %s, language: %s", args.model_config_data.name, args.code_language
             )
-            return {"code": "", "language": code_language, "error": f"An unexpected error occurred: {str(e)}"}
+            return {"code": "", "language": args.code_language, "error": f"An unexpected error occurred: {str(e)}"}
 
     @classmethod
     def generate_qa_document(cls, tenant_id: str, query, document_language: str):
@@ -353,20 +359,20 @@ class LLMGenerator:
         return answer.strip()
 
     @classmethod
-    def generate_structured_output(cls, tenant_id: str, instruction: str, model_config: dict):
+    def generate_structured_output(cls, tenant_id: str, args: RuleStructuredOutputPayload):
         model_manager = ModelManager()
         model_instance = model_manager.get_model_instance(
             tenant_id=tenant_id,
             model_type=ModelType.LLM,
-            provider=model_config.get("provider", ""),
-            model=model_config.get("name", ""),
+            provider=args.model_config_data.provider,
+            model=args.model_config_data.name,
         )
 
         prompt_messages = [
             SystemPromptMessage(content=SYSTEM_STRUCTURED_OUTPUT_GENERATE),
-            UserPromptMessage(content=instruction),
+            UserPromptMessage(content=args.instruction),
         ]
-        model_parameters = model_config.get("model_parameters", {})
+        model_parameters = args.model_config_data.completion_params
 
         try:
             response: LLMResult = model_instance.invoke_llm(
@@ -390,12 +396,17 @@ class LLMGenerator:
             error = str(e)
             return {"output": "", "error": f"Failed to generate JSON Schema. Error: {error}"}
         except Exception as e:
-            logger.exception("Failed to invoke LLM model, model: %s", model_config.get("name"))
+            logger.exception("Failed to invoke LLM model, model: %s", args.model_config_data.name)
             return {"output": "", "error": f"An unexpected error occurred: {str(e)}"}
 
     @staticmethod
     def instruction_modify_legacy(
-        tenant_id: str, flow_id: str, current: str, instruction: str, model_config: dict, ideal_output: str | None
+        tenant_id: str,
+        flow_id: str,
+        current: str,
+        instruction: str,
+        model_config: ModelConfig,
+        ideal_output: str | None,
     ):
         last_run: Message | None = (
             db.session.query(Message).where(Message.app_id == flow_id).order_by(Message.created_at.desc()).first()
@@ -434,7 +445,7 @@ class LLMGenerator:
         node_id: str,
         current: str,
         instruction: str,
-        model_config: dict,
+        model_config: ModelConfig,
         ideal_output: str | None,
         workflow_service: WorkflowServiceInterface,
     ):
@@ -505,7 +516,7 @@ class LLMGenerator:
     @staticmethod
     def __instruction_modify_common(
         tenant_id: str,
-        model_config: dict,
+        model_config: ModelConfig,
         last_run: dict | None,
         current: str | None,
         error_message: str | None,
@@ -526,8 +537,8 @@ class LLMGenerator:
         model_instance = ModelManager().get_model_instance(
             tenant_id=tenant_id,
             model_type=ModelType.LLM,
-            provider=model_config.get("provider", ""),
-            model=model_config.get("name", ""),
+            provider=model_config.provider,
+            model=model_config.name,
         )
         match node_type:
             case "llm" | "agent":
@@ -570,7 +581,5 @@ class LLMGenerator:
             error = str(e)
             return {"error": f"Failed to generate code. Error: {error}"}
         except Exception as e:
-            logger.exception(
-                "Failed to invoke LLM model, model: %s", json.dumps(model_config.get("name")), exc_info=True
-            )
+            logger.exception("Failed to invoke LLM model, model: %s", json.dumps(model_config.name), exc_info=True)
             return {"error": f"An unexpected error occurred: {str(e)}"}

api/core/llm_generator/prompts.py

@@ -441,11 +441,13 @@ DEFAULT_GENERATOR_SUMMARY_PROMPT = (
 
 Requirements:
 1. Write a concise summary in plain text
-2. Use the same language as the input content
+2. You must write in {language}. No language other than {language} should be used.
 3. Focus on important facts, concepts, and details
 4. If images are included, describe their key information
 5. Do not use words like "好的", "ok", "I understand", "This text discusses", "The content mentions"
 6. Write directly without extra words
+7. If there is not enough content to generate a meaningful summary, 
+   return an empty string without any explanation or prompt
 
 Output only the summary text. Start summarizing now:
 

api/core/mcp/session/base_session.py

@@ -347,7 +347,7 @@ class BaseSession(
                         message.message.root.model_dump(by_alias=True, mode="json", exclude_none=True)
                     )
 
-                    responder = RequestResponder(
+                    responder = RequestResponder[ReceiveRequestT, SendResultT](
                         request_id=message.message.root.id,
                         request_meta=validated_request.root.params.meta if validated_request.root.params else None,
                         request=validated_request,

api/core/model_runtime/entities/defaults.py

@@ -88,7 +88,7 @@ PARAMETER_RULE_TEMPLATE: dict[DefaultParameterName, dict] = {
     DefaultParameterName.MAX_TOKENS: {
         "label": {
             "en_US": "Max Tokens",
-            "zh_Hans": "最大标记",
+            "zh_Hans": "最大 Token 数",
         },
         "type": "int",
         "help": {

api/core/model_runtime/model_providers/__base/large_language_model.py

@@ -92,6 +92,10 @@ def _build_llm_result_from_first_chunk(
     Build a single `LLMResult` from the first returned chunk.
 
     This is used for `stream=False` because the plugin side may still implement the response via a chunked stream.
+
+    Note:
+        This function always drains the `chunks` iterator after reading the first chunk to ensure any underlying
+        streaming resources are released (e.g., HTTP connections owned by the plugin runtime).
     """
     content = ""
     content_list: list[PromptMessageContentUnionTypes] = []
@@ -99,18 +103,25 @@ def _build_llm_result_from_first_chunk(
     system_fingerprint: str | None = None
     tools_calls: list[AssistantPromptMessage.ToolCall] = []
 
-    first_chunk = next(chunks, None)
-    if first_chunk is not None:
-        if isinstance(first_chunk.delta.message.content, str):
-            content += first_chunk.delta.message.content
-        elif isinstance(first_chunk.delta.message.content, list):
-            content_list.extend(first_chunk.delta.message.content)
+    try:
+        first_chunk = next(chunks, None)
+        if first_chunk is not None:
+            if isinstance(first_chunk.delta.message.content, str):
+                content += first_chunk.delta.message.content
+            elif isinstance(first_chunk.delta.message.content, list):
+                content_list.extend(first_chunk.delta.message.content)
 
-        if first_chunk.delta.message.tool_calls:
-            _increase_tool_call(first_chunk.delta.message.tool_calls, tools_calls)
+            if first_chunk.delta.message.tool_calls:
+                _increase_tool_call(first_chunk.delta.message.tool_calls, tools_calls)
 
-        usage = first_chunk.delta.usage or LLMUsage.empty_usage()
-        system_fingerprint = first_chunk.system_fingerprint
+            usage = first_chunk.delta.usage or LLMUsage.empty_usage()
+            system_fingerprint = first_chunk.system_fingerprint
+    finally:
+        try:
+            for _ in chunks:
+                pass
+        except Exception:
+            logger.debug("Failed to drain non-stream plugin chunk iterator.", exc_info=True)
 
     return LLMResult(
         model=model,
@@ -283,7 +294,7 @@ class LargeLanguageModel(AIModel):
             # TODO
             raise self._transform_invoke_error(e)
 
-        if stream and isinstance(result, Generator):
+        if stream and not isinstance(result, LLMResult):
             return self._invoke_result_generator(
                 model=model,
                 result=result,

api/core/model_runtime/model_providers/model_provider_factory.py

@@ -314,6 +314,8 @@ class ModelProviderFactory:
         elif model_type == ModelType.TTS:
             return TTSModel.model_validate(init_params)
 
+        raise ValueError(f"Unsupported model type: {model_type}")
+
     def get_provider_icon(self, provider: str, icon_type: str, lang: str) -> tuple[bytes, str]:
         """
         Get provider icon

api/core/plugin/entities/marketplace.py

@@ -1,4 +1,4 @@
-from pydantic import BaseModel, Field, model_validator
+from pydantic import BaseModel, Field, computed_field, model_validator
 
 from core.model_runtime.entities.provider_entities import ProviderEntity
 from core.plugin.entities.endpoint import EndpointProviderDeclaration
@@ -48,3 +48,15 @@ class MarketplacePluginDeclaration(BaseModel):
         if "tool" in data and not data["tool"]:
             del data["tool"]
         return data
+
+
+class MarketplacePluginSnapshot(BaseModel):
+    org: str
+    name: str
+    latest_version: str
+    latest_package_identifier: str
+    latest_package_url: str
+
+    @computed_field
+    def plugin_id(self) -> str:
+        return f"{self.org}/{self.name}"

api/core/rag/index_processor/index_processor_base.py

@@ -48,12 +48,22 @@ class BaseIndexProcessor(ABC):
 
     @abstractmethod
     def generate_summary_preview(
-        self, tenant_id: str, preview_texts: list[PreviewDetail], summary_index_setting: dict
+        self,
+        tenant_id: str,
+        preview_texts: list[PreviewDetail],
+        summary_index_setting: dict,
+        doc_language: str | None = None,
     ) -> list[PreviewDetail]:
         """
         For each segment in preview_texts, generate a summary using LLM and attach it to the segment.
         The summary can be stored in a new attribute, e.g., summary.
         This method should be implemented by subclasses.
+
+        Args:
+            tenant_id: Tenant ID
+            preview_texts: List of preview details to generate summaries for
+            summary_index_setting: Summary index configuration
+            doc_language: Optional document language to ensure summary is generated in the correct language
         """
         raise NotImplementedError
 

api/core/rag/index_processor/processor/paragraph_index_processor.py

@@ -275,7 +275,11 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
             raise ValueError("Chunks is not a list")
 
     def generate_summary_preview(
-        self, tenant_id: str, preview_texts: list[PreviewDetail], summary_index_setting: dict
+        self,
+        tenant_id: str,
+        preview_texts: list[PreviewDetail],
+        summary_index_setting: dict,
+        doc_language: str | None = None,
     ) -> list[PreviewDetail]:
         """
         For each segment, concurrently call generate_summary to generate a summary
@@ -298,11 +302,15 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
             if flask_app:
                 # Ensure Flask app context in worker thread
                 with flask_app.app_context():
-                    summary, _ = self.generate_summary(tenant_id, preview.content, summary_index_setting)
+                    summary, _ = self.generate_summary(
+                        tenant_id, preview.content, summary_index_setting, document_language=doc_language
+                    )
                     preview.summary = summary
             else:
                 # Fallback: try without app context (may fail)
-                summary, _ = self.generate_summary(tenant_id, preview.content, summary_index_setting)
+                summary, _ = self.generate_summary(
+                    tenant_id, preview.content, summary_index_setting, document_language=doc_language
+                )
                 preview.summary = summary
 
         # Generate summaries concurrently using ThreadPoolExecutor
@@ -356,6 +364,7 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
         text: str,
         summary_index_setting: dict | None = None,
         segment_id: str | None = None,
+        document_language: str | None = None,
     ) -> tuple[str, LLMUsage]:
         """
         Generate summary for the given text using ModelInstance.invoke_llm and the default or custom summary prompt,
@@ -366,6 +375,8 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
             text: Text content to summarize
             summary_index_setting: Summary index configuration
             segment_id: Optional segment ID to fetch attachments from SegmentAttachmentBinding table
+            document_language: Optional document language (e.g., "Chinese", "English")
+                to ensure summary is generated in the correct language
 
         Returns:
             Tuple of (summary_content, llm_usage) where llm_usage is LLMUsage object
@@ -381,8 +392,22 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
             raise ValueError("model_name and model_provider_name are required in summary_index_setting")
 
         # Import default summary prompt
+        is_default_prompt = False
         if not summary_prompt:
             summary_prompt = DEFAULT_GENERATOR_SUMMARY_PROMPT
+            is_default_prompt = True
+
+        # Format prompt with document language only for default prompt
+        # Custom prompts are used as-is to avoid interfering with user-defined templates
+        # If document_language is provided, use it; otherwise, use "the same language as the input content"
+        # This is especially important for image-only chunks where text is empty or minimal
+        if is_default_prompt:
+            language_for_prompt = document_language or "the same language as the input content"
+            try:
+                summary_prompt = summary_prompt.format(language=language_for_prompt)
+            except KeyError:
+                # If default prompt doesn't have {language} placeholder, use it as-is
+                pass
 
         provider_manager = ProviderManager()
         provider_model_bundle = provider_manager.get_provider_model_bundle(

api/core/rag/index_processor/processor/parent_child_index_processor.py

@@ -358,7 +358,11 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
         }
 
     def generate_summary_preview(
-        self, tenant_id: str, preview_texts: list[PreviewDetail], summary_index_setting: dict
+        self,
+        tenant_id: str,
+        preview_texts: list[PreviewDetail],
+        summary_index_setting: dict,
+        doc_language: str | None = None,
     ) -> list[PreviewDetail]:
         """
         For each parent chunk in preview_texts, concurrently call generate_summary to generate a summary
@@ -389,6 +393,7 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
                         tenant_id=tenant_id,
                         text=preview.content,
                         summary_index_setting=summary_index_setting,
+                        document_language=doc_language,
                     )
                     preview.summary = summary
             else:
@@ -397,6 +402,7 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
                     tenant_id=tenant_id,
                     text=preview.content,
                     summary_index_setting=summary_index_setting,
+                    document_language=doc_language,
                 )
                 preview.summary = summary
 

api/core/rag/index_processor/processor/qa_index_processor.py

@@ -241,7 +241,11 @@ class QAIndexProcessor(BaseIndexProcessor):
         }
 
     def generate_summary_preview(
-        self, tenant_id: str, preview_texts: list[PreviewDetail], summary_index_setting: dict
+        self,
+        tenant_id: str,
+        preview_texts: list[PreviewDetail],
+        summary_index_setting: dict,
+        doc_language: str | None = None,
     ) -> list[PreviewDetail]:
         """
         QA model doesn't generate summaries, so this method returns preview_texts unchanged.

api/core/schemas/registry.py

@@ -35,6 +35,7 @@ class SchemaRegistry:
                     registry.load_all_versions()
 
                     cls._default_instance = registry
+            return cls._default_instance
 
         return cls._default_instance
 

api/core/tools/tool_manager.py

@@ -189,16 +189,13 @@ class ToolManager:
                 raise ToolProviderNotFoundError(f"builtin tool {tool_name} not found")
 
             if not provider_controller.need_credentials:
-                return cast(
-                    BuiltinTool,
-                    builtin_tool.fork_tool_runtime(
-                        runtime=ToolRuntime(
-                            tenant_id=tenant_id,
-                            credentials={},
-                            invoke_from=invoke_from,
-                            tool_invoke_from=tool_invoke_from,
-                        )
-                    ),
+                return builtin_tool.fork_tool_runtime(
+                    runtime=ToolRuntime(
+                        tenant_id=tenant_id,
+                        credentials={},
+                        invoke_from=invoke_from,
+                        tool_invoke_from=tool_invoke_from,
+                    )
                 )
             builtin_provider = None
             if isinstance(provider_controller, PluginToolProviderController):
@@ -300,18 +297,15 @@ class ToolManager:
                 decrypted_credentials = refreshed_credentials.credentials
                 cache.delete()
 
-            return cast(
-                BuiltinTool,
-                builtin_tool.fork_tool_runtime(
-                    runtime=ToolRuntime(
-                        tenant_id=tenant_id,
-                        credentials=dict(decrypted_credentials),
-                        credential_type=CredentialType.of(builtin_provider.credential_type),
-                        runtime_parameters={},
-                        invoke_from=invoke_from,
-                        tool_invoke_from=tool_invoke_from,
-                    )
-                ),
+            return builtin_tool.fork_tool_runtime(
+                runtime=ToolRuntime(
+                    tenant_id=tenant_id,
+                    credentials=dict(decrypted_credentials),
+                    credential_type=CredentialType.of(builtin_provider.credential_type),
+                    runtime_parameters={},
+                    invoke_from=invoke_from,
+                    tool_invoke_from=tool_invoke_from,
+                )
             )
 
         elif provider_type == ToolProviderType.API:

api/core/tools/utils/workflow_configuration_sync.py

@@ -7,11 +7,6 @@ from core.workflow.nodes.base.entities import OutputVariableEntity
 
 
 class WorkflowToolConfigurationUtils:
-    @classmethod
-    def check_parameter_configurations(cls, configurations: list[Mapping[str, Any]]):
-        for configuration in configurations:
-            WorkflowToolParameterConfiguration.model_validate(configuration)
-
     @classmethod
     def get_workflow_graph_variables(cls, graph: Mapping[str, Any]) -> Sequence[VariableEntity]:
         """

api/core/trigger/debug/event_bus.py

@@ -23,8 +23,8 @@ class TriggerDebugEventBus:
     """
 
     # LUA_SELECT: Atomic poll or register for event
-    # KEYS[1] = trigger_debug_inbox:{tenant_id}:{address_id}
-    # KEYS[2] = trigger_debug_waiting_pool:{tenant_id}:...
+    # KEYS[1] = trigger_debug_inbox:{<tenant_id>}:<address_id>
+    # KEYS[2] = trigger_debug_waiting_pool:{<tenant_id>}:...
     # ARGV[1] = address_id
     LUA_SELECT = (
         "local v=redis.call('GET',KEYS[1]);"
@@ -35,7 +35,7 @@ class TriggerDebugEventBus:
     )
 
     # LUA_DISPATCH: Dispatch event to all waiting addresses
-    # KEYS[1] = trigger_debug_waiting_pool:{tenant_id}:...
+    # KEYS[1] = trigger_debug_waiting_pool:{<tenant_id>}:...
     # ARGV[1] = tenant_id
     # ARGV[2] = event_json
     LUA_DISPATCH = (
@@ -43,7 +43,7 @@ class TriggerDebugEventBus:
         "if #a==0 then return 0 end;"
         "redis.call('DEL',KEYS[1]);"
         "for i=1,#a do "
-        f"redis.call('SET','trigger_debug_inbox:'..ARGV[1]..':'..a[i],ARGV[2],'EX',{TRIGGER_DEBUG_EVENT_TTL});"
+        f"redis.call('SET','trigger_debug_inbox:{{'..ARGV[1]..'}}'..':'..a[i],ARGV[2],'EX',{TRIGGER_DEBUG_EVENT_TTL});"
         "end;"
         "return #a"
     )
@@ -108,7 +108,7 @@ class TriggerDebugEventBus:
             Event object if available, None otherwise
         """
         address_id: str = hashlib.sha256(f"{user_id}|{app_id}|{node_id}".encode()).hexdigest()
-        address: str = f"trigger_debug_inbox:{tenant_id}:{address_id}"
+        address: str = f"trigger_debug_inbox:{{{tenant_id}}}:{address_id}"
 
         try:
             event_data = redis_client.eval(

api/core/trigger/debug/events.py

@@ -42,7 +42,7 @@ def build_webhook_pool_key(tenant_id: str, app_id: str, node_id: str) -> str:
         app_id: App ID
         node_id: Node ID
     """
-    return f"{TriggerDebugPoolKey.WEBHOOK}:{tenant_id}:{app_id}:{node_id}"
+    return f"{TriggerDebugPoolKey.WEBHOOK}:{{{tenant_id}}}:{app_id}:{node_id}"
 
 
 class PluginTriggerDebugEvent(BaseDebugEvent):
@@ -64,4 +64,4 @@ def build_plugin_pool_key(tenant_id: str, provider_id: str, subscription_id: str
         provider_id: Provider ID
         subscription_id: Subscription ID
     """
-    return f"{TriggerDebugPoolKey.PLUGIN}:{tenant_id}:{str(provider_id)}:{subscription_id}:{name}"
+    return f"{TriggerDebugPoolKey.PLUGIN}:{{{tenant_id}}}:{str(provider_id)}:{subscription_id}:{name}"

api/core/variables/variables.py

@@ -112,7 +112,7 @@ class ArrayBooleanVariable(ArrayBooleanSegment, ArrayVariable):
 
 class RAGPipelineVariable(BaseModel):
     belong_to_node_id: str = Field(description="belong to which node id, shared means public")
-    type: str = Field(description="variable type, text-input, paragraph, select, number,  file, file-list")
+    type: str = Field(description="variable type, text-input, paragraph, select, number, file, file-list")
     label: str = Field(description="label")
     description: str | None = Field(description="description", default="")
     variable: str = Field(description="variable key", default="")

api/core/workflow/entities/graph_config.py

@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+import sys
+
+from pydantic import TypeAdapter, with_config
+
+if sys.version_info >= (3, 12):
+    from typing import TypedDict
+else:
+    from typing_extensions import TypedDict
+
+
+@with_config(extra="allow")
+class NodeConfigData(TypedDict):
+    type: str
+
+
+@with_config(extra="allow")
+class NodeConfigDict(TypedDict):
+    id: str
+    data: NodeConfigData
+
+
+NodeConfigDictAdapter = TypeAdapter(NodeConfigDict)

api/core/workflow/graph/graph.py

@@ -5,15 +5,20 @@ from collections import defaultdict
 from collections.abc import Mapping, Sequence
 from typing import Protocol, cast, final
 
+from pydantic import TypeAdapter
+
+from core.workflow.entities.graph_config import NodeConfigDict
 from core.workflow.enums import ErrorStrategy, NodeExecutionType, NodeState, NodeType
 from core.workflow.nodes.base.node import Node
-from libs.typing import is_str, is_str_dict
+from libs.typing import is_str
 
 from .edge import Edge
 from .validation import get_graph_validator
 
 logger = logging.getLogger(__name__)
 
+_ListNodeConfigDict = TypeAdapter(list[NodeConfigDict])
+
 
 class NodeFactory(Protocol):
     """
@@ -23,7 +28,7 @@ class NodeFactory(Protocol):
     allowing for different node creation strategies while maintaining type safety.
     """
 
-    def create_node(self, node_config: dict[str, object]) -> Node:
+    def create_node(self, node_config: NodeConfigDict) -> Node:
         """
         Create a Node instance from node configuration data.
 
@@ -63,28 +68,24 @@ class Graph:
         self.root_node = root_node
 
     @classmethod
-    def _parse_node_configs(cls, node_configs: list[dict[str, object]]) -> dict[str, dict[str, object]]:
+    def _parse_node_configs(cls, node_configs: list[NodeConfigDict]) -> dict[str, NodeConfigDict]:
         """
         Parse node configurations and build a mapping of node IDs to configs.
 
         :param node_configs: list of node configuration dictionaries
         :return: mapping of node ID to node config
         """
-        node_configs_map: dict[str, dict[str, object]] = {}
+        node_configs_map: dict[str, NodeConfigDict] = {}
 
         for node_config in node_configs:
-            node_id = node_config.get("id")
-            if not node_id or not isinstance(node_id, str):
-                continue
-
-            node_configs_map[node_id] = node_config
+            node_configs_map[node_config["id"]] = node_config
 
         return node_configs_map
 
     @classmethod
     def _find_root_node_id(
         cls,
-        node_configs_map: Mapping[str, Mapping[str, object]],
+        node_configs_map: Mapping[str, NodeConfigDict],
         edge_configs: Sequence[Mapping[str, object]],
         root_node_id: str | None = None,
     ) -> str:
@@ -113,10 +114,8 @@ class Graph:
         # Prefer START node if available
         start_node_id = None
         for nid in root_candidates:
-            node_data = node_configs_map[nid].get("data")
-            if not is_str_dict(node_data):
-                continue
-            node_type = node_data.get("type")
+            node_data = node_configs_map[nid]["data"]
+            node_type = node_data["type"]
             if not isinstance(node_type, str):
                 continue
             if NodeType(node_type).is_start_node:
@@ -176,7 +175,7 @@ class Graph:
     @classmethod
     def _create_node_instances(
         cls,
-        node_configs_map: dict[str, dict[str, object]],
+        node_configs_map: dict[str, NodeConfigDict],
         node_factory: NodeFactory,
     ) -> dict[str, Node]:
         """
@@ -303,7 +302,7 @@ class Graph:
         node_configs = graph_config.get("nodes", [])
 
         edge_configs = cast(list[dict[str, object]], edge_configs)
-        node_configs = cast(list[dict[str, object]], node_configs)
+        node_configs = _ListNodeConfigDict.validate_python(node_configs)
 
         if not node_configs:
             raise ValueError("Graph must have at least one node")

api/core/workflow/graph_engine/graph_engine.py

@@ -46,7 +46,6 @@ from .graph_traversal import EdgeProcessor, SkipPropagator
 from .layers.base import GraphEngineLayer
 from .orchestration import Dispatcher, ExecutionCoordinator
 from .protocols.command_channel import CommandChannel
-from .ready_queue import ReadyQueue
 from .worker_management import WorkerPool
 
 if TYPE_CHECKING:
@@ -90,7 +89,7 @@ class GraphEngine:
         self._graph_execution.workflow_id = workflow_id
 
         # === Execution Queues ===
-        self._ready_queue = cast(ReadyQueue, self._graph_runtime_state.ready_queue)
+        self._ready_queue = self._graph_runtime_state.ready_queue
 
         # Queue for events generated during execution
         self._event_queue: queue.Queue[GraphNodeEventBase] = queue.Queue()

api/core/workflow/graph_engine/response_coordinator/coordinator.py

@@ -15,10 +15,10 @@ from uuid import uuid4
 from pydantic import BaseModel, Field
 
 from core.workflow.enums import NodeExecutionType, NodeState
-from core.workflow.graph import Graph
 from core.workflow.graph_events import NodeRunStreamChunkEvent, NodeRunSucceededEvent
 from core.workflow.nodes.base.template import TextSegment, VariableSegment
 from core.workflow.runtime import VariablePool
+from core.workflow.runtime.graph_runtime_state import GraphProtocol
 
 from .path import Path
 from .session import ResponseSession
@@ -75,7 +75,7 @@ class ResponseStreamCoordinator:
     Ensures ordered streaming of responses based on upstream node outputs and constants.
     """
 
-    def __init__(self, variable_pool: "VariablePool", graph: "Graph") -> None:
+    def __init__(self, variable_pool: "VariablePool", graph: GraphProtocol) -> None:
         """
         Initialize coordinator with variable pool.
 

api/core/workflow/graph_engine/response_coordinator/session.py

@@ -10,10 +10,10 @@ from __future__ import annotations
 from dataclasses import dataclass
 
 from core.workflow.nodes.answer.answer_node import AnswerNode
-from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.base.template import Template
 from core.workflow.nodes.end.end_node import EndNode
 from core.workflow.nodes.knowledge_index import KnowledgeIndexNode
+from core.workflow.runtime.graph_runtime_state import NodeProtocol
 
 
 @dataclass
@@ -29,21 +29,26 @@ class ResponseSession:
     index: int = 0  # Current position in the template segments
 
     @classmethod
-    def from_node(cls, node: Node) -> ResponseSession:
+    def from_node(cls, node: NodeProtocol) -> ResponseSession:
         """
-        Create a ResponseSession from an AnswerNode or EndNode.
+        Create a ResponseSession from a response-capable node.
+
+        The parameter is typed as `NodeProtocol` because the graph is exposed behind a protocol at the runtime layer,
+        but at runtime this must be an `AnswerNode`, `EndNode`, or `KnowledgeIndexNode` that provides:
+        - `id: str`
+        - `get_streaming_template() -> Template`
 
         Args:
-            node: Must be either an AnswerNode or EndNode instance
+            node: Node from the materialized workflow graph.
 
         Returns:
             ResponseSession configured with the node's streaming template
 
         Raises:
-            TypeError: If node is not an AnswerNode or EndNode
+            TypeError: If node is not a supported response node type.
         """
         if not isinstance(node, AnswerNode | EndNode | KnowledgeIndexNode):
-            raise TypeError
+            raise TypeError("ResponseSession.from_node only supports AnswerNode, EndNode, or KnowledgeIndexNode")
         return cls(
             node_id=node.id,
             template=node.get_streaming_template(),

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

@@ -192,32 +192,33 @@ class AgentNode(Node[AgentNodeData]):
                 result[parameter_name] = None
                 continue
             agent_input = node_data.agent_parameters[parameter_name]
-            if agent_input.type == "variable":
-                variable = variable_pool.get(agent_input.value)  # type: ignore
-                if variable is None:
-                    raise AgentVariableNotFoundError(str(agent_input.value))
-                parameter_value = variable.value
-            elif agent_input.type in {"mixed", "constant"}:
-                # variable_pool.convert_template expects a string template,
-                # but if passing a dict, convert to JSON string first before rendering
-                try:
-                    if not isinstance(agent_input.value, str):
-                        parameter_value = json.dumps(agent_input.value, ensure_ascii=False)
-                    else:
+            match agent_input.type:
+                case "variable":
+                    variable = variable_pool.get(agent_input.value)  # type: ignore
+                    if variable is None:
+                        raise AgentVariableNotFoundError(str(agent_input.value))
+                    parameter_value = variable.value
+                case "mixed" | "constant":
+                    # variable_pool.convert_template expects a string template,
+                    # but if passing a dict, convert to JSON string first before rendering
+                    try:
+                        if not isinstance(agent_input.value, str):
+                            parameter_value = json.dumps(agent_input.value, ensure_ascii=False)
+                        else:
+                            parameter_value = str(agent_input.value)
+                    except TypeError:
                         parameter_value = str(agent_input.value)
-                except TypeError:
-                    parameter_value = str(agent_input.value)
-                segment_group = variable_pool.convert_template(parameter_value)
-                parameter_value = segment_group.log if for_log else segment_group.text
-                # variable_pool.convert_template returns a string,
-                # so we need to convert it back to a dictionary
-                try:
-                    if not isinstance(agent_input.value, str):
-                        parameter_value = json.loads(parameter_value)
-                except json.JSONDecodeError:
-                    parameter_value = parameter_value
-            else:
-                raise AgentInputTypeError(agent_input.type)
+                    segment_group = variable_pool.convert_template(parameter_value)
+                    parameter_value = segment_group.log if for_log else segment_group.text
+                    # variable_pool.convert_template returns a string,
+                    # so we need to convert it back to a dictionary
+                    try:
+                        if not isinstance(agent_input.value, str):
+                            parameter_value = json.loads(parameter_value)
+                    except json.JSONDecodeError:
+                        parameter_value = parameter_value
+                case _:
+                    raise AgentInputTypeError(agent_input.type)
             value = parameter_value
             if parameter.type == "array[tools]":
                 value = cast(list[dict[str, Any]], value)
@@ -374,12 +375,13 @@ class AgentNode(Node[AgentNodeData]):
         result: dict[str, Any] = {}
         for parameter_name in typed_node_data.agent_parameters:
             input = typed_node_data.agent_parameters[parameter_name]
-            if input.type in ["mixed", "constant"]:
-                selectors = VariableTemplateParser(str(input.value)).extract_variable_selectors()
-                for selector in selectors:
-                    result[selector.variable] = selector.value_selector
-            elif input.type == "variable":
-                result[parameter_name] = input.value
+            match input.type:
+                case "mixed" | "constant":
+                    selectors = VariableTemplateParser(str(input.value)).extract_variable_selectors()
+                    for selector in selectors:
+                        result[selector.variable] = selector.value_selector
+                case "variable":
+                    result[parameter_name] = input.value
 
         result = {node_id + "." + key: value for key, value in result.items()}
 

api/core/workflow/nodes/base/entities.py

@@ -115,7 +115,7 @@ class DefaultValue(BaseModel):
     @model_validator(mode="after")
     def validate_value_type(self) -> DefaultValue:
         # Type validation configuration
-        type_validators = {
+        type_validators: dict[DefaultValueType, dict[str, Any]] = {
             DefaultValueType.STRING: {
                 "type": str,
                 "converter": lambda x: x,

api/core/workflow/nodes/code/entities.py

@@ -1,4 +1,4 @@
-from typing import Annotated, Literal, Self
+from typing import Annotated, Literal
 
 from pydantic import AfterValidator, BaseModel
 
@@ -34,7 +34,7 @@ class CodeNodeData(BaseNodeData):
 
     class Output(BaseModel):
         type: Annotated[SegmentType, AfterValidator(_validate_type)]
-        children: dict[str, Self] | None = None
+        children: dict[str, "CodeNodeData.Output"] | None = None
 
     class Dependency(BaseModel):
         name: str

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

@@ -69,11 +69,13 @@ class DatasourceNode(Node[DatasourceNodeData]):
         if datasource_type is None:
             raise DatasourceNodeError("Datasource type is not set")
 
+        datasource_type = DatasourceProviderType.value_of(datasource_type)
+
         datasource_runtime = DatasourceManager.get_datasource_runtime(
             provider_id=f"{node_data.plugin_id}/{node_data.provider_name}",
             datasource_name=node_data.datasource_name or "",
             tenant_id=self.tenant_id,
-            datasource_type=DatasourceProviderType.value_of(datasource_type),
+            datasource_type=datasource_type,
         )
         datasource_info["icon"] = datasource_runtime.get_icon_url(self.tenant_id)
 
@@ -268,15 +270,18 @@ class DatasourceNode(Node[DatasourceNodeData]):
         if typed_node_data.datasource_parameters:
             for parameter_name in typed_node_data.datasource_parameters:
                 input = typed_node_data.datasource_parameters[parameter_name]
-                if input.type == "mixed":
-                    assert isinstance(input.value, str)
-                    selectors = VariableTemplateParser(input.value).extract_variable_selectors()
-                    for selector in selectors:
-                        result[selector.variable] = selector.value_selector
-                elif input.type == "variable":
-                    result[parameter_name] = input.value
-                elif input.type == "constant":
-                    pass
+                match input.type:
+                    case "mixed":
+                        assert isinstance(input.value, str)
+                        selectors = VariableTemplateParser(input.value).extract_variable_selectors()
+                        for selector in selectors:
+                            result[selector.variable] = selector.value_selector
+                    case "variable":
+                        result[parameter_name] = input.value
+                    case "constant":
+                        pass
+                    case None:
+                        pass
 
             result = {node_id + "." + key: value for key, value in result.items()}
 
@@ -306,99 +311,107 @@ class DatasourceNode(Node[DatasourceNodeData]):
         variables: dict[str, Any] = {}
 
         for message in message_stream:
-            if message.type in {
-                DatasourceMessage.MessageType.IMAGE_LINK,
-                DatasourceMessage.MessageType.BINARY_LINK,
-                DatasourceMessage.MessageType.IMAGE,
-            }:
-                assert isinstance(message.message, DatasourceMessage.TextMessage)
+            match message.type:
+                case (
+                    DatasourceMessage.MessageType.IMAGE_LINK
+                    | DatasourceMessage.MessageType.BINARY_LINK
+                    | DatasourceMessage.MessageType.IMAGE
+                ):
+                    assert isinstance(message.message, DatasourceMessage.TextMessage)
 
-                url = message.message.text
-                transfer_method = FileTransferMethod.TOOL_FILE
+                    url = message.message.text
+                    transfer_method = FileTransferMethod.TOOL_FILE
 
-                datasource_file_id = str(url).split("/")[-1].split(".")[0]
+                    datasource_file_id = str(url).split("/")[-1].split(".")[0]
 
-                with Session(db.engine) as session:
-                    stmt = select(ToolFile).where(ToolFile.id == datasource_file_id)
-                    datasource_file = session.scalar(stmt)
-                    if datasource_file is None:
-                        raise ToolFileError(f"Tool file {datasource_file_id} does not exist")
+                    with Session(db.engine) as session:
+                        stmt = select(ToolFile).where(ToolFile.id == datasource_file_id)
+                        datasource_file = session.scalar(stmt)
+                        if datasource_file is None:
+                            raise ToolFileError(f"Tool file {datasource_file_id} does not exist")
 
-                mapping = {
-                    "tool_file_id": datasource_file_id,
-                    "type": file_factory.get_file_type_by_mime_type(datasource_file.mimetype),
-                    "transfer_method": transfer_method,
-                    "url": url,
-                }
-                file = file_factory.build_from_mapping(
-                    mapping=mapping,
-                    tenant_id=self.tenant_id,
-                )
-                files.append(file)
-            elif message.type == DatasourceMessage.MessageType.BLOB:
-                # get tool file id
-                assert isinstance(message.message, DatasourceMessage.TextMessage)
-                assert message.meta
-
-                datasource_file_id = message.message.text.split("/")[-1].split(".")[0]
-                with Session(db.engine) as session:
-                    stmt = select(ToolFile).where(ToolFile.id == datasource_file_id)
-                    datasource_file = session.scalar(stmt)
-                    if datasource_file is None:
-                        raise ToolFileError(f"datasource file {datasource_file_id} not exists")
-
-                mapping = {
-                    "tool_file_id": datasource_file_id,
-                    "transfer_method": FileTransferMethod.TOOL_FILE,
-                }
-
-                files.append(
-                    file_factory.build_from_mapping(
+                    mapping = {
+                        "tool_file_id": datasource_file_id,
+                        "type": file_factory.get_file_type_by_mime_type(datasource_file.mimetype),
+                        "transfer_method": transfer_method,
+                        "url": url,
+                    }
+                    file = file_factory.build_from_mapping(
                         mapping=mapping,
                         tenant_id=self.tenant_id,
                     )
-                )
-            elif message.type == DatasourceMessage.MessageType.TEXT:
-                assert isinstance(message.message, DatasourceMessage.TextMessage)
-                text += message.message.text
-                yield StreamChunkEvent(
-                    selector=[self._node_id, "text"],
-                    chunk=message.message.text,
-                    is_final=False,
-                )
-            elif message.type == DatasourceMessage.MessageType.JSON:
-                assert isinstance(message.message, DatasourceMessage.JsonMessage)
-                json.append(message.message.json_object)
-            elif message.type == DatasourceMessage.MessageType.LINK:
-                assert isinstance(message.message, DatasourceMessage.TextMessage)
-                stream_text = f"Link: {message.message.text}\n"
-                text += stream_text
-                yield StreamChunkEvent(
-                    selector=[self._node_id, "text"],
-                    chunk=stream_text,
-                    is_final=False,
-                )
-            elif message.type == DatasourceMessage.MessageType.VARIABLE:
-                assert isinstance(message.message, DatasourceMessage.VariableMessage)
-                variable_name = message.message.variable_name
-                variable_value = message.message.variable_value
-                if message.message.stream:
-                    if not isinstance(variable_value, str):
-                        raise ValueError("When 'stream' is True, 'variable_value' must be a string.")
-                    if variable_name not in variables:
-                        variables[variable_name] = ""
-                    variables[variable_name] += variable_value
+                    files.append(file)
+                case DatasourceMessage.MessageType.BLOB:
+                    # get tool file id
+                    assert isinstance(message.message, DatasourceMessage.TextMessage)
+                    assert message.meta
 
+                    datasource_file_id = message.message.text.split("/")[-1].split(".")[0]
+                    with Session(db.engine) as session:
+                        stmt = select(ToolFile).where(ToolFile.id == datasource_file_id)
+                        datasource_file = session.scalar(stmt)
+                        if datasource_file is None:
+                            raise ToolFileError(f"datasource file {datasource_file_id} not exists")
+
+                    mapping = {
+                        "tool_file_id": datasource_file_id,
+                        "transfer_method": FileTransferMethod.TOOL_FILE,
+                    }
+
+                    files.append(
+                        file_factory.build_from_mapping(
+                            mapping=mapping,
+                            tenant_id=self.tenant_id,
+                        )
+                    )
+                case DatasourceMessage.MessageType.TEXT:
+                    assert isinstance(message.message, DatasourceMessage.TextMessage)
+                    text += message.message.text
                     yield StreamChunkEvent(
-                        selector=[self._node_id, variable_name],
-                        chunk=variable_value,
+                        selector=[self._node_id, "text"],
+                        chunk=message.message.text,
                         is_final=False,
                     )
-                else:
-                    variables[variable_name] = variable_value
-            elif message.type == DatasourceMessage.MessageType.FILE:
-                assert message.meta is not None
-                files.append(message.meta["file"])
+                case DatasourceMessage.MessageType.JSON:
+                    assert isinstance(message.message, DatasourceMessage.JsonMessage)
+                    json.append(message.message.json_object)
+                case DatasourceMessage.MessageType.LINK:
+                    assert isinstance(message.message, DatasourceMessage.TextMessage)
+                    stream_text = f"Link: {message.message.text}\n"
+                    text += stream_text
+                    yield StreamChunkEvent(
+                        selector=[self._node_id, "text"],
+                        chunk=stream_text,
+                        is_final=False,
+                    )
+                case DatasourceMessage.MessageType.VARIABLE:
+                    assert isinstance(message.message, DatasourceMessage.VariableMessage)
+                    variable_name = message.message.variable_name
+                    variable_value = message.message.variable_value
+                    if message.message.stream:
+                        if not isinstance(variable_value, str):
+                            raise ValueError("When 'stream' is True, 'variable_value' must be a string.")
+                        if variable_name not in variables:
+                            variables[variable_name] = ""
+                        variables[variable_name] += variable_value
+
+                        yield StreamChunkEvent(
+                            selector=[self._node_id, variable_name],
+                            chunk=variable_value,
+                            is_final=False,
+                        )
+                    else:
+                        variables[variable_name] = variable_value
+                case DatasourceMessage.MessageType.FILE:
+                    assert message.meta is not None
+                    files.append(message.meta["file"])
+                case (
+                    DatasourceMessage.MessageType.BLOB_CHUNK
+                    | DatasourceMessage.MessageType.LOG
+                    | DatasourceMessage.MessageType.RETRIEVER_RESOURCES
+                ):
+                    pass
+
         # mark the end of the stream
         yield StreamChunkEvent(
             selector=[self._node_id, "text"],