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

@@ -99,7 +99,7 @@ jobs:
       - name: Set up dotenvs
         run: |
           cp docker/.env.example docker/.env
-          cp docker/middleware.env.example docker/middleware.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

@@ -57,7 +57,7 @@ jobs:
               - '.github/workflows/api-tests.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'
@@ -84,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:
@@ -94,7 +94,7 @@ jobs:
               - '.github/workflows/vdb-tests.yml'
               - '.github/workflows/expose_service_ports.sh'
               - 'docker/.env.example'
-              - 'docker/middleware.env.example'
+              - 'docker/envs/middleware.env.example'
               - 'docker/docker-compose.yaml'
               - 'docker/docker-compose-template.yaml'
               - 'docker/generate_docker_compose'
@@ -116,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

@@ -51,7 +51,7 @@ jobs:
       - name: Set up dotenvs
         run: |
           cp docker/.env.example docker/.env
-          cp docker/middleware.env.example docker/middleware.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

@@ -48,7 +48,7 @@ jobs:
       - name: Set up dotenvs
         run: |
           cp docker/.env.example docker/.env
-          cp docker/middleware.env.example docker/middleware.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,11 +76,10 @@ The easiest way to start the Dify server is through [Docker Compose](docker/dock
 ```bash
 cd dify
 cd docker
-./dify-compose up -d
+cp .env.example .env
+docker compose up -d
 ```
 
-On Windows PowerShell, run `.\dify-compose.ps1 up -d` from the `docker` directory.
-
 After running, you can access the Dify dashboard in your browser at [http://localhost/install](http://localhost/install) and start the initialization process.
 
 #### Seeking help
@@ -138,7 +137,7 @@ Star Dify on GitHub and be instantly notified of new releases.
 
 ### Custom configurations
 
-If you need to customize the configuration, add only the values you want to override to `docker/.env`. The default values live in [`docker/.env.default`](docker/.env.default), and the full reference remains in [`docker/.env.example`](docker/.env.example). After making any changes, re-run `./dify-compose up -d` or `.\dify-compose.ps1 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

@@ -74,11 +74,6 @@ class StorageConfig(BaseSettings):
         default="opendal",
     )
 
-    STORAGE_PATH_PREFIX: str = Field(
-        description="Global path prefix prepended to all storage object keys.",
-        default="",
-    )
-
     STORAGE_LOCAL_PATH: str = Field(
         description="Path for local storage when STORAGE_TYPE is set to 'local'.",
         default="storage",
@@ -119,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
 
 
@@ -228,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,
@@ -257,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/constants/recommended_apps.json

@@ -19,7 +19,7 @@
                         "name": "Website Generator"
                     },
                     "app_id": "b53545b1-79ea-4da3-b31a-c39391c6f041",
-                    "category": "Programming",
+                    "categories": ["Programming"],
                     "copyright": null,
                     "description": null,
                     "is_listed": true,
@@ -35,7 +35,7 @@
                         "name": "Investment Analysis Report Copilot"
                     },
                     "app_id": "a23b57fa-85da-49c0-a571-3aff375976c1",
-                    "category": "Agent",
+                    "categories": ["Agent"],
                     "copyright": "Dify.AI",
                     "description": "Welcome to your personalized Investment Analysis Copilot service, where we delve into the depths of stock analysis to provide you with comprehensive insights. \n",
                     "is_listed": true,
@@ -51,7 +51,7 @@
                         "name": "Workflow Planning Assistant "
                     },
                     "app_id": "f3303a7d-a81c-404e-b401-1f8711c998c1",
-                    "category": "Workflow",
+                    "categories": ["Workflow"],
                     "copyright": null,
                     "description": "An assistant that helps you plan and select the right node for a workflow (V0.6.0).  ",
                     "is_listed": true,
@@ -67,7 +67,7 @@
                         "name": "Automated Email Reply  "
                     },
                     "app_id": "e9d92058-7d20-4904-892f-75d90bef7587",
-                    "category": "Workflow",
+                    "categories": ["Workflow"],
                     "copyright": null,
                     "description": "Reply emails using Gmail API. It will automatically retrieve email in your inbox and create a response in Gmail. \nConfigure your Gmail API in Google Cloud Console. ",
                     "is_listed": true,
@@ -83,7 +83,7 @@
                         "name": "Book Translation "
                     },
                     "app_id": "98b87f88-bd22-4d86-8b74-86beba5e0ed4",
-                    "category": "Workflow",
+                    "categories": ["Workflow"],
                     "copyright": null,
                     "description": "A workflow designed to translate a full book up to 15000 tokens per run. Uses Code node to separate text into chunks and Iteration to translate each chunk. ",
                     "is_listed": true,
@@ -99,7 +99,7 @@
                         "name": "Python bug fixer"
                     },
                     "app_id": "cae337e6-aec5-4c7b-beca-d6f1a808bd5e",
-                    "category": "Programming",
+                    "categories": ["Programming"],
                     "copyright": null,
                     "description": null,
                     "is_listed": true,
@@ -115,7 +115,7 @@
                         "name": "Code Interpreter"
                     },
                     "app_id": "d077d587-b072-4f2c-b631-69ed1e7cdc0f",
-                    "category": "Programming",
+                    "categories": ["Programming"],
                     "copyright": "Copyright 2023 Dify",
                     "description": "Code interpreter, clarifying the syntax and semantics of the code.",
                     "is_listed": true,
@@ -131,7 +131,7 @@
                         "name": "SVG Logo Design "
                     },
                     "app_id": "73fbb5f1-c15d-4d74-9cc8-46d9db9b2cca",
-                    "category": "Agent",
+                    "categories": ["Agent"],
                     "copyright": "Dify.AI",
                     "description": "Hello, I am your creative partner in bringing ideas to vivid life! I can assist you in creating stunning designs by leveraging abilities of DALL·E 3.  ",
                     "is_listed": true,
@@ -147,7 +147,7 @@
                         "name": "Long Story Generator (Iteration) "
                     },
                     "app_id": "5efb98d7-176b-419c-b6ef-50767391ab62",
-                    "category": "Workflow",
+                    "categories": ["Workflow"],
                     "copyright": null,
                     "description": "A workflow demonstrating how to use Iteration node to generate long article that is longer than the context length of LLMs. ",
                     "is_listed": true,
@@ -163,7 +163,7 @@
                         "name": "Text Summarization Workflow"
                     },
                     "app_id": "f00c4531-6551-45ee-808f-1d7903099515",
-                    "category": "Workflow",
+                    "categories": ["Workflow"],
                     "copyright": null,
                     "description": "Based on users' choice, retrieve external knowledge to more accurately summarize articles.",
                     "is_listed": true,
@@ -179,7 +179,7 @@
                         "name": "YouTube Channel Data Analysis"
                     },
                     "app_id": "be591209-2ca8-410f-8f3b-ca0e530dd638",
-                    "category": "Agent",
+                    "categories": ["Agent"],
                     "copyright": "Dify.AI",
                     "description": "I am a YouTube Channel Data Analysis Copilot, I am here to provide expert data analysis tailored to your needs. ",
                     "is_listed": true,
@@ -195,7 +195,7 @@
                         "name": "Article Grading Bot"
                     },
                     "app_id": "a747f7b4-c48b-40d6-b313-5e628232c05f",
-                    "category": "Writing",
+                    "categories": ["Writing"],
                     "copyright": null,
                     "description": "Assess the quality of articles and text based on user defined criteria. ",
                     "is_listed": true,
@@ -211,7 +211,7 @@
                         "name": "SEO Blog Generator"
                     },
                     "app_id": "18f3bd03-524d-4d7a-8374-b30dbe7c69d5",
-                    "category": "Workflow",
+                    "categories": ["Workflow"],
                     "copyright": null,
                     "description": "Workflow for retrieving information from the internet, followed by segmented generation of SEO blogs.",
                     "is_listed": true,
@@ -227,7 +227,7 @@
                         "name": "SQL Creator"
                     },
                     "app_id": "050ef42e-3e0c-40c1-a6b6-a64f2c49d744",
-                    "category": "Programming",
+                    "categories": ["Programming"],
                     "copyright": "Copyright 2023 Dify",
                     "description": "Write SQL from natural language by pasting in your schema with the request.Please describe your query requirements in natural language and select the target database type.",
                     "is_listed": true,
@@ -243,7 +243,7 @@
                         "name": "Sentiment Analysis "
                     },
                     "app_id": "f06bf86b-d50c-4895-a942-35112dbe4189",
-                    "category": "Workflow",
+                    "categories": ["Workflow"],
                     "copyright": null,
                     "description": "Batch sentiment analysis of text, followed by JSON output of sentiment classification along with scores.",
                     "is_listed": true,
@@ -259,7 +259,7 @@
                         "name": "Strategic Consulting Expert"
                     },
                     "app_id": "7e8ca1ae-02f2-4b5f-979e-62d19133bee2",
-                    "category": "Assistant",
+                    "categories": ["Assistant"],
                     "copyright": "Copyright 2023 Dify",
                     "description": "I can answer your questions related to strategic marketing.",
                     "is_listed": true,
@@ -275,7 +275,7 @@
                         "name": "Code Converter"
                     },
                     "app_id": "4006c4b2-0735-4f37-8dbb-fb1a8c5bd87a",
-                    "category": "Programming",
+                    "categories": ["Programming"],
                     "copyright": "Copyright 2023 Dify",
                     "description": "This is an application that provides the ability to convert code snippets in multiple programming languages. You can input the code you wish to convert, select the target programming language, and get the desired output.",
                     "is_listed": true,
@@ -291,7 +291,7 @@
                         "name": "Question Classifier + Knowledge + Chatbot "
                     },
                     "app_id": "d9f6b733-e35d-4a40-9f38-ca7bbfa009f7",
-                    "category": "Workflow",
+                    "categories": ["Workflow"],
                     "copyright": null,
                     "description": "Basic Workflow Template, a chatbot capable of identifying intents alongside with a knowledge base.",
                     "is_listed": true,
@@ -307,7 +307,7 @@
                         "name": "AI Front-end interviewer"
                     },
                     "app_id": "127efead-8944-4e20-ba9d-12402eb345e0",
-                    "category": "HR",
+                    "categories": ["HR"],
                     "copyright": "Copyright 2023 Dify",
                     "description": "A simulated front-end interviewer that tests the skill level of front-end development through questioning.",
                     "is_listed": true,
@@ -323,7 +323,7 @@
                         "name": "Knowledge Retrieval + Chatbot "
                     },
                     "app_id": "e9870913-dd01-4710-9f06-15d4180ca1ce",
-                    "category": "Workflow",
+                    "categories": ["Workflow"],
                     "copyright": null,
                     "description": "Basic Workflow Template, A chatbot with a knowledge base. ",
                     "is_listed": true,
@@ -339,7 +339,7 @@
                         "name": "Email Assistant Workflow "
                     },
                     "app_id": "dd5b6353-ae9b-4bce-be6a-a681a12cf709",
-                    "category": "Workflow",
+                    "categories": ["Workflow"],
                     "copyright": null,
                     "description": "A multifunctional email assistant capable of summarizing, replying, composing, proofreading, and checking grammar.",
                     "is_listed": true,
@@ -355,7 +355,7 @@
                         "name": "Customer Review Analysis Workflow "
                     },
                     "app_id": "9c0cd31f-4b62-4005-adf5-e3888d08654a",
-                    "category": "Workflow",
+                    "categories": ["Workflow"],
                     "copyright": null,
                     "description": "Utilize LLM (Large Language Models) to classify customer reviews and forward them to the internal system.",
                     "is_listed": true,

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):
@@ -52,7 +53,7 @@ class RecommendedAppResponse(ResponseModel):
     copyright: str | None = None
     privacy_policy: str | None = None
     custom_disclaimer: str | None = None
-    category: str | None = None
+    categories: list[str] = Field(default_factory=list)
     position: int | None = None
     is_listed: bool | None = None
     can_trial: bool | None = None
@@ -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/tool_providers.py

@@ -876,10 +876,10 @@ class ToolBuiltinProviderSetDefaultApi(Resource):
     @login_required
     @account_initialization_required
     def post(self, provider):
-        current_user, current_tenant_id = current_account_with_tenant()
+        _, current_tenant_id = current_account_with_tenant()
         payload = BuiltinProviderDefaultCredentialPayload.model_validate(console_ns.payload or {})
         return BuiltinToolManageService.set_default_provider(
-            tenant_id=current_tenant_id, user_id=current_user.id, provider=provider, id=payload.id
+            tenant_id=current_tenant_id, provider=provider, id=payload.id
         )
 
 

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/extensions/ext_storage.py

@@ -1,5 +1,4 @@
 import logging
-import posixpath
 from collections.abc import Callable, Generator
 from typing import Literal, Union, overload
 
@@ -18,15 +17,6 @@ class Storage:
         storage_factory = self.get_storage_factory(dify_config.STORAGE_TYPE)
         with app.app_context():
             self.storage_runner = storage_factory()
-            prefix = dify_config.STORAGE_PATH_PREFIX.strip("/") if dify_config.STORAGE_PATH_PREFIX else ""
-            if prefix and ".." in prefix.split("/"):
-                raise ValueError(f"STORAGE_PATH_PREFIX must not contain '..': {dify_config.STORAGE_PATH_PREFIX}")
-            self._path_prefix = prefix
-
-    def _prefix(self, filename: str) -> str:
-        if not self._path_prefix:
-            return filename
-        return posixpath.join(self._path_prefix, filename)
 
     @staticmethod
     def get_storage_factory(storage_type: str) -> Callable[[], BaseStorage]:
@@ -96,7 +86,7 @@ class Storage:
                 raise ValueError(f"unsupported storage type {storage_type}")
 
     def save(self, filename: str, data: bytes):
-        self.storage_runner.save(self._prefix(filename), data)
+        self.storage_runner.save(filename, data)
 
     @overload
     def load(self, filename: str, /, *, stream: Literal[False] = False) -> bytes: ...
@@ -115,26 +105,22 @@ class Storage:
             return self.load_once(filename)
 
     def load_once(self, filename: str) -> bytes:
-        return self.storage_runner.load_once(self._prefix(filename))
+        return self.storage_runner.load_once(filename)
 
     def load_stream(self, filename: str) -> Generator:
-        return self.storage_runner.load_stream(self._prefix(filename))
+        return self.storage_runner.load_stream(filename)
 
     def download(self, filename, target_filepath):
-        self.storage_runner.download(self._prefix(filename), target_filepath)
+        self.storage_runner.download(filename, target_filepath)
 
     def exists(self, filename):
-        return self.storage_runner.exists(self._prefix(filename))
+        return self.storage_runner.exists(filename)
 
     def delete(self, filename: str):
-        return self.storage_runner.delete(self._prefix(filename))
+        return self.storage_runner.delete(filename)
 
     def scan(self, path: str, files: bool = True, directories: bool = False) -> list[str]:
-        results = self.storage_runner.scan(self._prefix(path), files=files, directories=directories)
-        if not self._path_prefix:
-            return results
-        prefix_with_slash = self._path_prefix + "/"
-        return [r.removeprefix(prefix_with_slash) for r in results]
+        return self.storage_runner.scan(path, files=files, directories=directories)
 
 
 storage = Storage()

api/extensions/storage/aliyun_oss_storage.py

@@ -13,7 +13,7 @@ class AliyunOssStorage(BaseStorage):
     def __init__(self):
         super().__init__()
         self.bucket_name = dify_config.ALIYUN_OSS_BUCKET_NAME
-        self.folder = None if dify_config.STORAGE_PATH_PREFIX else dify_config.ALIYUN_OSS_PATH
+        self.folder = dify_config.ALIYUN_OSS_PATH
         oss_auth_method = aliyun_s3.Auth
         region = None
         if dify_config.ALIYUN_OSS_AUTH_VERSION == "v4":

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/migrations/versions/2026_04_29_1200-a4f2d8c9b731_add_recommended_app_categories.py

@@ -0,0 +1,26 @@
+"""add recommended app categories
+
+Revision ID: a4f2d8c9b731
+Revises: 227822d22895
+Create Date: 2026-04-29 12:00:00.000000
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "a4f2d8c9b731"
+down_revision = "227822d22895"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    with op.batch_alter_table("recommended_apps", schema=None) as batch_op:
+        batch_op.add_column(sa.Column("categories", sa.JSON(), nullable=True))
+
+
+def downgrade():
+    with op.batch_alter_table("recommended_apps", schema=None) as batch_op:
+        batch_op.drop_column("categories")

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/models/model.py

@@ -878,6 +878,7 @@ class RecommendedApp(TypeBase):
     copyright: Mapped[str] = mapped_column(String(255), nullable=False)
     privacy_policy: Mapped[str] = mapped_column(String(255), nullable=False)
     category: Mapped[str] = mapped_column(String(255), nullable=False)
+    categories: Mapped[list[str] | None] = mapped_column(sa.JSON, nullable=True, default=None)
     custom_disclaimer: Mapped[str] = mapped_column(LongText, default="")
     position: Mapped[int] = mapped_column(sa.Integer, nullable=False, default=0)
     is_listed: Mapped[bool] = mapped_column(sa.Boolean, nullable=False, default=True)