fix(web): confirm app deletion with enter key

Co-authored-by: claude[bot] <41898282+claude[bot]@users.noreply.github.com>

.github/workflows/api-tests.yml

@@ -98,8 +98,8 @@ jobs:
 
       - name: Set up dotenvs
         run: |
-          ./docker/init-env.sh
-          cp docker/middleware.env.example docker/middleware.env
+          cp docker/.env.example docker/.env
+          cp docker/envs/middleware.env.example docker/middleware.env
 
       - name: Expose Service Ports
         run: sh .github/workflows/expose_service_ports.sh

.github/workflows/autofix.yml

@@ -116,6 +116,12 @@ jobs:
         if: github.event_name != 'merge_group'
         uses: ./.github/actions/setup-web
 
+      - name: Generate API docs
+        if: github.event_name != 'merge_group' && steps.api-changes.outputs.any_changed == 'true'
+        run: |
+          cd api
+          uv run dev/generate_swagger_markdown_docs.py --swagger-dir openapi --markdown-dir openapi/markdown
+
       - name: ESLint autofix
         if: github.event_name != 'merge_group' && steps.web-changes.outputs.any_changed == 'true'
         run: |

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

@@ -37,7 +37,7 @@ jobs:
       - name: Prepare middleware env
         run: |
           cd docker
-          cp middleware.env.example middleware.env
+          cp envs/middleware.env.example middleware.env
 
       - name: Set up Middlewares
         uses: hoverkraft-tech/compose-action@d2bee4f07e8ca410d6b196d00f90c12e7d48c33a # v2.6.0
@@ -87,7 +87,7 @@ jobs:
       - name: Prepare middleware env for MySQL
         run: |
           cd docker
-          cp middleware.env.example middleware.env
+          cp envs/middleware.env.example middleware.env
           sed -i 's/DB_TYPE=postgresql/DB_TYPE=mysql/' middleware.env
           sed -i 's/DB_HOST=db_postgres/DB_HOST=db_mysql/' middleware.env
           sed -i 's/DB_PORT=5432/DB_PORT=3306/' middleware.env

.github/workflows/main-ci.yml

@@ -56,10 +56,8 @@ jobs:
               - 'api/**'
               - '.github/workflows/api-tests.yml'
               - '.github/workflows/expose_service_ports.sh'
-              - 'docker/.env.all'
               - 'docker/.env.example'
-              - 'docker/init-env.sh'
-              - 'docker/middleware.env.example'
+              - 'docker/envs/middleware.env.example'
               - 'docker/docker-compose.middleware.yaml'
               - 'docker/docker-compose-template.yaml'
               - 'docker/generate_docker_compose'
@@ -86,7 +84,7 @@ jobs:
               - 'pnpm-workspace.yaml'
               - '.nvmrc'
               - 'docker/docker-compose.middleware.yaml'
-              - 'docker/middleware.env.example'
+              - 'docker/envs/middleware.env.example'
               - '.github/workflows/web-e2e.yml'
               - '.github/actions/setup-web/**'
             vdb:
@@ -95,10 +93,8 @@ jobs:
               - 'api/providers/vdb/*/tests/**'
               - '.github/workflows/vdb-tests.yml'
               - '.github/workflows/expose_service_ports.sh'
-              - 'docker/.env.all'
               - 'docker/.env.example'
-              - 'docker/init-env.sh'
-              - 'docker/middleware.env.example'
+              - 'docker/envs/middleware.env.example'
               - 'docker/docker-compose.yaml'
               - 'docker/docker-compose-template.yaml'
               - 'docker/generate_docker_compose'
@@ -120,7 +116,7 @@ jobs:
               - '.github/workflows/db-migration-test.yml'
               - '.github/workflows/expose_service_ports.sh'
               - 'docker/.env.example'
-              - 'docker/middleware.env.example'
+              - 'docker/envs/middleware.env.example'
               - 'docker/docker-compose.middleware.yaml'
               - 'docker/docker-compose-template.yaml'
               - 'docker/generate_docker_compose'

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

@@ -77,10 +77,28 @@ jobs:
             }
 
             if (diff.trim()) {
-              await github.rest.issues.createComment({
+              const body = '### Pyrefly Diff\n<details>\n<summary>base → PR</summary>\n\n```diff\n' + diff + '\n```\n</details>';
+              const marker = '### Pyrefly Diff';
+              const { data: comments } = await github.rest.issues.listComments({
                 issue_number: prNumber,
                 owner: context.repo.owner,
                 repo: context.repo.repo,
-                body: '### Pyrefly Diff\n<details>\n<summary>base → PR</summary>\n\n```diff\n' + diff + '\n```\n</details>',
               });
+              const existing = comments.find((comment) => comment.body.startsWith(marker));
+
+              if (existing) {
+                await github.rest.issues.updateComment({
+                  comment_id: existing.id,
+                  owner: context.repo.owner,
+                  repo: context.repo.repo,
+                  body,
+                });
+              } else {
+                await github.rest.issues.createComment({
+                  issue_number: prNumber,
+                  owner: context.repo.owner,
+                  repo: context.repo.repo,
+                  body,
+                });
+              }
             }

.github/workflows/pyrefly-diff.yml

@@ -103,9 +103,26 @@ jobs:
                 ].join('\n')
               : '### Pyrefly Diff\nNo changes detected.';
 
-            await github.rest.issues.createComment({
+            const marker = '### Pyrefly Diff';
+            const { data: comments } = await github.rest.issues.listComments({
               issue_number: prNumber,
               owner: context.repo.owner,
               repo: context.repo.repo,
-              body,
             });
+            const existing = comments.find((comment) => comment.body.startsWith(marker));
+
+            if (existing) {
+              await github.rest.issues.updateComment({
+                comment_id: existing.id,
+                owner: context.repo.owner,
+                repo: context.repo.repo,
+                body,
+              });
+            } else {
+              await github.rest.issues.createComment({
+                issue_number: prNumber,
+                owner: context.repo.owner,
+                repo: context.repo.repo,
+                body,
+              });
+            }

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

@@ -50,8 +50,8 @@ jobs:
 
       - name: Set up dotenvs
         run: |
-          ./docker/init-env.sh
-          cp docker/middleware.env.example docker/middleware.env
+          cp docker/.env.example docker/.env
+          cp docker/envs/middleware.env.example docker/middleware.env
 
       - name: Expose Service Ports
         run: sh .github/workflows/expose_service_ports.sh

.github/workflows/vdb-tests.yml

@@ -47,8 +47,8 @@ jobs:
 
       - name: Set up dotenvs
         run: |
-          ./docker/init-env.sh
-          cp docker/middleware.env.example docker/middleware.env
+          cp docker/.env.example docker/.env
+          cp docker/envs/middleware.env.example docker/middleware.env
 
       - name: Expose Service Ports
         run: sh .github/workflows/expose_service_ports.sh

Makefile

@@ -71,13 +71,13 @@ type-check:
 	@echo "📝 Running type checks (basedpyright + pyrefly + mypy)..."
 	@./dev/basedpyright-check $(PATH_TO_CHECK)
 	@./dev/pyrefly-check-local
-	@uv --directory api run mypy --exclude-gitignore --exclude 'tests/' --exclude 'migrations/' --check-untyped-defs --disable-error-code=import-untyped .
+	@uv --directory api run mypy --exclude-gitignore --exclude 'tests/' --exclude 'migrations/' --exclude 'dev/generate_swagger_specs.py' --check-untyped-defs --disable-error-code=import-untyped .
 	@echo "✅ Type checks complete"
 
 type-check-core:
 	@echo "📝 Running core type checks (basedpyright + mypy)..."
 	@./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 .
+	@uv --directory api run mypy --exclude-gitignore --exclude 'tests/' --exclude 'migrations/' --exclude 'dev/generate_swagger_specs.py'  --exclude 'dev/generate_fastopenapi_specs.py' --check-untyped-defs --disable-error-code=import-untyped .
 	@echo "✅ Core type checks complete"
 
 test:

README.md

@@ -76,14 +76,7 @@ The easiest way to start the Dify server is through [Docker Compose](docker/dock
 ```bash
 cd dify
 cd docker
-./init-env.sh
-docker compose up -d
-```
-
-On Windows PowerShell, initialize `.env`, then run `docker compose up -d` from the `docker` directory.
-
-```powershell
-.\init-env.ps1
+cp .env.example .env
 docker compose up -d
 ```
 
@@ -144,7 +137,7 @@ Star Dify on GitHub and be instantly notified of new releases.
 
 ### Custom configurations
 
-If you need to customize the configuration, edit `docker/.env` after running the initialization script. The full reference remains in [`docker/.env.all`](docker/.env.all). After making any changes, re-run `docker compose up -d` from the `docker` directory. You can find the full list of available environment variables [here](https://docs.dify.ai/getting-started/install-self-hosted/environments).
+If you need to customize the configuration, edit `docker/.env`. The essential startup defaults live in [`docker/.env.example`](docker/.env.example), and optional advanced variables are split under `docker/envs/` by theme. After making any changes, re-run `docker compose up -d` from the `docker` directory. You can find the full list of available environment variables [here](https://docs.dify.ai/getting-started/install-self-hosted/environments).
 
 ### Metrics Monitoring with Grafana
 

api/.env.example

@@ -98,6 +98,8 @@ DB_DATABASE=dify
 
 SQLALCHEMY_POOL_PRE_PING=true
 SQLALCHEMY_POOL_TIMEOUT=30
+# Connection pool reset behavior on return
+SQLALCHEMY_POOL_RESET_ON_RETURN=rollback
 
 # Storage configuration
 # use for store upload files, private keys...
@@ -381,7 +383,7 @@ VIKINGDB_ACCESS_KEY=your-ak
 VIKINGDB_SECRET_KEY=your-sk
 VIKINGDB_REGION=cn-shanghai
 VIKINGDB_HOST=api-vikingdb.xxx.volces.com
-VIKINGDB_SCHEMA=http
+VIKINGDB_SCHEME=http
 VIKINGDB_CONNECTION_TIMEOUT=30
 VIKINGDB_SOCKET_TIMEOUT=30
 
@@ -432,8 +434,6 @@ UPLOAD_FILE_EXTENSION_BLACKLIST=
 
 # Model configuration
 MULTIMODAL_SEND_FORMAT=base64
-PROMPT_GENERATION_MAX_TOKENS=512
-CODE_GENERATION_MAX_TOKENS=1024
 PLUGIN_BASED_TOKEN_COUNTING_ENABLED=false
 
 # Mail configuration, support: resend, smtp, sendgrid

api/AGENTS.md

@@ -193,6 +193,10 @@ Before opening a PR / submitting:
 - Controllers: parse input via Pydantic, invoke services, return serialised responses; no business logic.
 - Services: coordinate repositories, providers, background tasks; keep side effects explicit.
 - Document non-obvious behaviour with concise docstrings and comments.
+- For Flask-RESTX controller request, query, and response schemas, follow `controllers/API_SCHEMA_GUIDE.md`.
+  In short: use Pydantic models, document GET query params with `query_params_from_model(...)`, register response
+  DTOs with `register_response_schema_models(...)`, serialize with `ResponseModel.model_validate(...).model_dump(...)`,
+  and avoid adding new legacy `ns.model(...)`, `@marshal_with(...)`, or GET `@ns.expect(...)` patterns.
 
 ### Miscellaneous
 

api/configs/middleware/__init__.py

@@ -114,7 +114,7 @@ class SQLAlchemyEngineOptionsDict(TypedDict):
     pool_pre_ping: bool
     connect_args: dict[str, str]
     pool_use_lifo: bool
-    pool_reset_on_return: None
+    pool_reset_on_return: Literal["commit", "rollback", None]
     pool_timeout: int
 
 
@@ -223,6 +223,11 @@ class DatabaseConfig(BaseSettings):
         default=30,
     )
 
+    SQLALCHEMY_POOL_RESET_ON_RETURN: Literal["commit", "rollback", None] = Field(
+        description="Connection pool reset behavior on return. Options: 'commit', 'rollback', or None",
+        default="rollback",
+    )
+
     RETRIEVAL_SERVICE_EXECUTORS: NonNegativeInt = Field(
         description="Number of processes for the retrieval service, default to CPU cores.",
         default=os.cpu_count() or 1,
@@ -252,7 +257,7 @@ class DatabaseConfig(BaseSettings):
             "pool_pre_ping": self.SQLALCHEMY_POOL_PRE_PING,
             "connect_args": connect_args,
             "pool_use_lifo": self.SQLALCHEMY_POOL_USE_LIFO,
-            "pool_reset_on_return": None,
+            "pool_reset_on_return": self.SQLALCHEMY_POOL_RESET_ON_RETURN,
             "pool_timeout": self.SQLALCHEMY_POOL_TIMEOUT,
         }
         return result

api/controllers/API_SCHEMA_GUIDE.md

@@ -0,0 +1,193 @@
+# API Schema Guide
+
+This guide describes the expected Flask-RESTX + Pydantic pattern for controller request payloads, query
+parameters, response schemas, and Swagger documentation.
+
+## Principles
+
+- Use Pydantic `BaseModel` for request bodies and query parameters.
+- Use `fields.base.ResponseModel` for response DTOs.
+- Keep runtime validation and Swagger documentation wired to the same Pydantic model.
+- Prefer explicit validation and serialization in controller methods over Flask-RESTX marshalling.
+- Do not add new Flask-RESTX `fields.*` dictionaries, `Namespace.model(...)` exports, or `@marshal_with(...)` for migrated or new endpoints.
+- Do not use `@ns.expect(...)` for GET query parameters. Flask-RESTX documents that as a request body.
+
+## Naming
+
+- Request body models: use a `Payload` suffix.
+  - Example: `WorkflowRunPayload`, `DatasourceVariablesPayload`.
+- Query parameter models: use a `Query` suffix.
+  - Example: `WorkflowRunListQuery`, `MessageListQuery`.
+- Response models: use a `Response` suffix and inherit from `ResponseModel`.
+  - Example: `WorkflowRunDetailResponse`, `WorkflowRunNodeExecutionListResponse`.
+- Use `ListResponse` or `PaginationResponse` for wrapper responses.
+  - Example: `WorkflowRunNodeExecutionListResponse`, `WorkflowRunPaginationResponse`.
+- Keep these models near the controller when they are endpoint-specific. Move them to `fields/*_fields.py` only when shared by multiple controllers.
+
+## Registering Models For Swagger
+
+Use helpers from `controllers.common.schema`.
+
+```python
+from controllers.common.schema import (
+    query_params_from_model,
+    register_response_schema_models,
+    register_schema_models,
+)
+```
+
+Register request payload and query models with `register_schema_models(...)`:
+
+```python
+register_schema_models(
+    console_ns,
+    WorkflowRunPayload,
+    WorkflowRunListQuery,
+)
+```
+
+Register response models with `register_response_schema_models(...)`:
+
+```python
+register_response_schema_models(
+    console_ns,
+    WorkflowRunDetailResponse,
+    WorkflowRunPaginationResponse,
+)
+```
+
+Response models are registered in Pydantic serialization mode. This matters when a response model uses
+`validation_alias` to read internal object attributes but emits public API field names. For example, a response model
+can validate from `inputs_dict` while documenting and serializing `inputs`.
+
+## Request Bodies
+
+For non-GET request bodies:
+
+1. Define a Pydantic `Payload` model.
+2. Register it with `register_schema_models(...)`.
+3. Use `@ns.expect(ns.models[Payload.__name__])` for Swagger documentation.
+4. Validate from `ns.payload or {}` inside the controller.
+
+```python
+class DraftWorkflowNodeRunPayload(BaseModel):
+    inputs: dict[str, Any]
+    query: str = ""
+
+
+register_schema_models(console_ns, DraftWorkflowNodeRunPayload)
+
+
+@console_ns.expect(console_ns.models[DraftWorkflowNodeRunPayload.__name__])
+def post(self, app_model: App, node_id: str):
+    payload = DraftWorkflowNodeRunPayload.model_validate(console_ns.payload or {})
+    result = service.run(..., inputs=payload.inputs, query=payload.query)
+    return WorkflowRunNodeExecutionResponse.model_validate(result, from_attributes=True).model_dump(mode="json")
+```
+
+## Query Parameters
+
+For GET query parameters:
+
+1. Define a Pydantic `Query` model.
+2. Register it with `register_schema_models(...)` if it is referenced elsewhere in docs, or only use
+   `query_params_from_model(...)` if a body schema is not needed.
+3. Use `@ns.doc(params=query_params_from_model(QueryModel))`.
+4. Validate from `request.args.to_dict(flat=True)` or an explicit dict when type coercion is needed.
+
+```python
+class WorkflowRunListQuery(BaseModel):
+    last_id: str | None = Field(default=None, description="Last run ID for pagination")
+    limit: int = Field(default=20, ge=1, le=100, description="Number of items per page (1-100)")
+
+
+@console_ns.doc(params=query_params_from_model(WorkflowRunListQuery))
+def get(self, app_model: App):
+    query = WorkflowRunListQuery.model_validate(request.args.to_dict(flat=True))
+    result = service.list(..., limit=query.limit, last_id=query.last_id)
+    return WorkflowRunPaginationResponse.model_validate(result, from_attributes=True).model_dump(mode="json")
+```
+
+Do not do this for GET query parameters:
+
+```python
+@console_ns.expect(console_ns.models[WorkflowRunListQuery.__name__])
+def get(...):
+    ...
+```
+
+That documents a GET request body and is not the expected contract.
+
+## Responses
+
+Response models should inherit from `ResponseModel`:
+
+```python
+class WorkflowRunNodeExecutionResponse(ResponseModel):
+    id: str
+    inputs: Any = Field(default=None, validation_alias="inputs_dict")
+    process_data: Any = Field(default=None, validation_alias="process_data_dict")
+    outputs: Any = Field(default=None, validation_alias="outputs_dict")
+```
+
+Document response models with `@ns.response(...)`:
+
+```python
+@console_ns.response(
+    200,
+    "Node run started successfully",
+    console_ns.models[WorkflowRunNodeExecutionResponse.__name__],
+)
+def post(...):
+    ...
+```
+
+Serialize explicitly:
+
+```python
+return WorkflowRunNodeExecutionResponse.model_validate(
+    workflow_node_execution,
+    from_attributes=True,
+).model_dump(mode="json")
+```
+
+If the service can return `None`, translate that into the expected HTTP error before validation:
+
+```python
+workflow_run = service.get_workflow_run(...)
+if workflow_run is None:
+    raise NotFound("Workflow run not found")
+
+return WorkflowRunDetailResponse.model_validate(workflow_run, from_attributes=True).model_dump(mode="json")
+```
+
+## Legacy Flask-RESTX Patterns
+
+Avoid adding these patterns to new or migrated endpoints:
+
+- `ns.model(...)` for new request/response DTOs.
+- Module-level exported RESTX model objects such as `workflow_run_detail_model`.
+- `fields.Nested({...})` with raw inline dict field maps.
+- `@marshal_with(...)` for response serialization.
+- `@ns.expect(...)` for GET query params.
+
+Existing legacy field dictionaries may remain where an endpoint has not yet been migrated. Keep that compatibility local
+to the legacy area and avoid importing RESTX model objects from controllers.
+
+## Verifying Swagger
+
+For schema and documentation changes, run focused tests and generate Swagger JSON:
+
+```bash
+uv run --project . pytest tests/unit_tests/controllers/common/test_schema.py
+uv run --project . pytest tests/unit_tests/commands/test_generate_swagger_specs.py tests/unit_tests/controllers/test_swagger.py
+uv run --project . dev/generate_swagger_specs.py --output-dir /tmp/dify-openapi-check
+```
+
+Inspect affected endpoints with `jq`. Check that:
+
+- GET parameters are `in: query`.
+- Request bodies appear only where the endpoint has a body.
+- Responses reference the expected `*Response` schema.
+- Response schemas use public serialized names, not internal validation aliases like `inputs_dict`.
+

api/controllers/common/schema.py

@@ -1,6 +1,14 @@
-"""Helpers for registering Pydantic models with Flask-RESTX namespaces."""
+"""Helpers for registering Pydantic models with Flask-RESTX namespaces.
 
+Flask-RESTX treats `SchemaModel` bodies as opaque JSON schemas; it does not
+promote Pydantic's nested `$defs` into top-level Swagger `definitions`.
+These helpers keep that translation centralized so models registered through
+`register_schema_models` emit resolvable Swagger 2.0 references.
+"""
+
+from collections.abc import Mapping
 from enum import StrEnum
+from typing import Any, Literal, NotRequired, TypedDict
 
 from flask_restx import Namespace
 from pydantic import BaseModel, TypeAdapter
@@ -8,10 +16,59 @@ from pydantic import BaseModel, TypeAdapter
 DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
 
-def register_schema_model(namespace: Namespace, model: type[BaseModel]) -> None:
-    """Register a single BaseModel with a namespace for Swagger documentation."""
+QueryParamDoc = TypedDict(
+    "QueryParamDoc",
+    {
+        "in": NotRequired[str],
+        "type": NotRequired[str],
+        "items": NotRequired[dict[str, object]],
+        "required": NotRequired[bool],
+        "description": NotRequired[str],
+        "enum": NotRequired[list[object]],
+        "default": NotRequired[object],
+        "minimum": NotRequired[int | float],
+        "maximum": NotRequired[int | float],
+        "minLength": NotRequired[int],
+        "maxLength": NotRequired[int],
+        "minItems": NotRequired[int],
+        "maxItems": NotRequired[int],
+    },
+)
 
-    namespace.schema_model(model.__name__, model.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
+
+def _register_json_schema(namespace: Namespace, name: str, schema: dict) -> None:
+    """Register a JSON schema and promote any nested Pydantic `$defs`."""
+
+    nested_definitions = schema.get("$defs")
+    schema_to_register = dict(schema)
+    if isinstance(nested_definitions, dict):
+        schema_to_register.pop("$defs")
+
+    namespace.schema_model(name, schema_to_register)
+
+    if not isinstance(nested_definitions, dict):
+        return
+
+    for nested_name, nested_schema in nested_definitions.items():
+        if isinstance(nested_schema, dict):
+            _register_json_schema(namespace, nested_name, nested_schema)
+
+
+JsonSchemaMode = Literal["validation", "serialization"]
+
+
+def _register_schema_model(namespace: Namespace, model: type[BaseModel], *, mode: JsonSchemaMode) -> None:
+    _register_json_schema(
+        namespace,
+        model.__name__,
+        model.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0, mode=mode),
+    )
+
+
+def register_schema_model(namespace: Namespace, model: type[BaseModel]) -> None:
+    """Register a BaseModel and its nested schema definitions for Swagger documentation."""
+
+    _register_schema_model(namespace, model, mode="validation")
 
 
 def register_schema_models(namespace: Namespace, *models: type[BaseModel]) -> None:
@@ -21,6 +78,19 @@ def register_schema_models(namespace: Namespace, *models: type[BaseModel]) -> No
         register_schema_model(namespace, model)
 
 
+def register_response_schema_model(namespace: Namespace, model: type[BaseModel]) -> None:
+    """Register a BaseModel using its serialized response shape."""
+
+    _register_schema_model(namespace, model, mode="serialization")
+
+
+def register_response_schema_models(namespace: Namespace, *models: type[BaseModel]) -> None:
+    """Register multiple response BaseModels using their serialized response shape."""
+
+    for model in models:
+        register_response_schema_model(namespace, model)
+
+
 def get_or_create_model(model_name: str, field_def):
     # Import lazily to avoid circular imports between console controllers and schema helpers.
     from controllers.console import console_ns
@@ -34,15 +104,114 @@ def get_or_create_model(model_name: str, field_def):
 def register_enum_models(namespace: Namespace, *models: type[StrEnum]) -> None:
     """Register multiple StrEnum with a namespace."""
     for model in models:
-        namespace.schema_model(
-            model.__name__, TypeAdapter(model).json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+        _register_json_schema(
+            namespace,
+            model.__name__,
+            TypeAdapter(model).json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
         )
 
 
+def query_params_from_model(model: type[BaseModel]) -> dict[str, QueryParamDoc]:
+    """Build Flask-RESTX query parameter docs from a flat Pydantic model.
+
+    `Namespace.expect()` treats Pydantic schema models as request bodies, so GET
+    endpoints should keep runtime validation on the Pydantic model and feed this
+    derived mapping to `Namespace.doc(params=...)` for Swagger documentation.
+    """
+
+    schema = model.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+    properties = schema.get("properties", {})
+    if not isinstance(properties, Mapping):
+        return {}
+
+    required = schema.get("required", [])
+    required_names = set(required) if isinstance(required, list) else set()
+
+    params: dict[str, QueryParamDoc] = {}
+    for name, property_schema in properties.items():
+        if not isinstance(name, str) or not isinstance(property_schema, Mapping):
+            continue
+
+        params[name] = _query_param_from_property(property_schema, required=name in required_names)
+
+    return params
+
+
+def _query_param_from_property(property_schema: Mapping[str, Any], *, required: bool) -> QueryParamDoc:
+    param_schema = _nullable_property_schema(property_schema)
+    param_doc: QueryParamDoc = {"in": "query", "required": required}
+
+    description = param_schema.get("description")
+    if isinstance(description, str):
+        param_doc["description"] = description
+
+    schema_type = param_schema.get("type")
+    if isinstance(schema_type, str) and schema_type in {"array", "boolean", "integer", "number", "string"}:
+        param_doc["type"] = schema_type
+        if schema_type == "array":
+            items = param_schema.get("items")
+            if isinstance(items, Mapping):
+                item_type = items.get("type")
+                if isinstance(item_type, str):
+                    param_doc["items"] = {"type": item_type}
+
+    enum = param_schema.get("enum")
+    if isinstance(enum, list):
+        param_doc["enum"] = enum
+
+    default = param_schema.get("default")
+    if default is not None:
+        param_doc["default"] = default
+
+    minimum = param_schema.get("minimum")
+    if isinstance(minimum, int | float):
+        param_doc["minimum"] = minimum
+
+    maximum = param_schema.get("maximum")
+    if isinstance(maximum, int | float):
+        param_doc["maximum"] = maximum
+
+    min_length = param_schema.get("minLength")
+    if isinstance(min_length, int):
+        param_doc["minLength"] = min_length
+
+    max_length = param_schema.get("maxLength")
+    if isinstance(max_length, int):
+        param_doc["maxLength"] = max_length
+
+    min_items = param_schema.get("minItems")
+    if isinstance(min_items, int):
+        param_doc["minItems"] = min_items
+
+    max_items = param_schema.get("maxItems")
+    if isinstance(max_items, int):
+        param_doc["maxItems"] = max_items
+
+    return param_doc
+
+
+def _nullable_property_schema(property_schema: Mapping[str, Any]) -> Mapping[str, Any]:
+    any_of = property_schema.get("anyOf")
+    if not isinstance(any_of, list):
+        return property_schema
+
+    non_null_candidates = [
+        candidate for candidate in any_of if isinstance(candidate, Mapping) and candidate.get("type") != "null"
+    ]
+
+    if len(non_null_candidates) == 1:
+        return {**property_schema, **non_null_candidates[0]}
+
+    return property_schema
+
+
 __all__ = [
     "DEFAULT_REF_TEMPLATE_SWAGGER_2_0",
     "get_or_create_model",
+    "query_params_from_model",
     "register_enum_models",
+    "register_response_schema_model",
+    "register_response_schema_models",
     "register_schema_model",
     "register_schema_models",
 ]

api/controllers/console/admin.py

@@ -3,6 +3,7 @@ import io
 from collections.abc import Callable
 from functools import wraps
 from typing import cast
+from uuid import UUID
 
 from flask import request
 from flask_restx import Resource
@@ -12,6 +13,7 @@ from werkzeug.exceptions import BadRequest, NotFound, Unauthorized
 
 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.wraps import only_edition_cloud
 from core.db.session_factory import session_factory
@@ -20,8 +22,6 @@ from libs.token import extract_access_token
 from models.model import App, ExporleBanner, InstalledApp, RecommendedApp, TrialApp
 from services.billing_service import BillingService, LangContentDict
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
 
 class InsertExploreAppPayload(BaseModel):
     app_id: str = Field(...)
@@ -58,15 +58,7 @@ class InsertExploreBannerPayload(BaseModel):
     model_config = {"populate_by_name": True}
 
 
-console_ns.schema_model(
-    InsertExploreAppPayload.__name__,
-    InsertExploreAppPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
-
-console_ns.schema_model(
-    InsertExploreBannerPayload.__name__,
-    InsertExploreBannerPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
+register_schema_models(console_ns, InsertExploreAppPayload, InsertExploreBannerPayload)
 
 
 def admin_required[**P, R](view: Callable[P, R]) -> Callable[P, R]:
@@ -190,7 +182,7 @@ class InsertExploreAppApi(Resource):
     @console_ns.response(204, "App removed successfully")
     @only_edition_cloud
     @admin_required
-    def delete(self, app_id):
+    def delete(self, app_id: UUID):
         with session_factory.create_session() as session:
             recommended_app = session.execute(
                 select(RecommendedApp).where(RecommendedApp.app_id == str(app_id))
@@ -301,15 +293,7 @@ class BatchAddNotificationAccountsPayload(BaseModel):
     user_email: list[str] = Field(..., description="List of account email addresses")
 
 
-console_ns.schema_model(
-    UpsertNotificationPayload.__name__,
-    UpsertNotificationPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
-
-console_ns.schema_model(
-    BatchAddNotificationAccountsPayload.__name__,
-    BatchAddNotificationAccountsPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
+register_schema_models(console_ns, UpsertNotificationPayload, BatchAddNotificationAccountsPayload)
 
 
 @console_ns.route("/admin/upsert_notification")
@@ -411,11 +395,11 @@ class BatchAddNotificationAccountsApi(Resource):
             raise BadRequest("Invalid file type. Only CSV (.csv) and TXT (.txt) files are allowed.")
 
         try:
-            content = file.read().decode("utf-8")
+            content = file.stream.read().decode("utf-8")
         except UnicodeDecodeError:
             try:
-                file.seek(0)
-                content = file.read().decode("gbk")
+                file.stream.seek(0)
+                content = file.stream.read().decode("gbk")
             except UnicodeDecodeError:
                 raise BadRequest("Unable to decode the file. Please use UTF-8 or GBK encoding.")
 

api/controllers/console/app/advanced_prompt_template.py

@@ -34,7 +34,7 @@ class AdvancedPromptTemplateList(Resource):
     @login_required
     @account_initialization_required
     def get(self):
-        args = AdvancedPromptTemplateQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = AdvancedPromptTemplateQuery.model_validate(request.args.to_dict(flat=True))
         prompt_args: AdvancedPromptTemplateArgs = {
             "app_mode": args.app_mode,
             "model_mode": args.model_mode,

api/controllers/console/app/agent.py

@@ -2,6 +2,7 @@ from flask import request
 from flask_restx import Resource, fields
 from pydantic import BaseModel, Field, field_validator
 
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
@@ -10,8 +11,6 @@ from libs.login import login_required
 from models.model import AppMode
 from services.agent_service import AgentService
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
 
 class AgentLogQuery(BaseModel):
     message_id: str = Field(..., description="Message UUID")
@@ -23,9 +22,7 @@ class AgentLogQuery(BaseModel):
         return uuid_value(value)
 
 
-console_ns.schema_model(
-    AgentLogQuery.__name__, AgentLogQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
-)
+register_schema_models(console_ns, AgentLogQuery)
 
 
 @console_ns.route("/apps/<uuid:app_id>/agent/logs")
@@ -44,6 +41,6 @@ class AgentLogApi(Resource):
     @get_app_model(mode=[AppMode.AGENT_CHAT])
     def get(self, app_model):
         """Get agent logs"""
-        args = AgentLogQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = AgentLogQuery.model_validate(request.args.to_dict(flat=True))
 
         return AgentService.get_agent_logs(app_model, args.conversation_id, args.message_id)

api/controllers/console/app/annotation.py

@@ -1,4 +1,5 @@
 from typing import Any, Literal
+from uuid import UUID
 
 from flask import abort, make_response, request
 from flask_restx import Resource
@@ -33,8 +34,6 @@ from services.annotation_service import (
     UpsertAnnotationArgs,
 )
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
 
 class AnnotationReplyPayload(BaseModel):
     score_threshold: float = Field(..., description="Score threshold for annotation matching")
@@ -87,17 +86,6 @@ class AnnotationFilePayload(BaseModel):
         return uuid_value(value)
 
 
-def reg(model: type[BaseModel]) -> None:
-    console_ns.schema_model(model.__name__, model.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
-
-
-reg(AnnotationReplyPayload)
-reg(AnnotationSettingUpdatePayload)
-reg(AnnotationListQuery)
-reg(CreateAnnotationPayload)
-reg(UpdateAnnotationPayload)
-reg(AnnotationReplyStatusQuery)
-reg(AnnotationFilePayload)
 register_schema_models(
     console_ns,
     Annotation,
@@ -105,6 +93,13 @@ register_schema_models(
     AnnotationExportList,
     AnnotationHitHistory,
     AnnotationHitHistoryList,
+    AnnotationReplyPayload,
+    AnnotationSettingUpdatePayload,
+    AnnotationListQuery,
+    CreateAnnotationPayload,
+    UpdateAnnotationPayload,
+    AnnotationReplyStatusQuery,
+    AnnotationFilePayload,
 )
 
 
@@ -121,8 +116,7 @@ class AnnotationReplyActionApi(Resource):
     @account_initialization_required
     @cloud_edition_billing_resource_check("annotation")
     @edit_permission_required
-    def post(self, app_id, action: Literal["enable", "disable"]):
-        app_id = str(app_id)
+    def post(self, app_id: UUID, action: Literal["enable", "disable"]):
         args = AnnotationReplyPayload.model_validate(console_ns.payload)
         match action:
             case "enable":
@@ -131,9 +125,9 @@ class AnnotationReplyActionApi(Resource):
                     "embedding_provider_name": args.embedding_provider_name,
                     "embedding_model_name": args.embedding_model_name,
                 }
-                result = AppAnnotationService.enable_app_annotation(enable_args, app_id)
+                result = AppAnnotationService.enable_app_annotation(enable_args, str(app_id))
             case "disable":
-                result = AppAnnotationService.disable_app_annotation(app_id)
+                result = AppAnnotationService.disable_app_annotation(str(app_id))
         return result, 200
 
 
@@ -148,9 +142,8 @@ class AppAnnotationSettingDetailApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
-    def get(self, app_id):
-        app_id = str(app_id)
-        result = AppAnnotationService.get_app_annotation_setting_by_app_id(app_id)
+    def get(self, app_id: UUID):
+        result = AppAnnotationService.get_app_annotation_setting_by_app_id(str(app_id))
         return result, 200
 
 
@@ -166,14 +159,13 @@ class AppAnnotationSettingUpdateApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
-    def post(self, app_id, annotation_setting_id):
-        app_id = str(app_id)
+    def post(self, app_id: UUID, annotation_setting_id):
         annotation_setting_id = str(annotation_setting_id)
 
         args = AnnotationSettingUpdatePayload.model_validate(console_ns.payload)
 
         setting_args: UpdateAnnotationSettingArgs = {"score_threshold": args.score_threshold}
-        result = AppAnnotationService.update_app_annotation_setting(app_id, annotation_setting_id, setting_args)
+        result = AppAnnotationService.update_app_annotation_setting(str(app_id), annotation_setting_id, setting_args)
         return result, 200
 
 
@@ -189,7 +181,7 @@ class AnnotationReplyActionStatusApi(Resource):
     @account_initialization_required
     @cloud_edition_billing_resource_check("annotation")
     @edit_permission_required
-    def get(self, app_id, job_id, action):
+    def get(self, app_id: UUID, job_id, action):
         job_id = str(job_id)
         app_annotation_job_key = f"{action}_app_annotation_job_{str(job_id)}"
         cache_result = redis_client.get(app_annotation_job_key)
@@ -217,14 +209,13 @@ class AnnotationApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
-    def get(self, app_id):
-        args = AnnotationListQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+    def get(self, app_id: UUID):
+        args = AnnotationListQuery.model_validate(request.args.to_dict(flat=True))
         page = args.page
         limit = args.limit
         keyword = args.keyword
 
-        app_id = str(app_id)
-        annotation_list, total = AppAnnotationService.get_annotation_list_by_app_id(app_id, page, limit, keyword)
+        annotation_list, total = AppAnnotationService.get_annotation_list_by_app_id(str(app_id), page, limit, keyword)
         annotation_models = TypeAdapter(list[Annotation]).validate_python(annotation_list, from_attributes=True)
         response = AnnotationList(
             data=annotation_models,
@@ -246,8 +237,7 @@ class AnnotationApi(Resource):
     @account_initialization_required
     @cloud_edition_billing_resource_check("annotation")
     @edit_permission_required
-    def post(self, app_id):
-        app_id = str(app_id)
+    def post(self, app_id: UUID):
         args = CreateAnnotationPayload.model_validate(console_ns.payload)
         upsert_args: UpsertAnnotationArgs = {}
         if args.answer is not None:
@@ -258,15 +248,14 @@ class AnnotationApi(Resource):
             upsert_args["message_id"] = args.message_id
         if args.question is not None:
             upsert_args["question"] = args.question
-        annotation = AppAnnotationService.up_insert_app_annotation_from_message(upsert_args, app_id)
+        annotation = AppAnnotationService.up_insert_app_annotation_from_message(upsert_args, str(app_id))
         return Annotation.model_validate(annotation, from_attributes=True).model_dump(mode="json")
 
     @setup_required
     @login_required
     @account_initialization_required
     @edit_permission_required
-    def delete(self, app_id):
-        app_id = str(app_id)
+    def delete(self, app_id: UUID):
 
         # Use request.args.getlist to get annotation_ids array directly
         annotation_ids = request.args.getlist("annotation_id")
@@ -280,11 +269,11 @@ class AnnotationApi(Resource):
                     "message": "annotation_ids are required if the parameter is provided.",
                 }, 400
 
-            result = AppAnnotationService.delete_app_annotations_in_batch(app_id, annotation_ids)
+            result = AppAnnotationService.delete_app_annotations_in_batch(str(app_id), annotation_ids)
             return result, 204
         # If no annotation_ids are provided, handle clearing all annotations
         else:
-            AppAnnotationService.clear_all_annotations(app_id)
+            AppAnnotationService.clear_all_annotations(str(app_id))
             return {"result": "success"}, 204
 
 
@@ -303,9 +292,8 @@ class AnnotationExportApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
-    def get(self, app_id):
-        app_id = str(app_id)
-        annotation_list = AppAnnotationService.export_annotation_list_by_app_id(app_id)
+    def get(self, app_id: UUID):
+        annotation_list = AppAnnotationService.export_annotation_list_by_app_id(str(app_id))
         annotation_models = TypeAdapter(list[Annotation]).validate_python(annotation_list, from_attributes=True)
         response_data = AnnotationExportList(data=annotation_models).model_dump(mode="json")
 
@@ -331,26 +319,22 @@ class AnnotationUpdateDeleteApi(Resource):
     @account_initialization_required
     @cloud_edition_billing_resource_check("annotation")
     @edit_permission_required
-    def post(self, app_id, annotation_id):
-        app_id = str(app_id)
-        annotation_id = str(annotation_id)
+    def post(self, app_id: UUID, annotation_id: UUID):
         args = UpdateAnnotationPayload.model_validate(console_ns.payload)
         update_args: UpdateAnnotationArgs = {}
         if args.answer is not None:
             update_args["answer"] = args.answer
         if args.question is not None:
             update_args["question"] = args.question
-        annotation = AppAnnotationService.update_app_annotation_directly(update_args, app_id, annotation_id)
+        annotation = AppAnnotationService.update_app_annotation_directly(update_args, str(app_id), str(annotation_id))
         return Annotation.model_validate(annotation, from_attributes=True).model_dump(mode="json")
 
     @setup_required
     @login_required
     @account_initialization_required
     @edit_permission_required
-    def delete(self, app_id, annotation_id):
-        app_id = str(app_id)
-        annotation_id = str(annotation_id)
-        AppAnnotationService.delete_app_annotation(app_id, annotation_id)
+    def delete(self, app_id: UUID, annotation_id: UUID):
+        AppAnnotationService.delete_app_annotation(str(app_id), str(annotation_id))
         return {"result": "success"}, 204
 
 
@@ -371,11 +355,9 @@ class AnnotationBatchImportApi(Resource):
     @annotation_import_rate_limit
     @annotation_import_concurrency_limit
     @edit_permission_required
-    def post(self, app_id):
+    def post(self, app_id: UUID):
         from configs import dify_config
 
-        app_id = str(app_id)
-
         # check file
         if "file" not in request.files:
             raise NoFileUploadedError()
@@ -391,9 +373,9 @@ class AnnotationBatchImportApi(Resource):
             raise ValueError("Invalid file type. Only CSV files are allowed")
 
         # Check file size before processing
-        file.seek(0, 2)  # Seek to end of file
-        file_size = file.tell()
-        file.seek(0)  # Reset to beginning
+        file.stream.seek(0, 2)  # Seek to end of file
+        file_size = file.stream.tell()
+        file.stream.seek(0)  # Reset to beginning
 
         max_size_bytes = dify_config.ANNOTATION_IMPORT_FILE_SIZE_LIMIT * 1024 * 1024
         if file_size > max_size_bytes:
@@ -406,7 +388,7 @@ class AnnotationBatchImportApi(Resource):
         if file_size == 0:
             raise ValueError("The uploaded file is empty")
 
-        return AppAnnotationService.batch_import_app_annotations(app_id, file)
+        return AppAnnotationService.batch_import_app_annotations(str(app_id), file)
 
 
 @console_ns.route("/apps/<uuid:app_id>/annotations/batch-import-status/<uuid:job_id>")
@@ -421,8 +403,7 @@ class AnnotationBatchImportStatusApi(Resource):
     @account_initialization_required
     @cloud_edition_billing_resource_check("annotation")
     @edit_permission_required
-    def get(self, app_id, job_id):
-        job_id = str(job_id)
+    def get(self, app_id: UUID, job_id: UUID):
         indexing_cache_key = f"app_annotation_batch_import_{str(job_id)}"
         cache_result = redis_client.get(indexing_cache_key)
         if cache_result is None:
@@ -456,13 +437,11 @@ class AnnotationHitHistoryListApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
-    def get(self, app_id, annotation_id):
+    def get(self, app_id: UUID, annotation_id: UUID):
         page = request.args.get("page", default=1, type=int)
         limit = request.args.get("limit", default=20, type=int)
-        app_id = str(app_id)
-        annotation_id = str(annotation_id)
         annotation_hit_history_list, total = AppAnnotationService.get_annotation_hit_histories(
-            app_id, annotation_id, page, limit
+            str(app_id), str(annotation_id), page, limit
         )
         history_models = TypeAdapter(list[AnnotationHitHistory]).validate_python(
             annotation_hit_history_list, from_attributes=True

api/controllers/console/app/app.py

@@ -3,6 +3,7 @@ import re
 import uuid
 from datetime import datetime
 from typing import Any, Literal
+from uuid import UUID
 
 from flask import request
 from flask_restx import Resource
@@ -25,6 +26,7 @@ from controllers.console.wraps import (
     is_admin_or_owner_required,
     setup_required,
 )
+from core.db.session_factory import session_factory
 from core.ops.ops_trace_manager import OpsTraceManager
 from core.rag.entities import PreProcessingRule, Rule, Segmentation
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
@@ -700,7 +702,7 @@ class AppExportApi(Resource):
     @edit_permission_required
     def get(self, app_model):
         """Export app"""
-        args = AppExportQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = AppExportQuery.model_validate(request.args.to_dict(flat=True))
 
         payload = AppExportResponse(
             data=AppDslService.export_dsl(
@@ -839,9 +841,10 @@ class AppTraceApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    def get(self, app_id):
+    def get(self, app_id: UUID):
         """Get app trace"""
-        app_trace_config = OpsTraceManager.get_app_tracing_config(app_id=app_id)
+        with session_factory.create_session() as session:
+            app_trace_config = OpsTraceManager.get_app_tracing_config(str(app_id), session)
 
         return app_trace_config
 
@@ -855,12 +858,12 @@ class AppTraceApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
-    def post(self, app_id):
+    def post(self, app_id: UUID):
         # add app trace
         args = AppTracePayload.model_validate(console_ns.payload)
 
         OpsTraceManager.update_app_tracing_config(
-            app_id=app_id,
+            app_id=str(app_id),
             enabled=args.enabled,
             tracing_provider=args.tracing_provider,
         )

api/controllers/console/app/app_import.py

@@ -2,7 +2,7 @@ from flask_restx import Resource
 from pydantic import BaseModel, Field
 from sqlalchemy.orm import Session
 
-from controllers.common.schema import register_schema_models
+from controllers.common.schema import register_enum_models, register_schema_models
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import (
     account_initialization_required,
@@ -33,6 +33,7 @@ class AppImportPayload(BaseModel):
     app_id: str | None = Field(None)
 
 
+register_enum_models(console_ns, ImportStatus)
 register_schema_models(console_ns, AppImportPayload, Import, CheckDependenciesResult)
 
 

api/controllers/console/app/audio.py

@@ -173,7 +173,7 @@ class TextModesApi(Resource):
     @account_initialization_required
     def get(self, app_model):
         try:
-            args = TextToSpeechVoiceQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+            args = TextToSpeechVoiceQuery.model_validate(request.args.to_dict(flat=True))
 
             response = AudioService.transcript_tts_voices(
                 tenant_id=app_model.tenant_id,

api/controllers/console/app/completion.py

@@ -7,6 +7,7 @@ from pydantic import BaseModel, Field, field_validator
 from werkzeug.exceptions import InternalServerError, NotFound
 
 import services
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.error import (
     AppUnavailableError,
@@ -37,7 +38,6 @@ from services.app_task_service import AppTaskService
 from services.errors.llm import InvokeRateLimitError
 
 logger = logging.getLogger(__name__)
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
 
 class BaseMessagePayload(BaseModel):
@@ -65,13 +65,7 @@ class ChatMessagePayload(BaseMessagePayload):
         return uuid_value(value)
 
 
-console_ns.schema_model(
-    CompletionMessagePayload.__name__,
-    CompletionMessagePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
-console_ns.schema_model(
-    ChatMessagePayload.__name__, ChatMessagePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
-)
+register_schema_models(console_ns, CompletionMessagePayload, ChatMessagePayload)
 
 
 # define completion message api for user

api/controllers/console/app/conversation.py

@@ -39,8 +39,6 @@ from models.model import AppMode
 from services.conversation_service import ConversationService
 from services.errors.conversation import ConversationNotExistsError
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
 
 class BaseConversationQuery(BaseModel):
     keyword: str | None = Field(default=None, description="Search keyword")
@@ -70,15 +68,6 @@ class ChatConversationQuery(BaseConversationQuery):
     )
 
 
-console_ns.schema_model(
-    CompletionConversationQuery.__name__,
-    CompletionConversationQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
-console_ns.schema_model(
-    ChatConversationQuery.__name__,
-    ChatConversationQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
-
 register_schema_models(
     console_ns,
     CompletionConversationQuery,
@@ -89,6 +78,8 @@ register_schema_models(
     ConversationWithSummaryPaginationResponse,
     ConversationDetailResponse,
     ResultResponse,
+    CompletionConversationQuery,
+    ChatConversationQuery,
 )
 
 
@@ -107,7 +98,7 @@ class CompletionConversationApi(Resource):
     @edit_permission_required
     def get(self, app_model):
         current_user, _ = current_account_with_tenant()
-        args = CompletionConversationQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = CompletionConversationQuery.model_validate(request.args.to_dict(flat=True))
 
         query = sa.select(Conversation).where(
             Conversation.app_id == app_model.id, Conversation.mode == "completion", Conversation.is_deleted.is_(False)
@@ -221,7 +212,7 @@ class ChatConversationApi(Resource):
     @edit_permission_required
     def get(self, app_model):
         current_user, _ = current_account_with_tenant()
-        args = ChatConversationQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = ChatConversationQuery.model_validate(request.args.to_dict(flat=True))
 
         subquery = (
             sa.select(Conversation.id.label("conversation_id"), EndUser.session_id.label("from_end_user_session_id"))

api/controllers/console/app/conversation_variables.py

@@ -100,7 +100,7 @@ class ConversationVariablesApi(Resource):
     @account_initialization_required
     @get_app_model(mode=AppMode.ADVANCED_CHAT)
     def get(self, app_model):
-        args = ConversationVariablesQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = ConversationVariablesQuery.model_validate(request.args.to_dict(flat=True))
 
         stmt = (
             select(ConversationVariable)

api/controllers/console/app/generator.py

@@ -3,6 +3,7 @@ from collections.abc import Sequence
 from flask_restx import Resource
 from pydantic import BaseModel, Field
 
+from controllers.common.schema import register_enum_models, register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.error import (
     CompletionRequestError,
@@ -19,13 +20,12 @@ from core.helper.code_executor.python3.python3_code_provider import Python3CodeP
 from core.llm_generator.entities import RuleCodeGeneratePayload, RuleGeneratePayload, RuleStructuredOutputPayload
 from core.llm_generator.llm_generator import LLMGenerator
 from extensions.ext_database import db
+from graphon.model_runtime.entities.llm_entities import LLMMode
 from graphon.model_runtime.errors.invoke import InvokeError
 from libs.login import current_account_with_tenant, login_required
 from models import App
 from services.workflow_service import WorkflowService
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
 
 class InstructionGeneratePayload(BaseModel):
     flow_id: str = Field(..., description="Workflow/Flow ID")
@@ -41,16 +41,16 @@ class InstructionTemplatePayload(BaseModel):
     type: str = Field(..., description="Instruction template type")
 
 
-def reg(cls: type[BaseModel]):
-    console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
-
-
-reg(RuleGeneratePayload)
-reg(RuleCodeGeneratePayload)
-reg(RuleStructuredOutputPayload)
-reg(InstructionGeneratePayload)
-reg(InstructionTemplatePayload)
-reg(ModelConfig)
+register_enum_models(console_ns, LLMMode)
+register_schema_models(
+    console_ns,
+    RuleGeneratePayload,
+    RuleCodeGeneratePayload,
+    RuleStructuredOutputPayload,
+    InstructionGeneratePayload,
+    InstructionTemplatePayload,
+    ModelConfig,
+)
 
 
 @console_ns.route("/rule-generate")

api/controllers/console/app/ops_trace.py

@@ -1,18 +1,18 @@
 from typing import Any
+from uuid import UUID
 
 from flask import request
 from flask_restx import Resource, fields
 from pydantic import BaseModel, Field
 from werkzeug.exceptions import BadRequest
 
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.error import TracingConfigCheckError, TracingConfigIsExist, TracingConfigNotExist
 from controllers.console.wraps import account_initialization_required, setup_required
 from libs.login import login_required
 from services.ops_service import OpsService
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
 
 class TraceProviderQuery(BaseModel):
     tracing_provider: str = Field(..., description="Tracing provider name")
@@ -23,13 +23,7 @@ class TraceConfigPayload(BaseModel):
     tracing_config: dict[str, Any] = Field(..., description="Tracing configuration data")
 
 
-console_ns.schema_model(
-    TraceProviderQuery.__name__,
-    TraceProviderQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
-console_ns.schema_model(
-    TraceConfigPayload.__name__, TraceConfigPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
-)
+register_schema_models(console_ns, TraceProviderQuery, TraceConfigPayload)
 
 
 @console_ns.route("/apps/<uuid:app_id>/trace-config")
@@ -49,11 +43,11 @@ class TraceAppConfigApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    def get(self, app_id):
-        args = TraceProviderQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+    def get(self, app_id: UUID):
+        args = TraceProviderQuery.model_validate(request.args.to_dict(flat=True))
 
         try:
-            trace_config = OpsService.get_tracing_app_config(app_id=app_id, tracing_provider=args.tracing_provider)
+            trace_config = OpsService.get_tracing_app_config(app_id=str(app_id), tracing_provider=args.tracing_provider)
             if not trace_config:
                 return {"has_not_configured": True}
             return trace_config
@@ -71,13 +65,13 @@ class TraceAppConfigApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    def post(self, app_id):
+    def post(self, app_id: UUID):
         """Create a new trace app configuration"""
         args = TraceConfigPayload.model_validate(console_ns.payload)
 
         try:
             result = OpsService.create_tracing_app_config(
-                app_id=app_id, tracing_provider=args.tracing_provider, tracing_config=args.tracing_config
+                app_id=str(app_id), tracing_provider=args.tracing_provider, tracing_config=args.tracing_config
             )
             if not result:
                 raise TracingConfigIsExist()
@@ -96,13 +90,13 @@ class TraceAppConfigApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    def patch(self, app_id):
+    def patch(self, app_id: UUID):
         """Update an existing trace app configuration"""
         args = TraceConfigPayload.model_validate(console_ns.payload)
 
         try:
             result = OpsService.update_tracing_app_config(
-                app_id=app_id, tracing_provider=args.tracing_provider, tracing_config=args.tracing_config
+                app_id=str(app_id), tracing_provider=args.tracing_provider, tracing_config=args.tracing_config
             )
             if not result:
                 raise TracingConfigNotExist()
@@ -119,12 +113,12 @@ class TraceAppConfigApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    def delete(self, app_id):
+    def delete(self, app_id: UUID):
         """Delete an existing trace app configuration"""
-        args = TraceProviderQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = TraceProviderQuery.model_validate(request.args.to_dict(flat=True))
 
         try:
-            result = OpsService.delete_tracing_app_config(app_id=app_id, tracing_provider=args.tracing_provider)
+            result = OpsService.delete_tracing_app_config(app_id=str(app_id), tracing_provider=args.tracing_provider)
             if not result:
                 raise TracingConfigNotExist()
             return {"result": "success"}, 204

api/controllers/console/app/statistic.py

@@ -5,6 +5,7 @@ from flask import abort, jsonify, request
 from flask_restx import Resource, fields
 from pydantic import BaseModel, Field, field_validator
 
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
@@ -15,8 +16,6 @@ from libs.helper import convert_datetime_to_date
 from libs.login import current_account_with_tenant, login_required
 from models import AppMode
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
 
 class StatisticTimeRangeQuery(BaseModel):
     start: str | None = Field(default=None, description="Start date (YYYY-MM-DD HH:MM)")
@@ -30,10 +29,7 @@ class StatisticTimeRangeQuery(BaseModel):
         return value
 
 
-console_ns.schema_model(
-    StatisticTimeRangeQuery.__name__,
-    StatisticTimeRangeQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
+register_schema_models(console_ns, StatisticTimeRangeQuery)
 
 
 @console_ns.route("/apps/<uuid:app_id>/statistics/daily-messages")
@@ -54,7 +50,7 @@ class DailyMessageStatistic(Resource):
     def get(self, app_model):
         account, _ = current_account_with_tenant()
 
-        args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))
 
         converted_created_at = convert_datetime_to_date("created_at")
         sql_query = f"""SELECT
@@ -111,7 +107,7 @@ class DailyConversationStatistic(Resource):
     def get(self, app_model):
         account, _ = current_account_with_tenant()
 
-        args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))
 
         converted_created_at = convert_datetime_to_date("created_at")
         sql_query = f"""SELECT
@@ -167,7 +163,7 @@ class DailyTerminalsStatistic(Resource):
     def get(self, app_model):
         account, _ = current_account_with_tenant()
 
-        args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))
 
         converted_created_at = convert_datetime_to_date("created_at")
         sql_query = f"""SELECT
@@ -224,7 +220,7 @@ class DailyTokenCostStatistic(Resource):
     def get(self, app_model):
         account, _ = current_account_with_tenant()
 
-        args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))
 
         converted_created_at = convert_datetime_to_date("created_at")
         sql_query = f"""SELECT
@@ -284,7 +280,7 @@ class AverageSessionInteractionStatistic(Resource):
     def get(self, app_model):
         account, _ = current_account_with_tenant()
 
-        args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))
 
         converted_created_at = convert_datetime_to_date("c.created_at")
         sql_query = f"""SELECT
@@ -360,7 +356,7 @@ class UserSatisfactionRateStatistic(Resource):
     def get(self, app_model):
         account, _ = current_account_with_tenant()
 
-        args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))
 
         converted_created_at = convert_datetime_to_date("m.created_at")
         sql_query = f"""SELECT
@@ -426,7 +422,7 @@ class AverageResponseTimeStatistic(Resource):
     def get(self, app_model):
         account, _ = current_account_with_tenant()
 
-        args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))
 
         converted_created_at = convert_datetime_to_date("created_at")
         sql_query = f"""SELECT
@@ -482,7 +478,7 @@ class TokensPerSecondStatistic(Resource):
     @account_initialization_required
     def get(self, app_model):
         account, _ = current_account_with_tenant()
-        args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))
 
         converted_created_at = convert_datetime_to_date("created_at")
         sql_query = f"""SELECT

api/controllers/console/app/workflow.py

@@ -11,9 +11,9 @@ from werkzeug.exceptions import BadRequest, Forbidden, InternalServerError, NotF
 
 import services
 from controllers.common.controller_schemas import DefaultBlockConfigQuery, WorkflowListQuery, WorkflowUpdatePayload
+from controllers.common.schema import register_response_schema_model, register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.error import ConversationCompletedError, DraftWorkflowNotExist, DraftWorkflowNotSync
-from controllers.console.app.workflow_run import workflow_run_node_execution_model
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
 from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
@@ -37,6 +37,7 @@ from factories import file_factory, variable_factory
 from fields.member_fields import simple_account_fields
 from fields.online_user_fields import online_user_list_fields
 from fields.workflow_fields import workflow_fields, workflow_pagination_fields
+from fields.workflow_run_fields import WorkflowRunNodeExecutionResponse
 from graphon.enums import NodeType
 from graphon.file import File
 from graphon.file import helpers as file_helpers
@@ -56,9 +57,10 @@ from services.errors.llm import InvokeRateLimitError
 from services.workflow_service import DraftWorkflowDeletionError, WorkflowInUseError, WorkflowService
 
 logger = logging.getLogger(__name__)
+
 _file_access_controller = DatabaseFileAccessController()
 LISTENING_RETRY_IN = 2000
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
+
 RESTORE_SOURCE_WORKFLOW_MUST_BE_PUBLISHED_MESSAGE = "source workflow must be published"
 MAX_WORKFLOW_ONLINE_USERS_REQUEST_IDS = 1000
 WORKFLOW_ONLINE_USERS_REDIS_BATCH_SIZE = 50
@@ -176,25 +178,25 @@ class DraftWorkflowTriggerRunAllPayload(BaseModel):
     node_ids: list[str]
 
 
-def reg(cls: type[BaseModel]):
-    console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
-
-
-reg(SyncDraftWorkflowPayload)
-reg(AdvancedChatWorkflowRunPayload)
-reg(IterationNodeRunPayload)
-reg(LoopNodeRunPayload)
-reg(DraftWorkflowRunPayload)
-reg(DraftWorkflowNodeRunPayload)
-reg(PublishWorkflowPayload)
-reg(DefaultBlockConfigQuery)
-reg(ConvertToWorkflowPayload)
-reg(WorkflowListQuery)
-reg(WorkflowUpdatePayload)
-reg(WorkflowFeaturesPayload)
-reg(WorkflowOnlineUsersPayload)
-reg(DraftWorkflowTriggerRunPayload)
-reg(DraftWorkflowTriggerRunAllPayload)
+register_schema_models(
+    console_ns,
+    SyncDraftWorkflowPayload,
+    AdvancedChatWorkflowRunPayload,
+    IterationNodeRunPayload,
+    LoopNodeRunPayload,
+    DraftWorkflowRunPayload,
+    DraftWorkflowNodeRunPayload,
+    PublishWorkflowPayload,
+    DefaultBlockConfigQuery,
+    ConvertToWorkflowPayload,
+    WorkflowListQuery,
+    WorkflowUpdatePayload,
+    WorkflowFeaturesPayload,
+    WorkflowOnlineUsersPayload,
+    DraftWorkflowTriggerRunPayload,
+    DraftWorkflowTriggerRunAllPayload,
+)
+register_response_schema_model(console_ns, WorkflowRunNodeExecutionResponse)
 
 
 # TODO(QuantumGhost): Refactor existing node run API to handle file parameter parsing
@@ -540,9 +542,12 @@ class HumanInputDeliveryTestPayload(BaseModel):
     )
 
 
-reg(HumanInputFormPreviewPayload)
-reg(HumanInputFormSubmitPayload)
-reg(HumanInputDeliveryTestPayload)
+register_schema_models(
+    console_ns,
+    HumanInputFormPreviewPayload,
+    HumanInputFormSubmitPayload,
+    HumanInputDeliveryTestPayload,
+)
 
 
 @console_ns.route("/apps/<uuid:app_id>/advanced-chat/workflows/draft/human-input/nodes/<string:node_id>/form/preview")
@@ -760,14 +765,17 @@ class DraftWorkflowNodeRunApi(Resource):
     @console_ns.doc(description="Run draft workflow node")
     @console_ns.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
     @console_ns.expect(console_ns.models[DraftWorkflowNodeRunPayload.__name__])
-    @console_ns.response(200, "Node run started successfully", workflow_run_node_execution_model)
+    @console_ns.response(
+        200,
+        "Node run started successfully",
+        console_ns.models[WorkflowRunNodeExecutionResponse.__name__],
+    )
     @console_ns.response(403, "Permission denied")
     @console_ns.response(404, "Node not found")
     @setup_required
     @login_required
     @account_initialization_required
     @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    @marshal_with(workflow_run_node_execution_model)
     @edit_permission_required
     def post(self, app_model: App, node_id: str):
         """
@@ -799,7 +807,9 @@ class DraftWorkflowNodeRunApi(Resource):
             files=files,
         )
 
-        return workflow_node_execution
+        return WorkflowRunNodeExecutionResponse.model_validate(
+            workflow_node_execution, from_attributes=True
+        ).model_dump(mode="json")
 
 
 @console_ns.route("/apps/<uuid:app_id>/workflows/publish")
@@ -902,7 +912,7 @@ class DefaultBlockConfigApi(Resource):
         """
         Get default block config
         """
-        args = DefaultBlockConfigQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = DefaultBlockConfigQuery.model_validate(request.args.to_dict(flat=True))
 
         filters = None
         if args.q:
@@ -995,7 +1005,7 @@ class PublishedAllWorkflowApi(Resource):
         """
         current_user, _ = current_account_with_tenant()
 
-        args = WorkflowListQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = WorkflowListQuery.model_validate(request.args.to_dict(flat=True))
         page = args.page
         limit = args.limit
         user_id = args.user_id
@@ -1143,14 +1153,17 @@ class DraftWorkflowNodeLastRunApi(Resource):
     @console_ns.doc("get_draft_workflow_node_last_run")
     @console_ns.doc(description="Get last run result for draft workflow node")
     @console_ns.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
-    @console_ns.response(200, "Node last run retrieved successfully", workflow_run_node_execution_model)
+    @console_ns.response(
+        200,
+        "Node last run retrieved successfully",
+        console_ns.models[WorkflowRunNodeExecutionResponse.__name__],
+    )
     @console_ns.response(404, "Node last run not found")
     @console_ns.response(403, "Permission denied")
     @setup_required
     @login_required
     @account_initialization_required
     @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    @marshal_with(workflow_run_node_execution_model)
     def get(self, app_model: App, node_id: str):
         srv = WorkflowService()
         workflow = srv.get_draft_workflow(app_model)
@@ -1163,7 +1176,7 @@ class DraftWorkflowNodeLastRunApi(Resource):
         )
         if node_exec is None:
             raise NotFound("last run not found")
-        return node_exec
+        return WorkflowRunNodeExecutionResponse.model_validate(node_exec, from_attributes=True).model_dump(mode="json")
 
 
 @console_ns.route("/apps/<uuid:app_id>/workflows/draft/trigger/run")

api/controllers/console/app/workflow_app_log.py

@@ -185,7 +185,7 @@ class WorkflowAppLogApi(Resource):
         """
         Get workflow app logs
         """
-        args = WorkflowAppLogQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = WorkflowAppLogQuery.model_validate(request.args.to_dict(flat=True))
 
         # get paginate workflow app logs
         workflow_app_service = WorkflowAppService()
@@ -228,7 +228,7 @@ class WorkflowArchivedLogApi(Resource):
         """
         Get workflow archived logs
         """
-        args = WorkflowAppLogQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = WorkflowAppLogQuery.model_validate(request.args.to_dict(flat=True))
 
         workflow_app_service = WorkflowAppService()
         with sessionmaker(db.engine, expire_on_commit=False).begin() as session:

api/controllers/console/app/workflow_comment.py

@@ -23,7 +23,6 @@ from services.account_service import TenantService
 from services.workflow_comment_service import WorkflowCommentService
 
 logger = logging.getLogger(__name__)
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
 
 class WorkflowCommentCreatePayload(BaseModel):
@@ -52,13 +51,14 @@ class WorkflowCommentMentionUsersPayload(BaseModel):
     users: list[AccountWithRole]
 
 
-for model in (
+register_schema_models(
+    console_ns,
+    AccountWithRole,
+    WorkflowCommentMentionUsersPayload,
     WorkflowCommentCreatePayload,
     WorkflowCommentUpdatePayload,
     WorkflowCommentReplyPayload,
-):
-    console_ns.schema_model(model.__name__, model.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
-register_schema_models(console_ns, AccountWithRole, WorkflowCommentMentionUsersPayload)
+)
 
 workflow_comment_basic_model = console_ns.model("WorkflowCommentBasic", workflow_comment_basic_fields)
 workflow_comment_detail_model = console_ns.model("WorkflowCommentDetail", workflow_comment_detail_fields)

api/controllers/console/app/workflow_draft_variable.py

@@ -8,6 +8,7 @@ from flask_restx import Resource, fields, marshal, marshal_with
 from pydantic import BaseModel, Field
 from sqlalchemy.orm import sessionmaker
 
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.error import (
     DraftWorkflowNotExist,
@@ -33,7 +34,6 @@ from services.workflow_service import WorkflowService
 
 logger = logging.getLogger(__name__)
 _file_access_controller = DatabaseFileAccessController()
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
 
 class WorkflowDraftVariableListQuery(BaseModel):
@@ -56,21 +56,12 @@ class EnvironmentVariableUpdatePayload(BaseModel):
     environment_variables: list[dict[str, Any]] = Field(..., description="Environment variables for the draft workflow")
 
 
-console_ns.schema_model(
-    WorkflowDraftVariableListQuery.__name__,
-    WorkflowDraftVariableListQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
-console_ns.schema_model(
-    WorkflowDraftVariableUpdatePayload.__name__,
-    WorkflowDraftVariableUpdatePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
-console_ns.schema_model(
-    ConversationVariableUpdatePayload.__name__,
-    ConversationVariableUpdatePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
-console_ns.schema_model(
-    EnvironmentVariableUpdatePayload.__name__,
-    EnvironmentVariableUpdatePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+register_schema_models(
+    console_ns,
+    WorkflowDraftVariableListQuery,
+    WorkflowDraftVariableUpdatePayload,
+    ConversationVariableUpdatePayload,
+    EnvironmentVariableUpdatePayload,
 )
 
 
@@ -260,7 +251,7 @@ class WorkflowVariableCollectionApi(Resource):
         """
         Get draft workflow
         """
-        args = WorkflowDraftVariableListQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = WorkflowDraftVariableListQuery.model_validate(request.args.to_dict(flat=True))
 
         # fetch draft workflow by app_model
         workflow_service = WorkflowService()

api/controllers/console/app/workflow_run.py

@@ -1,30 +1,28 @@
 from datetime import UTC, datetime, timedelta
-from typing import Literal, TypedDict, cast
+from typing import Literal, cast
 
 from flask import request
-from flask_restx import Resource, fields, marshal_with
+from flask_restx import Resource
 from pydantic import BaseModel, Field, field_validator
 from sqlalchemy import select
 from sqlalchemy.orm import sessionmaker
 
 from configs import dify_config
+from controllers.common.schema import query_params_from_model, register_response_schema_models, register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
 from controllers.web.error import NotFoundError
 from core.workflow.human_input_forms import load_form_tokens_by_form_id as _load_form_tokens_by_form_id
 from extensions.ext_database import db
-from fields.end_user_fields import simple_end_user_fields
-from fields.member_fields import simple_account_fields
+from fields.base import ResponseModel
 from fields.workflow_run_fields import (
-    advanced_chat_workflow_run_for_list_fields,
-    advanced_chat_workflow_run_pagination_fields,
-    workflow_run_count_fields,
-    workflow_run_detail_fields,
-    workflow_run_for_list_fields,
-    workflow_run_node_execution_fields,
-    workflow_run_node_execution_list_fields,
-    workflow_run_pagination_fields,
+    AdvancedChatWorkflowRunPaginationResponse,
+    WorkflowRunCountResponse,
+    WorkflowRunDetailResponse,
+    WorkflowRunNodeExecutionListResponse,
+    WorkflowRunNodeExecutionResponse,
+    WorkflowRunPaginationResponse,
 )
 from graphon.entities.pause_reason import HumanInputRequired
 from graphon.enums import WorkflowExecutionStatus
@@ -52,82 +50,6 @@ def _build_backstage_input_url(form_token: str | None) -> str | None:
 WORKFLOW_RUN_STATUS_CHOICES = ["running", "succeeded", "failed", "stopped", "partial-succeeded"]
 EXPORT_SIGNED_URL_EXPIRE_SECONDS = 3600
 
-# Register models for flask_restx to avoid dict type issues in Swagger
-# Register in dependency order: base models first, then dependent models
-
-# Base models
-simple_account_model = console_ns.model("SimpleAccount", simple_account_fields)
-
-simple_end_user_model = console_ns.model("SimpleEndUser", simple_end_user_fields)
-
-# Models that depend on simple_account_fields
-workflow_run_for_list_fields_copy = workflow_run_for_list_fields.copy()
-workflow_run_for_list_fields_copy["created_by_account"] = fields.Nested(
-    simple_account_model, attribute="created_by_account", allow_null=True
-)
-workflow_run_for_list_model = console_ns.model("WorkflowRunForList", workflow_run_for_list_fields_copy)
-
-advanced_chat_workflow_run_for_list_fields_copy = advanced_chat_workflow_run_for_list_fields.copy()
-advanced_chat_workflow_run_for_list_fields_copy["created_by_account"] = fields.Nested(
-    simple_account_model, attribute="created_by_account", allow_null=True
-)
-advanced_chat_workflow_run_for_list_model = console_ns.model(
-    "AdvancedChatWorkflowRunForList", advanced_chat_workflow_run_for_list_fields_copy
-)
-
-workflow_run_detail_fields_copy = workflow_run_detail_fields.copy()
-workflow_run_detail_fields_copy["created_by_account"] = fields.Nested(
-    simple_account_model, attribute="created_by_account", allow_null=True
-)
-workflow_run_detail_fields_copy["created_by_end_user"] = fields.Nested(
-    simple_end_user_model, attribute="created_by_end_user", allow_null=True
-)
-workflow_run_detail_model = console_ns.model("WorkflowRunDetail", workflow_run_detail_fields_copy)
-
-workflow_run_node_execution_fields_copy = workflow_run_node_execution_fields.copy()
-workflow_run_node_execution_fields_copy["created_by_account"] = fields.Nested(
-    simple_account_model, attribute="created_by_account", allow_null=True
-)
-workflow_run_node_execution_fields_copy["created_by_end_user"] = fields.Nested(
-    simple_end_user_model, attribute="created_by_end_user", allow_null=True
-)
-workflow_run_node_execution_model = console_ns.model(
-    "WorkflowRunNodeExecution", workflow_run_node_execution_fields_copy
-)
-
-# Simple models without nested dependencies
-workflow_run_count_model = console_ns.model("WorkflowRunCount", workflow_run_count_fields)
-
-# Pagination models that depend on list models
-advanced_chat_workflow_run_pagination_fields_copy = advanced_chat_workflow_run_pagination_fields.copy()
-advanced_chat_workflow_run_pagination_fields_copy["data"] = fields.List(
-    fields.Nested(advanced_chat_workflow_run_for_list_model), attribute="data"
-)
-advanced_chat_workflow_run_pagination_model = console_ns.model(
-    "AdvancedChatWorkflowRunPagination", advanced_chat_workflow_run_pagination_fields_copy
-)
-
-workflow_run_pagination_fields_copy = workflow_run_pagination_fields.copy()
-workflow_run_pagination_fields_copy["data"] = fields.List(fields.Nested(workflow_run_for_list_model), attribute="data")
-workflow_run_pagination_model = console_ns.model("WorkflowRunPagination", workflow_run_pagination_fields_copy)
-
-workflow_run_node_execution_list_fields_copy = workflow_run_node_execution_list_fields.copy()
-workflow_run_node_execution_list_fields_copy["data"] = fields.List(fields.Nested(workflow_run_node_execution_model))
-workflow_run_node_execution_list_model = console_ns.model(
-    "WorkflowRunNodeExecutionList", workflow_run_node_execution_list_fields_copy
-)
-
-workflow_run_export_fields = console_ns.model(
-    "WorkflowRunExport",
-    {
-        "status": fields.String(description="Export status: success/failed"),
-        "presigned_url": fields.String(description="Pre-signed URL for download", required=False),
-        "presigned_url_expires_at": fields.String(description="Pre-signed URL expiration time", required=False),
-    },
-)
-
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
 
 class WorkflowRunListQuery(BaseModel):
     last_id: str | None = Field(default=None, description="Last run ID for pagination")
@@ -136,7 +58,7 @@ class WorkflowRunListQuery(BaseModel):
         default=None, description="Workflow run status filter"
     )
     triggered_from: Literal["debugging", "app-run"] | None = Field(
-        default=None, description="Filter by trigger source: debugging or app-run"
+        default=None, description="Filter by trigger source: debugging or app-run. Default: debugging"
     )
 
     @field_validator("last_id")
@@ -151,9 +73,15 @@ class WorkflowRunCountQuery(BaseModel):
     status: Literal["running", "succeeded", "failed", "stopped", "partial-succeeded"] | None = Field(
         default=None, description="Workflow run status filter"
     )
-    time_range: str | None = Field(default=None, description="Time range filter (e.g., 7d, 4h, 30m, 30s)")
+    time_range: str | None = Field(
+        default=None,
+        description=(
+            "Filter by time range (optional): e.g., 7d (7 days), 4h (4 hours), "
+            "30m (30 minutes), 30s (30 seconds). Filters by created_at field."
+        ),
+    )
     triggered_from: Literal["debugging", "app-run"] | None = Field(
-        default=None, description="Filter by trigger source: debugging or app-run"
+        default=None, description="Filter by trigger source: debugging or app-run. Default: debugging"
     )
 
     @field_validator("time_range")
@@ -164,56 +92,69 @@ class WorkflowRunCountQuery(BaseModel):
         return time_duration(value)
 
 
-console_ns.schema_model(
-    WorkflowRunListQuery.__name__, WorkflowRunListQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
-)
-console_ns.schema_model(
-    WorkflowRunCountQuery.__name__,
-    WorkflowRunCountQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
+class WorkflowRunExportResponse(ResponseModel):
+    status: str = Field(description="Export status: success/failed")
+    presigned_url: str | None = Field(default=None, description="Pre-signed URL for download")
+    presigned_url_expires_at: str | None = Field(default=None, description="Pre-signed URL expiration time")
 
 
-class HumanInputPauseTypeResponse(TypedDict):
+class HumanInputPauseTypeResponse(ResponseModel):
     type: Literal["human_input"]
     form_id: str
-    backstage_input_url: str | None
+    backstage_input_url: str | None = None
 
 
-class PausedNodeResponse(TypedDict):
+class PausedNodeResponse(ResponseModel):
     node_id: str
     node_title: str
     pause_type: HumanInputPauseTypeResponse
 
 
-class WorkflowPauseDetailsResponse(TypedDict):
-    paused_at: str | None
+class WorkflowPauseDetailsResponse(ResponseModel):
+    paused_at: str | None = None
     paused_nodes: list[PausedNodeResponse]
 
 
+register_schema_models(
+    console_ns,
+    WorkflowRunListQuery,
+    WorkflowRunCountQuery,
+)
+register_response_schema_models(
+    console_ns,
+    AdvancedChatWorkflowRunPaginationResponse,
+    WorkflowRunPaginationResponse,
+    WorkflowRunCountResponse,
+    WorkflowRunDetailResponse,
+    WorkflowRunNodeExecutionResponse,
+    WorkflowRunNodeExecutionListResponse,
+    WorkflowRunExportResponse,
+    HumanInputPauseTypeResponse,
+    PausedNodeResponse,
+    WorkflowPauseDetailsResponse,
+)
+
+
 @console_ns.route("/apps/<uuid:app_id>/advanced-chat/workflow-runs")
 class AdvancedChatAppWorkflowRunListApi(Resource):
     @console_ns.doc("get_advanced_chat_workflow_runs")
     @console_ns.doc(description="Get advanced chat workflow run list")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.doc(params={"last_id": "Last run ID for pagination", "limit": "Number of items per page (1-100)"})
-    @console_ns.doc(
-        params={"status": "Filter by status (optional): running, succeeded, failed, stopped, partial-succeeded"}
+    @console_ns.doc(params=query_params_from_model(WorkflowRunListQuery))
+    @console_ns.response(
+        200,
+        "Workflow runs retrieved successfully",
+        console_ns.models[AdvancedChatWorkflowRunPaginationResponse.__name__],
     )
-    @console_ns.doc(
-        params={"triggered_from": "Filter by trigger source (optional): debugging or app-run. Default: debugging"}
-    )
-    @console_ns.expect(console_ns.models[WorkflowRunListQuery.__name__])
-    @console_ns.response(200, "Workflow runs retrieved successfully", advanced_chat_workflow_run_pagination_model)
     @setup_required
     @login_required
     @account_initialization_required
     @get_app_model(mode=[AppMode.ADVANCED_CHAT])
-    @marshal_with(advanced_chat_workflow_run_pagination_model)
     def get(self, app_model: App):
         """
         Get advanced chat app workflow run list
         """
-        args_model = WorkflowRunListQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args_model = WorkflowRunListQuery.model_validate(request.args.to_dict(flat=True))
         args: WorkflowRunListArgs = {"limit": args_model.limit}
         if args_model.last_id is not None:
             args["last_id"] = args_model.last_id
@@ -232,7 +173,9 @@ class AdvancedChatAppWorkflowRunListApi(Resource):
             app_model=app_model, args=args, triggered_from=triggered_from
         )
 
-        return result
+        return AdvancedChatWorkflowRunPaginationResponse.model_validate(result, from_attributes=True).model_dump(
+            mode="json"
+        )
 
 
 @console_ns.route("/apps/<uuid:app_id>/workflow-runs/<uuid:run_id>/export")
@@ -240,7 +183,7 @@ class WorkflowRunExportApi(Resource):
     @console_ns.doc("get_workflow_run_export_url")
     @console_ns.doc(description="Generate a download URL for an archived workflow run.")
     @console_ns.doc(params={"app_id": "Application ID", "run_id": "Workflow run ID"})
-    @console_ns.response(200, "Export URL generated", workflow_run_export_fields)
+    @console_ns.response(200, "Export URL generated", console_ns.models[WorkflowRunExportResponse.__name__])
     @setup_required
     @login_required
     @account_initialization_required
@@ -278,11 +221,14 @@ class WorkflowRunExportApi(Resource):
             expires_in=EXPORT_SIGNED_URL_EXPIRE_SECONDS,
         )
         expires_at = datetime.now(UTC) + timedelta(seconds=EXPORT_SIGNED_URL_EXPIRE_SECONDS)
-        return {
-            "status": "success",
-            "presigned_url": presigned_url,
-            "presigned_url_expires_at": expires_at.isoformat(),
-        }, 200
+        response = WorkflowRunExportResponse.model_validate(
+            {
+                "status": "success",
+                "presigned_url": presigned_url,
+                "presigned_url_expires_at": expires_at.isoformat(),
+            }
+        )
+        return response.model_dump(mode="json"), 200
 
 
 @console_ns.route("/apps/<uuid:app_id>/advanced-chat/workflow-runs/count")
@@ -290,32 +236,21 @@ class AdvancedChatAppWorkflowRunCountApi(Resource):
     @console_ns.doc("get_advanced_chat_workflow_runs_count")
     @console_ns.doc(description="Get advanced chat workflow runs count statistics")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.doc(
-        params={"status": "Filter by status (optional): running, succeeded, failed, stopped, partial-succeeded"}
+    @console_ns.doc(params=query_params_from_model(WorkflowRunCountQuery))
+    @console_ns.response(
+        200,
+        "Workflow runs count retrieved successfully",
+        console_ns.models[WorkflowRunCountResponse.__name__],
     )
-    @console_ns.doc(
-        params={
-            "time_range": (
-                "Filter by time range (optional): e.g., 7d (7 days), 4h (4 hours), "
-                "30m (30 minutes), 30s (30 seconds). Filters by created_at field."
-            )
-        }
-    )
-    @console_ns.doc(
-        params={"triggered_from": "Filter by trigger source (optional): debugging or app-run. Default: debugging"}
-    )
-    @console_ns.response(200, "Workflow runs count retrieved successfully", workflow_run_count_model)
-    @console_ns.expect(console_ns.models[WorkflowRunCountQuery.__name__])
     @setup_required
     @login_required
     @account_initialization_required
     @get_app_model(mode=[AppMode.ADVANCED_CHAT])
-    @marshal_with(workflow_run_count_model)
     def get(self, app_model: App):
         """
         Get advanced chat workflow runs count statistics
         """
-        args_model = WorkflowRunCountQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args_model = WorkflowRunCountQuery.model_validate(request.args.to_dict(flat=True))
         args = args_model.model_dump(exclude_none=True)
 
         # Default to DEBUGGING if not specified
@@ -333,7 +268,7 @@ class AdvancedChatAppWorkflowRunCountApi(Resource):
             triggered_from=triggered_from,
         )
 
-        return result
+        return WorkflowRunCountResponse.model_validate(result).model_dump(mode="json")
 
 
 @console_ns.route("/apps/<uuid:app_id>/workflow-runs")
@@ -341,25 +276,21 @@ class WorkflowRunListApi(Resource):
     @console_ns.doc("get_workflow_runs")
     @console_ns.doc(description="Get workflow run list")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.doc(params={"last_id": "Last run ID for pagination", "limit": "Number of items per page (1-100)"})
-    @console_ns.doc(
-        params={"status": "Filter by status (optional): running, succeeded, failed, stopped, partial-succeeded"}
+    @console_ns.doc(params=query_params_from_model(WorkflowRunListQuery))
+    @console_ns.response(
+        200,
+        "Workflow runs retrieved successfully",
+        console_ns.models[WorkflowRunPaginationResponse.__name__],
     )
-    @console_ns.doc(
-        params={"triggered_from": "Filter by trigger source (optional): debugging or app-run. Default: debugging"}
-    )
-    @console_ns.response(200, "Workflow runs retrieved successfully", workflow_run_pagination_model)
-    @console_ns.expect(console_ns.models[WorkflowRunListQuery.__name__])
     @setup_required
     @login_required
     @account_initialization_required
     @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    @marshal_with(workflow_run_pagination_model)
     def get(self, app_model: App):
         """
         Get workflow run list
         """
-        args_model = WorkflowRunListQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args_model = WorkflowRunListQuery.model_validate(request.args.to_dict(flat=True))
         args: WorkflowRunListArgs = {"limit": args_model.limit}
         if args_model.last_id is not None:
             args["last_id"] = args_model.last_id
@@ -378,7 +309,7 @@ class WorkflowRunListApi(Resource):
             app_model=app_model, args=args, triggered_from=triggered_from
         )
 
-        return result
+        return WorkflowRunPaginationResponse.model_validate(result, from_attributes=True).model_dump(mode="json")
 
 
 @console_ns.route("/apps/<uuid:app_id>/workflow-runs/count")
@@ -386,32 +317,21 @@ class WorkflowRunCountApi(Resource):
     @console_ns.doc("get_workflow_runs_count")
     @console_ns.doc(description="Get workflow runs count statistics")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.doc(
-        params={"status": "Filter by status (optional): running, succeeded, failed, stopped, partial-succeeded"}
+    @console_ns.doc(params=query_params_from_model(WorkflowRunCountQuery))
+    @console_ns.response(
+        200,
+        "Workflow runs count retrieved successfully",
+        console_ns.models[WorkflowRunCountResponse.__name__],
     )
-    @console_ns.doc(
-        params={
-            "time_range": (
-                "Filter by time range (optional): e.g., 7d (7 days), 4h (4 hours), "
-                "30m (30 minutes), 30s (30 seconds). Filters by created_at field."
-            )
-        }
-    )
-    @console_ns.doc(
-        params={"triggered_from": "Filter by trigger source (optional): debugging or app-run. Default: debugging"}
-    )
-    @console_ns.response(200, "Workflow runs count retrieved successfully", workflow_run_count_model)
-    @console_ns.expect(console_ns.models[WorkflowRunCountQuery.__name__])
     @setup_required
     @login_required
     @account_initialization_required
     @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    @marshal_with(workflow_run_count_model)
     def get(self, app_model: App):
         """
         Get workflow runs count statistics
         """
-        args_model = WorkflowRunCountQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args_model = WorkflowRunCountQuery.model_validate(request.args.to_dict(flat=True))
         args = args_model.model_dump(exclude_none=True)
 
         # Default to DEBUGGING for workflow if not specified (backward compatibility)
@@ -429,7 +349,7 @@ class WorkflowRunCountApi(Resource):
             triggered_from=triggered_from,
         )
 
-        return result
+        return WorkflowRunCountResponse.model_validate(result).model_dump(mode="json")
 
 
 @console_ns.route("/apps/<uuid:app_id>/workflow-runs/<uuid:run_id>")
@@ -437,13 +357,16 @@ class WorkflowRunDetailApi(Resource):
     @console_ns.doc("get_workflow_run_detail")
     @console_ns.doc(description="Get workflow run detail")
     @console_ns.doc(params={"app_id": "Application ID", "run_id": "Workflow run ID"})
-    @console_ns.response(200, "Workflow run detail retrieved successfully", workflow_run_detail_model)
+    @console_ns.response(
+        200,
+        "Workflow run detail retrieved successfully",
+        console_ns.models[WorkflowRunDetailResponse.__name__],
+    )
     @console_ns.response(404, "Workflow run not found")
     @setup_required
     @login_required
     @account_initialization_required
     @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    @marshal_with(workflow_run_detail_model)
     def get(self, app_model: App, run_id):
         """
         Get workflow run detail
@@ -452,8 +375,10 @@ class WorkflowRunDetailApi(Resource):
 
         workflow_run_service = WorkflowRunService()
         workflow_run = workflow_run_service.get_workflow_run(app_model=app_model, run_id=run_id)
+        if workflow_run is None:
+            raise NotFoundError("Workflow run not found")
 
-        return workflow_run
+        return WorkflowRunDetailResponse.model_validate(workflow_run, from_attributes=True).model_dump(mode="json")
 
 
 @console_ns.route("/apps/<uuid:app_id>/workflow-runs/<uuid:run_id>/node-executions")
@@ -461,13 +386,16 @@ class WorkflowRunNodeExecutionListApi(Resource):
     @console_ns.doc("get_workflow_run_node_executions")
     @console_ns.doc(description="Get workflow run node execution list")
     @console_ns.doc(params={"app_id": "Application ID", "run_id": "Workflow run ID"})
-    @console_ns.response(200, "Node executions retrieved successfully", workflow_run_node_execution_list_model)
+    @console_ns.response(
+        200,
+        "Node executions retrieved successfully",
+        console_ns.models[WorkflowRunNodeExecutionListResponse.__name__],
+    )
     @console_ns.response(404, "Workflow run not found")
     @setup_required
     @login_required
     @account_initialization_required
     @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    @marshal_with(workflow_run_node_execution_list_model)
     def get(self, app_model: App, run_id):
         """
         Get workflow run node execution list
@@ -482,13 +410,24 @@ class WorkflowRunNodeExecutionListApi(Resource):
             user=user,
         )
 
-        return {"data": node_executions}
+        return WorkflowRunNodeExecutionListResponse.model_validate(
+            {"data": node_executions}, from_attributes=True
+        ).model_dump(mode="json")
 
 
 @console_ns.route("/workflow/<string:workflow_run_id>/pause-details")
 class ConsoleWorkflowPauseDetailsApi(Resource):
     """Console API for getting workflow pause details."""
 
+    @console_ns.doc("get_workflow_pause_details")
+    @console_ns.doc(description="Get workflow pause details")
+    @console_ns.doc(params={"workflow_run_id": "Workflow run ID"})
+    @console_ns.response(
+        200,
+        "Workflow pause details retrieved successfully",
+        console_ns.models[WorkflowPauseDetailsResponse.__name__],
+    )
+    @console_ns.response(404, "Workflow run not found")
     @setup_required
     @login_required
     @account_initialization_required
@@ -515,11 +454,8 @@ class ConsoleWorkflowPauseDetailsApi(Resource):
         # Check if workflow is suspended
         is_paused = workflow_run.status == WorkflowExecutionStatus.PAUSED
         if not is_paused:
-            empty_response: WorkflowPauseDetailsResponse = {
-                "paused_at": None,
-                "paused_nodes": [],
-            }
-            return empty_response, 200
+            empty_response = WorkflowPauseDetailsResponse(paused_at=None, paused_nodes=[])
+            return empty_response.model_dump(mode="json"), 200
 
         pause_entity = workflow_run_repo.get_workflow_pause(workflow_run_id)
         pause_reasons = pause_entity.get_pause_reasons() if pause_entity else []
@@ -530,27 +466,25 @@ class ConsoleWorkflowPauseDetailsApi(Resource):
         # Build response
         paused_at = pause_entity.paused_at if pause_entity else None
         paused_nodes: list[PausedNodeResponse] = []
-        response: WorkflowPauseDetailsResponse = {
-            "paused_at": paused_at.isoformat() + "Z" if paused_at else None,
-            "paused_nodes": paused_nodes,
-        }
 
         for reason in pause_reasons:
             if isinstance(reason, HumanInputRequired):
                 paused_nodes.append(
-                    {
-                        "node_id": reason.node_id,
-                        "node_title": reason.node_title,
-                        "pause_type": {
-                            "type": "human_input",
-                            "form_id": reason.form_id,
-                            "backstage_input_url": _build_backstage_input_url(
-                                form_tokens_by_form_id.get(reason.form_id)
-                            ),
-                        },
-                    }
+                    PausedNodeResponse(
+                        node_id=reason.node_id,
+                        node_title=reason.node_title,
+                        pause_type=HumanInputPauseTypeResponse(
+                            type="human_input",
+                            form_id=reason.form_id,
+                            backstage_input_url=_build_backstage_input_url(form_tokens_by_form_id.get(reason.form_id)),
+                        ),
+                    )
                 )
             else:
                 raise AssertionError("unimplemented.")
 
-        return response, 200
+        response = WorkflowPauseDetailsResponse(
+            paused_at=paused_at.isoformat() + "Z" if paused_at else None,
+            paused_nodes=paused_nodes,
+        )
+        return response.model_dump(mode="json"), 200

api/controllers/console/app/workflow_statistic.py

@@ -3,6 +3,7 @@ from flask_restx import Resource
 from pydantic import BaseModel, Field, field_validator
 from sqlalchemy.orm import sessionmaker
 
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
@@ -13,8 +14,6 @@ from models.enums import WorkflowRunTriggeredFrom
 from models.model import AppMode
 from repositories.factory import DifyAPIRepositoryFactory
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
 
 class WorkflowStatisticQuery(BaseModel):
     start: str | None = Field(default=None, description="Start date and time (YYYY-MM-DD HH:MM)")
@@ -28,10 +27,7 @@ class WorkflowStatisticQuery(BaseModel):
         return value
 
 
-console_ns.schema_model(
-    WorkflowStatisticQuery.__name__,
-    WorkflowStatisticQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
+register_schema_models(console_ns, WorkflowStatisticQuery)
 
 
 @console_ns.route("/apps/<uuid:app_id>/workflow/statistics/daily-conversations")
@@ -53,7 +49,7 @@ class WorkflowDailyRunsStatistic(Resource):
     def get(self, app_model):
         account, _ = current_account_with_tenant()
 
-        args = WorkflowStatisticQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = WorkflowStatisticQuery.model_validate(request.args.to_dict(flat=True))
 
         assert account.timezone is not None
 
@@ -93,7 +89,7 @@ class WorkflowDailyTerminalsStatistic(Resource):
     def get(self, app_model):
         account, _ = current_account_with_tenant()
 
-        args = WorkflowStatisticQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = WorkflowStatisticQuery.model_validate(request.args.to_dict(flat=True))
 
         assert account.timezone is not None
 
@@ -133,7 +129,7 @@ class WorkflowDailyTokenCostStatistic(Resource):
     def get(self, app_model):
         account, _ = current_account_with_tenant()
 
-        args = WorkflowStatisticQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = WorkflowStatisticQuery.model_validate(request.args.to_dict(flat=True))
 
         assert account.timezone is not None
 
@@ -173,7 +169,7 @@ class WorkflowAverageAppInteractionStatistic(Resource):
     def get(self, app_model):
         account, _ = current_account_with_tenant()
 
-        args = WorkflowStatisticQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = WorkflowStatisticQuery.model_validate(request.args.to_dict(flat=True))
 
         assert account.timezone is not None
 

api/controllers/console/app/workflow_trigger.py

@@ -94,7 +94,7 @@ class WebhookTriggerApi(Resource):
     @console_ns.response(200, "Success", console_ns.models[WebhookTriggerResponse.__name__])
     def get(self, app_model: App):
         """Get webhook trigger for a node"""
-        args = Parser.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = Parser.model_validate(request.args.to_dict(flat=True))
 
         node_id = args.node_id
 

api/controllers/console/auth/activate.py

@@ -63,7 +63,7 @@ class ActivateCheckApi(Resource):
         console_ns.models[ActivationCheckResponse.__name__],
     )
     def get(self):
-        args = ActivateCheckQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = ActivateCheckQuery.model_validate(request.args.to_dict(flat=True))
 
         workspaceId = args.workspace_id
         token = args.token

api/controllers/console/auth/data_source_bearer_auth.py

@@ -1,6 +1,7 @@
 from flask_restx import Resource
 from pydantic import BaseModel, Field
 
+from controllers.common.schema import register_schema_models
 from libs.login import current_account_with_tenant, login_required
 from services.auth.api_key_auth_service import ApiKeyAuthService
 
@@ -8,8 +9,6 @@ from .. import console_ns
 from ..auth.error import ApiKeyAuthFailedError
 from ..wraps import account_initialization_required, is_admin_or_owner_required, setup_required
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
 
 class ApiKeyAuthBindingPayload(BaseModel):
     category: str = Field(...)
@@ -17,10 +16,7 @@ class ApiKeyAuthBindingPayload(BaseModel):
     credentials: dict = Field(...)
 
 
-console_ns.schema_model(
-    ApiKeyAuthBindingPayload.__name__,
-    ApiKeyAuthBindingPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
+register_schema_models(console_ns, ApiKeyAuthBindingPayload)
 
 
 @console_ns.route("/api-key-auth/data-source")

api/controllers/console/auth/email_register.py

@@ -4,6 +4,7 @@ from pydantic import BaseModel, Field, field_validator
 
 from configs import dify_config
 from constants.languages import languages
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.auth.error import (
     EmailAlreadyInUseError,
@@ -23,8 +24,6 @@ from services.errors.account import AccountNotFoundError, AccountRegisterError
 from ..error import AccountInFreezeError, EmailSendIpLimitError
 from ..wraps import email_password_login_enabled, email_register_enabled, setup_required
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
 
 class EmailRegisterSendPayload(BaseModel):
     email: EmailStr = Field(..., description="Email address")
@@ -48,8 +47,7 @@ class EmailRegisterResetPayload(BaseModel):
         return valid_password(value)
 
 
-for model in (EmailRegisterSendPayload, EmailRegisterValidityPayload, EmailRegisterResetPayload):
-    console_ns.schema_model(model.__name__, model.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
+register_schema_models(console_ns, EmailRegisterSendPayload, EmailRegisterValidityPayload, EmailRegisterResetPayload)
 
 
 @console_ns.route("/email-register/send-email")

api/controllers/console/auth/forgot_password.py

@@ -28,8 +28,6 @@ from services.entities.auth_entities import (
 )
 from services.feature_service import FeatureService
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
 
 class ForgotPasswordEmailResponse(BaseModel):
     result: str = Field(description="Operation result")

api/controllers/console/auth/login.py

@@ -9,6 +9,7 @@ from werkzeug.exceptions import Unauthorized
 import services
 from configs import dify_config
 from constants.languages import get_valid_language
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.auth.error import (
     AuthenticationFailedError,
@@ -50,7 +51,6 @@ from services.errors.account import AccountRegisterError
 from services.errors.workspace import WorkSpaceNotAllowedCreateError, WorkspacesLimitExceededError
 from services.feature_service import FeatureService
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 logger = logging.getLogger(__name__)
 
 
@@ -71,13 +71,7 @@ class EmailCodeLoginPayload(BaseModel):
     language: str | None = Field(default=None)
 
 
-def reg(cls: type[BaseModel]):
-    console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
-
-
-reg(LoginPayload)
-reg(EmailPayload)
-reg(EmailCodeLoginPayload)
+register_schema_models(console_ns, LoginPayload, EmailPayload, EmailCodeLoginPayload)
 
 
 @console_ns.route("/login")

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

@@ -4,6 +4,7 @@ from flask_restx import (  # type: ignore
 from pydantic import BaseModel
 from werkzeug.exceptions import Forbidden
 
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.datasets.wraps import get_rag_pipeline
 from controllers.console.wraps import account_initialization_required, setup_required
@@ -12,8 +13,6 @@ from models import Account
 from models.dataset import Pipeline
 from services.rag_pipeline.rag_pipeline import RagPipelineService
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
 
 class Parser(BaseModel):
     inputs: dict
@@ -21,7 +20,7 @@ class Parser(BaseModel):
     credential_id: str | None = None
 
 
-console_ns.schema_model(Parser.__name__, Parser.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
+register_schema_models(console_ns, Parser)
 
 
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/published/datasource/nodes/<string:node_id>/preview")

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

@@ -10,7 +10,7 @@ from werkzeug.exceptions import BadRequest, Forbidden, InternalServerError, NotF
 
 import services
 from controllers.common.controller_schemas import DefaultBlockConfigQuery, WorkflowListQuery, WorkflowUpdatePayload
-from controllers.common.schema import register_schema_models
+from controllers.common.schema import register_response_schema_models, register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.error import (
     ConversationCompletedError,
@@ -22,12 +22,6 @@ from controllers.console.app.workflow import (
     workflow_model,
     workflow_pagination_model,
 )
-from controllers.console.app.workflow_run import (
-    workflow_run_detail_model,
-    workflow_run_node_execution_list_model,
-    workflow_run_node_execution_model,
-    workflow_run_pagination_model,
-)
 from controllers.console.datasets.wraps import get_rag_pipeline
 from controllers.console.wraps import (
     account_initialization_required,
@@ -40,6 +34,12 @@ from core.app.apps.pipeline.pipeline_generator import PipelineGenerator
 from core.app.entities.app_invoke_entities import InvokeFrom
 from extensions.ext_database import db
 from factories import variable_factory
+from fields.workflow_run_fields import (
+    WorkflowRunDetailResponse,
+    WorkflowRunNodeExecutionListResponse,
+    WorkflowRunNodeExecutionResponse,
+    WorkflowRunPaginationResponse,
+)
 from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs import helper
 from libs.helper import TimestampField, UUIDStrOrEmpty
@@ -131,6 +131,13 @@ register_schema_models(
     DatasourceVariablesPayload,
     RagPipelineRecommendedPluginQuery,
 )
+register_response_schema_models(
+    console_ns,
+    WorkflowRunDetailResponse,
+    WorkflowRunNodeExecutionListResponse,
+    WorkflowRunNodeExecutionResponse,
+    WorkflowRunPaginationResponse,
+)
 
 
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft")
@@ -415,12 +422,16 @@ class RagPipelineDraftDatasourceNodeRunApi(Resource):
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/nodes/<string:node_id>/run")
 class RagPipelineDraftNodeRunApi(Resource):
     @console_ns.expect(console_ns.models[NodeRunRequiredPayload.__name__])
+    @console_ns.response(
+        200,
+        "Node run started successfully",
+        console_ns.models[WorkflowRunNodeExecutionResponse.__name__],
+    )
     @setup_required
     @login_required
     @edit_permission_required
     @account_initialization_required
     @get_rag_pipeline
-    @marshal_with(workflow_run_node_execution_model)
     def post(self, pipeline: Pipeline, node_id: str):
         """
         Run draft workflow node
@@ -439,7 +450,9 @@ class RagPipelineDraftNodeRunApi(Resource):
         if workflow_node_execution is None:
             raise ValueError("Workflow node execution not found")
 
-        return workflow_node_execution
+        return WorkflowRunNodeExecutionResponse.model_validate(
+            workflow_node_execution, from_attributes=True
+        ).model_dump(mode="json")
 
 
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflow-runs/tasks/<string:task_id>/stop")
@@ -778,11 +791,15 @@ class DraftRagPipelineSecondStepApi(Resource):
 
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflow-runs")
 class RagPipelineWorkflowRunListApi(Resource):
+    @console_ns.response(
+        200,
+        "Workflow runs retrieved successfully",
+        console_ns.models[WorkflowRunPaginationResponse.__name__],
+    )
     @setup_required
     @login_required
     @account_initialization_required
     @get_rag_pipeline
-    @marshal_with(workflow_run_pagination_model)
     def get(self, pipeline: Pipeline):
         """
         Get workflow run list
@@ -801,16 +818,20 @@ class RagPipelineWorkflowRunListApi(Resource):
         rag_pipeline_service = RagPipelineService()
         result = rag_pipeline_service.get_rag_pipeline_paginate_workflow_runs(pipeline=pipeline, args=args)
 
-        return result
+        return WorkflowRunPaginationResponse.model_validate(result, from_attributes=True).model_dump(mode="json")
 
 
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflow-runs/<uuid:run_id>")
 class RagPipelineWorkflowRunDetailApi(Resource):
+    @console_ns.response(
+        200,
+        "Workflow run detail retrieved successfully",
+        console_ns.models[WorkflowRunDetailResponse.__name__],
+    )
     @setup_required
     @login_required
     @account_initialization_required
     @get_rag_pipeline
-    @marshal_with(workflow_run_detail_model)
     def get(self, pipeline: Pipeline, run_id):
         """
         Get workflow run detail
@@ -819,17 +840,23 @@ class RagPipelineWorkflowRunDetailApi(Resource):
 
         rag_pipeline_service = RagPipelineService()
         workflow_run = rag_pipeline_service.get_rag_pipeline_workflow_run(pipeline=pipeline, run_id=run_id)
+        if workflow_run is None:
+            raise NotFound("Workflow run not found")
 
-        return workflow_run
+        return WorkflowRunDetailResponse.model_validate(workflow_run, from_attributes=True).model_dump(mode="json")
 
 
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflow-runs/<uuid:run_id>/node-executions")
 class RagPipelineWorkflowRunNodeExecutionListApi(Resource):
+    @console_ns.response(
+        200,
+        "Node executions retrieved successfully",
+        console_ns.models[WorkflowRunNodeExecutionListResponse.__name__],
+    )
     @setup_required
     @login_required
     @account_initialization_required
     @get_rag_pipeline
-    @marshal_with(workflow_run_node_execution_list_model)
     def get(self, pipeline: Pipeline, run_id: str):
         """
         Get workflow run node execution list
@@ -844,7 +871,9 @@ class RagPipelineWorkflowRunNodeExecutionListApi(Resource):
             user=user,
         )
 
-        return {"data": node_executions}
+        return WorkflowRunNodeExecutionListResponse.model_validate(
+            {"data": node_executions}, from_attributes=True
+        ).model_dump(mode="json")
 
 
 @console_ns.route("/rag/pipelines/datasource-plugins")
@@ -859,11 +888,15 @@ class DatasourceListApi(Resource):
 
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/nodes/<string:node_id>/last-run")
 class RagPipelineWorkflowLastRunApi(Resource):
+    @console_ns.response(
+        200,
+        "Node last run retrieved successfully",
+        console_ns.models[WorkflowRunNodeExecutionResponse.__name__],
+    )
     @setup_required
     @login_required
     @account_initialization_required
     @get_rag_pipeline
-    @marshal_with(workflow_run_node_execution_model)
     def get(self, pipeline: Pipeline, node_id: str):
         rag_pipeline_service = RagPipelineService()
         workflow = rag_pipeline_service.get_draft_workflow(pipeline=pipeline)
@@ -876,7 +909,7 @@ class RagPipelineWorkflowLastRunApi(Resource):
         )
         if node_exec is None:
             raise NotFound("last run not found")
-        return node_exec
+        return WorkflowRunNodeExecutionResponse.model_validate(node_exec, from_attributes=True).model_dump(mode="json")
 
 
 @console_ns.route("/rag/pipelines/transform/datasets/<uuid:dataset_id>")
@@ -899,12 +932,16 @@ class RagPipelineTransformApi(Resource):
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/datasource/variables-inspect")
 class RagPipelineDatasourceVariableApi(Resource):
     @console_ns.expect(console_ns.models[DatasourceVariablesPayload.__name__])
+    @console_ns.response(
+        200,
+        "Datasource variables set successfully",
+        console_ns.models[WorkflowRunNodeExecutionResponse.__name__],
+    )
     @setup_required
     @login_required
     @account_initialization_required
     @get_rag_pipeline
     @edit_permission_required
-    @marshal_with(workflow_run_node_execution_model)
     def post(self, pipeline: Pipeline):
         """
         Set datasource variables
@@ -918,7 +955,9 @@ class RagPipelineDatasourceVariableApi(Resource):
             args=args,
             current_user=current_user,
         )
-        return workflow_node_execution
+        return WorkflowRunNodeExecutionResponse.model_validate(
+            workflow_node_execution, from_attributes=True
+        ).model_dump(mode="json")
 
 
 @console_ns.route("/rag/pipelines/recommended-plugins")

api/controllers/console/explore/recommended_app.py

@@ -1,11 +1,12 @@
 from typing import Any
+from uuid import UUID
 
 from flask import request
 from flask_restx import Resource
 from pydantic import BaseModel, Field, computed_field, field_validator
 
 from constants.languages import languages
-from controllers.common.schema import register_schema_models
+from controllers.common.schema import query_params_from_model, register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required
 from fields.base import ResponseModel
@@ -15,7 +16,7 @@ from services.recommended_app_service import RecommendedAppService
 
 
 class RecommendedAppsQuery(BaseModel):
-    language: str | None = Field(default=None)
+    language: str | None = Field(default=None, description="Language code for recommended app localization")
 
 
 class RecommendedAppInfoResponse(ResponseModel):
@@ -74,13 +75,13 @@ register_schema_models(
 
 @console_ns.route("/explore/apps")
 class RecommendedAppListApi(Resource):
-    @console_ns.expect(console_ns.models[RecommendedAppsQuery.__name__])
+    @console_ns.doc(params=query_params_from_model(RecommendedAppsQuery))
     @console_ns.response(200, "Success", console_ns.models[RecommendedAppListResponse.__name__])
     @login_required
     @account_initialization_required
     def get(self):
         # language args
-        args = RecommendedAppsQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = RecommendedAppsQuery.model_validate(request.args.to_dict(flat=True))
         language = args.language
         if language and language in languages:
             language_prefix = language
@@ -99,6 +100,5 @@ class RecommendedAppListApi(Resource):
 class RecommendedAppApi(Resource):
     @login_required
     @account_initialization_required
-    def get(self, app_id):
-        app_id = str(app_id)
-        return RecommendedAppService.get_recommend_app_detail(app_id)
+    def get(self, app_id: UUID):
+        return RecommendedAppService.get_recommend_app_detail(str(app_id))

api/controllers/console/explore/trial.py

@@ -10,7 +10,7 @@ from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
 import services
 from controllers.common.fields import Parameters as ParametersResponse
 from controllers.common.fields import Site as SiteResponse
-from controllers.common.schema import get_or_create_model
+from controllers.common.schema import get_or_create_model, register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.error import (
     AppUnavailableError,
@@ -120,10 +120,6 @@ 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
@@ -153,19 +149,7 @@ class CompletionRequest(BaseModel):
     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)
-)
+register_schema_models(console_ns, WorkflowRunRequest, ChatRequest, TextToSpeechRequest, CompletionRequest)
 
 
 class TrialAppWorkflowRunApi(TrialAppResource):

api/controllers/console/extension.py

@@ -89,7 +89,7 @@ class CodeBasedExtensionAPI(Resource):
     @login_required
     @account_initialization_required
     def get(self):
-        query = CodeBasedExtensionQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        query = CodeBasedExtensionQuery.model_validate(request.args.to_dict(flat=True))
 
         return CodeBasedExtensionResponse(
             module=query.module,

api/controllers/console/files.py

@@ -82,7 +82,7 @@ class FileApi(Resource):
         try:
             upload_file = FileService(db.engine).upload_file(
                 filename=file.filename,
-                content=file.read(),
+                content=file.stream.read(),
                 mimetype=file.mimetype,
                 user=current_user,
                 source=source,

api/controllers/console/workspace/account.py

@@ -52,8 +52,6 @@ from services.account_service import AccountService
 from services.billing_service import BillingService
 from services.errors.account import CurrentPasswordIncorrectError as ServiceCurrentPasswordIncorrectError
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
 
 class AccountInitPayload(BaseModel):
     interface_language: str
@@ -161,27 +159,26 @@ class CheckEmailUniquePayload(BaseModel):
     email: EmailStr
 
 
-def reg(cls: type[BaseModel]):
-    console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
-
-
-reg(AccountInitPayload)
-reg(AccountNamePayload)
-reg(AccountAvatarPayload)
-reg(AccountAvatarQuery)
-reg(AccountInterfaceLanguagePayload)
-reg(AccountInterfaceThemePayload)
-reg(AccountTimezonePayload)
-reg(AccountPasswordPayload)
-reg(AccountDeletePayload)
-reg(AccountDeletionFeedbackPayload)
-reg(EducationActivatePayload)
-reg(EducationAutocompleteQuery)
-reg(ChangeEmailSendPayload)
-reg(ChangeEmailValidityPayload)
-reg(ChangeEmailResetPayload)
-reg(CheckEmailUniquePayload)
-register_schema_models(console_ns, AccountResponse)
+register_schema_models(
+    console_ns,
+    AccountResponse,
+    AccountInitPayload,
+    AccountNamePayload,
+    AccountAvatarPayload,
+    AccountAvatarQuery,
+    AccountInterfaceLanguagePayload,
+    AccountInterfaceThemePayload,
+    AccountTimezonePayload,
+    AccountPasswordPayload,
+    AccountDeletePayload,
+    AccountDeletionFeedbackPayload,
+    EducationActivatePayload,
+    EducationAutocompleteQuery,
+    ChangeEmailSendPayload,
+    ChangeEmailValidityPayload,
+    ChangeEmailResetPayload,
+    CheckEmailUniquePayload,
+)
 
 
 def _serialize_account(account) -> dict[str, Any]:
@@ -326,7 +323,7 @@ class AccountAvatarApi(Resource):
     @account_initialization_required
     def get(self):
         current_user, current_tenant_id = current_account_with_tenant()
-        args = AccountAvatarQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = AccountAvatarQuery.model_validate(request.args.to_dict(flat=True))
         avatar = args.avatar
 
         if avatar.startswith(("http://", "https://")):

api/controllers/console/workspace/endpoint.py

@@ -20,8 +20,6 @@ from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs.login import current_account_with_tenant, login_required
 from services.plugin.endpoint_service import EndpointService
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
 
 class EndpointCreatePayload(BaseModel):
     plugin_unique_identifier: str
@@ -80,10 +78,6 @@ 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))
-
-
 register_schema_models(
     console_ns,
     EndpointCreatePayload,
@@ -215,7 +209,7 @@ class EndpointListApi(Resource):
     def get(self):
         user, tenant_id = current_account_with_tenant()
 
-        args = EndpointListQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = EndpointListQuery.model_validate(request.args.to_dict(flat=True))
 
         page = args.page
         page_size = args.page_size
@@ -248,7 +242,7 @@ class EndpointListForSinglePluginApi(Resource):
     def get(self):
         user, tenant_id = current_account_with_tenant()
 
-        args = EndpointListForPluginQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = EndpointListForPluginQuery.model_validate(request.args.to_dict(flat=True))
 
         page = args.page
         page_size = args.page_size

api/controllers/console/workspace/members.py

@@ -33,8 +33,6 @@ from services.account_service import AccountService, RegisterService, TenantServ
 from services.errors.account import AccountAlreadyInTenantError
 from services.feature_service import FeatureService
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
 
 class MemberInvitePayload(BaseModel):
     emails: list[str] = Field(default_factory=list)
@@ -59,17 +57,17 @@ class OwnerTransferPayload(BaseModel):
     token: str
 
 
-def reg(cls: type[BaseModel]):
-    console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
-
-
-reg(MemberInvitePayload)
-reg(MemberRoleUpdatePayload)
-reg(OwnerTransferEmailPayload)
-reg(OwnerTransferCheckPayload)
-reg(OwnerTransferPayload)
 register_enum_models(console_ns, TenantAccountRole)
-register_schema_models(console_ns, AccountWithRole, AccountWithRoleList)
+register_schema_models(
+    console_ns,
+    AccountWithRole,
+    AccountWithRoleList,
+    MemberInvitePayload,
+    MemberRoleUpdatePayload,
+    OwnerTransferEmailPayload,
+    OwnerTransferCheckPayload,
+    OwnerTransferPayload,
+)
 
 
 @console_ns.route("/workspaces/current/members")

api/controllers/console/workspace/model_providers.py

@@ -5,6 +5,7 @@ from flask import request, send_file
 from flask_restx import Resource
 from pydantic import BaseModel, Field, field_validator
 
+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 graphon.model_runtime.entities.model_entities import ModelType
@@ -15,8 +16,6 @@ from libs.login import current_account_with_tenant, login_required
 from services.billing_service import BillingService
 from services.model_provider_service import ModelProviderService
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
 
 class ParserModelList(BaseModel):
     model_type: ModelType | None = None
@@ -75,18 +74,17 @@ class ParserPreferredProviderType(BaseModel):
     preferred_provider_type: Literal["system", "custom"]
 
 
-def reg(cls: type[BaseModel]):
-    console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
-
-
-reg(ParserModelList)
-reg(ParserCredentialId)
-reg(ParserCredentialCreate)
-reg(ParserCredentialUpdate)
-reg(ParserCredentialDelete)
-reg(ParserCredentialSwitch)
-reg(ParserCredentialValidate)
-reg(ParserPreferredProviderType)
+register_schema_models(
+    console_ns,
+    ParserModelList,
+    ParserCredentialId,
+    ParserCredentialCreate,
+    ParserCredentialUpdate,
+    ParserCredentialDelete,
+    ParserCredentialSwitch,
+    ParserCredentialValidate,
+    ParserPreferredProviderType,
+)
 
 
 @console_ns.route("/workspaces/current/model-providers")

api/controllers/console/workspace/models.py

@@ -17,7 +17,6 @@ from services.model_load_balancing_service import ModelLoadBalancingService
 from services.model_provider_service import ModelProviderService
 
 logger = logging.getLogger(__name__)
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
 
 class ParserGetDefault(BaseModel):
@@ -107,6 +106,12 @@ class ParserParameter(BaseModel):
     model: str
 
 
+class ParserSwitch(BaseModel):
+    model: str
+    model_type: ModelType
+    credential_id: str
+
+
 register_schema_models(
     console_ns,
     ParserGetDefault,
@@ -119,6 +124,7 @@ register_schema_models(
     ParserDeleteCredential,
     ParserParameter,
     Inner,
+    ParserSwitch,
 )
 
 register_enum_models(console_ns, ModelType)
@@ -133,7 +139,7 @@ class DefaultModelApi(Resource):
     def get(self):
         _, tenant_id = current_account_with_tenant()
 
-        args = ParserGetDefault.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = ParserGetDefault.model_validate(request.args.to_dict(flat=True))
 
         model_provider_service = ModelProviderService()
         default_model_entity = model_provider_service.get_default_model_of_model_type(
@@ -261,7 +267,7 @@ class ModelProviderModelCredentialApi(Resource):
     def get(self, provider: str):
         _, tenant_id = current_account_with_tenant()
 
-        args = ParserGetCredentials.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = ParserGetCredentials.model_validate(request.args.to_dict(flat=True))
 
         model_provider_service = ModelProviderService()
         current_credential = model_provider_service.get_model_credential(
@@ -387,17 +393,6 @@ class ModelProviderModelCredentialApi(Resource):
         return {"result": "success"}, 204
 
 
-class ParserSwitch(BaseModel):
-    model: str
-    model_type: ModelType
-    credential_id: str
-
-
-console_ns.schema_model(
-    ParserSwitch.__name__, ParserSwitch.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
-)
-
-
 @console_ns.route("/workspaces/current/model-providers/<path:provider>/models/credentials/switch")
 class ModelProviderModelCredentialSwitchApi(Resource):
     @console_ns.expect(console_ns.models[ParserSwitch.__name__])
@@ -468,9 +463,7 @@ class ParserValidate(BaseModel):
     credentials: dict[str, Any]
 
 
-console_ns.schema_model(
-    ParserValidate.__name__, ParserValidate.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
-)
+register_schema_models(console_ns, ParserSwitch, ParserValidate)
 
 
 @console_ns.route("/workspaces/current/model-providers/<path:provider>/models/credentials/validate")
@@ -515,7 +508,7 @@ class ModelProviderModelParameterRuleApi(Resource):
     @login_required
     @account_initialization_required
     def get(self, provider: str):
-        args = ParserParameter.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = ParserParameter.model_validate(request.args.to_dict(flat=True))
         _, tenant_id = current_account_with_tenant()
 
         model_provider_service = ModelProviderService()

api/controllers/console/workspace/plugin.py

@@ -177,7 +177,7 @@ def _read_upload_content(file: FileStorage, max_size: int) -> bytes:
     FileStorage.content_length is not reliable for multipart test uploads and may be zero even when
     content exists, so the controllers validate against the loaded bytes instead.
     """
-    content = file.read()
+    content = file.stream.read()
     if len(content) > max_size:
         raise ValueError("File size exceeds the maximum allowed size")
 
@@ -211,7 +211,7 @@ class PluginListApi(Resource):
     @account_initialization_required
     def get(self):
         _, tenant_id = current_account_with_tenant()
-        args = ParserList.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = ParserList.model_validate(request.args.to_dict(flat=True))
         try:
             plugins_with_total = PluginService.list_with_total(tenant_id, args.page, args.page_size)
         except PluginDaemonClientSideError as e:
@@ -261,7 +261,7 @@ class PluginIconApi(Resource):
     @console_ns.expect(console_ns.models[ParserIcon.__name__])
     @setup_required
     def get(self):
-        args = ParserIcon.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = ParserIcon.model_validate(request.args.to_dict(flat=True))
 
         try:
             icon_bytes, mimetype = PluginService.get_asset(args.tenant_id, args.filename)
@@ -279,7 +279,7 @@ class PluginAssetApi(Resource):
     @login_required
     @account_initialization_required
     def get(self):
-        args = ParserAsset.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = ParserAsset.model_validate(request.args.to_dict(flat=True))
 
         _, tenant_id = current_account_with_tenant()
         try:
@@ -421,7 +421,7 @@ class PluginFetchMarketplacePkgApi(Resource):
     @plugin_permission_required(install_required=True)
     def get(self):
         _, tenant_id = current_account_with_tenant()
-        args = ParserPluginIdentifierQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = ParserPluginIdentifierQuery.model_validate(request.args.to_dict(flat=True))
 
         try:
             return jsonable_encoder(
@@ -446,7 +446,7 @@ class PluginFetchManifestApi(Resource):
     def get(self):
         _, tenant_id = current_account_with_tenant()
 
-        args = ParserPluginIdentifierQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = ParserPluginIdentifierQuery.model_validate(request.args.to_dict(flat=True))
 
         try:
             return jsonable_encoder(
@@ -466,7 +466,7 @@ class PluginFetchInstallTasksApi(Resource):
     def get(self):
         _, tenant_id = current_account_with_tenant()
 
-        args = ParserTasks.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = ParserTasks.model_validate(request.args.to_dict(flat=True))
 
         try:
             return jsonable_encoder({"tasks": PluginService.fetch_install_tasks(tenant_id, args.page, args.page_size)})
@@ -660,7 +660,7 @@ class PluginFetchDynamicSelectOptionsApi(Resource):
         current_user, tenant_id = current_account_with_tenant()
         user_id = current_user.id
 
-        args = ParserDynamicOptions.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = ParserDynamicOptions.model_validate(request.args.to_dict(flat=True))
 
         try:
             options = PluginParameterService.get_dynamic_select_options(
@@ -822,7 +822,7 @@ class PluginReadmeApi(Resource):
     @account_initialization_required
     def get(self):
         _, tenant_id = current_account_with_tenant()
-        args = ParserReadme.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = ParserReadme.model_validate(request.args.to_dict(flat=True))
         return jsonable_encoder(
             {"readme": PluginService.fetch_plugin_readme(tenant_id, args.plugin_unique_identifier, args.language)}
         )

api/controllers/console/workspace/workspace.py

@@ -16,6 +16,7 @@ from controllers.common.errors import (
     TooManyFilesError,
     UnsupportedFileTypeError,
 )
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.admin import admin_required
 from controllers.console.error import AccountNotLinkTenantError
@@ -39,7 +40,6 @@ from services.file_service import FileService
 from services.workspace_service import WorkspaceService
 
 logger = logging.getLogger(__name__)
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
 
 class WorkspaceListQuery(BaseModel):
@@ -91,15 +91,14 @@ class TenantInfoResponse(ResponseModel):
         return value
 
 
-def reg(cls: type[BaseModel]):
-    console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
-
-
-reg(WorkspaceListQuery)
-reg(SwitchWorkspacePayload)
-reg(WorkspaceCustomConfigPayload)
-reg(WorkspaceInfoPayload)
-reg(TenantInfoResponse)
+register_schema_models(
+    console_ns,
+    WorkspaceListQuery,
+    SwitchWorkspacePayload,
+    WorkspaceCustomConfigPayload,
+    WorkspaceInfoPayload,
+    TenantInfoResponse,
+)
 
 provider_fields = {
     "provider_name": fields.String,
@@ -322,7 +321,7 @@ class WebappLogoWorkspaceApi(Resource):
         try:
             upload_file = FileService(db.engine).upload_file(
                 filename=file.filename,
-                content=file.read(),
+                content=file.stream.read(),
                 mimetype=file.mimetype,
                 user=current_user,
             )

api/controllers/files/image_preview.py

@@ -8,13 +8,12 @@ from werkzeug.exceptions import NotFound
 import services
 from controllers.common.errors import UnsupportedFileTypeError
 from controllers.common.file_response import enforce_download_for_html
+from controllers.common.schema import register_schema_models
 from controllers.files import files_ns
 from extensions.ext_database import db
 from services.account_service import TenantService
 from services.file_service import FileService
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
 
 class FileSignatureQuery(BaseModel):
     timestamp: str = Field(..., description="Unix timestamp used in the signature")
@@ -26,12 +25,7 @@ class FilePreviewQuery(FileSignatureQuery):
     as_attachment: bool = Field(default=False, description="Whether to download as attachment")
 
 
-files_ns.schema_model(
-    FileSignatureQuery.__name__, FileSignatureQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
-)
-files_ns.schema_model(
-    FilePreviewQuery.__name__, FilePreviewQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
-)
+register_schema_models(files_ns, FileSignatureQuery, FilePreviewQuery)
 
 
 @files_ns.route("/<uuid:file_id>/image-preview")
@@ -58,7 +52,7 @@ class ImagePreviewApi(Resource):
     def get(self, file_id):
         file_id = str(file_id)
 
-        args = FileSignatureQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = FileSignatureQuery.model_validate(request.args.to_dict(flat=True))
         timestamp = args.timestamp
         nonce = args.nonce
         sign = args.sign
@@ -100,7 +94,7 @@ class FilePreviewApi(Resource):
     def get(self, file_id):
         file_id = str(file_id)
 
-        args = FilePreviewQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = FilePreviewQuery.model_validate(request.args.to_dict(flat=True))
 
         try:
             generator, upload_file = FileService(db.engine).get_file_generator_by_file_id(

api/controllers/files/tool_files.py

@@ -7,12 +7,11 @@ from werkzeug.exceptions import Forbidden, NotFound
 
 from controllers.common.errors import UnsupportedFileTypeError
 from controllers.common.file_response import enforce_download_for_html
+from controllers.common.schema import register_schema_models
 from controllers.files import files_ns
 from core.tools.signature import verify_tool_file_signature
 from core.tools.tool_file_manager import ToolFileManager
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
 
 class ToolFileQuery(BaseModel):
     timestamp: str = Field(..., description="Unix timestamp")
@@ -21,9 +20,7 @@ class ToolFileQuery(BaseModel):
     as_attachment: bool = Field(default=False, description="Download as attachment")
 
 
-files_ns.schema_model(
-    ToolFileQuery.__name__, ToolFileQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
-)
+register_schema_models(files_ns, ToolFileQuery)
 
 
 @files_ns.route("/tools/<uuid:file_id>.<string:extension>")

api/controllers/files/upload.py

@@ -20,8 +20,6 @@ from ..console.wraps import setup_required
 from ..files import files_ns
 from ..inner_api.plugin.wraps import get_user
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
 
 class PluginUploadQuery(BaseModel):
     timestamp: str = Field(..., description="Unix timestamp for signature verification")
@@ -31,9 +29,8 @@ class PluginUploadQuery(BaseModel):
     user_id: str | None = Field(default=None, description="User identifier")
 
 
-files_ns.schema_model(
-    PluginUploadQuery.__name__, PluginUploadQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
-)
+register_schema_models(files_ns, PluginUploadQuery)
+
 
 register_schema_models(files_ns, FileResponse)
 
@@ -69,7 +66,7 @@ class PluginUploadFileApi(Resource):
             FileTooLargeError: File exceeds size limit
             UnsupportedFileTypeError: File type not supported
         """
-        args = PluginUploadQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = PluginUploadQuery.model_validate(request.args.to_dict(flat=True))
 
         file = request.files.get("file")
         if file is None:
@@ -103,7 +100,7 @@ class PluginUploadFileApi(Resource):
             tool_file = ToolFileManager().create_file_by_raw(
                 user_id=user.id,
                 tenant_id=tenant_id,
-                file_binary=file.read(),
+                file_binary=file.stream.read(),
                 mimetype=mimetype,
                 filename=filename,
                 conversation_id=None,

api/controllers/service_api/app/file.py

@@ -58,7 +58,7 @@ class FileApi(Resource):
         try:
             upload_file = FileService(db.engine).upload_file(
                 filename=file.filename,
-                content=file.read(),
+                content=file.stream.read(),
                 mimetype=file.mimetype,
                 user=end_user,
             )

api/controllers/service_api/dataset/dataset.py

@@ -6,7 +6,7 @@ from pydantic import BaseModel, Field, TypeAdapter, field_validator, model_valid
 from werkzeug.exceptions import Forbidden, NotFound
 
 import services
-from controllers.common.schema import register_schema_models
+from controllers.common.schema import register_enum_models, register_schema_models
 from controllers.console.wraps import edit_permission_required
 from controllers.service_api import service_api_ns
 from controllers.service_api.dataset.error import DatasetInUseError, DatasetNameDuplicateError, InvalidActionError
@@ -34,13 +34,7 @@ from services.tag_service import (
     UpdateTagPayload,
 )
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
-
-service_api_ns.schema_model(
-    DatasetPermissionEnum.__name__,
-    TypeAdapter(DatasetPermissionEnum).json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
+register_enum_models(service_api_ns, DatasetPermissionEnum)
 
 
 class DatasetCreatePayload(BaseModel):

api/controllers/service_api/dataset/document.py

@@ -77,9 +77,6 @@ class DocumentTextCreatePayload(BaseModel):
         return value
 
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
-
 class DocumentTextUpdate(BaseModel):
     name: str | None = None
     text: str | None = None
@@ -435,7 +432,7 @@ class DocumentAddByFileApi(DatasetApiResource):
             raise ValueError("current_user is required")
         upload_file = FileService(db.engine).upload_file(
             filename=file.filename,
-            content=file.read(),
+            content=file.stream.read(),
             mimetype=file.mimetype,
             user=current_user,
             source="datasets",
@@ -509,7 +506,7 @@ def _update_document_by_file(tenant_id: str, dataset_id: UUID, document_id: UUID
         try:
             upload_file = FileService(db.engine).upload_file(
                 filename=file.filename,
-                content=file.read(),
+                content=file.stream.read(),
                 mimetype=file.mimetype,
                 user=current_user,
                 source="datasets",

api/controllers/service_api/dataset/rag_pipeline/rag_pipeline_workflow.py

@@ -241,7 +241,7 @@ class KnowledgebasePipelineFileUploadApi(DatasetApiResource):
         try:
             upload_file = FileService(db.engine).upload_file(
                 filename=file.filename,
-                content=file.read(),
+                content=file.stream.read(),
                 mimetype=file.mimetype,
                 user=current_user,
             )

api/controllers/web/files.py

@@ -73,7 +73,7 @@ class FileApi(WebApiResource):
         try:
             upload_file = FileService(db.engine).upload_file(
                 filename=file.filename,
-                content=file.read(),
+                content=file.stream.read(),
                 mimetype=file.mimetype,
                 user=end_user,
                 source="datasets" if source == "datasets" else None,

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

@@ -75,7 +75,7 @@ class PromptTemplateConfigManager:
         if not config.get("prompt_type"):
             config["prompt_type"] = PromptTemplateEntity.PromptType.SIMPLE
 
-        prompt_type_vals = [typ.value for typ in PromptTemplateEntity.PromptType]
+        prompt_type_vals = list(PromptTemplateEntity.PromptType)
         if config["prompt_type"] not in prompt_type_vals:
             raise ValueError(f"prompt_type must be in {prompt_type_vals}")
 

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

@@ -842,24 +842,24 @@ class WorkflowResponseConverter:
             return []
 
         files: list[Mapping[str, Any]] = []
-        if isinstance(value, FileSegment):
-            files.append(value.value.to_dict())
-        elif isinstance(value, ArrayFileSegment):
-            files.extend([i.to_dict() for i in value.value])
-        elif isinstance(value, File):
-            files.append(value.to_dict())
-        elif isinstance(value, list):
-            for item in value:
-                file = cls._get_file_var_from_value(item)
+        match value:
+            case FileSegment():
+                files.append(value.value.to_dict())
+            case ArrayFileSegment():
+                files.extend([i.to_dict() for i in value.value])
+            case File():
+                files.append(value.to_dict())
+            case list():
+                for item in value:
+                    file = cls._get_file_var_from_value(item)
+                    if file:
+                        files.append(file)
+            case dict():
+                file = cls._get_file_var_from_value(value)
                 if file:
                     files.append(file)
-        elif isinstance(
-            value,
-            dict,
-        ):
-            file = cls._get_file_var_from_value(value)
-            if file:
-                files.append(file)
+            case _:
+                pass
 
         return files
 

api/core/app/llm/__init__.py

@@ -1,5 +1,15 @@
 """LLM-related application services."""
 
-from .quota import deduct_llm_quota, ensure_llm_quota_available
+from .quota import (
+    deduct_llm_quota,
+    deduct_llm_quota_for_model,
+    ensure_llm_quota_available,
+    ensure_llm_quota_available_for_model,
+)
 
-__all__ = ["deduct_llm_quota", "ensure_llm_quota_available"]
+__all__ = [
+    "deduct_llm_quota",
+    "deduct_llm_quota_for_model",
+    "ensure_llm_quota_available",
+    "ensure_llm_quota_available_for_model",
+]

api/core/app/llm/quota.py

@@ -1,4 +1,14 @@
-from sqlalchemy import update
+"""Tenant-scoped helpers for checking and deducting LLM provider quota.
+
+System-hosted quota accounting is currently defined only for LLM models. Keep
+the public helpers LLM-specific so callers do not carry unused model-type
+plumbing, and fail loudly if the deprecated ``ModelInstance`` wrappers are used
+with a non-LLM model.
+"""
+
+import warnings
+
+from sqlalchemy import select
 from sqlalchemy.orm import sessionmaker
 
 from configs import dify_config
@@ -6,44 +16,47 @@ from core.entities.model_entities import ModelStatus
 from core.entities.provider_entities import ProviderQuotaType, QuotaUnit
 from core.errors.error import QuotaExceededError
 from core.model_manager import ModelInstance
+from core.plugin.impl.model_runtime_factory import create_plugin_provider_manager
 from extensions.ext_database import db
 from graphon.model_runtime.entities.llm_entities import LLMUsage
+from graphon.model_runtime.entities.model_entities import ModelType
 from libs.datetime_utils import naive_utc_now
 from models.provider import Provider, ProviderType
 from models.provider_ids import ModelProviderID
 
 
-def ensure_llm_quota_available(*, model_instance: ModelInstance) -> None:
-    provider_model_bundle = model_instance.provider_model_bundle
-    provider_configuration = provider_model_bundle.configuration
+def _get_provider_configuration(*, tenant_id: str, provider: str):
+    """Resolve the tenant-bound provider configuration for quota decisions."""
+    provider_manager = create_plugin_provider_manager(tenant_id=tenant_id)
+    provider_configuration = provider_manager.get_configurations(tenant_id).get(provider)
+    if provider_configuration is None:
+        raise ValueError(f"Provider {provider} does not exist.")
+    return provider_configuration
 
+
+def ensure_llm_quota_available_for_model(*, tenant_id: str, provider: str, model: str) -> None:
+    """Raise when a tenant-bound LLM model is already out of quota."""
+    provider_configuration = _get_provider_configuration(tenant_id=tenant_id, provider=provider)
     if provider_configuration.using_provider_type != ProviderType.SYSTEM:
         return
 
     provider_model = provider_configuration.get_provider_model(
-        model_type=model_instance.model_type_instance.model_type,
-        model=model_instance.model_name,
+        model_type=ModelType.LLM,
+        model=model,
     )
     if provider_model and provider_model.status == ModelStatus.QUOTA_EXCEEDED:
-        raise QuotaExceededError(f"Model provider {model_instance.provider} quota exceeded.")
+        raise QuotaExceededError(f"Model provider {provider} quota exceeded.")
 
 
-def deduct_llm_quota(*, tenant_id: str, model_instance: ModelInstance, usage: LLMUsage) -> None:
-    provider_model_bundle = model_instance.provider_model_bundle
-    provider_configuration = provider_model_bundle.configuration
-
-    if provider_configuration.using_provider_type != ProviderType.SYSTEM:
-        return
-
-    system_configuration = provider_configuration.system_configuration
-
+def _resolve_llm_used_quota(*, system_configuration, model: str, usage: LLMUsage) -> int | None:
+    """Compute the quota impact for an LLM invocation under the current quota mode."""
     quota_unit = None
     for quota_configuration in system_configuration.quota_configurations:
         if quota_configuration.quota_type == system_configuration.current_quota_type:
             quota_unit = quota_configuration.quota_unit
 
             if quota_configuration.quota_limit == -1:
-                return
+                return None
 
             break
 
@@ -52,42 +65,136 @@ def deduct_llm_quota(*, tenant_id: str, model_instance: ModelInstance, usage: LL
         if quota_unit == QuotaUnit.TOKENS:
             used_quota = usage.total_tokens
         elif quota_unit == QuotaUnit.CREDITS:
-            used_quota = dify_config.get_model_credits(model_instance.model_name)
+            used_quota = dify_config.get_model_credits(model)
         else:
             used_quota = 1
 
+    return used_quota
+
+
+def _deduct_free_llm_quota(
+    *,
+    tenant_id: str,
+    provider: str,
+    quota_type: ProviderQuotaType,
+    used_quota: int,
+) -> None:
+    """Deduct FREE provider quota, capping at the limit before reporting exhaustion."""
+    quota_exceeded = False
+    with sessionmaker(bind=db.engine).begin() as session:
+        provider_record = session.scalar(
+            select(Provider)
+            .where(
+                Provider.tenant_id == tenant_id,
+                # TODO: Use provider name with prefix after the data migration.
+                Provider.provider_name == ModelProviderID(provider).provider_name,
+                Provider.provider_type == ProviderType.SYSTEM.value,
+                Provider.quota_type == quota_type,
+            )
+            .with_for_update()
+        )
+        if (
+            provider_record is None
+            or provider_record.quota_limit is None
+            or provider_record.quota_used is None
+            or provider_record.quota_limit <= provider_record.quota_used
+        ):
+            quota_exceeded = True
+        else:
+            available_quota = provider_record.quota_limit - provider_record.quota_used
+            deducted_quota = min(used_quota, available_quota)
+            provider_record.quota_used += deducted_quota
+            provider_record.last_used = naive_utc_now()
+            quota_exceeded = deducted_quota < used_quota
+
+    if quota_exceeded:
+        raise QuotaExceededError(f"Model provider {provider} quota exceeded.")
+
+
+def _deduct_used_llm_quota(*, tenant_id: str, provider: str, provider_configuration, used_quota: int | None) -> None:
+    """Apply a resolved LLM quota charge against the current provider quota bucket."""
+    if provider_configuration.using_provider_type != ProviderType.SYSTEM:
+        return
+
+    system_configuration = provider_configuration.system_configuration
     if used_quota is not None and system_configuration.current_quota_type is not None:
         match system_configuration.current_quota_type:
             case ProviderQuotaType.TRIAL:
                 from services.credit_pool_service import CreditPoolService
 
-                CreditPoolService.check_and_deduct_credits(
+                CreditPoolService.deduct_credits_capped(
                     tenant_id=tenant_id,
                     credits_required=used_quota,
                 )
             case ProviderQuotaType.PAID:
                 from services.credit_pool_service import CreditPoolService
 
-                CreditPoolService.check_and_deduct_credits(
+                CreditPoolService.deduct_credits_capped(
                     tenant_id=tenant_id,
                     credits_required=used_quota,
                     pool_type="paid",
                 )
             case ProviderQuotaType.FREE:
-                with sessionmaker(bind=db.engine).begin() as session:
-                    stmt = (
-                        update(Provider)
-                        .where(
-                            Provider.tenant_id == tenant_id,
-                            # TODO: Use provider name with prefix after the data migration.
-                            Provider.provider_name == ModelProviderID(model_instance.provider).provider_name,
-                            Provider.provider_type == ProviderType.SYSTEM.value,
-                            Provider.quota_type == system_configuration.current_quota_type,
-                            Provider.quota_limit > Provider.quota_used,
-                        )
-                        .values(
-                            quota_used=Provider.quota_used + used_quota,
-                            last_used=naive_utc_now(),
-                        )
-                    )
-                    session.execute(stmt)
+                _deduct_free_llm_quota(
+                    tenant_id=tenant_id,
+                    provider=provider,
+                    quota_type=system_configuration.current_quota_type,
+                    used_quota=used_quota,
+                )
+            case _:
+                return
+
+
+def deduct_llm_quota_for_model(*, tenant_id: str, provider: str, model: str, usage: LLMUsage) -> None:
+    """Deduct tenant-bound quota for the resolved LLM model identity."""
+    provider_configuration = _get_provider_configuration(tenant_id=tenant_id, provider=provider)
+    used_quota = _resolve_llm_used_quota(
+        system_configuration=provider_configuration.system_configuration,
+        model=model,
+        usage=usage,
+    )
+    _deduct_used_llm_quota(
+        tenant_id=tenant_id,
+        provider=provider,
+        provider_configuration=provider_configuration,
+        used_quota=used_quota,
+    )
+
+
+def _require_llm_model_instance(model_instance: ModelInstance) -> None:
+    """Reject deprecated wrapper calls that pass a non-LLM model instance."""
+    if model_instance.model_type_instance.model_type != ModelType.LLM:
+        raise ValueError("LLM quota helpers only support LLM model instances.")
+
+
+def ensure_llm_quota_available(*, model_instance: ModelInstance) -> None:
+    """Deprecated compatibility wrapper for callers that still pass ModelInstance."""
+    warnings.warn(
+        "ensure_llm_quota_available(model_instance=...) is deprecated; "
+        "use ensure_llm_quota_available_for_model(...) instead.",
+        DeprecationWarning,
+        stacklevel=2,
+    )
+    _require_llm_model_instance(model_instance)
+    ensure_llm_quota_available_for_model(
+        tenant_id=model_instance.provider_model_bundle.configuration.tenant_id,
+        provider=model_instance.provider,
+        model=model_instance.model_name,
+    )
+
+
+def deduct_llm_quota(*, tenant_id: str, model_instance: ModelInstance, usage: LLMUsage) -> None:
+    """Deprecated compatibility wrapper for callers that still pass ModelInstance."""
+    warnings.warn(
+        "deduct_llm_quota(tenant_id=..., model_instance=..., usage=...) is deprecated; "
+        "use deduct_llm_quota_for_model(...) instead.",
+        DeprecationWarning,
+        stacklevel=2,
+    )
+    _require_llm_model_instance(model_instance)
+    deduct_llm_quota_for_model(
+        tenant_id=tenant_id,
+        provider=model_instance.provider,
+        model=model_instance.model_name,
+        usage=usage,
+    )

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

@@ -1,36 +1,48 @@
 """
 LLM quota deduction layer for GraphEngine.
 
-This layer centralizes model-quota deduction outside node implementations.
+This layer centralizes model-quota handling outside node implementations.
+
+Graphon LLM-backed nodes expose provider/model identity through public node
+configuration and, after execution, through ``node_run_result.inputs``. Resolve
+quota billing from that public identity instead of depending on
+``ModelInstance`` reconstruction inside the workflow layer. Missing identity on
+quota-tracked nodes is treated as a workflow bug and aborts execution so quota
+handling is never silently skipped.
 """
 
 import logging
-from typing import TYPE_CHECKING, cast, final, override
+from typing import final, override
 
-from core.app.entities.app_invoke_entities import DIFY_RUN_CONTEXT_KEY, DifyRunContext
-from core.app.llm import deduct_llm_quota, ensure_llm_quota_available
+from core.app.llm import deduct_llm_quota_for_model, ensure_llm_quota_available_for_model
 from core.errors.error import QuotaExceededError
-from core.model_manager import ModelInstance
-from graphon.enums import BuiltinNodeTypes
+from graphon.enums import BuiltinNodeTypes, WorkflowNodeExecutionStatus
 from graphon.graph_engine.entities.commands import AbortCommand, CommandType
 from graphon.graph_engine.layers import GraphEngineLayer
 from graphon.graph_events import GraphEngineEvent, GraphNodeEventBase, NodeRunSucceededEvent
+from graphon.node_events import NodeRunResult
 from graphon.nodes.base.node import Node
 
-if TYPE_CHECKING:
-    from graphon.nodes.llm.node import LLMNode
-    from graphon.nodes.parameter_extractor.parameter_extractor_node import ParameterExtractorNode
-    from graphon.nodes.question_classifier.question_classifier_node import QuestionClassifierNode
-
 logger = logging.getLogger(__name__)
+_QUOTA_NODE_TYPES = frozenset(
+    [
+        BuiltinNodeTypes.LLM,
+        BuiltinNodeTypes.PARAMETER_EXTRACTOR,
+        BuiltinNodeTypes.QUESTION_CLASSIFIER,
+    ]
+)
 
 
 @final
 class LLMQuotaLayer(GraphEngineLayer):
-    """Graph layer that applies LLM quota deduction after node execution."""
+    """Graph layer that applies tenant-scoped quota checks to LLM-backed nodes."""
 
-    def __init__(self) -> None:
+    tenant_id: str
+    _abort_sent: bool
+
+    def __init__(self, tenant_id: str) -> None:
         super().__init__()
+        self.tenant_id = tenant_id
         self._abort_sent = False
 
     @override
@@ -50,33 +62,49 @@ class LLMQuotaLayer(GraphEngineLayer):
         if self._abort_sent:
             return
 
-        model_instance = self._extract_model_instance(node)
-        if model_instance is None:
+        if not self._supports_quota(node):
             return
 
+        model_identity = self._extract_model_identity_from_node(node)
+        if model_identity is None:
+            reason = "LLM quota check requires public node model identity before execution."
+            self._abort_before_node_run(node=node, reason=reason, error_type="LLMQuotaIdentityError")
+            logger.error("LLM quota handling aborted, node_id=%s, reason=%s", node.id, reason)
+            return
+
+        provider, model_name = model_identity
         try:
-            ensure_llm_quota_available(model_instance=model_instance)
+            ensure_llm_quota_available_for_model(
+                tenant_id=self.tenant_id,
+                provider=provider,
+                model=model_name,
+            )
         except QuotaExceededError as exc:
-            self._set_stop_event(node)
-            self._send_abort_command(reason=str(exc))
+            self._abort_before_node_run(node=node, reason=str(exc), error_type=QuotaExceededError.__name__)
             logger.warning("LLM quota check failed, node_id=%s, error=%s", node.id, exc)
 
     @override
     def on_node_run_end(
         self, node: Node, error: Exception | None, result_event: GraphNodeEventBase | None = None
     ) -> None:
-        if error is not None or not isinstance(result_event, NodeRunSucceededEvent):
+        if error is not None or not isinstance(result_event, NodeRunSucceededEvent) or not self._supports_quota(node):
             return
 
-        model_instance = self._extract_model_instance(node)
-        if model_instance is None:
+        model_identity = self._extract_model_identity_from_result_event(result_event)
+        if model_identity is None:
+            self._abort_for_missing_model_identity(
+                node=node,
+                reason="LLM quota deduction requires model identity in the node result event.",
+            )
             return
 
+        provider, model_name = model_identity
+
         try:
-            dify_ctx = DifyRunContext.model_validate(node.require_run_context_value(DIFY_RUN_CONTEXT_KEY))
-            deduct_llm_quota(
-                tenant_id=dify_ctx.tenant_id,
-                model_instance=model_instance,
+            deduct_llm_quota_for_model(
+                tenant_id=self.tenant_id,
+                provider=provider,
+                model=model_name,
                 usage=result_event.node_run_result.llm_usage,
             )
         except QuotaExceededError as exc:
@@ -92,6 +120,27 @@ class LLMQuotaLayer(GraphEngineLayer):
         if stop_event is not None:
             stop_event.set()
 
+    def _abort_before_node_run(self, *, node: Node, reason: str, error_type: str) -> None:
+        self._set_stop_event(node)
+        node.node_data.error_strategy = None
+        node.node_data.retry_config.retry_enabled = False
+
+        def quota_aborted_run() -> NodeRunResult:
+            return NodeRunResult(
+                status=WorkflowNodeExecutionStatus.FAILED,
+                error=reason,
+                error_type=error_type,
+            )
+
+        # TODO: Push Graphon to expose a public pre-run failure/skip hook, then replace this private _run override.
+        node._run = quota_aborted_run  # type: ignore[method-assign]
+        self._send_abort_command(reason=reason)
+
+    def _abort_for_missing_model_identity(self, *, node: Node, reason: str) -> None:
+        self._set_stop_event(node)
+        self._send_abort_command(reason=reason)
+        logger.error("LLM quota handling aborted, node_id=%s, reason=%s", node.id, reason)
+
     def _send_abort_command(self, *, reason: str) -> None:
         if not self.command_channel or self._abort_sent:
             return
@@ -108,29 +157,38 @@ class LLMQuotaLayer(GraphEngineLayer):
             logger.exception("Failed to send quota abort command")
 
     @staticmethod
-    def _extract_model_instance(node: Node) -> ModelInstance | None:
-        try:
-            match node.node_type:
-                case BuiltinNodeTypes.LLM:
-                    model_instance = cast("LLMNode", node).model_instance
-                case BuiltinNodeTypes.PARAMETER_EXTRACTOR:
-                    model_instance = cast("ParameterExtractorNode", node).model_instance
-                case BuiltinNodeTypes.QUESTION_CLASSIFIER:
-                    model_instance = cast("QuestionClassifierNode", node).model_instance
-                case _:
-                    return None
-        except AttributeError:
+    def _supports_quota(node: Node) -> bool:
+        return node.node_type in _QUOTA_NODE_TYPES
+
+    @staticmethod
+    def _extract_model_identity_from_result_event(result_event: NodeRunSucceededEvent) -> tuple[str, str] | None:
+        provider = result_event.node_run_result.inputs.get("model_provider")
+        model_name = result_event.node_run_result.inputs.get("model_name")
+        if isinstance(provider, str) and provider and isinstance(model_name, str) and model_name:
+            return provider, model_name
+        return None
+
+    @staticmethod
+    def _extract_model_identity_from_node(node: Node) -> tuple[str, str] | None:
+        node_data = getattr(node, "node_data", None)
+        if node_data is None:
+            node_data = getattr(node, "data", None)
+
+        model_config = getattr(node_data, "model", None)
+        if model_config is None:
             logger.warning(
-                "LLMQuotaLayer skipped quota deduction because node does not expose a model instance, node_id=%s",
+                "LLMQuotaLayer skipped quota handling because node model config is missing, node_id=%s",
                 node.id,
             )
             return None
 
-        if isinstance(model_instance, ModelInstance):
-            return model_instance
-
-        raw_model_instance = getattr(model_instance, "_model_instance", None)
-        if isinstance(raw_model_instance, ModelInstance):
-            return raw_model_instance
+        provider = getattr(model_config, "provider", None)
+        model_name = getattr(model_config, "name", None)
+        if isinstance(provider, str) and provider and isinstance(model_name, str) and model_name:
+            return provider, model_name
 
+        logger.warning(
+            "LLMQuotaLayer skipped quota handling because node model identity is invalid, node_id=%s",
+            node.id,
+        )
         return None

api/core/entities/provider_configuration.py

@@ -23,7 +23,7 @@ from core.entities.provider_entities import (
 )
 from core.helper import encrypter
 from core.helper.model_provider_cache import ProviderCredentialsCache, ProviderCredentialsCacheType
-from core.plugin.impl.model_runtime_factory import create_plugin_model_provider_factory
+from core.plugin.impl.model_runtime_factory import create_model_type_instance, create_plugin_model_assembly
 from graphon.model_runtime.entities.model_entities import AIModelEntity, FetchFrom, ModelType
 from graphon.model_runtime.entities.provider_entities import (
     ConfigurateMethod,
@@ -33,7 +33,7 @@ from graphon.model_runtime.entities.provider_entities import (
 )
 from graphon.model_runtime.model_providers.base.ai_model import AIModel
 from graphon.model_runtime.model_providers.model_provider_factory import ModelProviderFactory
-from graphon.model_runtime.runtime import ModelRuntime
+from graphon.model_runtime.protocols.runtime import ModelRuntime
 from libs.datetime_utils import naive_utc_now
 from models.engine import db
 from models.enums import CredentialSourceType
@@ -106,11 +106,18 @@ class ProviderConfiguration(BaseModel):
         """Attach the already-composed runtime for request-bound call chains."""
         self._bound_model_runtime = model_runtime
 
+    def _get_runtime_and_provider_factory(self) -> tuple[ModelRuntime, ModelProviderFactory]:
+        """Resolve a provider factory that stays aligned with the runtime used by the caller."""
+        if self._bound_model_runtime is not None:
+            return self._bound_model_runtime, ModelProviderFactory(runtime=self._bound_model_runtime)
+
+        model_assembly = create_plugin_model_assembly(tenant_id=self.tenant_id)
+        return model_assembly.model_runtime, model_assembly.model_provider_factory
+
     def get_model_provider_factory(self) -> ModelProviderFactory:
         """Return a provider factory that preserves any request-bound runtime."""
-        if self._bound_model_runtime is not None:
-            return ModelProviderFactory(model_runtime=self._bound_model_runtime)
-        return create_plugin_model_provider_factory(tenant_id=self.tenant_id)
+        _, model_provider_factory = self._get_runtime_and_provider_factory()
+        return model_provider_factory
 
     def get_current_credentials(self, model_type: ModelType, model: str) -> dict[str, Any] | None:
         """
@@ -1392,10 +1399,13 @@ class ProviderConfiguration(BaseModel):
         :param model_type: model type
         :return:
         """
-        model_provider_factory = self.get_model_provider_factory()
-
-        # Get model instance of LLM
-        return model_provider_factory.get_model_type_instance(provider=self.provider.provider, model_type=model_type)
+        model_runtime, model_provider_factory = self._get_runtime_and_provider_factory()
+        provider_schema = model_provider_factory.get_provider_schema(provider=self.provider.provider)
+        return create_model_type_instance(
+            runtime=model_runtime,
+            provider_schema=provider_schema,
+            model_type=model_type,
+        )
 
     def get_model_schema(
         self, model_type: ModelType, model: str, credentials: dict[str, Any] | None

api/core/helper/moderation.py

@@ -4,7 +4,7 @@ from typing import cast
 
 from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
 from core.entities import DEFAULT_PLUGIN_ID
-from core.plugin.impl.model_runtime_factory import create_plugin_model_provider_factory
+from core.plugin.impl.model_runtime_factory import create_plugin_model_assembly
 from extensions.ext_hosting_provider import hosting_configuration
 from graphon.model_runtime.entities.model_entities import ModelType
 from graphon.model_runtime.errors.invoke import InvokeBadRequestError
@@ -41,10 +41,8 @@ def check_moderation(tenant_id: str, model_config: ModelConfigWithCredentialsEnt
             text_chunk = secrets.choice(text_chunks)
 
             try:
-                model_provider_factory = create_plugin_model_provider_factory(tenant_id=tenant_id)
-
-                # Get model instance of LLM
-                model_type_instance = model_provider_factory.get_model_type_instance(
+                model_assembly = create_plugin_model_assembly(tenant_id=tenant_id)
+                model_type_instance = model_assembly.create_model_type_instance(
                     provider=openai_provider_name, model_type=ModelType.MODERATION
                 )
                 model_type_instance = cast(ModerationModel, model_type_instance)

api/core/ops/ops_trace_manager.py

@@ -569,13 +569,13 @@ class OpsTraceManager:
         db.session.commit()
 
     @classmethod
-    def get_app_tracing_config(cls, app_id: str):
+    def get_app_tracing_config(cls, app_id: str, session: Session):
         """
         Get app tracing config
         :param app_id: app id
         :return:
         """
-        app: App | None = db.session.get(App, app_id)
+        app: App | None = session.get(App, app_id)
         if not app:
             raise ValueError("App not found")
         if not app.tracing:

api/core/plugin/impl/model_runtime.py

@@ -4,23 +4,32 @@ import hashlib
 import logging
 from collections.abc import Generator, Iterable, Sequence
 from threading import Lock
-from typing import IO, Any, Union
+from typing import IO, Any, Literal, cast, overload
 
 from pydantic import ValidationError
 from redis import RedisError
 
 from configs import dify_config
+from core.llm_generator.output_parser.structured_output import (
+    invoke_llm_with_structured_output as invoke_llm_with_structured_output_helper,
+)
 from core.plugin.entities.plugin_daemon import PluginModelProviderEntity
 from core.plugin.impl.asset import PluginAssetManager
 from core.plugin.impl.model import PluginModelClient
 from extensions.ext_redis import redis_client
-from graphon.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk
+from graphon.model_runtime.entities.llm_entities import (
+    LLMResult,
+    LLMResultChunk,
+    LLMResultChunkWithStructuredOutput,
+    LLMResultWithStructuredOutput,
+)
 from graphon.model_runtime.entities.message_entities import PromptMessage, PromptMessageTool
 from graphon.model_runtime.entities.model_entities import AIModelEntity, ModelType
 from graphon.model_runtime.entities.provider_entities import ProviderEntity
 from graphon.model_runtime.entities.rerank_entities import MultimodalRerankInput, RerankResult
 from graphon.model_runtime.entities.text_embedding_entities import EmbeddingInputType, EmbeddingResult
-from graphon.model_runtime.runtime import ModelRuntime
+from graphon.model_runtime.model_providers.base.large_language_model import normalize_non_stream_runtime_result
+from graphon.model_runtime.protocols.runtime import ModelRuntime
 from models.provider_ids import ModelProviderID
 
 logger = logging.getLogger(__name__)
@@ -29,6 +38,68 @@ logger = logging.getLogger(__name__)
 TENANT_SCOPE_SCHEMA_CACHE_USER_ID = "__DIFY_TS__"
 
 
+# TODO(-LAN-): Move native structured-output invocation into Graphon's LLM node.
+# TODO(-LAN-): Remove this Dify-side adapter once Graphon owns structured output end-to-end.
+class _PluginStructuredOutputModelInstance:
+    """Bind plugin model identity to the shared structured-output helper.
+
+    The structured-output parser is shared with legacy ``ModelInstance`` flows
+    and only needs an object exposing ``invoke_llm(...)``. ``PluginModelRuntime``
+    intentionally exposes a lower-level API where provider, model, and
+    credentials are passed per call. This adapter supplies the small bound
+    ``invoke_llm`` surface the helper needs without constructing a full
+    ``ModelInstance`` or reintroducing model-manager dependencies into the
+    plugin runtime path.
+    """
+
+    def __init__(
+        self,
+        *,
+        runtime: PluginModelRuntime,
+        provider: str,
+        model: str,
+        credentials: dict[str, Any],
+    ) -> None:
+        self._runtime = runtime
+        self._provider = provider
+        self._model = model
+        self._credentials = credentials
+
+    def invoke_llm(
+        self,
+        *,
+        prompt_messages: Sequence[PromptMessage],
+        model_parameters: dict[str, Any] | None = None,
+        tools: Sequence[PromptMessageTool] | None = None,
+        stop: Sequence[str] | None = None,
+        stream: bool = True,
+        callbacks: object | None = None,
+    ) -> LLMResult | Generator[LLMResultChunk, None, None]:
+        del callbacks
+        if stream:
+            return self._runtime.invoke_llm(
+                provider=self._provider,
+                model=self._model,
+                credentials=self._credentials,
+                model_parameters=model_parameters or {},
+                prompt_messages=prompt_messages,
+                tools=list(tools) if tools else None,
+                stop=stop,
+                stream=True,
+            )
+
+        return self._runtime.invoke_llm(
+            provider=self._provider,
+            model=self._model,
+            credentials=self._credentials,
+            model_parameters=model_parameters or {},
+            prompt_messages=prompt_messages,
+            tools=list(tools) if tools else None,
+            stop=stop,
+            stream=False,
+        )
+
+
 class PluginModelRuntime(ModelRuntime):
     """Plugin-backed runtime adapter bound to tenant context and optional caller scope."""
 
@@ -195,6 +266,34 @@ class PluginModelRuntime(ModelRuntime):
 
         return schema
 
+    @overload
+    def invoke_llm(
+        self,
+        *,
+        provider: str,
+        model: str,
+        credentials: dict[str, Any],
+        model_parameters: dict[str, Any],
+        prompt_messages: Sequence[PromptMessage],
+        tools: list[PromptMessageTool] | None,
+        stop: Sequence[str] | None,
+        stream: Literal[False],
+    ) -> LLMResult: ...
+
+    @overload
+    def invoke_llm(
+        self,
+        *,
+        provider: str,
+        model: str,
+        credentials: dict[str, Any],
+        model_parameters: dict[str, Any],
+        prompt_messages: Sequence[PromptMessage],
+        tools: list[PromptMessageTool] | None,
+        stop: Sequence[str] | None,
+        stream: Literal[True],
+    ) -> Generator[LLMResultChunk, None, None]: ...
+
     def invoke_llm(
         self,
         *,
@@ -206,9 +305,9 @@ class PluginModelRuntime(ModelRuntime):
         tools: list[PromptMessageTool] | None,
         stop: Sequence[str] | None,
         stream: bool,
-    ) -> Union[LLMResult, Generator[LLMResultChunk, None, None]]:
+    ) -> LLMResult | Generator[LLMResultChunk, None, None]:
         plugin_id, provider_name = self._split_provider(provider)
-        return self.client.invoke_llm(
+        result = self.client.invoke_llm(
             tenant_id=self.tenant_id,
             user_id=self.user_id,
             plugin_id=plugin_id,
@@ -221,6 +320,81 @@ class PluginModelRuntime(ModelRuntime):
             stop=list(stop) if stop else None,
             stream=stream,
         )
+        if stream:
+            return result
+
+        return normalize_non_stream_runtime_result(
+            model=model,
+            prompt_messages=prompt_messages,
+            result=result,
+        )
+
+    @overload
+    def invoke_llm_with_structured_output(
+        self,
+        *,
+        provider: str,
+        model: str,
+        credentials: dict[str, Any],
+        json_schema: dict[str, Any],
+        model_parameters: dict[str, Any],
+        prompt_messages: Sequence[PromptMessage],
+        stop: Sequence[str] | None,
+        stream: Literal[False],
+    ) -> LLMResultWithStructuredOutput: ...
+
+    @overload
+    def invoke_llm_with_structured_output(
+        self,
+        *,
+        provider: str,
+        model: str,
+        credentials: dict[str, Any],
+        json_schema: dict[str, Any],
+        model_parameters: dict[str, Any],
+        prompt_messages: Sequence[PromptMessage],
+        stop: Sequence[str] | None,
+        stream: Literal[True],
+    ) -> Generator[LLMResultChunkWithStructuredOutput, None, None]: ...
+
+    def invoke_llm_with_structured_output(
+        self,
+        *,
+        provider: str,
+        model: str,
+        credentials: dict[str, Any],
+        json_schema: dict[str, Any],
+        model_parameters: dict[str, Any],
+        prompt_messages: Sequence[PromptMessage],
+        stop: Sequence[str] | None,
+        stream: bool,
+    ) -> LLMResultWithStructuredOutput | Generator[LLMResultChunkWithStructuredOutput, None, None]:
+        model_schema = self.get_model_schema(
+            provider=provider,
+            model_type=ModelType.LLM,
+            model=model,
+            credentials=credentials,
+        )
+        if model_schema is None:
+            raise ValueError(f"Model schema not found for {model}")
+
+        adapter = _PluginStructuredOutputModelInstance(
+            runtime=self,
+            provider=provider,
+            model=model,
+            credentials=credentials,
+        )
+        return invoke_llm_with_structured_output_helper(
+            provider=provider,
+            model_schema=model_schema,
+            model_instance=cast(Any, adapter),
+            prompt_messages=prompt_messages,
+            json_schema=json_schema,
+            model_parameters=model_parameters,
+            tools=None,
+            stop=list(stop) if stop else None,
+            stream=stream,
+        )
 
     def get_llm_num_tokens(
         self,

api/core/plugin/impl/model_runtime_factory.py

@@ -3,13 +3,46 @@ from __future__ import annotations
 from typing import TYPE_CHECKING
 
 from core.plugin.impl.model import PluginModelClient
+from graphon.model_runtime.entities.model_entities import ModelType
+from graphon.model_runtime.entities.provider_entities import ProviderEntity
+from graphon.model_runtime.model_providers.base.ai_model import AIModel
+from graphon.model_runtime.model_providers.base.large_language_model import LargeLanguageModel
+from graphon.model_runtime.model_providers.base.moderation_model import ModerationModel
+from graphon.model_runtime.model_providers.base.rerank_model import RerankModel
+from graphon.model_runtime.model_providers.base.speech2text_model import Speech2TextModel
+from graphon.model_runtime.model_providers.base.text_embedding_model import TextEmbeddingModel
+from graphon.model_runtime.model_providers.base.tts_model import TTSModel
 from graphon.model_runtime.model_providers.model_provider_factory import ModelProviderFactory
+from graphon.model_runtime.protocols.runtime import ModelRuntime
 
 if TYPE_CHECKING:
     from core.model_manager import ModelManager
     from core.plugin.impl.model_runtime import PluginModelRuntime
     from core.provider_manager import ProviderManager
 
+_MODEL_CLASS_BY_TYPE: dict[ModelType, type[AIModel]] = {
+    ModelType.LLM: LargeLanguageModel,
+    ModelType.TEXT_EMBEDDING: TextEmbeddingModel,
+    ModelType.RERANK: RerankModel,
+    ModelType.SPEECH2TEXT: Speech2TextModel,
+    ModelType.MODERATION: ModerationModel,
+    ModelType.TTS: TTSModel,
+}
+
+
+def create_model_type_instance(
+    *,
+    runtime: ModelRuntime,
+    provider_schema: ProviderEntity,
+    model_type: ModelType,
+) -> AIModel:
+    """Build the graphon model wrapper explicitly against the request runtime."""
+    model_class = _MODEL_CLASS_BY_TYPE.get(model_type)
+    if model_class is None:
+        raise ValueError(f"Unsupported model type: {model_type}")
+
+    return model_class(provider_schema=provider_schema, model_runtime=runtime)
+
 
 class PluginModelAssembly:
     """Compose request-scoped model views on top of a single plugin runtime."""
@@ -38,9 +71,22 @@ class PluginModelAssembly:
     @property
     def model_provider_factory(self) -> ModelProviderFactory:
         if self._model_provider_factory is None:
-            self._model_provider_factory = ModelProviderFactory(model_runtime=self.model_runtime)
+            self._model_provider_factory = ModelProviderFactory(runtime=self.model_runtime)
         return self._model_provider_factory
 
+    def create_model_type_instance(
+        self,
+        *,
+        provider: str,
+        model_type: ModelType,
+    ) -> AIModel:
+        provider_schema = self.model_provider_factory.get_provider_schema(provider=provider)
+        return create_model_type_instance(
+            runtime=self.model_runtime,
+            provider_schema=provider_schema,
+            model_type=model_type,
+        )
+
     @property
     def provider_manager(self) -> ProviderManager:
         if self._provider_manager is None:

api/core/prompt/utils/prompt_message_util.py

@@ -53,24 +53,27 @@ class PromptMessageUtil:
                 files = []
                 if isinstance(prompt_message.content, list):
                     for content in prompt_message.content:
-                        if isinstance(content, TextPromptMessageContent):
-                            text += content.data
-                        elif isinstance(content, ImagePromptMessageContent):
-                            files.append(
-                                {
-                                    "type": "image",
-                                    "data": content.data[:10] + "...[TRUNCATED]..." + content.data[-10:],
-                                    "detail": content.detail.value,
-                                }
-                            )
-                        elif isinstance(content, AudioPromptMessageContent):
-                            files.append(
-                                {
-                                    "type": "audio",
-                                    "data": content.data[:10] + "...[TRUNCATED]..." + content.data[-10:],
-                                    "format": content.format,
-                                }
-                            )
+                        match content:
+                            case TextPromptMessageContent():
+                                text += content.data
+                            case ImagePromptMessageContent():
+                                files.append(
+                                    {
+                                        "type": "image",
+                                        "data": content.data[:10] + "...[TRUNCATED]..." + content.data[-10:],
+                                        "detail": content.detail.value,
+                                    }
+                                )
+                            case AudioPromptMessageContent():
+                                files.append(
+                                    {
+                                        "type": "audio",
+                                        "data": content.data[:10] + "...[TRUNCATED]..." + content.data[-10:],
+                                        "format": content.format,
+                                    }
+                                )
+                            case _:
+                                continue
                 else:
                     text = cast(str, prompt_message.content)
 

api/core/provider_manager.py

@@ -56,7 +56,7 @@ from models.provider_ids import ModelProviderID
 from services.feature_service import FeatureService
 
 if TYPE_CHECKING:
-    from graphon.model_runtime.runtime import ModelRuntime
+    from graphon.model_runtime.protocols.runtime import ModelRuntime
 
 _credentials_adapter: TypeAdapter[dict[str, Any]] = TypeAdapter(dict[str, Any])
 
@@ -165,7 +165,7 @@ class ProviderManager:
                 )
 
         # Get all provider entities
-        model_provider_factory = ModelProviderFactory(model_runtime=self._model_runtime)
+        model_provider_factory = ModelProviderFactory(runtime=self._model_runtime)
         provider_entities = model_provider_factory.get_providers()
 
         # Get All preferred provider types of the workspace
@@ -362,7 +362,7 @@ class ProviderManager:
         if not default_model:
             return None
 
-        model_provider_factory = ModelProviderFactory(model_runtime=self._model_runtime)
+        model_provider_factory = ModelProviderFactory(runtime=self._model_runtime)
         provider_schema = model_provider_factory.get_provider_schema(provider=default_model.provider_name)
 
         return DefaultModelEntity(

api/core/rag/datasource/retrieval_service.py

@@ -21,7 +21,7 @@ from core.rag.index_processor.constant.query_type import QueryType
 from core.rag.models.document import Document
 from core.rag.rerank.rerank_type import RerankMode
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
-from core.tools.signature import sign_upload_file
+from core.tools.signature import sign_upload_file_preview_url
 from extensions.ext_database import db
 from graphon.model_runtime.entities.model_entities import ModelType
 from models.dataset import (
@@ -893,7 +893,7 @@ class RetrievalService:
                     "name": upload_file.name,
                     "extension": "." + upload_file.extension,
                     "mime_type": upload_file.mime_type,
-                    "source_url": sign_upload_file(upload_file.id, upload_file.extension),
+                    "source_url": sign_upload_file_preview_url(upload_file.id, upload_file.extension),
                     "size": upload_file.size,
                 }
                 return {"attachment_info": attachment_info, "segment_id": attachment_binding.segment_id}
@@ -920,7 +920,7 @@ class RetrievalService:
                         "name": upload_file.name,
                         "extension": "." + upload_file.extension,
                         "mime_type": upload_file.mime_type,
-                        "source_url": sign_upload_file(upload_file.id, upload_file.extension),
+                        "source_url": sign_upload_file_preview_url(upload_file.id, upload_file.extension),
                         "size": upload_file.size,
                     }
                     if attachment_binding:

api/core/rag/extractor/pdf_extractor.py

@@ -115,7 +115,7 @@ class PdfExtractor(BaseExtractor):
         """
         image_content = []
         upload_files = []
-        base_url = dify_config.INTERNAL_FILES_URL or dify_config.FILES_URL
+        base_url = dify_config.FILES_URL
 
         try:
             image_objects = page.get_objects(filter=(pdfium_c.FPDF_PAGEOBJ_IMAGE,))

api/core/rag/extractor/word_extractor.py

@@ -110,7 +110,7 @@ class WordExtractor(BaseExtractor):
     def _extract_images_from_docx(self, doc):
         image_count = 0
         image_map = {}
-        base_url = dify_config.INTERNAL_FILES_URL or dify_config.FILES_URL
+        base_url = dify_config.FILES_URL
 
         for r_id, rel in doc.part.rels.items():
             if "image" in rel.target_ref:

api/core/rag/retrieval/dataset_retrieval.py

@@ -52,7 +52,7 @@ from core.rag.retrieval.template_prompts import (
     METADATA_FILTER_USER_PROMPT_2,
     METADATA_FILTER_USER_PROMPT_3,
 )
-from core.tools.signature import sign_upload_file
+from core.tools.signature import sign_upload_file_preview_url
 from core.tools.utils.dataset_retriever.dataset_retriever_base_tool import DatasetRetrieverBaseTool
 from core.workflow.file_reference import build_file_reference
 from core.workflow.nodes.knowledge_retrieval import exc
@@ -529,7 +529,7 @@ class DatasetRetrieval:
                                     ),
                                     size=upload_file.size,
                                     storage_key=upload_file.key,
-                                    url=sign_upload_file(upload_file.id, upload_file.extension),
+                                    url=sign_upload_file_preview_url(upload_file.id, upload_file.extension),
                                 )
                                 context_files.append(attachment_info)
                 if show_retrieve_source:

api/core/tools/signature.py

@@ -26,12 +26,14 @@ def sign_tool_file(tool_file_id: str, extension: str, for_external: bool = True)
     return f"{file_preview_url}?timestamp={timestamp}&nonce={nonce}&sign={encoded_sign}"
 
 
-def sign_upload_file(upload_file_id: str, extension: str) -> str:
+def sign_upload_file_preview_url(upload_file_id: str, extension: str) -> str:
     """
-    sign file to get a temporary url for plugin access
+    Sign an upload file to get a temporary image preview URL.
+
+    The URL generated by this function is only for external preview and download,
+    not for internal communication.
     """
-    # Use internal URL for plugin/tool file access in Docker environments
-    base_url = dify_config.INTERNAL_FILES_URL or dify_config.FILES_URL
+    base_url = dify_config.FILES_URL
     file_preview_url = f"{base_url}/files/{upload_file_id}/image-preview"
 
     timestamp = str(int(time.time()))

api/core/tools/utils/message_transformer.py

@@ -23,36 +23,37 @@ _TOOL_FILE_URL_PATTERN = re.compile(r"(?:^|/+)files/tools/(?P<tool_file_id>[^/?#
 
 
 def safe_json_value(v):
-    if isinstance(v, datetime):
-        tz_name = "UTC"
-        if isinstance(current_user, Account) and current_user.timezone is not None:
-            tz_name = current_user.timezone
-        return v.astimezone(pytz.timezone(tz_name)).isoformat()
-    elif isinstance(v, date):
-        return v.isoformat()
-    elif isinstance(v, UUID):
-        return str(v)
-    elif isinstance(v, Decimal):
-        return float(v)
-    elif isinstance(v, bytes):
-        try:
-            return v.decode("utf-8")
-        except UnicodeDecodeError:
-            return v.hex()
-    elif isinstance(v, memoryview):
-        return v.tobytes().hex()
-    elif isinstance(v, np.integer):
-        return int(v)
-    elif isinstance(v, np.floating):
-        return float(v)
-    elif isinstance(v, np.ndarray):
-        return v.tolist()
-    elif isinstance(v, dict):
-        return safe_json_dict(v)
-    elif isinstance(v, list | tuple | set):
-        return [safe_json_value(i) for i in v]
-    else:
-        return v
+    match v:
+        case datetime():
+            tz_name = "UTC"
+            if isinstance(current_user, Account) and current_user.timezone is not None:
+                tz_name = current_user.timezone
+            return v.astimezone(pytz.timezone(tz_name)).isoformat()
+        case date():
+            return v.isoformat()
+        case UUID():
+            return str(v)
+        case Decimal():
+            return float(v)
+        case bytes():
+            try:
+                return v.decode("utf-8")
+            except UnicodeDecodeError:
+                return v.hex()
+        case memoryview():
+            return v.tobytes().hex()
+        case np.integer():
+            return int(v)
+        case np.floating():
+            return float(v)
+        case np.ndarray():
+            return v.tolist()
+        case dict():
+            return safe_json_dict(v)
+        case list() | tuple() | set():
+            return [safe_json_value(i) for i in v]
+        case _:
+            return v
 
 
 def safe_json_dict(d: dict[str, Any]):

api/core/workflow/node_factory.py

@@ -374,11 +374,6 @@ class DifyNodeFactory(NodeFactory):
         # Re-validate using the resolved node class so workflow-local node schemas
         # stay explicit and constructors receive the concrete typed payload.
         resolved_node_data = self._validate_resolved_node_data(node_class, node_data)
-        config_for_node_init: BaseNodeData | dict[str, Any]
-        if isinstance(resolved_node_data, BaseNodeData):
-            config_for_node_init = resolved_node_data.model_dump(mode="python", by_alias=True)
-        else:
-            config_for_node_init = resolved_node_data
         node_type = node_data.type
         node_init_kwargs_factories: Mapping[NodeType, Callable[[], dict[str, object]]] = {
             BuiltinNodeTypes.CODE: lambda: {
@@ -446,9 +441,10 @@ class DifyNodeFactory(NodeFactory):
             },
         }
         node_init_kwargs = node_init_kwargs_factories.get(node_type, lambda: {})()
+        constructor_node_data = resolved_node_data.model_dump(mode="python", by_alias=True)
         return node_class(
             node_id=node_id,
-            config=config_for_node_init,
+            data=constructor_node_data,
             graph_init_params=self.graph_init_params,
             graph_runtime_state=self.graph_runtime_state,
             **node_init_kwargs,

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

@@ -35,7 +35,7 @@ class AgentNode(Node[AgentNodeData]):
     def __init__(
         self,
         node_id: str,
-        config: AgentNodeData,
+        data: AgentNodeData,
         *,
         graph_init_params: GraphInitParams,
         graph_runtime_state: GraphRuntimeState,
@@ -46,7 +46,7 @@ class AgentNode(Node[AgentNodeData]):
     ) -> None:
         super().__init__(
             node_id=node_id,
-            config=config,
+            data=data,
             graph_init_params=graph_init_params,
             graph_runtime_state=graph_runtime_state,
         )

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

@@ -36,14 +36,14 @@ class DatasourceNode(Node[DatasourceNodeData]):
     def __init__(
         self,
         node_id: str,
-        config: DatasourceNodeData,
+        data: DatasourceNodeData,
         *,
         graph_init_params: "GraphInitParams",
         graph_runtime_state: "GraphRuntimeState",
     ) -> None:
         super().__init__(
             node_id=node_id,
-            config=config,
+            data=data,
             graph_init_params=graph_init_params,
             graph_runtime_state=graph_runtime_state,
         )

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

@@ -32,14 +32,14 @@ class KnowledgeIndexNode(Node[KnowledgeIndexNodeData]):
     def __init__(
         self,
         node_id: str,
-        config: KnowledgeIndexNodeData,
+        data: KnowledgeIndexNodeData,
         *,
         graph_init_params: "GraphInitParams",
         graph_runtime_state: "GraphRuntimeState",
     ) -> None:
         super().__init__(
             node_id=node_id,
-            config=config,
+            data=data,
             graph_init_params=graph_init_params,
             graph_runtime_state=graph_runtime_state,
         )

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

@@ -71,14 +71,14 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
     def __init__(
         self,
         node_id: str,
-        config: KnowledgeRetrievalNodeData,
+        data: KnowledgeRetrievalNodeData,
         *,
         graph_init_params: "GraphInitParams",
         graph_runtime_state: "GraphRuntimeState",
     ) -> None:
         super().__init__(
             node_id=node_id,
-            config=config,
+            data=data,
             graph_init_params=graph_init_params,
             graph_runtime_state=graph_runtime_state,
         )

api/core/workflow/system_variables.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 from collections import defaultdict
 from collections.abc import Mapping, Sequence
 from enum import StrEnum
-from typing import Any, Protocol, cast
+from typing import Any, Protocol
 from uuid import uuid4
 
 from graphon.enums import BuiltinNodeTypes
@@ -82,13 +82,10 @@ def build_system_variables(values: Mapping[str, Any] | None = None, /, **kwargs:
     normalized = _normalize_system_variable_values(values, **kwargs)
 
     return [
-        cast(
-            Variable,
-            segment_to_variable(
-                segment=build_segment(value),
-                selector=system_variable_selector(key),
-                name=key,
-            ),
+        segment_to_variable(
+            segment=build_segment(value),
+            selector=system_variable_selector(key),
+            name=key,
         )
         for key, value in normalized.items()
     ]
@@ -130,13 +127,10 @@ def build_bootstrap_variables(
 
     for node_id, value in rag_pipeline_variables_map.items():
         variables.append(
-            cast(
-                Variable,
-                segment_to_variable(
-                    segment=build_segment(value),
-                    selector=(RAG_PIPELINE_VARIABLE_NODE_ID, node_id),
-                    name=node_id,
-                ),
+            segment_to_variable(
+                segment=build_segment(value),
+                selector=(RAG_PIPELINE_VARIABLE_NODE_ID, node_id),
+                name=node_id,
             )
         )
 

api/core/workflow/workflow_entry.py

@@ -46,6 +46,11 @@ _file_access_controller = DatabaseFileAccessController()
 
 
 class _WorkflowChildEngineBuilder:
+    tenant_id: str
+
+    def __init__(self, *, tenant_id: str) -> None:
+        self.tenant_id = tenant_id
+
     @staticmethod
     def _has_node_id(graph_config: Mapping[str, Any], node_id: str) -> bool | None:
         """
@@ -107,7 +112,7 @@ class _WorkflowChildEngineBuilder:
             config=config,
             child_engine_builder=self,
         )
-        child_engine.layer(LLMQuotaLayer())
+        child_engine.layer(LLMQuotaLayer(tenant_id=self.tenant_id))
         return child_engine
 
 
@@ -176,7 +181,7 @@ class WorkflowEntry:
         self.command_channel = command_channel
         execution_context = capture_current_context()
         graph_runtime_state.execution_context = execution_context
-        self._child_engine_builder = _WorkflowChildEngineBuilder()
+        self._child_engine_builder = _WorkflowChildEngineBuilder(tenant_id=tenant_id)
         self.graph_engine = GraphEngine(
             workflow_id=workflow_id,
             graph=graph,
@@ -208,7 +213,7 @@ class WorkflowEntry:
             max_steps=dify_config.WORKFLOW_MAX_EXECUTION_STEPS, max_time=dify_config.WORKFLOW_MAX_EXECUTION_TIME
         )
         self.graph_engine.layer(limits_layer)
-        self.graph_engine.layer(LLMQuotaLayer())
+        self.graph_engine.layer(LLMQuotaLayer(tenant_id=tenant_id))
 
         # Add observability layer when OTel is enabled
         if dify_config.ENABLE_OTEL or is_instrument_flag_enabled():

api/dev/generate_fastopenapi_specs.py

@@ -0,0 +1,95 @@
+"""Generate FastOpenAPI OpenAPI 3.0 specs without booting the full backend."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import sys
+from dataclasses import dataclass
+from pathlib import Path
+
+API_ROOT = Path(__file__).resolve().parents[1]
+if str(API_ROOT) not in sys.path:
+    sys.path.insert(0, str(API_ROOT))
+
+from dev.generate_swagger_specs import apply_runtime_defaults, drop_null_values, sort_openapi_arrays
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class FastOpenApiSpecTarget:
+    route: str
+    filename: str
+
+
+FASTOPENAPI_SPEC_TARGETS: tuple[FastOpenApiSpecTarget, ...] = (
+    FastOpenApiSpecTarget(route="/fastopenapi/openapi.json", filename="fastopenapi-console-openapi.json"),
+)
+
+
+def create_fastopenapi_spec_app():
+    """Build a minimal Flask app that only mounts FastOpenAPI docs routes."""
+
+    apply_runtime_defaults()
+
+    from app_factory import create_flask_app_with_configs
+    from extensions import ext_fastopenapi
+
+    app = create_flask_app_with_configs()
+    ext_fastopenapi.init_app(app)
+    return app
+
+
+def generate_fastopenapi_specs(output_dir: Path) -> list[Path]:
+    """Write FastOpenAPI specs to `output_dir` and return the written paths."""
+
+    output_dir.mkdir(parents=True, exist_ok=True)
+
+    app = create_fastopenapi_spec_app()
+    client = app.test_client()
+
+    written_paths: list[Path] = []
+    for target in FASTOPENAPI_SPEC_TARGETS:
+        response = client.get(target.route)
+        if response.status_code != 200:
+            raise RuntimeError(f"failed to fetch {target.route}: {response.status_code}")
+
+        payload = response.get_json()
+        if not isinstance(payload, dict):
+            raise RuntimeError(f"unexpected response payload for {target.route}")
+        payload = drop_null_values(payload)
+        payload = sort_openapi_arrays(payload)
+
+        output_path = output_dir / target.filename
+        output_path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+        written_paths.append(output_path)
+
+    return written_paths
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "-o",
+        "--output-dir",
+        type=Path,
+        default=Path("openapi"),
+        help="Directory where the OpenAPI JSON files will be written.",
+    )
+    return parser.parse_args()
+
+
+def main() -> int:
+    args = parse_args()
+    written_paths = generate_fastopenapi_specs(args.output_dir)
+
+    for path in written_paths:
+        logger.debug(path)
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

api/dev/generate_swagger_markdown_docs.py

@@ -0,0 +1,182 @@
+"""Generate OpenAPI JSON specs and split Markdown API docs.
+
+The Markdown step uses `swagger-markdown`, the same converter family as the
+Swagger Markdown UI, so CI and local regeneration catch converter-incompatible
+OpenAPI output early.
+"""
+
+from __future__ import annotations
+
+import argparse
+import logging
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+
+API_ROOT = Path(__file__).resolve().parents[1]
+if str(API_ROOT) not in sys.path:
+    sys.path.insert(0, str(API_ROOT))
+
+from dev.generate_fastopenapi_specs import FASTOPENAPI_SPEC_TARGETS, generate_fastopenapi_specs
+from dev.generate_swagger_specs import SPEC_TARGETS, generate_specs
+
+logger = logging.getLogger(__name__)
+
+SWAGGER_MARKDOWN_PACKAGE = "swagger-markdown@3.0.0"
+CONSOLE_SWAGGER_FILENAME = "console-swagger.json"
+STALE_COMBINED_MARKDOWN_FILENAME = "api-reference.md"
+
+
+def _convert_spec_to_markdown(spec_path: Path, markdown_path: Path) -> None:
+    markdown_path.parent.mkdir(parents=True, exist_ok=True)
+    with tempfile.TemporaryDirectory(prefix=f"{markdown_path.stem}-", dir=markdown_path.parent) as temp_dir:
+        temp_markdown_path = Path(temp_dir) / markdown_path.name
+        result = subprocess.run(
+            [
+                "npx",
+                "--yes",
+                SWAGGER_MARKDOWN_PACKAGE,
+                "-i",
+                str(spec_path),
+                "-o",
+                str(temp_markdown_path),
+            ],
+            check=False,
+            capture_output=True,
+            text=True,
+        )
+        if result.returncode != 0:
+            raise subprocess.CalledProcessError(
+                result.returncode,
+                result.args,
+                output=result.stdout,
+                stderr=result.stderr,
+            )
+        if not temp_markdown_path.exists():
+            converter_output = "\n".join(item for item in (result.stdout, result.stderr) if item).strip()
+            raise RuntimeError(f"swagger-markdown did not write {markdown_path}: {converter_output}")
+
+        converted_markdown = temp_markdown_path.read_text(encoding="utf-8")
+        if not converted_markdown.strip():
+            raise RuntimeError(f"swagger-markdown wrote an empty document for {markdown_path}")
+
+    markdown_path.write_text(converted_markdown, encoding="utf-8")
+
+
+def _demote_markdown_headings(markdown: str, *, levels: int = 1) -> str:
+    """Nest generated Markdown under another Markdown section."""
+
+    heading_prefix = "#" * levels
+    lines = []
+    for line in markdown.splitlines():
+        if line.startswith("#"):
+            lines.append(f"{heading_prefix}{line}")
+        else:
+            lines.append(line)
+    return "\n".join(lines).strip()
+
+
+def _append_fastopenapi_markdown(console_markdown_path: Path, fastopenapi_markdown_path: Path) -> None:
+    """Append FastOpenAPI console docs to the existing console API Markdown."""
+
+    console_markdown = console_markdown_path.read_text(encoding="utf-8").rstrip()
+    fastopenapi_markdown = _demote_markdown_headings(
+        fastopenapi_markdown_path.read_text(encoding="utf-8"),
+        levels=2,
+    )
+    console_markdown_path.write_text(
+        "\n\n".join(
+            [
+                console_markdown,
+                "## FastOpenAPI Preview (OpenAPI 3.0)",
+                fastopenapi_markdown,
+            ]
+        )
+        + "\n",
+        encoding="utf-8",
+    )
+
+
+def generate_markdown_docs(
+    swagger_dir: Path,
+    markdown_dir: Path,
+    *,
+    keep_swagger_json: bool = False,
+) -> list[Path]:
+    """Generate intermediate specs, convert them to split Markdown API docs, and return Markdown paths."""
+
+    swagger_paths = generate_specs(swagger_dir)
+    fastopenapi_paths = generate_fastopenapi_specs(swagger_dir)
+    spec_paths = [*swagger_paths, *fastopenapi_paths]
+    swagger_paths_by_name = {path.name: path for path in swagger_paths}
+    fastopenapi_paths_by_name = {path.name: path for path in fastopenapi_paths}
+
+    markdown_dir.mkdir(parents=True, exist_ok=True)
+
+    written_paths: list[Path] = []
+    try:
+        with tempfile.TemporaryDirectory(prefix="dify-api-docs-") as temp_dir:
+            temp_markdown_dir = Path(temp_dir)
+
+            for target in SPEC_TARGETS:
+                swagger_path = swagger_paths_by_name[target.filename]
+                markdown_path = markdown_dir / f"{swagger_path.stem}.md"
+                _convert_spec_to_markdown(swagger_path, markdown_path)
+                written_paths.append(markdown_path)
+
+            for target in FASTOPENAPI_SPEC_TARGETS:  # type: ignore
+                fastopenapi_path = fastopenapi_paths_by_name[target.filename]
+                markdown_path = temp_markdown_dir / f"{fastopenapi_path.stem}.md"
+                _convert_spec_to_markdown(fastopenapi_path, markdown_path)
+
+                console_markdown_path = markdown_dir / f"{Path(CONSOLE_SWAGGER_FILENAME).stem}.md"
+                _append_fastopenapi_markdown(console_markdown_path, markdown_path)
+
+            (markdown_dir / STALE_COMBINED_MARKDOWN_FILENAME).unlink(missing_ok=True)
+    finally:
+        if not keep_swagger_json:
+            for path in spec_paths:
+                path.unlink(missing_ok=True)
+
+    return written_paths
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--swagger-dir",
+        type=Path,
+        default=Path("openapi"),
+        help="Directory where intermediate JSON spec files will be written.",
+    )
+    parser.add_argument(
+        "--markdown-dir",
+        type=Path,
+        default=Path("openapi/markdown"),
+        help="Directory where split Markdown API docs will be written.",
+    )
+    parser.add_argument(
+        "--keep-swagger-json",
+        action="store_true",
+        help="Keep intermediate JSON spec files after Markdown generation.",
+    )
+    return parser.parse_args()
+
+
+def main() -> int:
+    args = parse_args()
+    written_paths = generate_markdown_docs(
+        args.swagger_dir,
+        args.markdown_dir,
+        keep_swagger_json=args.keep_swagger_json,
+    )
+
+    for path in written_paths:
+        logger.debug(path)
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

api/dev/generate_swagger_specs.py

@@ -9,15 +9,17 @@ which is unnecessary when the goal is only to serialize the Flask-RESTX
 from __future__ import annotations
 
 import argparse
+import hashlib
 import json
 import logging
 import os
 import sys
+from collections.abc import MutableMapping
 from dataclasses import dataclass
 from pathlib import Path
+from typing import Protocol, TypeGuard
 
 from flask import Flask
-from flask_restx.swagger import Swagger
 
 logger = logging.getLogger(__name__)
 
@@ -30,19 +32,107 @@ if str(API_ROOT) not in sys.path:
 class SpecTarget:
     route: str
     filename: str
+    namespace: str
+
+
+class RestxApi(Protocol):
+    models: MutableMapping[str, object]
+
+    def model(self, name: str, model: dict[object, object]) -> object: ...
 
 
 SPEC_TARGETS: tuple[SpecTarget, ...] = (
-    SpecTarget(route="/console/api/swagger.json", filename="console-swagger.json"),
-    SpecTarget(route="/api/swagger.json", filename="web-swagger.json"),
-    SpecTarget(route="/v1/swagger.json", filename="service-swagger.json"),
+    SpecTarget(route="/console/api/swagger.json", filename="console-swagger.json", namespace="console"),
+    SpecTarget(route="/api/swagger.json", filename="web-swagger.json", namespace="web"),
+    SpecTarget(route="/v1/swagger.json", filename="service-swagger.json", namespace="service"),
 )
 
-_ORIGINAL_REGISTER_MODEL = Swagger.register_model
-_ORIGINAL_REGISTER_FIELD = Swagger.register_field
+
+def _is_inline_field_map(value: object) -> TypeGuard[dict[object, object]]:
+    """Return whether a nested field map is an anonymous inline mapping."""
+
+    from flask_restx.model import Model, OrderedModel
+
+    return isinstance(value, dict) and not isinstance(value, (Model, OrderedModel))
 
 
-def _apply_runtime_defaults() -> None:
+def _jsonable_schema_value(value: object) -> object:
+    """Return a deterministic JSON-serializable representation for schema fingerprints."""
+
+    if value is None or isinstance(value, str | int | float | bool):
+        return value
+    if isinstance(value, list | tuple):
+        return [_jsonable_schema_value(item) for item in value]
+    if isinstance(value, dict):
+        return {str(key): _jsonable_schema_value(item) for key, item in value.items()}
+    value_type = type(value)
+    return f"<{value_type.__module__}.{value_type.__qualname__}>"
+
+
+def _field_signature(field: object) -> object:
+    """Build a stable signature for a Flask-RESTX field object."""
+
+    from flask_restx import fields
+    from flask_restx.model import instance
+
+    field_instance = instance(field)
+    signature: dict[str, object] = {
+        "class": f"{field_instance.__class__.__module__}.{field_instance.__class__.__qualname__}"
+    }
+
+    if isinstance(field_instance, fields.Nested):
+        nested = getattr(field_instance, "nested", None)
+        if _is_inline_field_map(nested):
+            signature["nested"] = _inline_model_signature(nested)
+        else:
+            signature["nested"] = getattr(
+                nested,
+                "name",
+                f"<{type(nested).__module__}.{type(nested).__qualname__}>",
+            )
+    elif hasattr(field_instance, "container"):
+        signature["container"] = _field_signature(field_instance.container)
+    else:
+        schema = getattr(field_instance, "__schema__", None)
+        if isinstance(schema, dict):
+            signature["schema"] = _jsonable_schema_value(schema)
+
+    for attr_name in (
+        "attribute",
+        "default",
+        "description",
+        "example",
+        "max",
+        "min",
+        "nullable",
+        "readonly",
+        "required",
+        "title",
+    ):
+        if hasattr(field_instance, attr_name):
+            signature[attr_name] = _jsonable_schema_value(getattr(field_instance, attr_name))
+
+    return signature
+
+
+def _inline_model_signature(nested_fields: dict[object, object]) -> object:
+    """Build a stable signature for an anonymous inline model."""
+
+    return [
+        (str(field_name), _field_signature(field))
+        for field_name, field in sorted(nested_fields.items(), key=lambda item: str(item[0]))
+    ]
+
+
+def _inline_model_name(nested_fields: dict[object, object]) -> str:
+    """Return a stable Swagger model name for an anonymous inline field map."""
+
+    signature = json.dumps(_inline_model_signature(nested_fields), sort_keys=True, separators=(",", ":"))
+    digest = hashlib.sha1(signature.encode("utf-8")).hexdigest()[:12]
+    return f"_AnonymousInlineModel_{digest}"
+
+
+def apply_runtime_defaults() -> None:
     """Force the small config surface required for Swagger generation."""
 
     os.environ.setdefault("SECRET_KEY", "spec-export")
@@ -58,69 +148,175 @@ def _apply_runtime_defaults() -> None:
     dify_config.SWAGGER_UI_ENABLED = os.environ["SWAGGER_UI_ENABLED"].lower() == "true"
 
 
-def _patch_swagger_for_inline_nested_dicts() -> None:
-    """Teach Flask-RESTX Swagger generation to tolerate inline nested field maps.
-
-    Some existing controllers use `fields.Nested({...})` with a raw field mapping
-    instead of a named `api.model(...)`. Flask-RESTX crashes on those anonymous
-    dicts during schema registration, so this helper upgrades them into temporary
-    named models at export time.
-    """
-
-    if getattr(Swagger, "_dify_inline_nested_dict_patch", False):
-        return
-
-    def get_or_create_inline_model(self: Swagger, nested_fields: dict[object, object]) -> object:
-        anonymous_models = getattr(self, "_anonymous_inline_models", None)
-        if anonymous_models is None:
-            anonymous_models = {}
-            self._anonymous_inline_models = anonymous_models
-
-        anonymous_name = anonymous_models.get(id(nested_fields))
-        if anonymous_name is None:
-            anonymous_name = f"_AnonymousInlineModel{len(anonymous_models) + 1}"
-            anonymous_models[id(nested_fields)] = anonymous_name
-            self.api.model(anonymous_name, nested_fields)
-
-        return self.api.models[anonymous_name]
-
-    def register_model_with_inline_dict_support(self: Swagger, model: object) -> dict[str, str]:
-        if isinstance(model, dict):
-            model = get_or_create_inline_model(self, model)
-
-        return _ORIGINAL_REGISTER_MODEL(self, model)
-
-    def register_field_with_inline_dict_support(self: Swagger, field: object) -> None:
-        nested = getattr(field, "nested", None)
-        if isinstance(nested, dict):
-            field.model = get_or_create_inline_model(self, nested)  # type: ignore
-
-        _ORIGINAL_REGISTER_FIELD(self, field)
-
-    Swagger.register_model = register_model_with_inline_dict_support
-    Swagger.register_field = register_field_with_inline_dict_support
-    Swagger._dify_inline_nested_dict_patch = True
-
-
 def create_spec_app() -> Flask:
     """Build a minimal Flask app that only mounts the Swagger-producing blueprints."""
 
-    _apply_runtime_defaults()
-    _patch_swagger_for_inline_nested_dicts()
+    apply_runtime_defaults()
+
+    from libs.flask_restx_compat import patch_swagger_for_inline_nested_dicts
+
+    patch_swagger_for_inline_nested_dicts()
 
     app = Flask(__name__)
 
     from controllers.console import bp as console_bp
+    from controllers.console import console_ns
     from controllers.service_api import bp as service_api_bp
+    from controllers.service_api import service_api_ns
     from controllers.web import bp as web_bp
+    from controllers.web import web_ns
 
     app.register_blueprint(console_bp)
     app.register_blueprint(web_bp)
     app.register_blueprint(service_api_bp)
 
+    for namespace in (console_ns, web_ns, service_api_ns):
+        for api in namespace.apis:
+            _materialize_inline_model_definitions(api)
+
     return app
 
 
+def _registered_models(namespace: str) -> dict[str, object]:
+    """Return the Flask-RESTX models registered for a Swagger namespace."""
+
+    if namespace == "console":
+        from controllers.console import console_ns
+
+        models = dict(console_ns.models)
+        for api in console_ns.apis:
+            models.update(api.models)
+        return models
+    if namespace == "web":
+        from controllers.web import web_ns
+
+        models = dict(web_ns.models)
+        for api in web_ns.apis:
+            models.update(api.models)
+        return models
+    if namespace == "service":
+        from controllers.service_api import service_api_ns
+
+        models = dict(service_api_ns.models)
+        for api in service_api_ns.apis:
+            models.update(api.models)
+        return models
+
+    raise ValueError(f"unknown Swagger namespace: {namespace}")
+
+
+def _materialize_inline_model_definitions(api: RestxApi) -> None:
+    """Convert inline `fields.Nested({...})` maps into named API models."""
+
+    from flask_restx import fields
+    from flask_restx.model import Model, OrderedModel, instance
+
+    inline_models: dict[int, dict[object, object]] = {}
+    inline_model_names: dict[int, str] = {}
+
+    def collect_field(field: object) -> None:
+        field_instance = instance(field)
+        if isinstance(field_instance, fields.Nested):
+            nested = getattr(field_instance, "nested", None)
+            if _is_inline_field_map(nested) and id(nested) not in inline_models:
+                inline_models[id(nested)] = nested
+                for nested_field in nested.values():
+                    collect_field(nested_field)
+
+        container = getattr(field_instance, "container", None)
+        if container is not None:
+            collect_field(container)
+
+    for model in list(api.models.values()):
+        if isinstance(model, (Model, OrderedModel)):
+            for field in model.values():
+                collect_field(field)
+
+    for nested_fields in sorted(inline_models.values(), key=_inline_model_name):
+        anonymous_name = _inline_model_name(nested_fields)
+        inline_model_names[id(nested_fields)] = anonymous_name
+        if anonymous_name not in api.models:
+            api.model(anonymous_name, nested_fields)
+
+    def model_name_for(nested_fields: dict[object, object]) -> str:
+        anonymous_name = inline_model_names.get(id(nested_fields))
+        if anonymous_name is None:
+            anonymous_name = _inline_model_name(nested_fields)
+            inline_model_names[id(nested_fields)] = anonymous_name
+            if anonymous_name not in api.models:
+                api.model(anonymous_name, nested_fields)
+        return anonymous_name
+
+    def materialize_field(field: object) -> None:
+        field_instance = instance(field)
+        if isinstance(field_instance, fields.Nested):
+            nested = getattr(field_instance, "nested", None)
+            if _is_inline_field_map(nested):
+                field_instance.model = api.models[model_name_for(nested)]  # type: ignore[attr-defined]
+
+        container = getattr(field_instance, "container", None)
+        if container is not None:
+            materialize_field(container)
+
+    index = 0
+    while index < len(api.models):
+        model = list(api.models.values())[index]
+        index += 1
+        if isinstance(model, (Model, OrderedModel)):
+            for field in model.values():
+                materialize_field(field)
+
+
+def drop_null_values(value: object) -> object:
+    """Remove JSON null values that make the Markdown converter crash."""
+
+    if isinstance(value, dict):
+        return {key: drop_null_values(item) for key, item in value.items() if item is not None}
+    if isinstance(value, list):
+        return [drop_null_values(item) for item in value]
+    return value
+
+
+def sort_openapi_arrays(value: object, *, parent_key: str | None = None) -> object:
+    """Sort order-insensitive Swagger arrays so generated Markdown is stable."""
+
+    if isinstance(value, dict):
+        return {key: sort_openapi_arrays(item, parent_key=key) for key, item in value.items()}
+    if not isinstance(value, list):
+        return value
+
+    sorted_items = [sort_openapi_arrays(item, parent_key=parent_key) for item in value]
+    if parent_key == "parameters":
+        return sorted(
+            sorted_items,
+            key=lambda item: (
+                item.get("in", "") if isinstance(item, dict) else "",
+                item.get("name", "") if isinstance(item, dict) else "",
+                json.dumps(item, sort_keys=True, default=str),
+            ),
+        )
+    if parent_key in {"enum", "required", "schemes", "tags"}:
+        string_items = [item for item in sorted_items if isinstance(item, str)]
+        if len(string_items) == len(sorted_items):
+            return sorted(string_items)
+    return sorted_items
+
+
+def _merge_registered_definitions(payload: dict[str, object], namespace: str) -> dict[str, object]:
+    """Include registered but route-indirect models in the exported Swagger definitions."""
+
+    definitions = payload.setdefault("definitions", {})
+    if not isinstance(definitions, dict):
+        raise RuntimeError("unexpected Swagger definitions payload")
+
+    for name, model in _registered_models(namespace).items():
+        schema = getattr(model, "__schema__", None)
+        if isinstance(schema, dict):
+            definitions.setdefault(name, schema)
+
+    return payload
+
+
 def generate_specs(output_dir: Path) -> list[Path]:
     """Write all Swagger specs to `output_dir` and return the written paths."""
 
@@ -138,6 +334,9 @@ def generate_specs(output_dir: Path) -> list[Path]:
         payload = response.get_json()
         if not isinstance(payload, dict):
             raise RuntimeError(f"unexpected response payload for {target.route}")
+        payload = _merge_registered_definitions(payload, target.namespace)
+        payload = drop_null_values(payload)
+        payload = sort_openapi_arrays(payload)
 
         output_path = output_dir / target.filename
         output_path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")

api/events/event_handlers/update_provider_when_message_created.py

@@ -137,17 +137,13 @@ def handle(sender: Message, **kwargs):
         if used_quota is not None:
             match provider_configuration.system_configuration.current_quota_type:
                 case ProviderQuotaType.TRIAL:
-                    from services.credit_pool_service import CreditPoolService
-
-                    CreditPoolService.check_and_deduct_credits(
+                    _deduct_credit_pool_quota_capped(
                         tenant_id=tenant_id,
                         credits_required=used_quota,
                         pool_type="trial",
                     )
                 case ProviderQuotaType.PAID:
-                    from services.credit_pool_service import CreditPoolService
-
-                    CreditPoolService.check_and_deduct_credits(
+                    _deduct_credit_pool_quota_capped(
                         tenant_id=tenant_id,
                         credits_required=used_quota,
                         pool_type="paid",
@@ -200,6 +196,26 @@ def handle(sender: Message, **kwargs):
         raise
 
 
+def _deduct_credit_pool_quota_capped(*, tenant_id: str, credits_required: int, pool_type: str) -> None:
+    """Apply post-generation credit accounting without failing message persistence on quota exhaustion."""
+    from services.credit_pool_service import CreditPoolService
+
+    deducted_credits = CreditPoolService.deduct_credits_capped(
+        tenant_id=tenant_id,
+        credits_required=credits_required,
+        pool_type=pool_type,
+    )
+    if deducted_credits < credits_required:
+        logger.warning(
+            "Credit pool exhausted during message-created accounting, "
+            "tenant_id=%s, pool_type=%s, credits_required=%s, credits_deducted=%s",
+            tenant_id,
+            pool_type,
+            credits_required,
+            deducted_credits,
+        )
+
+
 def _calculate_quota_usage(
     *, message: Message, system_configuration: SystemConfiguration, model_name: str
 ) -> int | None:

api/fields/workflow_run_fields.py

@@ -1,14 +1,21 @@
+"""Workflow run response schemas for console APIs.
+
+Most workflow-run endpoints should document and serialize responses with the
+Pydantic models in this module. The remaining Flask-RESTX field dictionaries are
+kept only for workflow app-log endpoints that still build legacy log models.
+"""
+
 from __future__ import annotations
 
 from datetime import datetime
 from typing import Any
 
 from flask_restx import Namespace, fields
-from pydantic import Field, field_validator
+from pydantic import AliasChoices, Field, field_validator
 
 from fields.base import ResponseModel
-from fields.end_user_fields import SimpleEndUser, simple_end_user_fields
-from fields.member_fields import SimpleAccount, simple_account_fields
+from fields.end_user_fields import SimpleEndUser
+from fields.member_fields import SimpleAccount
 from libs.helper import TimestampField
 
 workflow_run_for_log_fields = {
@@ -43,119 +50,6 @@ def build_workflow_run_for_archived_log_model(api_or_ns: Namespace):
     return api_or_ns.model("WorkflowRunForArchivedLog", workflow_run_for_archived_log_fields)
 
 
-workflow_run_for_list_fields = {
-    "id": fields.String,
-    "version": fields.String,
-    "status": fields.String,
-    "elapsed_time": fields.Float,
-    "total_tokens": fields.Integer,
-    "total_steps": fields.Integer,
-    "created_by_account": fields.Nested(simple_account_fields, attribute="created_by_account", allow_null=True),
-    "created_at": TimestampField,
-    "finished_at": TimestampField,
-    "exceptions_count": fields.Integer,
-    "retry_index": fields.Integer,
-}
-
-advanced_chat_workflow_run_for_list_fields = {
-    "id": fields.String,
-    "conversation_id": fields.String,
-    "message_id": fields.String,
-    "version": fields.String,
-    "status": fields.String,
-    "elapsed_time": fields.Float,
-    "total_tokens": fields.Integer,
-    "total_steps": fields.Integer,
-    "created_by_account": fields.Nested(simple_account_fields, attribute="created_by_account", allow_null=True),
-    "created_at": TimestampField,
-    "finished_at": TimestampField,
-    "exceptions_count": fields.Integer,
-    "retry_index": fields.Integer,
-}
-
-advanced_chat_workflow_run_pagination_fields = {
-    "limit": fields.Integer(attribute="limit"),
-    "has_more": fields.Boolean(attribute="has_more"),
-    "data": fields.List(fields.Nested(advanced_chat_workflow_run_for_list_fields), attribute="data"),
-}
-
-workflow_run_pagination_fields = {
-    "limit": fields.Integer(attribute="limit"),
-    "has_more": fields.Boolean(attribute="has_more"),
-    "data": fields.List(fields.Nested(workflow_run_for_list_fields), attribute="data"),
-}
-
-workflow_run_count_fields = {
-    "total": fields.Integer,
-    "running": fields.Integer,
-    "succeeded": fields.Integer,
-    "failed": fields.Integer,
-    "stopped": fields.Integer,
-    "partial_succeeded": fields.Integer(attribute="partial-succeeded"),
-}
-
-workflow_run_detail_fields = {
-    "id": fields.String,
-    "version": fields.String,
-    "graph": fields.Raw(attribute="graph_dict"),
-    "inputs": fields.Raw(attribute="inputs_dict"),
-    "status": fields.String,
-    "outputs": fields.Raw(attribute="outputs_dict"),
-    "error": fields.String,
-    "elapsed_time": fields.Float,
-    "total_tokens": fields.Integer,
-    "total_steps": fields.Integer,
-    "created_by_role": fields.String,
-    "created_by_account": fields.Nested(simple_account_fields, attribute="created_by_account", allow_null=True),
-    "created_by_end_user": fields.Nested(simple_end_user_fields, attribute="created_by_end_user", allow_null=True),
-    "created_at": TimestampField,
-    "finished_at": TimestampField,
-    "exceptions_count": fields.Integer,
-}
-
-retry_event_field = {
-    "elapsed_time": fields.Float,
-    "status": fields.String,
-    "inputs": fields.Raw(attribute="inputs"),
-    "process_data": fields.Raw(attribute="process_data"),
-    "outputs": fields.Raw(attribute="outputs"),
-    "metadata": fields.Raw(attribute="metadata"),
-    "llm_usage": fields.Raw(attribute="llm_usage"),
-    "error": fields.String,
-    "retry_index": fields.Integer,
-}
-
-
-workflow_run_node_execution_fields = {
-    "id": fields.String,
-    "index": fields.Integer,
-    "predecessor_node_id": fields.String,
-    "node_id": fields.String,
-    "node_type": fields.String,
-    "title": fields.String,
-    "inputs": fields.Raw(attribute="inputs_dict"),
-    "process_data": fields.Raw(attribute="process_data_dict"),
-    "outputs": fields.Raw(attribute="outputs_dict"),
-    "status": fields.String,
-    "error": fields.String,
-    "elapsed_time": fields.Float,
-    "execution_metadata": fields.Raw(attribute="execution_metadata_dict"),
-    "extras": fields.Raw,
-    "created_at": TimestampField,
-    "created_by_role": fields.String,
-    "created_by_account": fields.Nested(simple_account_fields, attribute="created_by_account", allow_null=True),
-    "created_by_end_user": fields.Nested(simple_end_user_fields, attribute="created_by_end_user", allow_null=True),
-    "finished_at": TimestampField,
-    "inputs_truncated": fields.Boolean,
-    "outputs_truncated": fields.Boolean,
-    "process_data_truncated": fields.Boolean,
-}
-
-workflow_run_node_execution_list_fields = {
-    "data": fields.List(fields.Nested(workflow_run_node_execution_fields)),
-}
-
-
 def _to_timestamp(value: datetime | int | None) -> int | None:
     if isinstance(value, datetime):
         return int(value.timestamp())
@@ -252,7 +146,10 @@ class WorkflowRunCountResponse(ResponseModel):
     succeeded: int
     failed: int
     stopped: int
-    partial_succeeded: int = Field(validation_alias="partial-succeeded")
+    partial_succeeded: int = Field(
+        alias="partial_succeeded",
+        validation_alias=AliasChoices("partial_succeeded", "partial-succeeded"),
+    )
 
 
 class WorkflowRunDetailResponse(ResponseModel):

api/libs/external_api.py

@@ -9,6 +9,7 @@ from werkzeug.http import HTTP_STATUS_CODES
 
 from configs import dify_config
 from core.errors.error import AppInvokeQuotaExceededError
+from libs.flask_restx_compat import patch_swagger_for_inline_nested_dicts
 from libs.token import build_force_logout_cookie_headers
 
 
@@ -120,6 +121,7 @@ class ExternalApi(Api):
     }
 
     def __init__(self, app: Blueprint | Flask, *args, **kwargs):
+        patch_swagger_for_inline_nested_dicts()
         kwargs.setdefault("authorizations", self._authorizations)
         kwargs.setdefault("security", "Bearer")
         kwargs["add_specs"] = dify_config.SWAGGER_UI_ENABLED

api/libs/flask_restx_compat.py

@@ -0,0 +1,149 @@
+"""Compatibility helpers for Dify's Flask-RESTX Swagger integration.
+
+These helpers are temporary bridges for legacy Flask-RESTX field contracts
+while controllers migrate their request and response documentation to Pydantic
+models. Keep the behavior centralized so live Swagger endpoints and offline
+spec export fail or succeed in the same way.
+"""
+
+import hashlib
+import json
+from typing import TypeGuard
+
+from flask import current_app
+from flask_restx import fields
+from flask_restx.model import Model, OrderedModel, instance
+from flask_restx.swagger import Swagger
+
+
+def _is_inline_field_map(value: object) -> TypeGuard[dict[object, object]]:
+    """Return whether a nested field map is an anonymous inline mapping."""
+
+    return isinstance(value, dict) and not isinstance(value, (Model, OrderedModel))
+
+
+def _jsonable_schema_value(value: object) -> object:
+    """Return a deterministic JSON-serializable representation for schema fingerprints."""
+
+    if value is None or isinstance(value, str | int | float | bool):
+        return value
+    if isinstance(value, list | tuple):
+        return [_jsonable_schema_value(item) for item in value]
+    if isinstance(value, dict):
+        return {str(key): _jsonable_schema_value(item) for key, item in value.items()}
+    value_type = type(value)
+    return f"<{value_type.__module__}.{value_type.__qualname__}>"
+
+
+def _field_signature(field: object) -> object:
+    """Build a stable signature for a Flask-RESTX field object."""
+
+    field_instance = instance(field)
+    signature: dict[str, object] = {
+        "class": f"{field_instance.__class__.__module__}.{field_instance.__class__.__qualname__}"
+    }
+
+    if isinstance(field_instance, fields.Nested):
+        nested = getattr(field_instance, "nested", None)
+        if _is_inline_field_map(nested):
+            signature["nested"] = _inline_model_signature(nested)
+        else:
+            signature["nested"] = getattr(
+                nested,
+                "name",
+                f"<{type(nested).__module__}.{type(nested).__qualname__}>",
+            )
+    elif hasattr(field_instance, "container"):
+        signature["container"] = _field_signature(field_instance.container)
+    else:
+        schema = getattr(field_instance, "__schema__", None)
+        if isinstance(schema, dict):
+            signature["schema"] = _jsonable_schema_value(schema)
+
+    for attr_name in (
+        "attribute",
+        "default",
+        "description",
+        "example",
+        "max",
+        "max_items",
+        "min",
+        "min_items",
+        "nullable",
+        "readonly",
+        "required",
+        "title",
+        "unique",
+    ):
+        if hasattr(field_instance, attr_name):
+            signature[attr_name] = _jsonable_schema_value(getattr(field_instance, attr_name))
+
+    return signature
+
+
+def _inline_model_signature(nested_fields: dict[object, object]) -> object:
+    """Build a stable signature for an anonymous inline model."""
+
+    return [
+        (str(field_name), _field_signature(field))
+        for field_name, field in sorted(nested_fields.items(), key=lambda item: str(item[0]))
+    ]
+
+
+def _inline_model_name(nested_fields: dict[object, object]) -> str:
+    """Return a stable Swagger model name for an anonymous inline field map."""
+
+    signature = json.dumps(_inline_model_signature(nested_fields), sort_keys=True, separators=(",", ":"))
+    digest = hashlib.sha1(signature.encode("utf-8")).hexdigest()[:12]
+    return f"_AnonymousInlineModel_{digest}"
+
+
+def patch_swagger_for_inline_nested_dicts() -> None:
+    """Allow Swagger generation to handle legacy inline Flask-RESTX field dicts.
+
+    Some existing controllers use raw field mappings in `fields.Nested({...})`
+    or directly in `@namespace.response(...)`. Runtime marshalling accepts that,
+    but Flask-RESTX Swagger registration expects a named model. Convert those
+    anonymous mappings into temporary named models during docs generation.
+    """
+
+    if getattr(Swagger, "_dify_inline_nested_dict_patch", False):
+        return
+
+    original_register_model = Swagger.register_model
+    original_register_field = Swagger.register_field
+    original_as_dict = Swagger.as_dict
+
+    def get_or_create_inline_model(self: Swagger, nested_fields: dict[object, object]) -> object:
+        anonymous_name = _inline_model_name(nested_fields)
+        if anonymous_name not in self.api.models:
+            self.api.model(anonymous_name, nested_fields)
+
+        return self.api.models[anonymous_name]
+
+    def register_model_with_inline_dict_support(self: Swagger, model: object) -> dict[str, str]:
+        if _is_inline_field_map(model):
+            model = get_or_create_inline_model(self, model)
+
+        return original_register_model(self, model)
+
+    def register_field_with_inline_dict_support(self: Swagger, field: object) -> None:
+        nested = getattr(field, "nested", None)
+        if _is_inline_field_map(nested):
+            field.model = get_or_create_inline_model(self, nested)  # type: ignore[attr-defined]
+
+        original_register_field(self, field)
+
+    def as_dict_with_inline_dict_support(self: Swagger):
+        # Temporary set RESTX_INCLUDE_ALL_MODELS = false to prevent "length changed while iterating" error
+        include_all_models = current_app.config.get("RESTX_INCLUDE_ALL_MODELS", False)
+        current_app.config["RESTX_INCLUDE_ALL_MODELS"] = False
+        try:
+            return original_as_dict(self)
+        finally:
+            current_app.config["RESTX_INCLUDE_ALL_MODELS"] = include_all_models
+
+    Swagger.register_model = register_model_with_inline_dict_support
+    Swagger.register_field = register_field_with_inline_dict_support
+    Swagger.as_dict = as_dict_with_inline_dict_support
+    Swagger._dify_inline_nested_dict_patch = True

api/models/dataset.py

@@ -24,7 +24,7 @@ from core.rag.index_processor.constant.built_in_field import BuiltInField, Metad
 from core.rag.index_processor.constant.index_type import IndexStructureType, IndexTechniqueType
 from core.rag.index_processor.constant.query_type import QueryType
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
-from core.tools.signature import sign_upload_file
+from core.tools.signature import sign_upload_file_preview_url
 from extensions.ext_storage import storage
 from libs.uuid_utils import uuidv7
 
@@ -1020,7 +1020,7 @@ class DocumentSegment(Base):
             encoded_sign = base64.urlsafe_b64encode(sign).decode()
 
             params = f"timestamp={timestamp}&nonce={nonce}&sign={encoded_sign}"
-            reference_url = dify_config.CONSOLE_API_URL or ""
+            reference_url = dify_config.FILES_URL or dify_config.CONSOLE_API_URL or ""
             base_url = f"{reference_url}/files/{upload_file_id}/image-preview"
             source_url = f"{base_url}?{params}"
             attachment_list.append(
@@ -1162,7 +1162,7 @@ class DatasetQuery(TypeBase):
                                 "size": file_info.size,
                                 "extension": file_info.extension,
                                 "mime_type": file_info.mime_type,
-                                "source_url": sign_upload_file(file_info.id, file_info.extension),
+                                "source_url": sign_upload_file_preview_url(file_info.id, file_info.extension),
                             }
                     else:
                         query["file_info"] = None

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

@@ -3,6 +3,7 @@ from collections.abc import Mapping
 from typing import Any, cast
 from unittest.mock import MagicMock
 
+import pytest
 from dify_trace_aliyun.entities.semconv import (
     GEN_AI_FRAMEWORK,
     GEN_AI_SESSION_ID,
@@ -31,7 +32,7 @@ from graphon.enums import WorkflowNodeExecutionStatus
 from models import EndUser
 
 
-def test_get_user_id_from_message_data_no_end_user(monkeypatch):
+def test_get_user_id_from_message_data_no_end_user(monkeypatch: pytest.MonkeyPatch):
     message_data = MagicMock()
     message_data.from_account_id = "account_id"
     message_data.from_end_user_id = None
@@ -39,7 +40,7 @@ def test_get_user_id_from_message_data_no_end_user(monkeypatch):
     assert get_user_id_from_message_data(message_data) == "account_id"
 
 
-def test_get_user_id_from_message_data_with_end_user(monkeypatch):
+def test_get_user_id_from_message_data_with_end_user(monkeypatch: pytest.MonkeyPatch):
     message_data = MagicMock()
     message_data.from_account_id = "account_id"
     message_data.from_end_user_id = "end_user_id"
@@ -57,7 +58,7 @@ def test_get_user_id_from_message_data_with_end_user(monkeypatch):
     assert get_user_id_from_message_data(message_data) == "session_id"
 
 
-def test_get_user_id_from_message_data_end_user_not_found(monkeypatch):
+def test_get_user_id_from_message_data_end_user_not_found(monkeypatch: pytest.MonkeyPatch):
     message_data = MagicMock()
     message_data.from_account_id = "account_id"
     message_data.from_end_user_id = "end_user_id"
@@ -111,7 +112,7 @@ def test_get_workflow_node_status():
     assert status.status_code == StatusCode.UNSET
 
 
-def test_create_links_from_trace_id(monkeypatch):
+def test_create_links_from_trace_id(monkeypatch: pytest.MonkeyPatch):
     # Mock create_link
     mock_link = MagicMock(spec=Link)
     import dify_trace_aliyun.data_exporter.traceclient

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

@@ -40,7 +40,7 @@ def langfuse_config():
 
 
 @pytest.fixture
-def trace_instance(langfuse_config, monkeypatch):
+def trace_instance(langfuse_config, monkeypatch: pytest.MonkeyPatch):
     # Mock Langfuse client to avoid network calls
     mock_client = MagicMock()
     monkeypatch.setattr("dify_trace_langfuse.langfuse_trace.Langfuse", lambda **kwargs: mock_client)
@@ -49,7 +49,7 @@ def trace_instance(langfuse_config, monkeypatch):
     return instance
 
 
-def test_init(langfuse_config, monkeypatch):
+def test_init(langfuse_config, monkeypatch: pytest.MonkeyPatch):
     mock_langfuse = MagicMock()
     monkeypatch.setattr("dify_trace_langfuse.langfuse_trace.Langfuse", mock_langfuse)
     monkeypatch.setenv("FILES_URL", "http://test.url")
@@ -64,7 +64,7 @@ def test_init(langfuse_config, monkeypatch):
     assert instance.file_base_url == "http://test.url"
 
 
-def test_trace_dispatch(trace_instance, monkeypatch):
+def test_trace_dispatch(trace_instance, monkeypatch: pytest.MonkeyPatch):
     methods = [
         "workflow_trace",
         "message_trace",
@@ -114,7 +114,7 @@ def test_trace_dispatch(trace_instance, monkeypatch):
     mocks["generate_name_trace"].assert_called_once_with(info)
 
 
-def test_workflow_trace_with_message_id(trace_instance, monkeypatch):
+def test_workflow_trace_with_message_id(trace_instance, monkeypatch: pytest.MonkeyPatch):
     # Setup trace info
     trace_info = WorkflowTraceInfo(
         workflow_id="wf-1",
@@ -218,7 +218,7 @@ def test_workflow_trace_with_message_id(trace_instance, monkeypatch):
     assert other_span.level == LevelEnum.ERROR
 
 
-def test_workflow_trace_no_message_id(trace_instance, monkeypatch):
+def test_workflow_trace_no_message_id(trace_instance, monkeypatch: pytest.MonkeyPatch):
     trace_info = WorkflowTraceInfo(
         workflow_id="wf-1",
         tenant_id="tenant-1",
@@ -259,7 +259,7 @@ def test_workflow_trace_no_message_id(trace_instance, monkeypatch):
     assert trace_data.name == TraceTaskName.WORKFLOW_TRACE
 
 
-def test_workflow_trace_missing_app_id(trace_instance, monkeypatch):
+def test_workflow_trace_missing_app_id(trace_instance, monkeypatch: pytest.MonkeyPatch):
     trace_info = WorkflowTraceInfo(
         workflow_id="wf-1",
         tenant_id="tenant-1",
@@ -287,7 +287,7 @@ def test_workflow_trace_missing_app_id(trace_instance, monkeypatch):
         trace_instance.workflow_trace(trace_info)
 
 
-def test_message_trace_basic(trace_instance, monkeypatch):
+def test_message_trace_basic(trace_instance, monkeypatch: pytest.MonkeyPatch):
     message_data = MagicMock()
     message_data.id = "msg-1"
     message_data.from_account_id = "acc-1"
@@ -331,7 +331,7 @@ def test_message_trace_basic(trace_instance, monkeypatch):
     assert gen_data.usage.total == 30
 
 
-def test_message_trace_with_end_user(trace_instance, monkeypatch):
+def test_message_trace_with_end_user(trace_instance, monkeypatch: pytest.MonkeyPatch):
     message_data = MagicMock()
     message_data.id = "msg-1"
     message_data.from_account_id = "acc-1"
@@ -636,7 +636,7 @@ def test_langfuse_trace_entity_with_list_dict_input():
     assert data.input[0]["content"] == "hello"
 
 
-def test_workflow_trace_handles_usage_extraction_error(trace_instance, monkeypatch, caplog):
+def test_workflow_trace_handles_usage_extraction_error(trace_instance, monkeypatch: pytest.MonkeyPatch, caplog):
     # Setup trace info to trigger LLM node usage extraction
     trace_info = WorkflowTraceInfo(
         workflow_id="wf-1",

api/providers/trace/trace-langsmith/tests/unit_tests/langsmith_trace/test_langsmith_trace.py

@@ -35,7 +35,7 @@ def langsmith_config():
 
 
 @pytest.fixture
-def trace_instance(langsmith_config, monkeypatch):
+def trace_instance(langsmith_config, monkeypatch: pytest.MonkeyPatch):
     # Mock LangSmith client
     mock_client = MagicMock()
     monkeypatch.setattr("dify_trace_langsmith.langsmith_trace.Client", lambda **kwargs: mock_client)
@@ -44,7 +44,7 @@ def trace_instance(langsmith_config, monkeypatch):
     return instance
 
 
-def test_init(langsmith_config, monkeypatch):
+def test_init(langsmith_config, monkeypatch: pytest.MonkeyPatch):
     mock_client_class = MagicMock()
     monkeypatch.setattr("dify_trace_langsmith.langsmith_trace.Client", mock_client_class)
     monkeypatch.setenv("FILES_URL", "http://test.url")
@@ -57,7 +57,7 @@ def test_init(langsmith_config, monkeypatch):
     assert instance.file_base_url == "http://test.url"
 
 
-def test_trace_dispatch(trace_instance, monkeypatch):
+def test_trace_dispatch(trace_instance, monkeypatch: pytest.MonkeyPatch):
     methods = [
         "workflow_trace",
         "message_trace",
@@ -107,7 +107,7 @@ def test_trace_dispatch(trace_instance, monkeypatch):
     mocks["generate_name_trace"].assert_called_once_with(info)
 
 
-def test_workflow_trace(trace_instance, monkeypatch):
+def test_workflow_trace(trace_instance, monkeypatch: pytest.MonkeyPatch):
     # Setup trace info
     workflow_data = MagicMock()
     workflow_data.created_at = _dt()
@@ -223,7 +223,7 @@ def test_workflow_trace(trace_instance, monkeypatch):
     assert call_args[4].run_type == LangSmithRunType.retriever
 
 
-def test_workflow_trace_no_start_time(trace_instance, monkeypatch):
+def test_workflow_trace_no_start_time(trace_instance, monkeypatch: pytest.MonkeyPatch):
     workflow_data = MagicMock()
     workflow_data.created_at = _dt()
     workflow_data.finished_at = _dt() + timedelta(seconds=1)
@@ -266,7 +266,7 @@ def test_workflow_trace_no_start_time(trace_instance, monkeypatch):
     assert trace_instance.add_run.called
 
 
-def test_workflow_trace_missing_app_id(trace_instance, monkeypatch):
+def test_workflow_trace_missing_app_id(trace_instance, monkeypatch: pytest.MonkeyPatch):
     trace_info = MagicMock(spec=WorkflowTraceInfo)
     trace_info.trace_id = "trace-1"
     trace_info.message_id = None
@@ -290,7 +290,7 @@ def test_workflow_trace_missing_app_id(trace_instance, monkeypatch):
         trace_instance.workflow_trace(trace_info)
 
 
-def test_message_trace(trace_instance, monkeypatch):
+def test_message_trace(trace_instance, monkeypatch: pytest.MonkeyPatch):
     message_data = MagicMock()
     message_data.id = "msg-1"
     message_data.from_account_id = "acc-1"
@@ -516,7 +516,7 @@ def test_update_run_error(trace_instance):
         trace_instance.update_run(update_data)
 
 
-def test_workflow_trace_usage_extraction_error(trace_instance, monkeypatch, caplog):
+def test_workflow_trace_usage_extraction_error(trace_instance, monkeypatch: pytest.MonkeyPatch, caplog):
     workflow_data = MagicMock()
     workflow_data.created_at = _dt()
     workflow_data.finished_at = _dt() + timedelta(seconds=1)