fix: can not create empty knowledge (#36336)

Co-authored-by: Joel <iamjoel007@gmail.com>

.github/actions/setup-web/action.yml

@@ -1,8 +1,13 @@
 name: Setup Web Environment
+description: Set up Node.js, Vite+, pnpm, and web dependencies
 
 runs:
   using: composite
   steps:
+    - name: Setup pnpm
+      uses: pnpm/action-setup@8912a9102ac27614460f54aedde9e1e7f9aec20d # v6.0.5
+      with:
+        run_install: false
     - name: Setup Vite+
       uses: voidzero-dev/setup-vp@4f5aa3e38c781f1b01e78fb9255527cee8a6efa6 # v1.8.0
       with:

.github/workflows/autofix.yml

@@ -120,7 +120,11 @@ jobs:
         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
+          uv run dev/generate_swagger_markdown_docs.py --swagger-dir ../packages/contracts/openapi --markdown-dir openapi/markdown --keep-swagger-json
+
+      - name: Generate frontend contracts
+        if: github.event_name != 'merge_group' && steps.api-changes.outputs.any_changed == 'true'
+        run: pnpm --dir packages/contracts gen-api-contract-from-openapi
 
       - name: ESLint autofix
         if: github.event_name != 'merge_group' && steps.web-changes.outputs.any_changed == 'true'

.gitignore

@@ -250,5 +250,5 @@ scripts/stress-test/reports/
 
 # Code Agent Folder
 .qoder/*
-
+.context/*
 .eslintcache

AGENTS.md

@@ -9,6 +9,7 @@ The codebase is split into:
 - **Backend API** (`/api`): Python Flask application organized with Domain-Driven Design
 - **Frontend Web** (`/web`): Next.js application using TypeScript and React
 - **Docker deployment** (`/docker`): Containerized deployment configurations
+- **Dify Agent Backend** (`/dify-agent`): Backend services for managing and executing agent
 
 ## Backend Workflow
 

Makefile

@@ -83,16 +83,15 @@ lint:
 	@echo "✅ Linting complete"
 
 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/' --exclude 'dev/generate_swagger_specs.py' --check-untyped-defs --disable-error-code=import-untyped .
+	@echo "📝 Running type checks (pyrefly + mypy)..."
+	@./dev/pyrefly-check-local $(PATH_TO_CHECK)
+	@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 "✅ 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/' --exclude 'dev/generate_swagger_specs.py'  --exclude 'dev/generate_fastopenapi_specs.py' --check-untyped-defs --disable-error-code=import-untyped .
+	@echo "📝 Running core type checks (pyrefly + mypy)..."
+	@./dev/pyrefly-check-local $(PATH_TO_CHECK)
+	@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:
@@ -153,8 +152,8 @@ help:
 	@echo "  make format         - Format code with ruff"
 	@echo "  make check          - Check code with ruff"
 	@echo "  make lint           - Format, fix, and lint code (ruff, imports, dotenv)"
-	@echo "  make type-check     - Run type checks (basedpyright, pyrefly, mypy)"
-	@echo "  make type-check-core - Run core type checks (basedpyright, mypy)"
+	@echo "  make type-check     - Run type checks (pyrefly, mypy)"
+	@echo "  make type-check-core - Run core type checks (pyrefly, mypy)"
 	@echo "  make test           - Run backend unit tests (or TARGET_TESTS=./api/tests/<target_tests>)"
 	@echo ""
 	@echo "Docker Build Targets:"

api/.env.example

@@ -557,7 +557,7 @@ MAX_VARIABLE_SIZE=204800
 
 # GraphEngine Worker Pool Configuration
 # Minimum number of workers per GraphEngine instance (default: 1)
-GRAPH_ENGINE_MIN_WORKERS=1
+GRAPH_ENGINE_MIN_WORKERS=3
 # Maximum number of workers per GraphEngine instance (default: 10)
 GRAPH_ENGINE_MAX_WORKERS=10
 # Queue depth threshold that triggers worker scale up (default: 3)

api/AGENTS.md

@@ -195,7 +195,7 @@ Before opening a PR / submitting:
 - 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(...)`,
+  DTOs with `register_response_schema_models(...)`, serialize response DTOs with `dump_response(...)`,
   and avoid adding new legacy `ns.model(...)`, `@marshal_with(...)`, or GET `@ns.expect(...)` patterns.
 
 ### Miscellaneous

api/Dockerfile

@@ -24,7 +24,8 @@ RUN apt-get update \
 # Install Python dependencies (workspace members under providers/vdb/)
 COPY pyproject.toml uv.lock ./
 COPY providers ./providers
-RUN uv sync --locked --no-dev
+# Trust the checked-in lock during image builds; dev-only path sources live outside the api/ context.
+RUN uv sync --frozen --no-dev
 
 # production stage
 FROM base AS production

api/README.md

@@ -99,7 +99,7 @@ The scripts resolve paths relative to their location, so you can run them from a
    ./dev/reformat               # Run all formatters and linters
    uv run ruff check --fix ./   # Fix linting issues
    uv run ruff format ./        # Format code
-   uv run basedpyright .        # Type checking
+   uv run pyrefly check         # Type checking
    ```
 
 ## Generate TS stub

api/app_factory.py

@@ -117,7 +117,7 @@ def create_flask_app_with_configs() -> DifyApp:
             logger.warning("Failed to add trace headers to response", exc_info=True)
         return response
 
-    # Capture the decorator's return value to avoid pyright reportUnusedFunction
+    # Capture the decorator return values so static checkers do not treat the hooks as unused.
     _ = before_request
     _ = add_trace_headers
 

api/commands/plugin.py

@@ -185,9 +185,9 @@ def transform_datasource_credentials(environment: str):
         firecrawl_plugin_id = "langgenius/firecrawl_datasource"
         jina_plugin_id = "langgenius/jina_datasource"
         if environment == "online":
-            notion_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(notion_plugin_id)  # pyright: ignore[reportPrivateUsage]
-            firecrawl_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(firecrawl_plugin_id)  # pyright: ignore[reportPrivateUsage]
-            jina_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(jina_plugin_id)  # pyright: ignore[reportPrivateUsage]
+            notion_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(notion_plugin_id)
+            firecrawl_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(firecrawl_plugin_id)
+            jina_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(jina_plugin_id)
         else:
             notion_plugin_unique_identifier = None
             firecrawl_plugin_unique_identifier = None

api/commands/system.py

@@ -14,6 +14,7 @@ from libs.rsa import generate_key_pair
 from models import Tenant
 from models.model import App, AppMode, Conversation
 from models.provider import Provider, ProviderModel
+from models.tools import ApiToolProvider, BuiltinToolProvider, MCPToolProvider
 
 logger = logging.getLogger(__name__)
 
@@ -23,13 +24,16 @@ DB_UPGRADE_LOCK_TTL_SECONDS = 60
 @click.command(
     "reset-encrypt-key-pair",
     help="Reset the asymmetric key pair of workspace for encrypt LLM credentials. "
-    "After the reset, all LLM credentials will become invalid, "
-    "requiring re-entry."
+    "After the reset, all LLM credentials and tool provider credentials "
+    "(builtin / API / MCP) will be purged, requiring re-entry. "
     "Only support SELF_HOSTED mode.",
 )
 @click.confirmation_option(
     prompt=click.style(
-        "Are you sure you want to reset encrypt key pair? This operation cannot be rolled back!", fg="red"
+        "Are you sure you want to reset encrypt key pair? "
+        "This will also purge builtin / API / MCP tool provider records for every tenant. "
+        "This operation cannot be rolled back!",
+        fg="red",
     )
 )
 def reset_encrypt_key_pair():
@@ -53,6 +57,13 @@ def reset_encrypt_key_pair():
             session.execute(delete(Provider).where(Provider.provider_type == "custom", Provider.tenant_id == tenant.id))
             session.execute(delete(ProviderModel).where(ProviderModel.tenant_id == tenant.id))
 
+            # Purge tool provider records that hold credentials encrypted under the
+            # tenant key. Leaving them in place causes /console/api/workspaces/current/
+            # tool-providers to 500 because decryption fails on stale ciphertext (#35396).
+            session.execute(delete(BuiltinToolProvider).where(BuiltinToolProvider.tenant_id == tenant.id))
+            session.execute(delete(ApiToolProvider).where(ApiToolProvider.tenant_id == tenant.id))
+            session.execute(delete(MCPToolProvider).where(MCPToolProvider.tenant_id == tenant.id))
+
             click.echo(
                 click.style(
                     f"Congratulations! The asymmetric key pair of workspace {tenant.id} has been reset.",

api/configs/enterprise/__init__.py

@@ -23,6 +23,12 @@ class EnterpriseFeatureConfig(BaseSettings):
         ge=1, description="Maximum timeout in seconds for enterprise requests", default=5
     )
 
+    ENTERPRISE_DISABLE_RUNTIME_CREDENTIAL_CHECK: bool = Field(
+        default=False,
+        description="If disabled, credential policy check is only performed when saving workflows."
+        "This helps gain runtime performance by trading off consistency.",
+    )
+
 
 class EnterpriseTelemetryConfig(BaseSettings):
     """

api/configs/feature/__init__.py

@@ -761,7 +761,7 @@ class WorkflowConfig(BaseSettings):
     # GraphEngine Worker Pool Configuration
     GRAPH_ENGINE_MIN_WORKERS: PositiveInt = Field(
         description="Minimum number of workers per GraphEngine instance",
-        default=1,
+        default=3,
     )
 
     GRAPH_ENGINE_MAX_WORKERS: PositiveInt = Field(

api/configs/middleware/__init__.py

@@ -1,5 +1,5 @@
 import os
-from typing import Any, Literal, TypedDict
+from typing import Any, Literal, TypedDict, cast
 from urllib.parse import parse_qsl, quote_plus
 
 from pydantic import Field, NonNegativeFloat, NonNegativeInt, PositiveFloat, PositiveInt, computed_field
@@ -50,28 +50,30 @@ from .vdb.vastbase_vector_config import VastbaseVectorConfig
 from .vdb.vikingdb_config import VikingDBConfig
 from .vdb.weaviate_config import WeaviateConfig
 
+_VALID_STORAGE_TYPE = Literal[
+    "opendal",
+    "s3",
+    "aliyun-oss",
+    "azure-blob",
+    "baidu-obs",
+    "clickzetta-volume",
+    "google-storage",
+    "huawei-obs",
+    "oci-storage",
+    "tencent-cos",
+    "volcengine-tos",
+    "supabase",
+    "local",
+]
+
 
 class StorageConfig(BaseSettings):
-    STORAGE_TYPE: Literal[
-        "opendal",
-        "s3",
-        "aliyun-oss",
-        "azure-blob",
-        "baidu-obs",
-        "clickzetta-volume",
-        "google-storage",
-        "huawei-obs",
-        "oci-storage",
-        "tencent-cos",
-        "volcengine-tos",
-        "supabase",
-        "local",
-    ] = Field(
+    STORAGE_TYPE: _VALID_STORAGE_TYPE = Field(
         description="Type of storage to use."
         " Options: 'opendal', '(deprecated) local', 's3', 'aliyun-oss', 'azure-blob', 'baidu-obs', "
         "'clickzetta-volume', 'google-storage', 'huawei-obs', 'oci-storage', 'tencent-cos', "
         "'volcengine-tos', 'supabase'. Default is 'opendal'.",
-        default="opendal",
+        default=cast(_VALID_STORAGE_TYPE, "opendal"),
     )
 
     STORAGE_LOCAL_PATH: str = Field(

api/controllers/API_SCHEMA_GUIDE.md

@@ -34,6 +34,7 @@ from controllers.common.schema import (
     register_response_schema_models,
     register_schema_models,
 )
+from libs.helper import dump_response
 ```
 
 Register request payload and query models with `register_schema_models(...)`:
@@ -82,7 +83,7 @@ register_schema_models(console_ns, DraftWorkflowNodeRunPayload)
 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")
+    return dump_response(WorkflowRunNodeExecutionResponse, result)
 ```
 
 ## Query Parameters
@@ -105,7 +106,7 @@ class WorkflowRunListQuery(BaseModel):
 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")
+    return dump_response(WorkflowRunPaginationResponse, result)
 ```
 
 Do not do this for GET query parameters:
@@ -145,10 +146,25 @@ def post(...):
 Serialize explicitly:
 
 ```python
-return WorkflowRunNodeExecutionResponse.model_validate(
-    workflow_node_execution,
-    from_attributes=True,
-).model_dump(mode="json")
+return dump_response(WorkflowRunNodeExecutionResponse, workflow_node_execution)
+```
+
+`dump_response(...)` is the preferred response serialization helper for a single Pydantic response DTO. It validates
+with `from_attributes=True` and returns `model_dump(mode="json")`, so SQLAlchemy models, plain objects, dictionaries,
+Pydantic aliases, computed fields, and `datetime` values are serialized consistently.
+
+For wrapper responses, pass a dictionary with the public wrapper fields:
+
+```python
+return dump_response(
+    WorkflowRunPaginationResponse,
+    {
+        "data": workflow_runs,
+        "page": page,
+        "limit": limit,
+        "has_more": has_more,
+    },
+)
 ```
 
 If the service can return `None`, translate that into the expected HTTP error before validation:
@@ -158,9 +174,12 @@ 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")
+return dump_response(WorkflowRunDetailResponse, workflow_run)
 ```
 
+Use manual `model_validate(...).model_dump(...)` only when the endpoint needs behavior that `dump_response(...)` does
+not provide, such as returning a non-dict payload, intentionally excluding fields, or composing a `(body, status)` tuple.
+
 ## Legacy Flask-RESTX Patterns
 
 Avoid adding these patterns to new or migrated endpoints:
@@ -190,4 +209,3 @@ Inspect affected endpoints with `jq`. Check that:
 - 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/human_input.py

@@ -1,6 +1,21 @@
+import json
+
 from pydantic import BaseModel, JsonValue
 
 
 class HumanInputFormSubmitPayload(BaseModel):
     inputs: dict[str, JsonValue]
     action: str
+
+
+def stringify_form_default_values(values: dict[str, object]) -> dict[str, str]:
+    """Serialize default values into strings expected by human-input form clients."""
+    result: dict[str, str] = {}
+    for key, value in values.items():
+        if value is None:
+            result[key] = ""
+        elif isinstance(value, (dict, list)):
+            result[key] = json.dumps(value, ensure_ascii=False)
+        else:
+            result[key] = str(value)
+    return result

api/controllers/common/schema.py

@@ -39,6 +39,7 @@ QueryParamDoc = TypedDict(
 def _register_json_schema(namespace: Namespace, name: str, schema: dict) -> None:
     """Register a JSON schema and promote any nested Pydantic `$defs`."""
 
+    schema = _swagger_2_compatible_schema(schema)
     nested_definitions = schema.get("$defs")
     schema_to_register = dict(schema)
     if isinstance(nested_definitions, dict):
@@ -65,6 +66,35 @@ def _register_schema_model(namespace: Namespace, model: type[BaseModel], *, mode
     )
 
 
+def _swagger_2_compatible_schema(value: Any) -> Any:
+    if isinstance(value, list):
+        return [_swagger_2_compatible_schema(item) for item in value]
+
+    if not isinstance(value, dict):
+        return value
+
+    converted = {key: _swagger_2_compatible_schema(child) for key, child in value.items()}
+    any_of = value.get("anyOf")
+    if not isinstance(any_of, list):
+        return converted
+
+    non_null_candidates = [
+        candidate for candidate in any_of if isinstance(candidate, Mapping) and candidate.get("type") != "null"
+    ]
+    has_null_candidate = any(isinstance(candidate, Mapping) and candidate.get("type") == "null" for candidate in any_of)
+    if not has_null_candidate or len(non_null_candidates) != 1:
+        return converted
+
+    non_null_schema = _swagger_2_compatible_schema(dict(non_null_candidates[0]))
+    if not isinstance(non_null_schema, dict):
+        return converted
+
+    converted.pop("anyOf", None)
+    converted.update(non_null_schema)
+    converted["x-nullable"] = True
+    return converted
+
+
 def register_schema_model(namespace: Namespace, model: type[BaseModel]) -> None:
     """Register a BaseModel and its nested schema definitions for Swagger documentation."""
 

api/controllers/console/__init__.py

@@ -33,7 +33,6 @@ for module_name in RESOURCE_MODULES:
 # Ensure resource modules are imported so route decorators are evaluated.
 # Import other controllers
 from . import (
-    admin,
     apikey,
     extension,
     feature,
@@ -117,7 +116,7 @@ from .explore import (
     saved_message,
     trial,
 )
-from .socketio import workflow as socketio_workflow  # pyright: ignore[reportUnusedImport]
+from .socketio import workflow as socketio_workflow
 
 # Import tag controllers
 from .tag import tags
@@ -142,7 +141,6 @@ api.add_namespace(console_ns)
 __all__ = [
     "account",
     "activate",
-    "admin",
     "advanced_prompt_template",
     "agent",
     "agent_providers",

api/controllers/console/admin.py

@@ -1,64 +1,11 @@
-import csv
-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
-from pydantic import BaseModel, Field, field_validator
-from sqlalchemy import select
-from werkzeug.exceptions import BadRequest, NotFound, Unauthorized
+from werkzeug.exceptions import 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
-from extensions.ext_database import db
 from libs.token import extract_access_token
-from models.model import App, ExporleBanner, InstalledApp, RecommendedApp, TrialApp
-from services.billing_service import BillingService, LangContentDict
-
-
-class InsertExploreAppPayload(BaseModel):
-    app_id: str = Field(...)
-    desc: str | None = None
-    copyright: str | None = None
-    privacy_policy: str | None = None
-    custom_disclaimer: str | None = None
-    language: str = Field(...)
-    category: str = Field(...)
-    position: int = Field(...)
-    can_trial: bool = Field(default=False)
-    trial_limit: int = Field(default=0)
-
-    @field_validator("language")
-    @classmethod
-    def validate_language(cls, value: str) -> str:
-        return supported_language(value)
-
-
-class InsertExploreBannerPayload(BaseModel):
-    category: str = Field(...)
-    title: str = Field(...)
-    description: str = Field(...)
-    img_src: str = Field(..., alias="img-src")
-    language: str = Field(default="en-US")
-    link: str = Field(...)
-    sort: int = Field(...)
-
-    @field_validator("language")
-    @classmethod
-    def validate_language(cls, value: str) -> str:
-        return supported_language(value)
-
-    model_config = {"populate_by_name": True}
-
-
-register_schema_models(console_ns, InsertExploreAppPayload, InsertExploreBannerPayload)
 
 
 def admin_required[**P, R](view: Callable[P, R]) -> Callable[P, R]:
@@ -76,353 +23,3 @@ def admin_required[**P, R](view: Callable[P, R]) -> Callable[P, R]:
         return view(*args, **kwargs)
 
     return decorated
-
-
-@console_ns.route("/admin/insert-explore-apps")
-class InsertExploreAppListApi(Resource):
-    @console_ns.doc("insert_explore_app")
-    @console_ns.doc(description="Insert or update an app in the explore list")
-    @console_ns.expect(console_ns.models[InsertExploreAppPayload.__name__])
-    @console_ns.response(200, "App updated successfully")
-    @console_ns.response(201, "App inserted successfully")
-    @console_ns.response(404, "App not found")
-    @only_edition_cloud
-    @admin_required
-    def post(self):
-        payload = InsertExploreAppPayload.model_validate(console_ns.payload)
-
-        app = db.session.execute(select(App).where(App.id == payload.app_id)).scalar_one_or_none()
-        if not app:
-            raise NotFound(f"App '{payload.app_id}' is not found")
-
-        site = app.site
-        if not site:
-            desc = payload.desc or ""
-            copy_right = payload.copyright or ""
-            privacy_policy = payload.privacy_policy or ""
-            custom_disclaimer = payload.custom_disclaimer or ""
-        else:
-            desc = site.description or payload.desc or ""
-            copy_right = site.copyright or payload.copyright or ""
-            privacy_policy = site.privacy_policy or payload.privacy_policy or ""
-            custom_disclaimer = site.custom_disclaimer or payload.custom_disclaimer or ""
-
-        with session_factory.create_session() as session:
-            recommended_app = session.execute(
-                select(RecommendedApp).where(RecommendedApp.app_id == payload.app_id)
-            ).scalar_one_or_none()
-
-            if not recommended_app:
-                recommended_app = RecommendedApp(
-                    app_id=app.id,
-                    description=desc,
-                    copyright=copy_right,
-                    privacy_policy=privacy_policy,
-                    custom_disclaimer=custom_disclaimer,
-                    language=payload.language,
-                    category=payload.category,
-                    position=payload.position,
-                )
-
-                db.session.add(recommended_app)
-                if payload.can_trial:
-                    trial_app = db.session.execute(
-                        select(TrialApp).where(TrialApp.app_id == payload.app_id)
-                    ).scalar_one_or_none()
-                    if not trial_app:
-                        db.session.add(
-                            TrialApp(
-                                app_id=payload.app_id,
-                                tenant_id=app.tenant_id,
-                                trial_limit=payload.trial_limit,
-                            )
-                        )
-                    else:
-                        trial_app.trial_limit = payload.trial_limit
-
-                app.is_public = True
-                db.session.commit()
-
-                return {"result": "success"}, 201
-            else:
-                recommended_app.description = desc
-                recommended_app.copyright = copy_right
-                recommended_app.privacy_policy = privacy_policy
-                recommended_app.custom_disclaimer = custom_disclaimer
-                recommended_app.language = payload.language
-                recommended_app.category = payload.category
-                recommended_app.position = payload.position
-
-                if payload.can_trial:
-                    trial_app = db.session.execute(
-                        select(TrialApp).where(TrialApp.app_id == payload.app_id)
-                    ).scalar_one_or_none()
-                    if not trial_app:
-                        db.session.add(
-                            TrialApp(
-                                app_id=payload.app_id,
-                                tenant_id=app.tenant_id,
-                                trial_limit=payload.trial_limit,
-                            )
-                        )
-                    else:
-                        trial_app.trial_limit = payload.trial_limit
-                app.is_public = True
-
-                db.session.commit()
-
-                return {"result": "success"}, 200
-
-
-@console_ns.route("/admin/insert-explore-apps/<uuid:app_id>")
-class InsertExploreAppApi(Resource):
-    @console_ns.doc("delete_explore_app")
-    @console_ns.doc(description="Remove an app from the explore list")
-    @console_ns.doc(params={"app_id": "Application ID to remove"})
-    @console_ns.response(204, "App removed successfully")
-    @only_edition_cloud
-    @admin_required
-    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))
-            ).scalar_one_or_none()
-
-        if not recommended_app:
-            return {"result": "success"}, 204
-
-        with session_factory.create_session() as session:
-            app = session.execute(select(App).where(App.id == recommended_app.app_id)).scalar_one_or_none()
-
-        if app:
-            app.is_public = False
-
-        with session_factory.create_session() as session:
-            installed_apps = (
-                session.execute(
-                    select(InstalledApp).where(
-                        InstalledApp.app_id == recommended_app.app_id,
-                        InstalledApp.tenant_id != InstalledApp.app_owner_tenant_id,
-                    )
-                )
-                .scalars()
-                .all()
-            )
-
-            for installed_app in installed_apps:
-                session.delete(installed_app)
-
-            trial_app = session.execute(
-                select(TrialApp).where(TrialApp.app_id == recommended_app.app_id)
-            ).scalar_one_or_none()
-            if trial_app:
-                session.delete(trial_app)
-
-        db.session.delete(recommended_app)
-        db.session.commit()
-
-        return {"result": "success"}, 204
-
-
-@console_ns.route("/admin/insert-explore-banner")
-class InsertExploreBannerApi(Resource):
-    @console_ns.doc("insert_explore_banner")
-    @console_ns.doc(description="Insert an explore banner")
-    @console_ns.expect(console_ns.models[InsertExploreBannerPayload.__name__])
-    @console_ns.response(201, "Banner inserted successfully")
-    @only_edition_cloud
-    @admin_required
-    def post(self):
-        payload = InsertExploreBannerPayload.model_validate(console_ns.payload)
-
-        banner = ExporleBanner(
-            content={
-                "category": payload.category,
-                "title": payload.title,
-                "description": payload.description,
-                "img-src": payload.img_src,
-            },
-            link=payload.link,
-            sort=payload.sort,
-            language=payload.language,
-        )
-        db.session.add(banner)
-        db.session.commit()
-
-        return {"result": "success"}, 201
-
-
-@console_ns.route("/admin/delete-explore-banner/<uuid:banner_id>")
-class DeleteExploreBannerApi(Resource):
-    @console_ns.doc("delete_explore_banner")
-    @console_ns.doc(description="Delete an explore banner")
-    @console_ns.doc(params={"banner_id": "Banner ID to delete"})
-    @console_ns.response(204, "Banner deleted successfully")
-    @only_edition_cloud
-    @admin_required
-    def delete(self, banner_id):
-        banner = db.session.execute(select(ExporleBanner).where(ExporleBanner.id == banner_id)).scalar_one_or_none()
-        if not banner:
-            raise NotFound(f"Banner '{banner_id}' is not found")
-
-        db.session.delete(banner)
-        db.session.commit()
-
-        return {"result": "success"}, 204
-
-
-class LangContentPayload(BaseModel):
-    lang: str = Field(..., description="Language tag: 'zh' | 'en' | 'jp'")
-    title: str = Field(...)
-    subtitle: str | None = Field(default=None)
-    body: str = Field(...)
-    title_pic_url: str | None = Field(default=None)
-
-
-class UpsertNotificationPayload(BaseModel):
-    notification_id: str | None = Field(default=None, description="Omit to create; supply UUID to update")
-    contents: list[LangContentPayload] = Field(..., min_length=1)
-    start_time: str | None = Field(default=None, description="RFC3339, e.g. 2026-03-01T00:00:00Z")
-    end_time: str | None = Field(default=None, description="RFC3339, e.g. 2026-03-20T23:59:59Z")
-    frequency: str = Field(default="once", description="'once' | 'every_page_load'")
-    status: str = Field(default="active", description="'active' | 'inactive'")
-
-
-class BatchAddNotificationAccountsPayload(BaseModel):
-    notification_id: str = Field(...)
-    user_email: list[str] = Field(..., description="List of account email addresses")
-
-
-register_schema_models(console_ns, UpsertNotificationPayload, BatchAddNotificationAccountsPayload)
-
-
-@console_ns.route("/admin/upsert_notification")
-class UpsertNotificationApi(Resource):
-    @console_ns.doc("upsert_notification")
-    @console_ns.doc(
-        description=(
-            "Create or update an in-product notification. "
-            "Supply notification_id to update an existing one; omit it to create a new one. "
-            "Pass at least one language variant in contents (zh / en / jp)."
-        )
-    )
-    @console_ns.expect(console_ns.models[UpsertNotificationPayload.__name__])
-    @console_ns.response(200, "Notification upserted successfully")
-    @only_edition_cloud
-    @admin_required
-    def post(self):
-        payload = UpsertNotificationPayload.model_validate(console_ns.payload)
-        result = BillingService.upsert_notification(
-            contents=[cast(LangContentDict, c.model_dump()) for c in payload.contents],
-            frequency=payload.frequency,
-            status=payload.status,
-            notification_id=payload.notification_id,
-            start_time=payload.start_time,
-            end_time=payload.end_time,
-        )
-        return {"result": "success", "notification_id": result.get("notificationId")}, 200
-
-
-@console_ns.route("/admin/batch_add_notification_accounts")
-class BatchAddNotificationAccountsApi(Resource):
-    @console_ns.doc("batch_add_notification_accounts")
-    @console_ns.doc(
-        description=(
-            "Register target accounts for a notification by email address. "
-            'JSON body: {"notification_id": "...", "user_email": ["a@example.com", ...]}. '
-            "File upload: multipart/form-data with a 'file' field (CSV or TXT, one email per line) "
-            "plus a 'notification_id' field. "
-            "Emails that do not match any account are silently skipped."
-        )
-    )
-    @console_ns.response(200, "Accounts added successfully")
-    @only_edition_cloud
-    @admin_required
-    def post(self):
-        from models.account import Account
-
-        if "file" in request.files:
-            notification_id = request.form.get("notification_id", "").strip()
-            if not notification_id:
-                raise BadRequest("notification_id is required.")
-            emails = self._parse_emails_from_file()
-        else:
-            payload = BatchAddNotificationAccountsPayload.model_validate(console_ns.payload)
-            notification_id = payload.notification_id
-            emails = payload.user_email
-
-        if not emails:
-            raise BadRequest("No valid email addresses provided.")
-
-        # Resolve emails → account IDs in chunks to avoid large IN-clause
-        account_ids: list[str] = []
-        chunk_size = 500
-        for i in range(0, len(emails), chunk_size):
-            chunk = emails[i : i + chunk_size]
-            rows = db.session.execute(select(Account.id, Account.email).where(Account.email.in_(chunk))).all()
-            account_ids.extend(str(row.id) for row in rows)
-
-        if not account_ids:
-            raise BadRequest("None of the provided emails matched an existing account.")
-
-        # Send to dify-saas in batches of 1000
-        total_count = 0
-        batch_size = 1000
-        for i in range(0, len(account_ids), batch_size):
-            batch = account_ids[i : i + batch_size]
-            result = BillingService.batch_add_notification_accounts(
-                notification_id=notification_id,
-                account_ids=batch,
-            )
-            total_count += result.get("count", 0)
-
-        return {
-            "result": "success",
-            "emails_provided": len(emails),
-            "accounts_matched": len(account_ids),
-            "count": total_count,
-        }, 200
-
-    @staticmethod
-    def _parse_emails_from_file() -> list[str]:
-        """Parse email addresses from an uploaded CSV or TXT file."""
-        file = request.files["file"]
-        if not file.filename:
-            raise BadRequest("Uploaded file has no filename.")
-
-        filename_lower = file.filename.lower()
-        if not filename_lower.endswith((".csv", ".txt")):
-            raise BadRequest("Invalid file type. Only CSV (.csv) and TXT (.txt) files are allowed.")
-
-        try:
-            content = file.stream.read().decode("utf-8")
-        except UnicodeDecodeError:
-            try:
-                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.")
-
-        emails: list[str] = []
-        if filename_lower.endswith(".csv"):
-            reader = csv.reader(io.StringIO(content))
-            for row in reader:
-                for cell in row:
-                    cell = cell.strip()
-                    if cell:
-                        emails.append(cell)
-        else:
-            for line in content.splitlines():
-                line = line.strip()
-                if line:
-                    emails.append(line)
-
-        # Deduplicate while preserving order
-        seen: set[str] = set()
-        unique_emails: list[str] = []
-        for email in emails:
-            if email.lower() not in seen:
-                seen.add(email.lower())
-                unique_emails.append(email)
-
-        return unique_emails

api/controllers/console/apikey.py

@@ -11,6 +11,7 @@ from werkzeug.exceptions import Forbidden
 from controllers.common.schema import register_schema_models
 from extensions.ext_database import db
 from fields.base import ResponseModel
+from libs.helper import to_timestamp
 from libs.login import current_account_with_tenant, login_required
 from models.dataset import Dataset
 from models.enums import ApiTokenType
@@ -21,12 +22,6 @@ from . import console_ns
 from .wraps import account_initialization_required, edit_permission_required, setup_required
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 class ApiKeyItem(ResponseModel):
     id: str
     type: str
@@ -37,7 +32,7 @@ class ApiKeyItem(ResponseModel):
     @field_validator("last_used_at", "created_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class ApiKeyList(ResponseModel):

api/controllers/console/app/app.py

@@ -3,7 +3,6 @@ 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
@@ -34,7 +33,7 @@ from core.trigger.constants import TRIGGER_NODE_TYPES
 from extensions.ext_database import db
 from fields.base import ResponseModel
 from graphon.enums import WorkflowExecutionStatus
-from libs.helper import build_icon_url
+from libs.helper import build_icon_url, to_timestamp
 from libs.login import current_account_with_tenant, login_required
 from models import App, DatasetPermissionEnum, Workflow
 from models.model import IconType
@@ -178,12 +177,6 @@ class AppTracePayload(BaseModel):
 type JSONValue = Any
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 class Tag(ResponseModel):
     id: str
     name: str
@@ -200,7 +193,7 @@ class WorkflowPartial(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class ModelConfigPartial(ResponseModel):
@@ -214,7 +207,7 @@ class ModelConfigPartial(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class ModelConfig(ResponseModel):
@@ -275,7 +268,7 @@ class ModelConfig(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class Site(ResponseModel):
@@ -318,7 +311,7 @@ class Site(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class DeletedTool(ResponseModel):
@@ -361,7 +354,7 @@ class AppPartial(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class AppDetail(ResponseModel):
@@ -391,7 +384,7 @@ class AppDetail(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class AppDetailWithSite(AppDetail):
@@ -856,10 +849,11 @@ class AppTraceApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    def get(self, app_id: UUID):
+    @get_app_model
+    def get(self, app_model):
         """Get app trace"""
         with session_factory.create_session() as session:
-            app_trace_config = OpsTraceManager.get_app_tracing_config(str(app_id), session)
+            app_trace_config = OpsTraceManager.get_app_tracing_config(app_model.id, session)
 
         return app_trace_config
 
@@ -873,12 +867,13 @@ class AppTraceApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
-    def post(self, app_id: UUID):
+    @get_app_model
+    def post(self, app_model):
         # add app trace
         args = AppTracePayload.model_validate(console_ns.payload)
 
         OpsTraceManager.update_app_tracing_config(
-            app_id=str(app_id),
+            app_id=app_model.id,
             enabled=args.enabled,
             tracing_provider=args.tracing_provider,
         )

api/controllers/console/app/conversation_variables.py

@@ -16,6 +16,7 @@ from controllers.console.wraps import account_initialization_required, setup_req
 from extensions.ext_database import db
 from fields._value_type_serializer import serialize_value_type
 from fields.base import ResponseModel
+from libs.helper import to_timestamp
 from libs.login import login_required
 from models import ConversationVariable
 from models.model import AppMode
@@ -25,12 +26,6 @@ class ConversationVariablesQuery(BaseModel):
     conversation_id: str = Field(..., description="Conversation ID to filter variables")
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 class ConversationVariableResponse(ResponseModel):
     id: str
     name: str
@@ -65,7 +60,7 @@ class ConversationVariableResponse(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class PaginatedConversationVariableResponse(ResponseModel):

api/controllers/console/app/mcp_server.py

@@ -13,6 +13,7 @@ from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
 from extensions.ext_database import db
 from fields.base import ResponseModel
+from libs.helper import to_timestamp
 from libs.login import current_account_with_tenant, login_required
 from models.enums import AppMCPServerStatus
 from models.model import AppMCPServer
@@ -30,12 +31,6 @@ class MCPServerUpdatePayload(BaseModel):
     status: str | None = Field(default=None, description="Server status")
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 class AppMCPServerResponse(ResponseModel):
     id: str
     name: str
@@ -59,7 +54,7 @@ class AppMCPServerResponse(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 register_schema_models(console_ns, MCPServerCreatePayload, MCPServerUpdatePayload, AppMCPServerResponse)

api/controllers/console/app/message.py

@@ -37,10 +37,9 @@ from fields.conversation_fields import (
     JSONValue,
     MessageFile,
     format_files_contained,
-    to_timestamp,
 )
 from graphon.model_runtime.errors.invoke import InvokeError
-from libs.helper import uuid_value
+from libs.helper import to_timestamp, uuid_value
 from libs.infinite_scroll_pagination import InfiniteScrollPagination
 from libs.login import current_account_with_tenant, login_required
 from models.enums import FeedbackFromSource, FeedbackRating
@@ -144,9 +143,7 @@ class MessageDetailResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return to_timestamp(value)
-        return value
+        return to_timestamp(value)
 
 
 class MessageInfiniteScrollPaginationResponse(ResponseModel):

api/controllers/console/app/ops_trace.py

@@ -1,5 +1,4 @@
 from typing import Any
-from uuid import UUID
 
 from flask import request
 from flask_restx import Resource, fields
@@ -9,8 +8,10 @@ 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.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
 from libs.login import login_required
+from models import App
 from services.ops_service import OpsService
 
 
@@ -43,11 +44,14 @@ class TraceAppConfigApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    def get(self, app_id: UUID):
-        args = TraceProviderQuery.model_validate(request.args.to_dict(flat=True))
+    @get_app_model
+    def get(self, app_model: App):
+        args = TraceProviderQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
 
         try:
-            trace_config = OpsService.get_tracing_app_config(app_id=str(app_id), tracing_provider=args.tracing_provider)
+            trace_config = OpsService.get_tracing_app_config(
+                app_id=app_model.id, tracing_provider=args.tracing_provider
+            )
             if not trace_config:
                 return {"has_not_configured": True}
             return trace_config
@@ -65,13 +69,14 @@ class TraceAppConfigApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    def post(self, app_id: UUID):
+    @get_app_model
+    def post(self, app_model: App):
         """Create a new trace app configuration"""
         args = TraceConfigPayload.model_validate(console_ns.payload)
 
         try:
             result = OpsService.create_tracing_app_config(
-                app_id=str(app_id), tracing_provider=args.tracing_provider, tracing_config=args.tracing_config
+                app_id=app_model.id, tracing_provider=args.tracing_provider, tracing_config=args.tracing_config
             )
             if not result:
                 raise TracingConfigIsExist()
@@ -90,13 +95,14 @@ class TraceAppConfigApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    def patch(self, app_id: UUID):
+    @get_app_model
+    def patch(self, app_model: App):
         """Update an existing trace app configuration"""
         args = TraceConfigPayload.model_validate(console_ns.payload)
 
         try:
             result = OpsService.update_tracing_app_config(
-                app_id=str(app_id), tracing_provider=args.tracing_provider, tracing_config=args.tracing_config
+                app_id=app_model.id, tracing_provider=args.tracing_provider, tracing_config=args.tracing_config
             )
             if not result:
                 raise TracingConfigNotExist()
@@ -113,12 +119,13 @@ class TraceAppConfigApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    def delete(self, app_id: UUID):
+    @get_app_model
+    def delete(self, app_model: App):
         """Delete an existing trace app configuration"""
         args = TraceProviderQuery.model_validate(request.args.to_dict(flat=True))
 
         try:
-            result = OpsService.delete_tracing_app_config(app_id=str(app_id), tracing_provider=args.tracing_provider)
+            result = OpsService.delete_tracing_app_config(app_id=app_model.id, tracing_provider=args.tracing_provider)
             if not result:
                 raise TracingConfigNotExist()
             return {"result": "success"}, 204

api/controllers/console/app/workflow.py

@@ -1,17 +1,22 @@
 import json
 import logging
 from collections.abc import Sequence
-from typing import Any
+from datetime import datetime
+from typing import Any, NotRequired, TypedDict
 
 from flask import abort, request
-from flask_restx import Resource, fields, marshal, marshal_with
-from pydantic import BaseModel, Field, ValidationError, field_validator
+from flask_restx import Resource, fields
+from pydantic import AliasChoices, BaseModel, Field, ValidationError, field_validator
 from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import BadRequest, Forbidden, InternalServerError, NotFound
 
 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.common.schema import (
+    register_response_schema_model,
+    register_response_schema_models,
+    register_schema_models,
+)
 from controllers.console import console_ns
 from controllers.console.app.error import ConversationCompletedError, DraftWorkflowNotExist, DraftWorkflowNotSync
 from controllers.console.app.wraps import get_app_model
@@ -22,6 +27,7 @@ from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.apps.workflow.app_generator import SKIP_PREPARE_USER_INPUTS_KEY
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.app.file_access import DatabaseFileAccessController
+from core.helper import encrypter
 from core.helper.trace_id_helper import get_external_trace_id
 from core.plugin.impl.exc import PluginInvokeError
 from core.trigger.constants import TRIGGER_SCHEDULE_NODE_TYPE
@@ -34,18 +40,18 @@ from core.trigger.debug.event_selectors import (
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 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.base import ResponseModel
+from fields.member_fields import SimpleAccount
 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
 from graphon.graph_engine.manager import GraphEngineManager
 from graphon.model_runtime.utils.encoders import jsonable_encoder
+from graphon.variables import SecretVariable, SegmentType, VariableBase
 from libs import helper
 from libs.datetime_utils import naive_utc_now
-from libs.helper import TimestampField, uuid_value
+from libs.helper import TimestampField, dump_response, to_timestamp, uuid_value
 from libs.login import current_account_with_tenant, login_required
 from models import App
 from models.model import AppMode
@@ -64,42 +70,15 @@ LISTENING_RETRY_IN = 2000
 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
+ENVIRONMENT_VARIABLE_SUPPORTED_TYPES = (SegmentType.STRING, SegmentType.NUMBER, SegmentType.SECRET)
 
-# 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)
-
-from fields.workflow_fields import pipeline_variable_fields, serialize_value_type
-
-conversation_variable_model = console_ns.model(
-    "ConversationVariable",
-    {
-        "id": fields.String,
-        "name": fields.String,
-        "value_type": fields.String(attribute=serialize_value_type),
-        "value": fields.Raw,
-        "description": fields.String,
-    },
-)
-
-pipeline_variable_model = console_ns.model("PipelineVariable", pipeline_variable_fields)
-
-# Workflow model with nested dependencies
-workflow_fields_copy = workflow_fields.copy()
-workflow_fields_copy["created_by"] = fields.Nested(simple_account_model, attribute="created_by_account")
-workflow_fields_copy["updated_by"] = fields.Nested(
-    simple_account_model, attribute="updated_by_account", allow_null=True
-)
-workflow_fields_copy["conversation_variables"] = fields.List(fields.Nested(conversation_variable_model))
-workflow_fields_copy["rag_pipeline_variables"] = fields.List(fields.Nested(pipeline_variable_model))
-workflow_model = console_ns.model("Workflow", workflow_fields_copy)
-
-# Workflow pagination model
-workflow_pagination_fields_copy = workflow_pagination_fields.copy()
-workflow_pagination_fields_copy["items"] = fields.List(fields.Nested(workflow_model), attribute="items")
-workflow_pagination_model = console_ns.model("WorkflowPagination", workflow_pagination_fields_copy)
+class EnvironmentVariableResponseDict(TypedDict):
+    value_type: str
+    id: NotRequired[str]
+    name: NotRequired[str]
+    value: NotRequired[Any]
+    description: NotRequired[str | None]
 
 
 class SyncDraftWorkflowPayload(BaseModel):
@@ -170,6 +149,110 @@ class WorkflowOnlineUsersPayload(BaseModel):
         return list(dict.fromkeys(app_id.strip() for app_id in app_ids if app_id.strip()))
 
 
+class WorkflowConversationVariableResponse(ResponseModel):
+    id: str
+    name: str
+    value_type: str
+    value: Any = Field(json_schema_extra={"type": "object"})
+    description: str
+
+    @field_validator("value_type", mode="before")
+    @classmethod
+    def _serialize_value_type(cls, value: Any) -> str:
+        if hasattr(value, "exposed_type"):
+            return str(value.exposed_type())
+        return str(value)
+
+
+class PipelineVariableResponse(ResponseModel):
+    label: str
+    variable: str
+    type: str
+    belong_to_node_id: str
+    max_length: int | None = None
+    required: bool
+    unit: str | None = None
+    default_value: Any = Field(default=None, json_schema_extra={"type": "object"})
+    options: list[str] | None = None
+    placeholder: str | None = None
+    tooltips: str | None = None
+    allowed_file_types: list[str] | None = None
+    allowed_file_extensions: list[str] | None = Field(
+        default=None, validation_alias=AliasChoices("allowed_file_extensions", "allow_file_extension")
+    )
+    allowed_file_upload_methods: list[str] | None = Field(
+        default=None, validation_alias=AliasChoices("allowed_file_upload_methods", "allow_file_upload_methods")
+    )
+
+
+class WorkflowEnvironmentVariableResponse(ResponseModel):
+    value_type: str
+    id: str
+    name: str
+    value: Any = Field(json_schema_extra={"type": "object"})
+    description: str
+
+
+class WorkflowResponse(ResponseModel):
+    id: str
+    graph: dict[str, Any] = Field(validation_alias=AliasChoices("graph_dict", "graph"))
+    features: dict[str, Any] = Field(validation_alias=AliasChoices("features_dict", "features"))
+    hash: str = Field(validation_alias=AliasChoices("unique_hash", "hash"))
+    version: str
+    marked_name: str
+    marked_comment: str
+    created_by: SimpleAccount | None = Field(
+        default=None, validation_alias=AliasChoices("created_by_account", "created_by")
+    )
+    created_at: int
+    updated_by: SimpleAccount | None = Field(
+        default=None, validation_alias=AliasChoices("updated_by_account", "updated_by")
+    )
+    updated_at: int
+    tool_published: bool
+    environment_variables: list[WorkflowEnvironmentVariableResponse]
+    conversation_variables: list[WorkflowConversationVariableResponse]
+    rag_pipeline_variables: list[PipelineVariableResponse]
+
+    @field_validator("created_at", "updated_at", mode="before")
+    @classmethod
+    def _normalize_timestamp(cls, value: datetime | int | None) -> int:
+        timestamp = to_timestamp(value)
+        if timestamp is None:
+            raise ValueError("timestamp is required")
+        return timestamp
+
+    @field_validator("environment_variables", mode="before")
+    @classmethod
+    def _serialize_environment_variables(cls, value: Any) -> list[Any]:
+        if value is None:
+            return []
+
+        return [_serialize_environment_variable(item) for item in value]
+
+
+class WorkflowPaginationResponse(ResponseModel):
+    items: list[WorkflowResponse]
+    page: int
+    limit: int
+    has_more: bool
+
+
+class WorkflowOnlineUser(ResponseModel):
+    user_id: str
+    username: str
+    avatar: str | None = None
+
+
+class WorkflowOnlineUsersByApp(ResponseModel):
+    app_id: str
+    users: list[WorkflowOnlineUser]
+
+
+class WorkflowOnlineUsersResponse(ResponseModel):
+    data: list[WorkflowOnlineUsersByApp]
+
+
 class DraftWorkflowTriggerRunPayload(BaseModel):
     node_id: str
 
@@ -197,6 +280,17 @@ register_schema_models(
     DraftWorkflowTriggerRunAllPayload,
 )
 register_response_schema_model(console_ns, WorkflowRunNodeExecutionResponse)
+register_response_schema_models(
+    console_ns,
+    WorkflowConversationVariableResponse,
+    PipelineVariableResponse,
+    WorkflowEnvironmentVariableResponse,
+    WorkflowResponse,
+    WorkflowPaginationResponse,
+    WorkflowOnlineUser,
+    WorkflowOnlineUsersByApp,
+    WorkflowOnlineUsersResponse,
+)
 
 
 # TODO(QuantumGhost): Refactor existing node run API to handle file parameter parsing
@@ -218,18 +312,56 @@ def _parse_file(workflow: Workflow, files: list[dict] | None = None) -> Sequence
     return file_objs
 
 
+def _serialize_environment_variable(value: Any) -> EnvironmentVariableResponseDict | Any:
+    match value:
+        case SecretVariable():
+            return {
+                "id": value.id,
+                "name": value.name,
+                "value": encrypter.full_mask_token(),
+                "value_type": value.value_type.value,
+                "description": value.description,
+            }
+
+        case VariableBase():
+            return {
+                "id": value.id,
+                "name": value.name,
+                "value": value.value,
+                "value_type": str(value.value_type.exposed_type()),
+                "description": value.description,
+            }
+
+        case dict():
+            value_type_str = value.get("value_type")
+            if not isinstance(value_type_str, str):
+                raise TypeError(
+                    f"unexpected type for value_type field, value={value_type_str}, type={type(value_type_str)}"
+                )
+            value_type = SegmentType(value_type_str).exposed_type()
+            if value_type not in ENVIRONMENT_VARIABLE_SUPPORTED_TYPES:
+                raise ValueError(f"Unsupported environment variable value type: {value_type}")
+            return value
+
+        case _:
+            return value
+
+
 @console_ns.route("/apps/<uuid:app_id>/workflows/draft")
 class DraftWorkflowApi(Resource):
     @console_ns.doc("get_draft_workflow")
     @console_ns.doc(description="Get draft workflow for an application")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.response(200, "Draft workflow retrieved successfully", workflow_model)
+    @console_ns.response(
+        200,
+        "Draft workflow retrieved successfully",
+        console_ns.models[WorkflowResponse.__name__],
+    )
     @console_ns.response(404, "Draft workflow not found")
     @setup_required
     @login_required
     @account_initialization_required
     @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    @marshal_with(workflow_model)
     @edit_permission_required
     def get(self, app_model: App):
         """
@@ -242,8 +374,8 @@ class DraftWorkflowApi(Resource):
         if not workflow:
             raise DraftWorkflowNotExist()
 
-        # return workflow, if not found, return None (initiate graph by frontend)
-        return workflow
+        # return workflow, if not found, return 404
+        return dump_response(WorkflowResponse, workflow)
 
     @setup_required
     @login_required
@@ -817,13 +949,15 @@ class PublishedWorkflowApi(Resource):
     @console_ns.doc("get_published_workflow")
     @console_ns.doc(description="Get published workflow for an application")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.response(200, "Published workflow retrieved successfully", workflow_model)
-    @console_ns.response(404, "Published workflow not found")
+    @console_ns.response(
+        200,
+        "Published workflow retrieved successfully, or null if not found",
+        console_ns.models[WorkflowResponse.__name__],
+    )
     @setup_required
     @login_required
     @account_initialization_required
     @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    @marshal_with(workflow_model)
     @edit_permission_required
     def get(self, app_model: App):
         """
@@ -834,7 +968,10 @@ class PublishedWorkflowApi(Resource):
         workflow = workflow_service.get_published_workflow(app_model=app_model)
 
         # return workflow, if not found, return None
-        return workflow
+        if workflow is None:
+            return None
+
+        return dump_response(WorkflowResponse, workflow)
 
     @console_ns.expect(console_ns.models[PublishWorkflowPayload.__name__])
     @setup_required
@@ -993,7 +1130,11 @@ class PublishedAllWorkflowApi(Resource):
     @console_ns.doc("get_all_published_workflows")
     @console_ns.doc(description="Get all published workflows for an application")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.response(200, "Published workflows retrieved successfully", workflow_pagination_model)
+    @console_ns.response(
+        200,
+        "Published workflows retrieved successfully",
+        console_ns.models[WorkflowPaginationResponse.__name__],
+    )
     @setup_required
     @login_required
     @account_initialization_required
@@ -1025,14 +1166,14 @@ class PublishedAllWorkflowApi(Resource):
                 user_id=user_id,
                 named_only=named_only,
             )
-            serialized_workflows = marshal(workflows, workflow_fields_copy)
-
-            return {
-                "items": serialized_workflows,
-                "page": page,
-                "limit": limit,
-                "has_more": has_more,
-            }
+            return WorkflowPaginationResponse.model_validate(
+                {
+                    "items": workflows,
+                    "page": page,
+                    "limit": limit,
+                    "has_more": has_more,
+                }
+            ).model_dump(mode="json")
 
 
 @console_ns.route("/apps/<uuid:app_id>/workflows/<string:workflow_id>/restore")
@@ -1078,14 +1219,13 @@ class WorkflowByIdApi(Resource):
     @console_ns.doc(description="Update workflow by ID")
     @console_ns.doc(params={"app_id": "Application ID", "workflow_id": "Workflow ID"})
     @console_ns.expect(console_ns.models[WorkflowUpdatePayload.__name__])
-    @console_ns.response(200, "Workflow updated successfully", workflow_model)
+    @console_ns.response(200, "Workflow updated successfully", console_ns.models[WorkflowResponse.__name__])
     @console_ns.response(404, "Workflow 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_model)
     @edit_permission_required
     def patch(self, app_model: App, workflow_id: str):
         """
@@ -1119,7 +1259,7 @@ class WorkflowByIdApi(Resource):
             if not workflow:
                 raise NotFound("Workflow not found")
 
-        return workflow
+        return dump_response(WorkflowResponse, workflow)
 
     @setup_required
     @login_required
@@ -1404,12 +1544,16 @@ class DraftWorkflowTriggerRunAllApi(Resource):
 @console_ns.route("/apps/workflows/online-users")
 class WorkflowOnlineUsersApi(Resource):
     @console_ns.expect(console_ns.models[WorkflowOnlineUsersPayload.__name__])
+    @console_ns.response(
+        200,
+        "Workflow online users retrieved successfully",
+        console_ns.models[WorkflowOnlineUsersResponse.__name__],
+    )
     @console_ns.doc("get_workflow_online_users")
     @console_ns.doc(description="Get workflow online users")
     @setup_required
     @login_required
     @account_initialization_required
-    @marshal_with(online_user_list_fields)
     def post(self):
         args = WorkflowOnlineUsersPayload.model_validate(console_ns.payload or {})
 
@@ -1452,10 +1596,18 @@ class WorkflowOnlineUsersApi(Resource):
                 if not isinstance(user_info, dict):
                     continue
 
+                user_id = user_info.get("user_id")
+                username = user_info.get("username")
+                if not isinstance(user_id, str) or not isinstance(username, str):
+                    continue
+
                 avatar = user_info.get("avatar")
+                if avatar is not None and not isinstance(avatar, str):
+                    avatar = None
+
                 if isinstance(avatar, str) and avatar and not avatar.startswith(("http://", "https://")):
                     try:
-                        user_info["avatar"] = file_helpers.get_signed_file_url(avatar)
+                        avatar = file_helpers.get_signed_file_url(avatar)
                     except Exception as exc:
                         logger.warning(
                             "Failed to sign workflow online user avatar; using original value. "
@@ -1465,7 +1617,7 @@ class WorkflowOnlineUsersApi(Resource):
                             exc,
                         )
 
-                users.append(user_info)
+                users.append({"user_id": user_id, "username": username, "avatar": avatar})
             results.append({"app_id": app_id, "users": users})
 
-        return {"data": results}
+        return WorkflowOnlineUsersResponse.model_validate({"data": results}).model_dump(mode="json")

api/controllers/console/app/workflow_app_log.py

@@ -16,6 +16,7 @@ from fields.base import ResponseModel
 from fields.end_user_fields import SimpleEndUser
 from fields.member_fields import SimpleAccount
 from graphon.enums import WorkflowExecutionStatus
+from libs.helper import to_timestamp
 from libs.login import login_required
 from models import App
 from models.model import AppMode
@@ -82,9 +83,7 @@ class WorkflowRunForLogResponse(ResponseModel):
     @field_validator("created_at", "finished_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return int(value.timestamp())
-        return value
+        return to_timestamp(value)
 
 
 class WorkflowRunForArchivedLogResponse(ResponseModel):
@@ -117,9 +116,7 @@ class WorkflowAppLogPartialResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return int(value.timestamp())
-        return value
+        return to_timestamp(value)
 
 
 class WorkflowArchivedLogPartialResponse(ResponseModel):
@@ -133,9 +130,7 @@ class WorkflowArchivedLogPartialResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return int(value.timestamp())
-        return value
+        return to_timestamp(value)
 
 
 class WorkflowAppLogPaginationResponse(ResponseModel):

api/controllers/console/app/workflow_comment.py

@@ -1,22 +1,16 @@
 import logging
+from datetime import datetime
 
-from flask_restx import Resource, marshal_with
-from pydantic import BaseModel, Field, TypeAdapter
+from flask_restx import Resource
+from pydantic import BaseModel, Field, TypeAdapter, computed_field, field_validator
 
-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.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
+from fields.base import ResponseModel
 from fields.member_fields import AccountWithRole
-from fields.workflow_comment_fields import (
-    workflow_comment_basic_fields,
-    workflow_comment_create_fields,
-    workflow_comment_detail_fields,
-    workflow_comment_reply_create_fields,
-    workflow_comment_reply_update_fields,
-    workflow_comment_resolve_fields,
-    workflow_comment_update_fields,
-)
+from libs.helper import build_avatar_url, dump_response, to_timestamp
 from libs.login import current_user, login_required
 from models import App
 from services.account_service import TenantService
@@ -51,6 +45,138 @@ class WorkflowCommentMentionUsersPayload(BaseModel):
     users: list[AccountWithRole]
 
 
+class WorkflowCommentAccount(ResponseModel):
+    id: str
+    name: str
+    email: str
+    avatar: str | None = Field(default=None, exclude=True)
+
+    @computed_field(return_type=str | None)  # type: ignore[prop-decorator]
+    @property
+    def avatar_url(self) -> str | None:
+        return build_avatar_url(self.avatar)
+
+
+class WorkflowCommentReply(ResponseModel):
+    id: str
+    content: str
+    created_by: str
+    created_by_account: WorkflowCommentAccount | None = None
+    created_at: int | None = None
+
+    @field_validator("created_at", mode="before")
+    @classmethod
+    def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
+        return to_timestamp(value)
+
+
+class WorkflowCommentMention(ResponseModel):
+    mentioned_user_id: str
+    mentioned_user_account: WorkflowCommentAccount | None = None
+    reply_id: str | None = None
+
+
+class WorkflowCommentBasic(ResponseModel):
+    id: str
+    position_x: float
+    position_y: float
+    content: str
+    created_by: str
+    created_by_account: WorkflowCommentAccount | None = None
+    created_at: int | None = None
+    updated_at: int | None = None
+    resolved: bool
+    resolved_at: int | None = None
+    resolved_by: str | None = None
+    resolved_by_account: WorkflowCommentAccount | None = None
+    reply_count: int
+    mention_count: int
+    participants: list[WorkflowCommentAccount]
+
+    @field_validator("created_at", "updated_at", "resolved_at", mode="before")
+    @classmethod
+    def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
+        return to_timestamp(value)
+
+
+class WorkflowCommentBasicList(ResponseModel):
+    data: list[WorkflowCommentBasic]
+
+
+class WorkflowCommentDetail(ResponseModel):
+    id: str
+    position_x: float
+    position_y: float
+    content: str
+    created_by: str
+    created_by_account: WorkflowCommentAccount | None = None
+    created_at: int | None = None
+    updated_at: int | None = None
+    resolved: bool
+    resolved_at: int | None = None
+    resolved_by: str | None = None
+    resolved_by_account: WorkflowCommentAccount | None = None
+    replies: list[WorkflowCommentReply]
+    mentions: list[WorkflowCommentMention]
+
+    @field_validator("created_at", "updated_at", "resolved_at", mode="before")
+    @classmethod
+    def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
+        return to_timestamp(value)
+
+
+class WorkflowCommentCreate(ResponseModel):
+    id: str
+    created_at: int | None = None
+
+    @field_validator("created_at", mode="before")
+    @classmethod
+    def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
+        return to_timestamp(value)
+
+
+class WorkflowCommentUpdate(ResponseModel):
+    id: str
+    updated_at: int | None = None
+
+    @field_validator("updated_at", mode="before")
+    @classmethod
+    def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
+        return to_timestamp(value)
+
+
+class WorkflowCommentResolve(ResponseModel):
+    id: str
+    resolved: bool
+    resolved_at: int | None = None
+    resolved_by: str | None = None
+
+    @field_validator("resolved_at", mode="before")
+    @classmethod
+    def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
+        return to_timestamp(value)
+
+
+class WorkflowCommentReplyCreate(ResponseModel):
+    id: str
+    created_at: int | None = None
+
+    @field_validator("created_at", mode="before")
+    @classmethod
+    def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
+        return to_timestamp(value)
+
+
+class WorkflowCommentReplyUpdate(ResponseModel):
+    id: str
+    updated_at: int | None = None
+
+    @field_validator("updated_at", mode="before")
+    @classmethod
+    def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
+        return to_timestamp(value)
+
+
 register_schema_models(
     console_ns,
     AccountWithRole,
@@ -59,17 +185,19 @@ register_schema_models(
     WorkflowCommentUpdatePayload,
     WorkflowCommentReplyPayload,
 )
-
-workflow_comment_basic_model = console_ns.model("WorkflowCommentBasic", workflow_comment_basic_fields)
-workflow_comment_detail_model = console_ns.model("WorkflowCommentDetail", workflow_comment_detail_fields)
-workflow_comment_create_model = console_ns.model("WorkflowCommentCreate", workflow_comment_create_fields)
-workflow_comment_update_model = console_ns.model("WorkflowCommentUpdate", workflow_comment_update_fields)
-workflow_comment_resolve_model = console_ns.model("WorkflowCommentResolve", workflow_comment_resolve_fields)
-workflow_comment_reply_create_model = console_ns.model(
-    "WorkflowCommentReplyCreate", workflow_comment_reply_create_fields
-)
-workflow_comment_reply_update_model = console_ns.model(
-    "WorkflowCommentReplyUpdate", workflow_comment_reply_update_fields
+register_response_schema_models(
+    console_ns,
+    WorkflowCommentAccount,
+    WorkflowCommentReply,
+    WorkflowCommentMention,
+    WorkflowCommentBasic,
+    WorkflowCommentBasicList,
+    WorkflowCommentDetail,
+    WorkflowCommentCreate,
+    WorkflowCommentUpdate,
+    WorkflowCommentResolve,
+    WorkflowCommentReplyCreate,
+    WorkflowCommentReplyUpdate,
 )
 
 
@@ -80,28 +208,26 @@ class WorkflowCommentListApi(Resource):
     @console_ns.doc("list_workflow_comments")
     @console_ns.doc(description="Get all comments for a workflow")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.response(200, "Comments retrieved successfully", workflow_comment_basic_model)
+    @console_ns.response(200, "Comments retrieved successfully", console_ns.models[WorkflowCommentBasicList.__name__])
     @login_required
     @setup_required
     @account_initialization_required
     @get_app_model()
-    @marshal_with(workflow_comment_basic_model, envelope="data")
     def get(self, app_model: App):
         """Get all comments for a workflow."""
         comments = WorkflowCommentService.get_comments(tenant_id=current_user.current_tenant_id, app_id=app_model.id)
 
-        return comments
+        return WorkflowCommentBasicList.model_validate({"data": comments}).model_dump(mode="json")
 
     @console_ns.doc("create_workflow_comment")
     @console_ns.doc(description="Create a new workflow comment")
     @console_ns.doc(params={"app_id": "Application ID"})
     @console_ns.expect(console_ns.models[WorkflowCommentCreatePayload.__name__])
-    @console_ns.response(201, "Comment created successfully", workflow_comment_create_model)
+    @console_ns.response(201, "Comment created successfully", console_ns.models[WorkflowCommentCreate.__name__])
     @login_required
     @setup_required
     @account_initialization_required
     @get_app_model()
-    @marshal_with(workflow_comment_create_model)
     @edit_permission_required
     def post(self, app_model: App):
         """Create a new workflow comment."""
@@ -117,7 +243,7 @@ class WorkflowCommentListApi(Resource):
             mentioned_user_ids=payload.mentioned_user_ids,
         )
 
-        return result, 201
+        return dump_response(WorkflowCommentCreate, result), 201
 
 
 @console_ns.route("/apps/<uuid:app_id>/workflow/comments/<string:comment_id>")
@@ -127,30 +253,28 @@ class WorkflowCommentDetailApi(Resource):
     @console_ns.doc("get_workflow_comment")
     @console_ns.doc(description="Get a specific workflow comment")
     @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID"})
-    @console_ns.response(200, "Comment retrieved successfully", workflow_comment_detail_model)
+    @console_ns.response(200, "Comment retrieved successfully", console_ns.models[WorkflowCommentDetail.__name__])
     @login_required
     @setup_required
     @account_initialization_required
     @get_app_model()
-    @marshal_with(workflow_comment_detail_model)
     def get(self, app_model: App, comment_id: str):
         """Get a specific workflow comment."""
         comment = WorkflowCommentService.get_comment(
             tenant_id=current_user.current_tenant_id, app_id=app_model.id, comment_id=comment_id
         )
 
-        return comment
+        return dump_response(WorkflowCommentDetail, comment)
 
     @console_ns.doc("update_workflow_comment")
     @console_ns.doc(description="Update a workflow comment")
     @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID"})
     @console_ns.expect(console_ns.models[WorkflowCommentUpdatePayload.__name__])
-    @console_ns.response(200, "Comment updated successfully", workflow_comment_update_model)
+    @console_ns.response(200, "Comment updated successfully", console_ns.models[WorkflowCommentUpdate.__name__])
     @login_required
     @setup_required
     @account_initialization_required
     @get_app_model()
-    @marshal_with(workflow_comment_update_model)
     @edit_permission_required
     def put(self, app_model: App, comment_id: str):
         """Update a workflow comment."""
@@ -167,7 +291,7 @@ class WorkflowCommentDetailApi(Resource):
             mentioned_user_ids=payload.mentioned_user_ids,
         )
 
-        return result
+        return dump_response(WorkflowCommentUpdate, result)
 
     @console_ns.doc("delete_workflow_comment")
     @console_ns.doc(description="Delete a workflow comment")
@@ -197,12 +321,11 @@ class WorkflowCommentResolveApi(Resource):
     @console_ns.doc("resolve_workflow_comment")
     @console_ns.doc(description="Resolve a workflow comment")
     @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID"})
-    @console_ns.response(200, "Comment resolved successfully", workflow_comment_resolve_model)
+    @console_ns.response(200, "Comment resolved successfully", console_ns.models[WorkflowCommentResolve.__name__])
     @login_required
     @setup_required
     @account_initialization_required
     @get_app_model()
-    @marshal_with(workflow_comment_resolve_model)
     @edit_permission_required
     def post(self, app_model: App, comment_id: str):
         """Resolve a workflow comment."""
@@ -213,7 +336,7 @@ class WorkflowCommentResolveApi(Resource):
             user_id=current_user.id,
         )
 
-        return comment
+        return dump_response(WorkflowCommentResolve, comment)
 
 
 @console_ns.route("/apps/<uuid:app_id>/workflow/comments/<string:comment_id>/replies")
@@ -224,12 +347,11 @@ class WorkflowCommentReplyApi(Resource):
     @console_ns.doc(description="Add a reply to a workflow comment")
     @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID"})
     @console_ns.expect(console_ns.models[WorkflowCommentReplyPayload.__name__])
-    @console_ns.response(201, "Reply created successfully", workflow_comment_reply_create_model)
+    @console_ns.response(201, "Reply created successfully", console_ns.models[WorkflowCommentReplyCreate.__name__])
     @login_required
     @setup_required
     @account_initialization_required
     @get_app_model()
-    @marshal_with(workflow_comment_reply_create_model)
     @edit_permission_required
     def post(self, app_model: App, comment_id: str):
         """Add a reply to a workflow comment."""
@@ -247,7 +369,7 @@ class WorkflowCommentReplyApi(Resource):
             mentioned_user_ids=payload.mentioned_user_ids,
         )
 
-        return result, 201
+        return dump_response(WorkflowCommentReplyCreate, result), 201
 
 
 @console_ns.route("/apps/<uuid:app_id>/workflow/comments/<string:comment_id>/replies/<string:reply_id>")
@@ -258,12 +380,11 @@ class WorkflowCommentReplyDetailApi(Resource):
     @console_ns.doc(description="Update a comment reply")
     @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID", "reply_id": "Reply ID"})
     @console_ns.expect(console_ns.models[WorkflowCommentReplyPayload.__name__])
-    @console_ns.response(200, "Reply updated successfully", workflow_comment_reply_update_model)
+    @console_ns.response(200, "Reply updated successfully", console_ns.models[WorkflowCommentReplyUpdate.__name__])
     @login_required
     @setup_required
     @account_initialization_required
     @get_app_model()
-    @marshal_with(workflow_comment_reply_update_model)
     @edit_permission_required
     def put(self, app_model: App, comment_id: str, reply_id: str):
         """Update a comment reply."""
@@ -284,7 +405,7 @@ class WorkflowCommentReplyDetailApi(Resource):
             mentioned_user_ids=payload.mentioned_user_ids,
         )
 
-        return reply
+        return dump_response(WorkflowCommentReplyUpdate, reply)
 
     @console_ns.doc("delete_workflow_comment_reply")
     @console_ns.doc(description="Delete a comment reply")

api/controllers/console/auth/email_register.py

@@ -3,7 +3,7 @@ from flask_restx import Resource
 from pydantic import BaseModel, Field, field_validator
 
 from configs import dify_config
-from constants.languages import languages
+from constants.languages import get_valid_language, languages
 from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.auth.error import (
@@ -15,11 +15,12 @@ from controllers.console.auth.error import (
     PasswordMismatchError,
 )
 from libs.helper import EmailStr, extract_remote_ip
+from libs.helper import timezone as validate_timezone_string
 from libs.password import valid_password
 from models import Account
 from services.account_service import AccountService
 from services.billing_service import BillingService
-from services.errors.account import AccountNotFoundError, AccountRegisterError
+from services.errors.account import AccountRegisterError
 
 from ..error import AccountInFreezeError, EmailSendIpLimitError
 from ..wraps import email_password_login_enabled, email_register_enabled, setup_required
@@ -40,12 +41,21 @@ class EmailRegisterResetPayload(BaseModel):
     token: str = Field(...)
     new_password: str = Field(...)
     password_confirm: str = Field(...)
+    language: str | None = Field(default=None)
+    timezone: str | None = Field(default=None)
 
     @field_validator("new_password", "password_confirm")
     @classmethod
     def validate_password(cls, value: str) -> str:
         return valid_password(value)
 
+    @field_validator("timezone")
+    @classmethod
+    def validate_timezone(cls, value: str | None) -> str | None:
+        if value is None:
+            return None
+        return validate_timezone_string(value)
+
 
 register_schema_models(console_ns, EmailRegisterSendPayload, EmailRegisterValidityPayload, EmailRegisterResetPayload)
 
@@ -144,26 +154,32 @@ class EmailRegisterResetApi(Resource):
 
         if account:
             raise EmailAlreadyInUseError()
-        else:
-            account = self._create_new_account(normalized_email, args.password_confirm)
-            if not account:
-                raise AccountNotFoundError()
-            token_pair = AccountService.login(account=account, ip_address=extract_remote_ip(request))
-            AccountService.reset_login_error_rate_limit(normalized_email)
+
+        account = self._create_new_account(
+            email=normalized_email,
+            password=args.password_confirm,
+            timezone=args.timezone,
+            language=args.language,
+        )
+        token_pair = AccountService.login(account=account, ip_address=extract_remote_ip(request))
+        AccountService.reset_login_error_rate_limit(normalized_email)
 
         return {"result": "success", "data": token_pair.model_dump()}
 
-    def _create_new_account(self, email: str, password: str) -> Account | None:
-        # Create new account if allowed
-        account = None
+    def _create_new_account(
+        self,
+        email: str,
+        password: str,
+        timezone: str | None = None,
+        language: str | None = None,
+    ) -> Account:
         try:
-            account = AccountService.create_account_and_tenant(
+            return AccountService.create_account_and_tenant(
                 email=email,
                 name=email,
                 password=password,
-                interface_language=languages[0],
+                interface_language=get_valid_language(language),
+                timezone=timezone,
             )
         except AccountRegisterError:
             raise AccountInFreezeError()
-
-        return account

api/controllers/console/auth/login.py

@@ -3,7 +3,7 @@ import logging
 import flask_login
 from flask import make_response, request
 from flask_restx import Resource
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, field_validator
 from werkzeug.exceptions import Unauthorized
 
 import services
@@ -34,6 +34,7 @@ from controllers.console.wraps import (
 )
 from events.tenant_event import tenant_was_created
 from libs.helper import EmailStr, extract_remote_ip
+from libs.helper import timezone as validate_timezone_string
 from libs.login import current_account_with_tenant
 from libs.token import (
     clear_access_token_from_cookie,
@@ -69,6 +70,14 @@ class EmailCodeLoginPayload(BaseModel):
     code: str = Field(...)
     token: str = Field(...)
     language: str | None = Field(default=None)
+    timezone: str | None = Field(default=None)
+
+    @field_validator("timezone")
+    @classmethod
+    def validate_timezone(cls, value: str | None) -> str | None:
+        if value is None:
+            return None
+        return validate_timezone_string(value)
 
 
 register_schema_models(console_ns, LoginPayload, EmailPayload, EmailCodeLoginPayload)
@@ -288,6 +297,7 @@ class EmailCodeLoginApi(Resource):
                     email=user_email,
                     name=user_email,
                     interface_language=get_valid_language(language),
+                    timezone=args.timezone,
                 )
             except WorkSpaceNotAllowedCreateError:
                 raise NotAllowedCreateWorkspace()

api/controllers/console/auth/oauth.py

@@ -12,7 +12,8 @@ from events.tenant_event import tenant_was_created
 from extensions.ext_database import db
 from libs.datetime_utils import naive_utc_now
 from libs.helper import extract_remote_ip
-from libs.oauth import GitHubOAuth, GoogleOAuth, OAuthUserInfo
+from libs.helper import timezone as validate_timezone_string
+from libs.oauth import GitHubOAuth, GoogleOAuth, OAuthUserInfo, decode_oauth_state
 from libs.token import (
     set_access_token_to_cookie,
     set_csrf_token_to_cookie,
@@ -53,6 +54,31 @@ def get_oauth_providers():
         return OAUTH_PROVIDERS
 
 
+def _validated_timezone(value: str | None) -> str | None:
+    if not value:
+        return None
+    try:
+        return validate_timezone_string(value)
+    except ValueError:
+        return None
+
+
+def _validated_language(value: str | None) -> str | None:
+    if value and value in languages:
+        return value
+    return None
+
+
+def _preferred_interface_language(language: str | None = None) -> str:
+    if language:
+        return language
+
+    preferred_lang = request.accept_languages.best_match(languages)
+    if preferred_lang and preferred_lang in languages:
+        return preferred_lang
+    return languages[0]
+
+
 @console_ns.route("/oauth/login/<provider>")
 class OAuthLogin(Resource):
     @console_ns.doc("oauth_login")
@@ -64,13 +90,19 @@ class OAuthLogin(Resource):
     @console_ns.response(400, "Invalid provider")
     def get(self, provider: str):
         invite_token = request.args.get("invite_token") or None
+        timezone = _validated_timezone(request.args.get("timezone") or None)
+        language = _validated_language(request.args.get("language") or None)
         OAUTH_PROVIDERS = get_oauth_providers()
         with current_app.app_context():
             oauth_provider = OAUTH_PROVIDERS.get(provider)
         if not oauth_provider:
             return {"error": "Invalid provider"}, 400
 
-        auth_url = oauth_provider.get_authorization_url(invite_token=invite_token)
+        auth_url = oauth_provider.get_authorization_url(
+            invite_token=invite_token,
+            timezone=timezone,
+            language=language,
+        )
         return redirect(auth_url)
 
 
@@ -96,9 +128,10 @@ class OAuthCallback(Resource):
 
         code = request.args.get("code")
         state = request.args.get("state")
-        invite_token = None
-        if state:
-            invite_token = state
+        oauth_state = decode_oauth_state(state)
+        invite_token = oauth_state.get("invite_token")
+        timezone = _validated_timezone(oauth_state.get("timezone"))
+        language = _validated_language(oauth_state.get("language"))
 
         if not code:
             return {"error": "Authorization code is required"}, 400
@@ -129,7 +162,7 @@ class OAuthCallback(Resource):
             return redirect(f"{dify_config.CONSOLE_WEB_URL}/signin/invite-settings?invite_token={invite_token}")
 
         try:
-            account, oauth_new_user = _generate_account(provider, user_info)
+            account, oauth_new_user = _generate_account(provider, user_info, timezone=timezone, language=language)
         except AccountNotFoundError:
             return redirect(f"{dify_config.CONSOLE_WEB_URL}/signin?message=Account not found.")
         except (WorkSpaceNotFoundError, WorkSpaceNotAllowedCreateError):
@@ -184,7 +217,12 @@ def _get_account_by_openid_or_email(provider: str, user_info: OAuthUserInfo) ->
     return account
 
 
-def _generate_account(provider: str, user_info: OAuthUserInfo) -> tuple[Account, bool]:
+def _generate_account(
+    provider: str,
+    user_info: OAuthUserInfo,
+    timezone: str | None = None,
+    language: str | None = None,
+) -> tuple[Account, bool]:
     # Get account by openid or email.
     account = _get_account_by_openid_or_email(provider, user_info)
     oauth_new_user = False
@@ -211,26 +249,19 @@ def _generate_account(provider: str, user_info: OAuthUserInfo) -> tuple[Account,
                         "30 days and is temporarily unavailable for new account registration"
                     )
                 )
-            else:
-                raise AccountRegisterError(description=("Invalid email or password"))
+            raise AccountRegisterError(description=("Invalid email or password"))
         account_name = user_info.name or "Dify"
+        interface_language = _preferred_interface_language(language)
         account = RegisterService.register(
             email=normalized_email,
             name=account_name,
             password=None,
             open_id=user_info.id,
             provider=provider,
+            language=interface_language,
+            timezone=timezone,
         )
 
-        # Set interface language
-        preferred_lang = request.accept_languages.best_match(languages)
-        if preferred_lang and preferred_lang in languages:
-            interface_language = preferred_lang
-        else:
-            interface_language = languages[0]
-        account.interface_language = interface_language
-        db.session.commit()
-
     # Link account
     AccountService.link_account_integrate(provider, user_info.id, account)
 

api/controllers/console/datasets/datasets_document.py

@@ -39,6 +39,7 @@ from fields.document_fields import (
 from graphon.model_runtime.entities.model_entities import ModelType
 from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
 from libs.datetime_utils import naive_utc_now
+from libs.helper import to_timestamp
 from libs.login import current_account_with_tenant, login_required
 from models import DatasetProcessRule, Document, DocumentSegment, UploadFile
 from models.dataset import DocumentPipelineExecutionLog
@@ -71,12 +72,6 @@ from ..wraps import (
 logger = logging.getLogger(__name__)
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 def _normalize_enum(value: Any) -> Any:
     if isinstance(value, str) or value is None:
         return value
@@ -101,7 +96,7 @@ class DatasetResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class DocumentMetadataResponse(ResponseModel):
@@ -152,7 +147,7 @@ class DocumentResponse(ResponseModel):
     @field_validator("created_at", "disabled_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class DocumentWithSegmentsResponse(DocumentResponse):

api/controllers/console/datasets/hit_testing.py

@@ -8,6 +8,7 @@ from pydantic import Field, field_validator
 
 from controllers.common.schema import register_schema_models
 from fields.base import ResponseModel
+from libs.helper import to_timestamp
 from libs.login import login_required
 
 from .. import console_ns
@@ -19,12 +20,6 @@ from ..wraps import (
 )
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 class HitTestingDocument(ResponseModel):
     id: str | None = None
     data_source_type: str | None = None
@@ -61,7 +56,7 @@ class HitTestingSegment(ResponseModel):
     @field_validator("disabled_at", "created_at", "indexing_at", "completed_at", "stopped_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class HitTestingChildChunk(ResponseModel):

api/controllers/console/datasets/hit_testing_base.py

@@ -39,11 +39,8 @@ class HitTestingPayload(BaseModel):
 
 class DatasetsHitTestingBase:
     @staticmethod
-    def _normalize_hit_testing_query(query: Any) -> str:
-        """Return the user-visible query string from legacy and current response shapes."""
-        if isinstance(query, str):
-            return query
-
+    def _extract_hit_testing_query(query: Any) -> str:
+        """Return the query string from the service response shape."""
         if isinstance(query, dict):
             content = query.get("content")
             if isinstance(content, str):
@@ -52,15 +49,15 @@ class DatasetsHitTestingBase:
         raise ValueError("Invalid hit testing query response")
 
     @staticmethod
-    def _normalize_hit_testing_records(records: Any) -> list[dict[str, Any]]:
-        """Coerce nullable collection fields into lists before response validation."""
+    def _prepare_hit_testing_records(records: Any) -> list[dict[str, Any]]:
+        """Ensure collection fields match the API schema before response validation."""
         if not isinstance(records, list):
-            return []
+            raise ValueError("Invalid hit testing records response")
 
         normalized_records: list[dict[str, Any]] = []
         for record in records:
             if not isinstance(record, dict):
-                continue
+                raise ValueError("Invalid hit testing record response")
 
             normalized_record = dict(record)
             segment = normalized_record.get("segment")
@@ -118,8 +115,8 @@ class DatasetsHitTestingBase:
                 limit=10,
             )
             return {
-                "query": DatasetsHitTestingBase._normalize_hit_testing_query(response.get("query")),
-                "records": DatasetsHitTestingBase._normalize_hit_testing_records(
+                "query": DatasetsHitTestingBase._extract_hit_testing_query(response.get("query")),
+                "records": DatasetsHitTestingBase._prepare_hit_testing_records(
                     marshal(response.get("records", []), hit_testing_record_fields)
                 ),
             }

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

@@ -3,7 +3,7 @@ import logging
 from typing import Any, Literal, cast
 
 from flask import abort, request
-from flask_restx import Resource, marshal_with  # type: ignore
+from flask_restx import Resource
 from pydantic import BaseModel, Field, ValidationError
 from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import BadRequest, Forbidden, InternalServerError, NotFound
@@ -19,8 +19,8 @@ from controllers.console.app.error import (
 )
 from controllers.console.app.workflow import (
     RESTORE_SOURCE_WORKFLOW_MUST_BE_PUBLISHED_MESSAGE,
-    workflow_model,
-    workflow_pagination_model,
+    WorkflowPaginationResponse,
+    WorkflowResponse,
 )
 from controllers.console.datasets.wraps import get_rag_pipeline
 from controllers.console.wraps import (
@@ -42,7 +42,7 @@ from fields.workflow_run_fields import (
 )
 from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs import helper
-from libs.helper import TimestampField, UUIDStrOrEmpty
+from libs.helper import TimestampField, UUIDStrOrEmpty, dump_response
 from libs.login import current_account_with_tenant, current_user, login_required
 from models import Account
 from models.dataset import Pipeline
@@ -142,12 +142,17 @@ register_response_schema_models(
 
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft")
 class DraftRagPipelineApi(Resource):
+    @console_ns.response(
+        200,
+        "Draft workflow retrieved successfully",
+        console_ns.models[WorkflowResponse.__name__],
+    )
+    @console_ns.response(404, "Draft workflow not found")
     @setup_required
     @login_required
     @account_initialization_required
     @get_rag_pipeline
     @edit_permission_required
-    @marshal_with(workflow_model)
     def get(self, pipeline: Pipeline):
         """
         Get draft rag pipeline's workflow
@@ -159,8 +164,8 @@ class DraftRagPipelineApi(Resource):
         if not workflow:
             raise DraftWorkflowNotExist()
 
-        # return workflow, if not found, return None (initiate graph by frontend)
-        return workflow
+        # return workflow, if not found, return 404
+        return dump_response(WorkflowResponse, workflow)
 
     @setup_required
     @login_required
@@ -476,12 +481,16 @@ class RagPipelineTaskStopApi(Resource):
 
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/publish")
 class PublishedRagPipelineApi(Resource):
+    @console_ns.response(
+        200,
+        "Published workflow retrieved successfully, or null if not exist",
+        console_ns.models[WorkflowResponse.__name__],
+    )
     @setup_required
     @login_required
     @account_initialization_required
     @edit_permission_required
     @get_rag_pipeline
-    @marshal_with(workflow_model)
     def get(self, pipeline: Pipeline):
         """
         Get published pipeline
@@ -494,7 +503,10 @@ class PublishedRagPipelineApi(Resource):
         workflow = rag_pipeline_service.get_published_workflow(pipeline=pipeline)
 
         # return workflow, if not found, return None
-        return workflow
+        if workflow is None:
+            return None
+
+        return dump_response(WorkflowResponse, workflow)
 
     @setup_required
     @login_required
@@ -567,12 +579,17 @@ class DefaultRagPipelineBlockConfigApi(Resource):
 
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows")
 class PublishedAllRagPipelineApi(Resource):
+    @console_ns.response(
+        200,
+        "Published workflows retrieved successfully",
+        console_ns.models[WorkflowPaginationResponse.__name__],
+    )
+    @console_ns.response(403, "Permission denied")
     @setup_required
     @login_required
     @account_initialization_required
     @edit_permission_required
     @get_rag_pipeline
-    @marshal_with(workflow_pagination_model)
     def get(self, pipeline: Pipeline):
         """
         Get published workflows
@@ -601,12 +618,14 @@ class PublishedAllRagPipelineApi(Resource):
                 named_only=named_only,
             )
 
-            return {
-                "items": workflows,
-                "page": page,
-                "limit": limit,
-                "has_more": has_more,
-            }
+            return WorkflowPaginationResponse.model_validate(
+                {
+                    "items": workflows,
+                    "page": page,
+                    "limit": limit,
+                    "has_more": has_more,
+                }
+            ).model_dump(mode="json")
 
 
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/<string:workflow_id>/restore")
@@ -641,12 +660,15 @@ class RagPipelineDraftWorkflowRestoreApi(Resource):
 
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/<string:workflow_id>")
 class RagPipelineByIdApi(Resource):
+    @console_ns.response(200, "Workflow updated successfully", console_ns.models[WorkflowResponse.__name__])
+    @console_ns.response(400, "No valid fields to update")
+    @console_ns.response(403, "Permission denied")
+    @console_ns.response(404, "Workflow not found")
     @setup_required
     @login_required
     @account_initialization_required
     @edit_permission_required
     @get_rag_pipeline
-    @marshal_with(workflow_model)
     def patch(self, pipeline: Pipeline, workflow_id: str):
         """
         Update workflow attributes
@@ -675,7 +697,7 @@ class RagPipelineByIdApi(Resource):
             if not workflow:
                 raise NotFound("Workflow not found")
 
-            return workflow
+            return dump_response(WorkflowResponse, workflow)
 
     @setup_required
     @login_required

api/controllers/console/explore/installed_app.py

@@ -16,6 +16,7 @@ from extensions.ext_database import db
 from fields.base import ResponseModel
 from graphon.file import helpers as file_helpers
 from libs.datetime_utils import naive_utc_now
+from libs.helper import to_timestamp
 from libs.login import current_account_with_tenant, login_required
 from models import App, InstalledApp, RecommendedApp
 from models.model import IconType
@@ -105,9 +106,7 @@ class InstalledAppResponse(ResponseModel):
     @field_validator("last_used_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return int(value.timestamp())
-        return value
+        return to_timestamp(value)
 
 
 class InstalledAppListResponse(ResponseModel):

api/controllers/console/extension.py

@@ -7,6 +7,7 @@ from pydantic import BaseModel, Field, TypeAdapter, field_validator
 
 from constants import HIDDEN_VALUE
 from fields.base import ResponseModel
+from libs.helper import to_timestamp
 from libs.login import current_account_with_tenant, login_required
 from models.api_based_extension import APIBasedExtension
 from services.api_based_extension_service import APIBasedExtensionService
@@ -40,12 +41,6 @@ def _mask_api_key(api_key: str) -> str:
     return api_key[:3] + "******" + api_key[-3:]
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 class APIBasedExtensionResponse(ResponseModel):
     id: str
     name: str
@@ -61,7 +56,7 @@ class APIBasedExtensionResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 register_schema_models(console_ns, APIBasedExtensionPayload, CodeBasedExtensionResponse, APIBasedExtensionResponse)

api/controllers/console/files.py

@@ -105,7 +105,8 @@ class FilePreviewApi(Resource):
     @account_initialization_required
     def get(self, file_id):
         file_id = str(file_id)
-        text = FileService(db.engine).get_file_preview(file_id)
+        _, tenant_id = current_account_with_tenant()
+        text = FileService(db.engine).get_file_preview(file_id, tenant_id)
         return {"content": text}
 
 

api/controllers/console/tag/tags.py

@@ -25,6 +25,10 @@ class TagBasePayload(BaseModel):
     type: TagType = Field(description="Tag type")
 
 
+class TagUpdateRequestPayload(BaseModel):
+    name: str = Field(description="Tag name", min_length=1, max_length=50)
+
+
 class TagBindingPayload(BaseModel):
     tag_ids: list[str] = Field(description="Tag IDs to bind")
     target_id: str = Field(description="Target ID to bind tags to")
@@ -68,6 +72,7 @@ class TagResponse(ResponseModel):
 register_schema_models(
     console_ns,
     TagBasePayload,
+    TagUpdateRequestPayload,
     TagBindingPayload,
     TagBindingRemovePayload,
     TagListQueryParam,
@@ -118,7 +123,7 @@ class TagListApi(Resource):
 
 @console_ns.route("/tags/<uuid:tag_id>")
 class TagUpdateDeleteApi(Resource):
-    @console_ns.expect(console_ns.models[TagBasePayload.__name__])
+    @console_ns.expect(console_ns.models[TagUpdateRequestPayload.__name__])
     @setup_required
     @login_required
     @account_initialization_required
@@ -129,8 +134,8 @@ class TagUpdateDeleteApi(Resource):
         if not (current_user.has_edit_permission or current_user.is_dataset_editor):
             raise Forbidden()
 
-        payload = TagBasePayload.model_validate(console_ns.payload or {})
-        tag = TagService.update_tags(UpdateTagPayload(name=payload.name, type=payload.type), tag_id)
+        payload = TagUpdateRequestPayload.model_validate(console_ns.payload or {})
+        tag = TagService.update_tags(UpdateTagPayload(name=payload.name), tag_id)
 
         binding_count = TagService.get_tag_binding_count(tag_id)
 

api/controllers/console/workspace/account.py

@@ -42,7 +42,7 @@ from fields.base import ResponseModel
 from fields.member_fields import Account as AccountResponse
 from graphon.file import helpers as file_helpers
 from libs.datetime_utils import naive_utc_now
-from libs.helper import EmailStr, extract_remote_ip, timezone
+from libs.helper import EmailStr, extract_remote_ip, timezone, to_timestamp
 from libs.login import current_account_with_tenant, login_required
 from models import AccountIntegrate, InvitationCode
 from models.account import AccountStatus, InvitationCodeStatus
@@ -185,12 +185,6 @@ def _serialize_account(account) -> dict[str, Any]:
     return AccountResponse.model_validate(account, from_attributes=True).model_dump(mode="json")
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 class AccountIntegrateResponse(ResponseModel):
     provider: str
     created_at: int | None = None
@@ -200,7 +194,7 @@ class AccountIntegrateResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class AccountIntegrateListResponse(ResponseModel):
@@ -220,7 +214,7 @@ class EducationStatusResponse(ResponseModel):
     @field_validator("expire_at", mode="before")
     @classmethod
     def _normalize_expire_at(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class EducationAutocompleteResponse(ResponseModel):

api/controllers/console/workspace/members.py

@@ -70,6 +70,12 @@ register_schema_models(
 )
 
 
+def _is_role_enabled(role: TenantAccountRole | str, tenant_id: str) -> bool:
+    if role != TenantAccountRole.DATASET_OPERATOR:
+        return True
+    return FeatureService.get_features(tenant_id=tenant_id).dataset_operator_enabled
+
+
 @console_ns.route("/workspaces/current/members")
 class MemberListApi(Resource):
     """List all members of current tenant."""
@@ -110,6 +116,8 @@ class MemberInviteEmailApi(Resource):
         inviter = current_user
         if not inviter.current_tenant:
             raise ValueError("No current tenant")
+        if not _is_role_enabled(invitee_role, inviter.current_tenant.id):
+            return {"code": "invalid-role", "message": "Invalid role"}, 400
 
         # Check workspace permission for member invitations
         from libs.workspace_permission import check_workspace_member_invite_permission
@@ -208,6 +216,8 @@ class MemberUpdateRoleApi(Resource):
         current_user, _ = current_account_with_tenant()
         if not current_user.current_tenant:
             raise ValueError("No current tenant")
+        if not _is_role_enabled(new_role, current_user.current_tenant.id):
+            return {"code": "invalid-role", "message": "Invalid role"}, 400
         member = db.session.get(Account, str(member_id))
         if not member:
             abort(404)
@@ -215,11 +225,17 @@ class MemberUpdateRoleApi(Resource):
         try:
             assert member is not None, "Member not found"
             TenantService.update_member_role(current_user.current_tenant, member, new_role, current_user)
+        except services.errors.account.CannotOperateSelfError as e:
+            return {"code": "cannot-operate-self", "message": str(e)}, 400
+        except services.errors.account.NoPermissionError as e:
+            return {"code": "forbidden", "message": str(e)}, 403
+        except services.errors.account.MemberNotInTenantError as e:
+            return {"code": "member-not-found", "message": str(e)}, 404
+        except services.errors.account.RoleAlreadyAssignedError as e:
+            return {"code": "role-already-assigned", "message": str(e)}, 400
         except Exception as e:
             raise ValueError(str(e))
 
-        # todo: 403
-
         return {"result": "success"}
 
 

api/controllers/console/workspace/workspace.py

@@ -29,7 +29,7 @@ from controllers.console.wraps import (
 from enums.cloud_plan import CloudPlan
 from extensions.ext_database import db
 from fields.base import ResponseModel
-from libs.helper import TimestampField
+from libs.helper import TimestampField, to_timestamp
 from libs.login import current_account_with_tenant, login_required
 from models.account import Tenant, TenantCustomConfigDict, TenantStatus
 from services.account_service import TenantService
@@ -86,9 +86,7 @@ class TenantInfoResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None):
-        if isinstance(value, datetime):
-            return int(value.timestamp())
-        return value
+        return to_timestamp(value)
 
 
 register_schema_models(

api/controllers/service_api/app/conversation.py

@@ -22,7 +22,7 @@ from fields.conversation_fields import (
     SimpleConversation,
 )
 from graphon.variables.types import SegmentType
-from libs.helper import UUIDStrOrEmpty
+from libs.helper import UUIDStrOrEmpty, to_timestamp
 from models.model import App, AppMode, EndUser
 from services.conversation_service import ConversationService
 
@@ -115,9 +115,7 @@ class ConversationVariableResponse(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return int(value.timestamp())
-        return value
+        return to_timestamp(value)
 
 
 class ConversationVariableInfiniteScrollPaginationResponse(ResponseModel):

api/controllers/service_api/app/human_input_form.py

@@ -7,18 +7,18 @@ paused human input forms in workflow/chatflow runs.
 
 import json
 import logging
-from datetime import datetime
 
 from flask import Response
 from flask_restx import Resource
 from werkzeug.exceptions import BadRequest, NotFound
 
-from controllers.common.human_input import HumanInputFormSubmitPayload
+from controllers.common.human_input import HumanInputFormSubmitPayload, stringify_form_default_values
 from controllers.common.schema import register_schema_models
 from controllers.service_api import service_api_ns
 from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate_app_token
 from core.workflow.human_input_policy import HumanInputSurface, is_recipient_type_allowed_for_surface
 from extensions.ext_database import db
+from libs.helper import to_timestamp
 from models.model import App, EndUser
 from services.human_input_service import Form, FormNotFoundError, HumanInputService
 
@@ -28,30 +28,14 @@ logger = logging.getLogger(__name__)
 register_schema_models(service_api_ns, HumanInputFormSubmitPayload)
 
 
-def _stringify_default_values(values: dict[str, object]) -> dict[str, str]:
-    result: dict[str, str] = {}
-    for key, value in values.items():
-        if value is None:
-            result[key] = ""
-        elif isinstance(value, (dict, list)):
-            result[key] = json.dumps(value, ensure_ascii=False)
-        else:
-            result[key] = str(value)
-    return result
-
-
-def _to_timestamp(value: datetime) -> int:
-    return int(value.timestamp())
-
-
 def _jsonify_form_definition(form: Form) -> Response:
     definition_payload = form.get_definition().model_dump()
     payload = {
         "form_content": definition_payload["rendered_content"],
         "inputs": definition_payload["inputs"],
-        "resolved_default_values": _stringify_default_values(definition_payload["default_values"]),
+        "resolved_default_values": stringify_form_default_values(definition_payload["default_values"]),
         "user_actions": definition_payload["user_actions"],
-        "expiration_time": _to_timestamp(form.expiration_time),
+        "expiration_time": to_timestamp(form.expiration_time),
     }
     return Response(json.dumps(payload, ensure_ascii=False), mimetype="application/json")
 

api/controllers/service_api/app/workflow.py

@@ -39,6 +39,7 @@ from graphon.enums import WorkflowExecutionStatus
 from graphon.graph_engine.manager import GraphEngineManager
 from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
+from libs.helper import to_timestamp
 from models.model import App, AppMode, EndUser
 from models.workflow import WorkflowRun
 from repositories.factory import DifyAPIRepositoryFactory
@@ -68,12 +69,6 @@ class WorkflowLogQuery(BaseModel):
 register_schema_models(service_api_ns, WorkflowRunPayload, WorkflowLogQuery)
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 def _enum_value(value):
     return getattr(value, "value", value)
 
@@ -109,7 +104,7 @@ class WorkflowRunResponse(ResponseModel):
     @field_validator("created_at", "finished_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class WorkflowRunForLogResponse(ResponseModel):
@@ -133,7 +128,7 @@ class WorkflowRunForLogResponse(ResponseModel):
     @field_validator("created_at", "finished_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class WorkflowAppLogPartialResponse(ResponseModel):
@@ -154,7 +149,7 @@ class WorkflowAppLogPartialResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class WorkflowAppLogPaginationResponse(ResponseModel):

api/controllers/service_api/dataset/dataset.py

@@ -31,7 +31,9 @@ from services.tag_service import (
     TagBindingCreatePayload,
     TagBindingDeletePayload,
     TagService,
-    UpdateTagPayload,
+)
+from services.tag_service import (
+    UpdateTagPayload as UpdateTagServicePayload,
 )
 
 register_enum_models(service_api_ns, DatasetPermissionEnum)
@@ -556,7 +558,7 @@ class DatasetTagsApi(DatasetApiResource):
 
         payload = TagUpdatePayload.model_validate(service_api_ns.payload or {})
         tag_id = payload.tag_id
-        tag = TagService.update_tags(UpdateTagPayload(name=payload.name, type=TagType.KNOWLEDGE), tag_id)
+        tag = TagService.update_tags(UpdateTagServicePayload(name=payload.name), tag_id)
 
         binding_count = TagService.get_tag_binding_count(tag_id)
 

api/controllers/web/human_input_form.py

@@ -4,7 +4,6 @@ Web App Human Input Form APIs.
 
 import json
 import logging
-from datetime import datetime
 from typing import Any, NotRequired, TypedDict
 
 from flask import Response, request
@@ -13,12 +12,12 @@ from sqlalchemy import select
 from werkzeug.exceptions import Forbidden
 
 from configs import dify_config
-from controllers.common.human_input import HumanInputFormSubmitPayload
+from controllers.common.human_input import HumanInputFormSubmitPayload, stringify_form_default_values
 from controllers.web import web_ns
 from controllers.web.error import NotFoundError, WebFormRateLimitExceededError
 from controllers.web.site import serialize_app_site_payload
 from extensions.ext_database import db
-from libs.helper import RateLimiter, extract_remote_ip
+from libs.helper import RateLimiter, extract_remote_ip, to_timestamp
 from models.account import TenantStatus
 from models.model import App, Site
 from services.human_input_service import Form, FormNotFoundError, HumanInputService
@@ -38,22 +37,6 @@ _FORM_ACCESS_RATE_LIMITER = RateLimiter(
 )
 
 
-def _stringify_default_values(values: dict[str, object]) -> dict[str, str]:
-    result: dict[str, str] = {}
-    for key, value in values.items():
-        if value is None:
-            result[key] = ""
-        elif isinstance(value, (dict, list)):
-            result[key] = json.dumps(value, ensure_ascii=False)
-        else:
-            result[key] = str(value)
-    return result
-
-
-def _to_timestamp(value: datetime) -> int:
-    return int(value.timestamp())
-
-
 class FormDefinitionPayload(TypedDict):
     form_content: Any
     inputs: Any
@@ -69,9 +52,9 @@ def _jsonify_form_definition(form: Form, site_payload: dict | None = None) -> Re
     payload: FormDefinitionPayload = {
         "form_content": definition_payload["rendered_content"],
         "inputs": definition_payload["inputs"],
-        "resolved_default_values": _stringify_default_values(definition_payload["default_values"]),
+        "resolved_default_values": stringify_form_default_values(definition_payload["default_values"]),
         "user_actions": definition_payload["user_actions"],
-        "expiration_time": _to_timestamp(form.expiration_time),
+        "expiration_time": to_timestamp(form.expiration_time),
     }
     if site_payload is not None:
         payload["site"] = site_payload

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

@@ -253,7 +253,20 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
     ):
         """
         Resume a paused advanced chat execution.
+
+        ``trace_manager`` is transient and excluded from generate-entity serialization,
+        so resumed executions rebuild it here before persistence layers receive the entity.
         """
+        if application_generate_entity.trace_manager is None:
+            application_generate_entity = application_generate_entity.model_copy(
+                update={
+                    "trace_manager": TraceQueueManager(
+                        app_id=app_model.id,
+                        user_id=user.id if isinstance(user, Account) else user.session_id,
+                    )
+                }
+            )
+
         return self._generate(
             workflow=workflow,
             user=user,

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

@@ -9,7 +9,7 @@ from datetime import datetime
 from threading import Thread
 from typing import Any, Union
 
-from sqlalchemy import select
+from sqlalchemy import select, update
 from sqlalchemy.orm import Session, sessionmaker
 
 from constants.tts_auto_play_timeout import TTS_AUTO_PLAY_TIMEOUT, TTS_AUTO_PLAY_YIELD_CPU_TIME
@@ -245,49 +245,50 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         """
         human_input_responses: list[HumanInputRequiredResponse] = []
         for stream_response in generator:
-            if isinstance(stream_response, ErrorStreamResponse):
-                raise stream_response.err
-            elif isinstance(stream_response, HumanInputRequiredResponse):
-                human_input_responses.append(stream_response)
-            elif isinstance(stream_response, WorkflowPauseStreamResponse):
-                return AdvancedChatPausedBlockingResponse(
-                    task_id=stream_response.task_id,
-                    data=AdvancedChatPausedBlockingResponse.Data(
-                        id=self._message_id,
-                        mode=self._conversation_mode,
-                        conversation_id=self._conversation_id,
-                        message_id=self._message_id,
-                        workflow_run_id=stream_response.data.workflow_run_id,
-                        answer=self._task_state.answer,
-                        metadata=self._message_end_to_stream_response().metadata,
-                        created_at=self._message_created_at,
-                        paused_nodes=stream_response.data.paused_nodes,
-                        reasons=stream_response.data.reasons,
-                        status=stream_response.data.status,
-                        elapsed_time=stream_response.data.elapsed_time,
-                        total_tokens=stream_response.data.total_tokens,
-                        total_steps=stream_response.data.total_steps,
-                    ),
-                )
-            elif isinstance(stream_response, MessageEndStreamResponse):
-                extras = {}
-                if stream_response.metadata:
-                    extras["metadata"] = stream_response.metadata
+            match stream_response:
+                case ErrorStreamResponse():
+                    raise stream_response.err
+                case HumanInputRequiredResponse():
+                    human_input_responses.append(stream_response)
+                case WorkflowPauseStreamResponse():
+                    return AdvancedChatPausedBlockingResponse(
+                        task_id=stream_response.task_id,
+                        data=AdvancedChatPausedBlockingResponse.Data(
+                            id=self._message_id,
+                            mode=self._conversation_mode,
+                            conversation_id=self._conversation_id,
+                            message_id=self._message_id,
+                            workflow_run_id=stream_response.data.workflow_run_id,
+                            answer=self._task_state.answer,
+                            metadata=self._message_end_to_stream_response().metadata,
+                            created_at=self._message_created_at,
+                            paused_nodes=stream_response.data.paused_nodes,
+                            reasons=stream_response.data.reasons,
+                            status=stream_response.data.status,
+                            elapsed_time=stream_response.data.elapsed_time,
+                            total_tokens=stream_response.data.total_tokens,
+                            total_steps=stream_response.data.total_steps,
+                        ),
+                    )
+                case MessageEndStreamResponse():
+                    extras = {}
+                    if stream_response.metadata:
+                        extras["metadata"] = stream_response.metadata
 
-                return ChatbotAppBlockingResponse(
-                    task_id=stream_response.task_id,
-                    data=ChatbotAppBlockingResponse.Data(
-                        id=self._message_id,
-                        mode=self._conversation_mode,
-                        conversation_id=self._conversation_id,
-                        message_id=self._message_id,
-                        answer=self._task_state.answer,
-                        created_at=self._message_created_at,
-                        **extras,
-                    ),
-                )
-            else:
-                continue
+                    return ChatbotAppBlockingResponse(
+                        task_id=stream_response.task_id,
+                        data=ChatbotAppBlockingResponse.Data(
+                            id=self._message_id,
+                            mode=self._conversation_mode,
+                            conversation_id=self._conversation_id,
+                            message_id=self._message_id,
+                            answer=self._task_state.answer,
+                            created_at=self._message_created_at,
+                            **extras,
+                        ),
+                    )
+                case _:
+                    continue
 
         if human_input_responses:
             return self._build_paused_blocking_response_from_human_input(human_input_responses)
@@ -425,11 +426,7 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         self._workflow_run_id = run_id
 
         with self._database_session() as session:
-            message = self._get_message(session=session)
-            if not message:
-                raise ValueError(f"Message not found: {self._message_id}")
-
-            message.workflow_run_id = run_id
+            session.execute(update(Message).where(Message.id == self._message_id).values(workflow_run_id=run_id))
 
         workflow_start_resp = self._workflow_response_converter.workflow_start_to_stream_response(
             task_id=self._application_generate_entity.task_id,

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

@@ -178,7 +178,7 @@ class AgentChatAppConfigManager(BaseAppConfigManager):
         if not isinstance(agent_mode, dict):
             raise ValueError("agent_mode must be of object type")
 
-        # FIXME(-LAN-): Cast needed due to basedpyright limitation with dict type narrowing
+        # FIXME(-LAN-): Cast needed because static checkers do not narrow this dict value.
         agent_mode = cast(dict[str, Any], agent_mode)
 
         if "enabled" not in agent_mode or not agent_mode["enabled"]:

api/core/app/apps/base_app_generator.py

@@ -195,22 +195,23 @@ class BaseAppGenerator:
             )
 
         if variable_entity.type == VariableEntityType.NUMBER:
-            if isinstance(value, (int, float)):
-                return value
-            elif isinstance(value, str):
-                # handle empty string case
-                if not value.strip():
-                    return None
-                # may raise ValueError if user_input_value is not a valid number
-                try:
-                    if "." in value:
-                        return float(value)
-                    else:
-                        return int(value)
-                except ValueError:
-                    raise ValueError(f"{variable_entity.variable} in input form must be a valid number")
-            else:
-                raise TypeError(f"expected value type int, float or str, got {type(value)}, value: {value}")
+            match value:
+                case int() | float():
+                    return value
+                case str():
+                    # handle empty string case
+                    if not value.strip():
+                        return None
+                    # may raise ValueError if user_input_value is not a valid number
+                    try:
+                        if "." in value:
+                            return float(value)
+                        else:
+                            return int(value)
+                    except ValueError:
+                        raise ValueError(f"{variable_entity.variable} in input form must be a valid number")
+                case _:
+                    raise TypeError(f"expected value type int, float or str, got {type(value)}, value: {value}")
 
         match variable_entity.type:
             case VariableEntityType.SELECT:
@@ -241,17 +242,18 @@ class BaseAppGenerator:
                         f"{variable_entity.variable} in input form must be less than {variable_entity.max_length} files"
                     )
             case VariableEntityType.CHECKBOX:
-                if isinstance(value, str):
-                    normalized_value = value.strip().lower()
-                    if normalized_value in {"true", "1", "yes", "on"}:
-                        value = True
-                    elif normalized_value in {"false", "0", "no", "off"}:
-                        value = False
-                elif isinstance(value, (int, float)):
-                    if value == 1:
-                        value = True
-                    elif value == 0:
-                        value = False
+                match value:
+                    case str():
+                        normalized_value = value.strip().lower()
+                        if normalized_value in {"true", "1", "yes", "on"}:
+                            value = True
+                        elif normalized_value in {"false", "0", "no", "off"}:
+                            value = False
+                    case int() | float():
+                        if value == 1:
+                            value = True
+                        elif value == 0:
+                            value = False
             case VariableEntityType.JSON_OBJECT:
                 if value and not isinstance(value, dict):
                     raise ValueError(f"{variable_entity.variable} in input form must be a dict")

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

@@ -45,20 +45,24 @@ class WorkflowAppGenerateResponseConverter(AppGenerateResponseConverter[Workflow
             chunk = cast(WorkflowAppStreamResponse, chunk)
             sub_stream_response = chunk.stream_response
 
-            if isinstance(sub_stream_response, PingStreamResponse):
-                yield "ping"
-                continue
+            match sub_stream_response:
+                case PingStreamResponse():
+                    yield "ping"
+                    continue
+                case ErrorStreamResponse():
+                    response_chunk = {
+                        "event": sub_stream_response.event.value,
+                        "workflow_run_id": chunk.workflow_run_id,
+                    }
+                    data = cls._error_to_stream_response(sub_stream_response.err)
+                    response_chunk.update(cast(dict, data))
+                case _:
+                    response_chunk = {
+                        "event": sub_stream_response.event.value,
+                        "workflow_run_id": chunk.workflow_run_id,
+                    }
+                    response_chunk.update(sub_stream_response.model_dump())
 
-            response_chunk = {
-                "event": sub_stream_response.event.value,
-                "workflow_run_id": chunk.workflow_run_id,
-            }
-
-            if isinstance(sub_stream_response, ErrorStreamResponse):
-                data = cls._error_to_stream_response(sub_stream_response.err)
-                response_chunk.update(cast(dict, data))
-            else:
-                response_chunk.update(sub_stream_response.model_dump())
             yield response_chunk
 
     @classmethod
@@ -74,20 +78,28 @@ class WorkflowAppGenerateResponseConverter(AppGenerateResponseConverter[Workflow
             chunk = cast(WorkflowAppStreamResponse, chunk)
             sub_stream_response = chunk.stream_response
 
-            if isinstance(sub_stream_response, PingStreamResponse):
-                yield "ping"
-                continue
+            match sub_stream_response:
+                case PingStreamResponse():
+                    yield "ping"
+                    continue
+                case ErrorStreamResponse():
+                    response_chunk = {
+                        "event": sub_stream_response.event.value,
+                        "workflow_run_id": chunk.workflow_run_id,
+                    }
+                    data = cls._error_to_stream_response(sub_stream_response.err)
+                    response_chunk.update(cast(dict, data))
+                case NodeStartStreamResponse() | NodeFinishStreamResponse():
+                    response_chunk = {
+                        "event": sub_stream_response.event.value,
+                        "workflow_run_id": chunk.workflow_run_id,
+                    }
+                    response_chunk.update(cast(dict, sub_stream_response.to_ignore_detail_dict()))
+                case _:
+                    response_chunk = {
+                        "event": sub_stream_response.event.value,
+                        "workflow_run_id": chunk.workflow_run_id,
+                    }
+                    response_chunk.update(sub_stream_response.model_dump())
 
-            response_chunk = {
-                "event": sub_stream_response.event.value,
-                "workflow_run_id": chunk.workflow_run_id,
-            }
-
-            if isinstance(sub_stream_response, ErrorStreamResponse):
-                data = cls._error_to_stream_response(sub_stream_response.err)
-                response_chunk.update(cast(dict, data))
-            elif isinstance(sub_stream_response, NodeStartStreamResponse | NodeFinishStreamResponse):
-                response_chunk.update(cast(dict, sub_stream_response.to_ignore_detail_dict()))
-            else:
-                response_chunk.update(sub_stream_response.model_dump())
             yield response_chunk

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

@@ -253,7 +253,20 @@ class WorkflowAppGenerator(BaseAppGenerator):
     ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any], None, None]:
         """
         Resume a paused workflow execution using the persisted runtime state.
+
+        ``trace_manager`` is transient and excluded from generate-entity serialization,
+        so resumed executions rebuild it here before persistence layers receive the entity.
         """
+        if application_generate_entity.trace_manager is None:
+            application_generate_entity = application_generate_entity.model_copy(
+                update={
+                    "trace_manager": TraceQueueManager(
+                        app_id=app_model.id,
+                        user_id=user.id if isinstance(user, Account) else user.session_id,
+                    )
+                }
+            )
+
         return self._generate(
             app_model=app_model,
             workflow=workflow,

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

@@ -52,20 +52,24 @@ class WorkflowAppGenerateResponseConverter(
             chunk = cast(WorkflowAppStreamResponse, chunk)
             sub_stream_response = chunk.stream_response
 
-            if isinstance(sub_stream_response, PingStreamResponse):
-                yield "ping"
-                continue
+            match sub_stream_response:
+                case PingStreamResponse():
+                    yield "ping"
+                    continue
+                case ErrorStreamResponse():
+                    response_chunk: dict[str, object] = {
+                        "event": sub_stream_response.event.value,
+                        "workflow_run_id": chunk.workflow_run_id,
+                    }
+                    data = cls._error_to_stream_response(sub_stream_response.err)
+                    response_chunk.update(data)
+                case _:
+                    response_chunk = {
+                        "event": sub_stream_response.event.value,
+                        "workflow_run_id": chunk.workflow_run_id,
+                    }
+                    response_chunk.update(sub_stream_response.model_dump(mode="json"))
 
-            response_chunk: dict[str, object] = {
-                "event": sub_stream_response.event.value,
-                "workflow_run_id": chunk.workflow_run_id,
-            }
-
-            if isinstance(sub_stream_response, ErrorStreamResponse):
-                data = cls._error_to_stream_response(sub_stream_response.err)
-                response_chunk.update(data)
-            else:
-                response_chunk.update(sub_stream_response.model_dump(mode="json"))
             yield response_chunk
 
     @classmethod
@@ -81,20 +85,28 @@ class WorkflowAppGenerateResponseConverter(
             chunk = cast(WorkflowAppStreamResponse, chunk)
             sub_stream_response = chunk.stream_response
 
-            if isinstance(sub_stream_response, PingStreamResponse):
-                yield "ping"
-                continue
+            match sub_stream_response:
+                case PingStreamResponse():
+                    yield "ping"
+                    continue
+                case ErrorStreamResponse():
+                    response_chunk: dict[str, object] = {
+                        "event": sub_stream_response.event.value,
+                        "workflow_run_id": chunk.workflow_run_id,
+                    }
+                    data = cls._error_to_stream_response(sub_stream_response.err)
+                    response_chunk.update(data)
+                case NodeStartStreamResponse() | NodeFinishStreamResponse():
+                    response_chunk = {
+                        "event": sub_stream_response.event.value,
+                        "workflow_run_id": chunk.workflow_run_id,
+                    }
+                    response_chunk.update(sub_stream_response.to_ignore_detail_dict())
+                case _:
+                    response_chunk = {
+                        "event": sub_stream_response.event.value,
+                        "workflow_run_id": chunk.workflow_run_id,
+                    }
+                    response_chunk.update(sub_stream_response.model_dump(mode="json"))
 
-            response_chunk: dict[str, object] = {
-                "event": sub_stream_response.event.value,
-                "workflow_run_id": chunk.workflow_run_id,
-            }
-
-            if isinstance(sub_stream_response, ErrorStreamResponse):
-                data = cls._error_to_stream_response(sub_stream_response.err)
-                response_chunk.update(data)
-            elif isinstance(sub_stream_response, NodeStartStreamResponse | NodeFinishStreamResponse):
-                response_chunk.update(sub_stream_response.to_ignore_detail_dict())
-            else:
-                response_chunk.update(sub_stream_response.model_dump(mode="json"))
             yield response_chunk

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

@@ -145,50 +145,51 @@ class WorkflowAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         """
         human_input_responses: list[HumanInputRequiredResponse] = []
         for stream_response in generator:
-            if isinstance(stream_response, ErrorStreamResponse):
-                raise stream_response.err
-            elif isinstance(stream_response, HumanInputRequiredResponse):
-                human_input_responses.append(stream_response)
-            elif isinstance(stream_response, WorkflowPauseStreamResponse):
-                return WorkflowAppPausedBlockingResponse(
-                    task_id=self._application_generate_entity.task_id,
-                    workflow_run_id=stream_response.data.workflow_run_id,
-                    data=WorkflowAppPausedBlockingResponse.Data(
-                        id=stream_response.data.workflow_run_id,
-                        workflow_id=self._workflow.id,
-                        status=stream_response.data.status,
-                        outputs=stream_response.data.outputs or {},
-                        error=None,
-                        elapsed_time=stream_response.data.elapsed_time,
-                        total_tokens=stream_response.data.total_tokens,
-                        total_steps=stream_response.data.total_steps,
-                        created_at=stream_response.data.created_at,
-                        finished_at=None,
-                        paused_nodes=stream_response.data.paused_nodes,
-                        reasons=stream_response.data.reasons,
-                    ),
-                )
-
-            elif isinstance(stream_response, WorkflowFinishStreamResponse):
-                return WorkflowAppBlockingResponse(
-                    task_id=self._application_generate_entity.task_id,
-                    workflow_run_id=stream_response.data.id,
-                    data=WorkflowAppBlockingResponse.Data(
-                        id=stream_response.data.id,
-                        workflow_id=stream_response.data.workflow_id,
-                        status=stream_response.data.status,
-                        outputs=stream_response.data.outputs,
-                        error=stream_response.data.error,
-                        elapsed_time=stream_response.data.elapsed_time,
-                        total_tokens=stream_response.data.total_tokens,
-                        total_steps=stream_response.data.total_steps,
-                        created_at=int(stream_response.data.created_at),
-                        finished_at=int(stream_response.data.finished_at) if stream_response.data.finished_at else None,
-                    ),
-                )
-
-            else:
-                continue
+            match stream_response:
+                case ErrorStreamResponse():
+                    raise stream_response.err
+                case HumanInputRequiredResponse():
+                    human_input_responses.append(stream_response)
+                case WorkflowPauseStreamResponse():
+                    return WorkflowAppPausedBlockingResponse(
+                        task_id=self._application_generate_entity.task_id,
+                        workflow_run_id=stream_response.data.workflow_run_id,
+                        data=WorkflowAppPausedBlockingResponse.Data(
+                            id=stream_response.data.workflow_run_id,
+                            workflow_id=self._workflow.id,
+                            status=stream_response.data.status,
+                            outputs=stream_response.data.outputs or {},
+                            error=None,
+                            elapsed_time=stream_response.data.elapsed_time,
+                            total_tokens=stream_response.data.total_tokens,
+                            total_steps=stream_response.data.total_steps,
+                            created_at=stream_response.data.created_at,
+                            finished_at=None,
+                            paused_nodes=stream_response.data.paused_nodes,
+                            reasons=stream_response.data.reasons,
+                        ),
+                    )
+                case WorkflowFinishStreamResponse():
+                    return WorkflowAppBlockingResponse(
+                        task_id=self._application_generate_entity.task_id,
+                        workflow_run_id=stream_response.data.id,
+                        data=WorkflowAppBlockingResponse.Data(
+                            id=stream_response.data.id,
+                            workflow_id=stream_response.data.workflow_id,
+                            status=stream_response.data.status,
+                            outputs=stream_response.data.outputs,
+                            error=stream_response.data.error,
+                            elapsed_time=stream_response.data.elapsed_time,
+                            total_tokens=stream_response.data.total_tokens,
+                            total_steps=stream_response.data.total_steps,
+                            created_at=int(stream_response.data.created_at),
+                            finished_at=int(stream_response.data.finished_at)
+                            if stream_response.data.finished_at
+                            else None,
+                        ),
+                    )
+                case _:
+                    continue
 
         if human_input_responses:
             return self._build_paused_blocking_response_from_human_input(human_input_responses)

api/core/app/apps/workflow_app_runner.py

@@ -399,278 +399,281 @@ class WorkflowBasedAppRunner:
         :param workflow_entry: workflow entry
         :param event: event
         """
-        if isinstance(event, GraphRunStartedEvent):
-            self._publish_event(QueueWorkflowStartedEvent(reason=event.reason))
-        elif isinstance(event, GraphRunSucceededEvent):
-            self._publish_event(QueueWorkflowSucceededEvent(outputs=event.outputs))
-        elif isinstance(event, GraphRunPartialSucceededEvent):
-            self._publish_event(
-                QueueWorkflowPartialSuccessEvent(outputs=event.outputs, exceptions_count=event.exceptions_count)
-            )
-        elif isinstance(event, GraphRunFailedEvent):
-            self._publish_event(QueueWorkflowFailedEvent(error=event.error, exceptions_count=event.exceptions_count))
-        elif isinstance(event, GraphRunAbortedEvent):
-            self._publish_event(QueueWorkflowFailedEvent(error=event.reason or "Unknown error", exceptions_count=0))
-        elif isinstance(event, GraphRunPausedEvent):
-            runtime_state = workflow_entry.graph_engine.graph_runtime_state
-            paused_nodes = runtime_state.get_paused_nodes()
-            self._enqueue_human_input_notifications(event.reasons)
-            self._publish_event(
-                QueueWorkflowPausedEvent(
-                    reasons=event.reasons,
-                    outputs=event.outputs,
-                    paused_nodes=paused_nodes,
+        match event:
+            case GraphRunStartedEvent():
+                self._publish_event(QueueWorkflowStartedEvent(reason=event.reason))
+            case GraphRunSucceededEvent():
+                self._publish_event(QueueWorkflowSucceededEvent(outputs=event.outputs))
+            case GraphRunPartialSucceededEvent():
+                self._publish_event(
+                    QueueWorkflowPartialSuccessEvent(outputs=event.outputs, exceptions_count=event.exceptions_count)
                 )
-            )
-        elif isinstance(event, NodeRunHumanInputFormFilledEvent):
-            self._publish_event(
-                QueueHumanInputFormFilledEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
-                    node_type=event.node_type,
-                    node_title=event.node_title,
-                    rendered_content=event.rendered_content,
-                    action_id=event.action_id,
-                    action_text=event.action_text,
+            case GraphRunFailedEvent():
+                self._publish_event(
+                    QueueWorkflowFailedEvent(error=event.error, exceptions_count=event.exceptions_count)
                 )
-            )
-        elif isinstance(event, NodeRunHumanInputFormTimeoutEvent):
-            self._publish_event(
-                QueueHumanInputFormTimeoutEvent(
-                    node_id=event.node_id,
-                    node_type=event.node_type,
-                    node_title=event.node_title,
-                    expiration_time=event.expiration_time,
+            case GraphRunAbortedEvent():
+                self._publish_event(QueueWorkflowFailedEvent(error=event.reason or "Unknown error", exceptions_count=0))
+            case GraphRunPausedEvent():
+                runtime_state = workflow_entry.graph_engine.graph_runtime_state
+                paused_nodes = runtime_state.get_paused_nodes()
+                self._enqueue_human_input_notifications(event.reasons)
+                self._publish_event(
+                    QueueWorkflowPausedEvent(
+                        reasons=event.reasons,
+                        outputs=event.outputs,
+                        paused_nodes=paused_nodes,
+                    )
                 )
-            )
-        elif isinstance(event, NodeRunRetryEvent):
-            node_run_result = event.node_run_result
-            inputs = node_run_result.inputs
-            process_data = node_run_result.process_data
-            outputs = project_node_outputs_for_workflow_run(
-                node_type=event.node_type,
-                inputs=inputs,
-                outputs=node_run_result.outputs,
-            )
-            execution_metadata = node_run_result.metadata
-            self._publish_event(
-                QueueNodeRetryEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
-                    node_title=event.node_title,
+            case NodeRunHumanInputFormFilledEvent():
+                self._publish_event(
+                    QueueHumanInputFormFilledEvent(
+                        node_execution_id=event.id,
+                        node_id=event.node_id,
+                        node_type=event.node_type,
+                        node_title=event.node_title,
+                        rendered_content=event.rendered_content,
+                        action_id=event.action_id,
+                        action_text=event.action_text,
+                    )
+                )
+            case NodeRunHumanInputFormTimeoutEvent():
+                self._publish_event(
+                    QueueHumanInputFormTimeoutEvent(
+                        node_id=event.node_id,
+                        node_type=event.node_type,
+                        node_title=event.node_title,
+                        expiration_time=event.expiration_time,
+                    )
+                )
+            case NodeRunRetryEvent():
+                node_run_result = event.node_run_result
+                inputs = node_run_result.inputs
+                process_data = node_run_result.process_data
+                outputs = project_node_outputs_for_workflow_run(
                     node_type=event.node_type,
-                    start_at=event.start_at,
-                    in_iteration_id=event.in_iteration_id,
-                    in_loop_id=event.in_loop_id,
                     inputs=inputs,
-                    process_data=process_data,
-                    outputs=outputs,
-                    error=event.error,
-                    execution_metadata=execution_metadata,
-                    retry_index=event.retry_index,
-                    provider_type=event.provider_type,
-                    provider_id=event.provider_id,
+                    outputs=node_run_result.outputs,
                 )
-            )
-        elif isinstance(event, NodeRunStartedEvent):
-            self._publish_event(
-                QueueNodeStartedEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
-                    node_title=event.node_title,
-                    node_type=event.node_type,
-                    start_at=event.start_at,
-                    in_iteration_id=event.in_iteration_id,
-                    in_loop_id=event.in_loop_id,
-                    agent_strategy=self._build_agent_strategy_info(event),
-                    provider_type=event.provider_type,
-                    provider_id=event.provider_id,
+                execution_metadata = node_run_result.metadata
+                self._publish_event(
+                    QueueNodeRetryEvent(
+                        node_execution_id=event.id,
+                        node_id=event.node_id,
+                        node_title=event.node_title,
+                        node_type=event.node_type,
+                        start_at=event.start_at,
+                        in_iteration_id=event.in_iteration_id,
+                        in_loop_id=event.in_loop_id,
+                        inputs=inputs,
+                        process_data=process_data,
+                        outputs=outputs,
+                        error=event.error,
+                        execution_metadata=execution_metadata,
+                        retry_index=event.retry_index,
+                        provider_type=event.provider_type,
+                        provider_id=event.provider_id,
+                    )
                 )
-            )
-        elif isinstance(event, NodeRunSucceededEvent):
-            node_run_result = event.node_run_result
-            inputs = node_run_result.inputs
-            process_data = node_run_result.process_data
-            outputs = project_node_outputs_for_workflow_run(
-                node_type=event.node_type,
-                inputs=inputs,
-                outputs=node_run_result.outputs,
-            )
-            execution_metadata = node_run_result.metadata
-            self._publish_event(
-                QueueNodeSucceededEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
+            case NodeRunStartedEvent():
+                self._publish_event(
+                    QueueNodeStartedEvent(
+                        node_execution_id=event.id,
+                        node_id=event.node_id,
+                        node_title=event.node_title,
+                        node_type=event.node_type,
+                        start_at=event.start_at,
+                        in_iteration_id=event.in_iteration_id,
+                        in_loop_id=event.in_loop_id,
+                        agent_strategy=self._build_agent_strategy_info(event),
+                        provider_type=event.provider_type,
+                        provider_id=event.provider_id,
+                    )
+                )
+            case NodeRunSucceededEvent():
+                node_run_result = event.node_run_result
+                inputs = node_run_result.inputs
+                process_data = node_run_result.process_data
+                outputs = project_node_outputs_for_workflow_run(
                     node_type=event.node_type,
-                    start_at=event.start_at,
-                    finished_at=event.finished_at,
                     inputs=inputs,
-                    process_data=process_data,
-                    outputs=outputs,
-                    execution_metadata=execution_metadata,
-                    in_iteration_id=event.in_iteration_id,
-                    in_loop_id=event.in_loop_id,
+                    outputs=node_run_result.outputs,
                 )
-            )
-        elif isinstance(event, NodeRunFailedEvent):
-            outputs = project_node_outputs_for_workflow_run(
-                node_type=event.node_type,
-                inputs=event.node_run_result.inputs,
-                outputs=event.node_run_result.outputs,
-            )
-            self._publish_event(
-                QueueNodeFailedEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
+                execution_metadata = node_run_result.metadata
+                self._publish_event(
+                    QueueNodeSucceededEvent(
+                        node_execution_id=event.id,
+                        node_id=event.node_id,
+                        node_type=event.node_type,
+                        start_at=event.start_at,
+                        finished_at=event.finished_at,
+                        inputs=inputs,
+                        process_data=process_data,
+                        outputs=outputs,
+                        execution_metadata=execution_metadata,
+                        in_iteration_id=event.in_iteration_id,
+                        in_loop_id=event.in_loop_id,
+                    )
+                )
+            case NodeRunFailedEvent():
+                outputs = project_node_outputs_for_workflow_run(
                     node_type=event.node_type,
-                    start_at=event.start_at,
-                    finished_at=event.finished_at,
                     inputs=event.node_run_result.inputs,
-                    process_data=event.node_run_result.process_data,
-                    outputs=outputs,
-                    error=event.node_run_result.error or "Unknown error",
-                    execution_metadata=event.node_run_result.metadata,
-                    in_iteration_id=event.in_iteration_id,
-                    in_loop_id=event.in_loop_id,
+                    outputs=event.node_run_result.outputs,
                 )
-            )
-        elif isinstance(event, NodeRunExceptionEvent):
-            outputs = project_node_outputs_for_workflow_run(
-                node_type=event.node_type,
-                inputs=event.node_run_result.inputs,
-                outputs=event.node_run_result.outputs,
-            )
-            self._publish_event(
-                QueueNodeExceptionEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
+                self._publish_event(
+                    QueueNodeFailedEvent(
+                        node_execution_id=event.id,
+                        node_id=event.node_id,
+                        node_type=event.node_type,
+                        start_at=event.start_at,
+                        finished_at=event.finished_at,
+                        inputs=event.node_run_result.inputs,
+                        process_data=event.node_run_result.process_data,
+                        outputs=outputs,
+                        error=event.node_run_result.error or "Unknown error",
+                        execution_metadata=event.node_run_result.metadata,
+                        in_iteration_id=event.in_iteration_id,
+                        in_loop_id=event.in_loop_id,
+                    )
+                )
+            case NodeRunExceptionEvent():
+                outputs = project_node_outputs_for_workflow_run(
                     node_type=event.node_type,
-                    start_at=event.start_at,
-                    finished_at=event.finished_at,
                     inputs=event.node_run_result.inputs,
-                    process_data=event.node_run_result.process_data,
-                    outputs=outputs,
-                    error=event.node_run_result.error or "Unknown error",
-                    execution_metadata=event.node_run_result.metadata,
-                    in_iteration_id=event.in_iteration_id,
-                    in_loop_id=event.in_loop_id,
+                    outputs=event.node_run_result.outputs,
                 )
-            )
-        elif isinstance(event, NodeRunStreamChunkEvent):
-            self._publish_event(
-                QueueTextChunkEvent(
-                    text=event.chunk,
-                    from_variable_selector=list(event.selector),
-                    in_iteration_id=event.in_iteration_id,
-                    in_loop_id=event.in_loop_id,
+                self._publish_event(
+                    QueueNodeExceptionEvent(
+                        node_execution_id=event.id,
+                        node_id=event.node_id,
+                        node_type=event.node_type,
+                        start_at=event.start_at,
+                        finished_at=event.finished_at,
+                        inputs=event.node_run_result.inputs,
+                        process_data=event.node_run_result.process_data,
+                        outputs=outputs,
+                        error=event.node_run_result.error or "Unknown error",
+                        execution_metadata=event.node_run_result.metadata,
+                        in_iteration_id=event.in_iteration_id,
+                        in_loop_id=event.in_loop_id,
+                    )
                 )
-            )
-        elif isinstance(event, NodeRunRetrieverResourceEvent):
-            self._publish_event(
-                QueueRetrieverResourcesEvent(
-                    retriever_resources=[
-                        RetrievalSourceMetadata.model_validate(resource) for resource in event.retriever_resources
-                    ],
-                    in_iteration_id=event.in_iteration_id,
-                    in_loop_id=event.in_loop_id,
+            case NodeRunStreamChunkEvent():
+                self._publish_event(
+                    QueueTextChunkEvent(
+                        text=event.chunk,
+                        from_variable_selector=list(event.selector),
+                        in_iteration_id=event.in_iteration_id,
+                        in_loop_id=event.in_loop_id,
+                    )
                 )
-            )
-        elif isinstance(event, NodeRunAgentLogEvent):
-            self._publish_event(
-                QueueAgentLogEvent(
-                    id=event.message_id,
-                    label=event.label,
-                    node_execution_id=event.node_execution_id,
-                    parent_id=event.parent_id,
-                    error=event.error,
-                    status=event.status,
-                    data=event.data,
-                    metadata=event.metadata,
-                    node_id=event.node_id,
+            case NodeRunRetrieverResourceEvent():
+                self._publish_event(
+                    QueueRetrieverResourcesEvent(
+                        retriever_resources=[
+                            RetrievalSourceMetadata.model_validate(resource) for resource in event.retriever_resources
+                        ],
+                        in_iteration_id=event.in_iteration_id,
+                        in_loop_id=event.in_loop_id,
+                    )
                 )
-            )
-        elif isinstance(event, NodeRunIterationStartedEvent):
-            self._publish_event(
-                QueueIterationStartEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
-                    node_type=event.node_type,
-                    node_title=event.node_title,
-                    start_at=event.start_at,
-                    node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
-                    inputs=event.inputs,
-                    metadata=event.metadata,
+            case NodeRunAgentLogEvent():
+                self._publish_event(
+                    QueueAgentLogEvent(
+                        id=event.message_id,
+                        label=event.label,
+                        node_execution_id=event.node_execution_id,
+                        parent_id=event.parent_id,
+                        error=event.error,
+                        status=event.status,
+                        data=event.data,
+                        metadata=event.metadata,
+                        node_id=event.node_id,
+                    )
                 )
-            )
-        elif isinstance(event, NodeRunIterationNextEvent):
-            self._publish_event(
-                QueueIterationNextEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
-                    node_type=event.node_type,
-                    node_title=event.node_title,
-                    index=event.index,
-                    node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
-                    output=event.pre_iteration_output,
+            case NodeRunIterationStartedEvent():
+                self._publish_event(
+                    QueueIterationStartEvent(
+                        node_execution_id=event.id,
+                        node_id=event.node_id,
+                        node_type=event.node_type,
+                        node_title=event.node_title,
+                        start_at=event.start_at,
+                        node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
+                        inputs=event.inputs,
+                        metadata=event.metadata,
+                    )
                 )
-            )
-        elif isinstance(event, (NodeRunIterationSucceededEvent | NodeRunIterationFailedEvent)):
-            self._publish_event(
-                QueueIterationCompletedEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
-                    node_type=event.node_type,
-                    node_title=event.node_title,
-                    start_at=event.start_at,
-                    node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
-                    inputs=event.inputs,
-                    outputs=event.outputs,
-                    metadata=event.metadata,
-                    steps=event.steps,
-                    error=event.error if isinstance(event, NodeRunIterationFailedEvent) else None,
+            case NodeRunIterationNextEvent():
+                self._publish_event(
+                    QueueIterationNextEvent(
+                        node_execution_id=event.id,
+                        node_id=event.node_id,
+                        node_type=event.node_type,
+                        node_title=event.node_title,
+                        index=event.index,
+                        node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
+                        output=event.pre_iteration_output,
+                    )
                 )
-            )
-        elif isinstance(event, NodeRunLoopStartedEvent):
-            self._publish_event(
-                QueueLoopStartEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
-                    node_type=event.node_type,
-                    node_title=event.node_title,
-                    start_at=event.start_at,
-                    node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
-                    inputs=event.inputs,
-                    metadata=event.metadata,
+            case NodeRunIterationSucceededEvent() | NodeRunIterationFailedEvent():
+                self._publish_event(
+                    QueueIterationCompletedEvent(
+                        node_execution_id=event.id,
+                        node_id=event.node_id,
+                        node_type=event.node_type,
+                        node_title=event.node_title,
+                        start_at=event.start_at,
+                        node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
+                        inputs=event.inputs,
+                        outputs=event.outputs,
+                        metadata=event.metadata,
+                        steps=event.steps,
+                        error=event.error if isinstance(event, NodeRunIterationFailedEvent) else None,
+                    )
                 )
-            )
-        elif isinstance(event, NodeRunLoopNextEvent):
-            self._publish_event(
-                QueueLoopNextEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
-                    node_type=event.node_type,
-                    node_title=event.node_title,
-                    index=event.index,
-                    node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
-                    output=event.pre_loop_output,
+            case NodeRunLoopStartedEvent():
+                self._publish_event(
+                    QueueLoopStartEvent(
+                        node_execution_id=event.id,
+                        node_id=event.node_id,
+                        node_type=event.node_type,
+                        node_title=event.node_title,
+                        start_at=event.start_at,
+                        node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
+                        inputs=event.inputs,
+                        metadata=event.metadata,
+                    )
                 )
-            )
-        elif isinstance(event, (NodeRunLoopSucceededEvent | NodeRunLoopFailedEvent)):
-            self._publish_event(
-                QueueLoopCompletedEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
-                    node_type=event.node_type,
-                    node_title=event.node_title,
-                    start_at=event.start_at,
-                    node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
-                    inputs=event.inputs,
-                    outputs=event.outputs,
-                    metadata=event.metadata,
-                    steps=event.steps,
-                    error=event.error if isinstance(event, NodeRunLoopFailedEvent) else None,
+            case NodeRunLoopNextEvent():
+                self._publish_event(
+                    QueueLoopNextEvent(
+                        node_execution_id=event.id,
+                        node_id=event.node_id,
+                        node_type=event.node_type,
+                        node_title=event.node_title,
+                        index=event.index,
+                        node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
+                        output=event.pre_loop_output,
+                    )
+                )
+            case NodeRunLoopSucceededEvent() | NodeRunLoopFailedEvent():
+                self._publish_event(
+                    QueueLoopCompletedEvent(
+                        node_execution_id=event.id,
+                        node_id=event.node_id,
+                        node_type=event.node_type,
+                        node_title=event.node_title,
+                        start_at=event.start_at,
+                        node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
+                        inputs=event.inputs,
+                        outputs=event.outputs,
+                        metadata=event.metadata,
+                        steps=event.steps,
+                        error=event.error if isinstance(event, NodeRunLoopFailedEvent) else None,
+                    )
                 )
-            )
 
     def _enqueue_human_input_notifications(self, reasons: Sequence[object]) -> None:
         for reason in reasons:

api/core/app/entities/task_entities.py

@@ -10,7 +10,7 @@ from graphon.entities import WorkflowStartReason
 from graphon.entities.pause_reason import PauseReasonType
 from graphon.enums import WorkflowExecutionStatus, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
 from graphon.model_runtime.entities.llm_entities import LLMResult, LLMUsage
-from graphon.nodes.human_input.entities import FormInput, UserAction
+from graphon.nodes.human_input.entities import FormInputConfig, UserActionConfig
 
 
 class AnnotationReplyAccount(BaseModel):
@@ -284,8 +284,8 @@ class HumanInputRequiredResponse(StreamResponse):
         node_id: str
         node_title: str
         form_content: str
-        inputs: Sequence[FormInput] = Field(default_factory=list)
-        actions: Sequence[UserAction] = Field(default_factory=list)
+        inputs: Sequence[FormInputConfig] = Field(default_factory=list)
+        actions: Sequence[UserActionConfig] = Field(default_factory=list)
         display_in_ui: bool = False
         form_token: str | None = None
         resolved_default_values: Mapping[str, Any] = Field(default_factory=dict)
@@ -307,8 +307,8 @@ class HumanInputRequiredPauseReasonPayload(BaseModel):
     node_id: str
     node_title: str
     form_content: str
-    inputs: Sequence[FormInput] = Field(default_factory=list)
-    actions: Sequence[UserAction] = Field(default_factory=list)
+    inputs: Sequence[FormInputConfig] = Field(default_factory=list)
+    actions: Sequence[UserActionConfig] = Field(default_factory=list)
     display_in_ui: bool = False
     form_token: str | None = None
     resolved_default_values: Mapping[str, Any] = Field(default_factory=dict)

api/core/app/file_access/__init__.py

@@ -1,6 +1,13 @@
 from .controller import DatabaseFileAccessController
 from .protocols import FileAccessControllerProtocol
-from .scope import FileAccessScope, bind_file_access_scope, get_current_file_access_scope
+from .scope import (
+    FileAccessScope,
+    bind_file_access_scope,
+    get_current_file_access_scope,
+    grant_retriever_segment_access,
+    grant_upload_file_access,
+    is_retriever_segment_access_granted,
+)
 
 __all__ = [
     "DatabaseFileAccessController",
@@ -8,4 +15,7 @@ __all__ = [
     "FileAccessScope",
     "bind_file_access_scope",
     "get_current_file_access_scope",
+    "grant_retriever_segment_access",
+    "grant_upload_file_access",
+    "is_retriever_segment_access_granted",
 ]

api/core/app/file_access/controller.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from collections.abc import Callable
 
-from sqlalchemy import select
+from sqlalchemy import and_, or_, select
 from sqlalchemy.orm import Session
 from sqlalchemy.sql import Select
 
@@ -18,7 +18,8 @@ class DatabaseFileAccessController(FileAccessControllerProtocol):
 
     Tenant scoping remains mandatory. When the current execution belongs to an
     end user, the lookup is additionally constrained to that end user's file
-    ownership markers.
+    ownership markers, plus upload files explicitly granted by the current
+    execution context.
     """
 
     _scope_getter: Callable[[], FileAccessScope | None]
@@ -47,10 +48,19 @@ class DatabaseFileAccessController(FileAccessControllerProtocol):
         if not resolved_scope.requires_user_ownership:
             return scoped_stmt
 
-        return scoped_stmt.where(
+        user_owned_filter = and_(
             UploadFile.created_by_role == CreatorUserRole.END_USER,
             UploadFile.created_by == resolved_scope.user_id,
         )
+        if not resolved_scope.granted_upload_file_ids:
+            return scoped_stmt.where(user_owned_filter)
+
+        return scoped_stmt.where(
+            or_(
+                user_owned_filter,
+                UploadFile.id.in_(resolved_scope.granted_upload_file_ids),
+            )
+        )
 
     def apply_tool_file_filters(
         self,

api/core/app/file_access/scope.py

@@ -1,9 +1,9 @@
 from __future__ import annotations
 
-from collections.abc import Generator  # Changed from Iterator
+from collections.abc import Generator, Iterable
 from contextlib import contextmanager
 from contextvars import ContextVar
-from dataclasses import dataclass
+from dataclasses import dataclass, field, replace
 
 from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom
 
@@ -15,12 +15,23 @@ _current_file_access_scope: ContextVar[FileAccessScope | None] = ContextVar(
 
 @dataclass(frozen=True, slots=True)
 class FileAccessScope:
-    """Request-scoped ownership context used by workflow-layer file lookups."""
+    """Request-scoped ownership context used by workflow-layer file lookups.
+
+    ``granted_upload_file_ids`` is execution-local: callers may add upload files
+    that were returned by trusted retrieval paths without changing persistent
+    ownership markers.
+
+    ``granted_retriever_segment_ids`` gates lazy attachment loading by segment
+    ID, so user-provided context cannot make a later LLM node load arbitrary
+    same-tenant knowledge attachments.
+    """
 
     tenant_id: str
     user_id: str
     user_from: UserFrom
     invoke_from: InvokeFrom
+    granted_upload_file_ids: frozenset[str] = field(default_factory=frozenset)
+    granted_retriever_segment_ids: frozenset[str] = field(default_factory=frozenset)
 
     @property
     def requires_user_ownership(self) -> bool:
@@ -31,8 +42,49 @@ def get_current_file_access_scope() -> FileAccessScope | None:
     return _current_file_access_scope.get()
 
 
+def grant_upload_file_access(upload_file_ids: Iterable[str]) -> None:
+    scope = _current_file_access_scope.get()
+    if scope is None:
+        return
+
+    granted_upload_file_ids = frozenset(str(file_id) for file_id in upload_file_ids if file_id)
+    if not granted_upload_file_ids:
+        return
+
+    _current_file_access_scope.set(
+        replace(
+            scope,
+            granted_upload_file_ids=scope.granted_upload_file_ids | granted_upload_file_ids,
+        )
+    )
+
+
+def grant_retriever_segment_access(segment_ids: Iterable[str]) -> None:
+    scope = _current_file_access_scope.get()
+    if scope is None:
+        return
+
+    granted_segment_ids = frozenset(str(segment_id) for segment_id in segment_ids if segment_id)
+    if not granted_segment_ids:
+        return
+
+    _current_file_access_scope.set(
+        replace(
+            scope,
+            granted_retriever_segment_ids=scope.granted_retriever_segment_ids | granted_segment_ids,
+        )
+    )
+
+
+def is_retriever_segment_access_granted(segment_id: str) -> bool:
+    scope = _current_file_access_scope.get()
+    if scope is None or not scope.requires_user_ownership:
+        return True
+    return str(segment_id) in scope.granted_retriever_segment_ids
+
+
 @contextmanager
-def bind_file_access_scope(scope: FileAccessScope) -> Generator[None, None, None]:  # Changed from Iterator[None]
+def bind_file_access_scope(scope: FileAccessScope) -> Generator[None, None, None]:
     token = _current_file_access_scope.set(scope)
     try:
         yield

api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py

@@ -140,42 +140,43 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
         :return:
         """
         for stream_response in generator:
-            if isinstance(stream_response, ErrorStreamResponse):
-                raise stream_response.err
-            elif isinstance(stream_response, MessageEndStreamResponse):
-                extras = {"usage": self._task_state.llm_result.usage.model_dump()}
-                if self._task_state.metadata:
-                    extras["metadata"] = self._task_state.metadata.model_dump()
-                response: ChatbotAppBlockingResponse | CompletionAppBlockingResponse
-                if self._conversation_mode == AppMode.COMPLETION:
-                    response = CompletionAppBlockingResponse(
-                        task_id=self._application_generate_entity.task_id,
-                        data=CompletionAppBlockingResponse.Data(
-                            id=self._message_id,
-                            mode=self._conversation_mode,
-                            message_id=self._message_id,
-                            answer=self._task_state.llm_result.message.get_text_content(),
-                            created_at=self._message_created_at,
-                            **extras,
-                        ),
-                    )
-                else:
-                    response = ChatbotAppBlockingResponse(
-                        task_id=self._application_generate_entity.task_id,
-                        data=ChatbotAppBlockingResponse.Data(
-                            id=self._message_id,
-                            mode=self._conversation_mode,
-                            conversation_id=self._conversation_id,
-                            message_id=self._message_id,
-                            answer=self._task_state.llm_result.message.get_text_content(),
-                            created_at=self._message_created_at,
-                            **extras,
-                        ),
-                    )
+            match stream_response:
+                case ErrorStreamResponse():
+                    raise stream_response.err
+                case MessageEndStreamResponse():
+                    extras = {"usage": self._task_state.llm_result.usage.model_dump()}
+                    if self._task_state.metadata:
+                        extras["metadata"] = self._task_state.metadata.model_dump()
+                    response: ChatbotAppBlockingResponse | CompletionAppBlockingResponse
+                    if self._conversation_mode == AppMode.COMPLETION:
+                        response = CompletionAppBlockingResponse(
+                            task_id=self._application_generate_entity.task_id,
+                            data=CompletionAppBlockingResponse.Data(
+                                id=self._message_id,
+                                mode=self._conversation_mode,
+                                message_id=self._message_id,
+                                answer=self._task_state.llm_result.message.get_text_content(),
+                                created_at=self._message_created_at,
+                                **extras,
+                            ),
+                        )
+                    else:
+                        response = ChatbotAppBlockingResponse(
+                            task_id=self._application_generate_entity.task_id,
+                            data=ChatbotAppBlockingResponse.Data(
+                                id=self._message_id,
+                                mode=self._conversation_mode,
+                                conversation_id=self._conversation_id,
+                                message_id=self._message_id,
+                                answer=self._task_state.llm_result.message.get_text_content(),
+                                created_at=self._message_created_at,
+                                **extras,
+                            ),
+                        )
 
-                return response
-            else:
-                continue
+                    return response
+                case _:
+                    continue
 
         raise RuntimeError("queue listening stopped unexpectedly.")
 
@@ -265,104 +266,107 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
                 publisher.publish(message)
             event = message.event
 
-            if isinstance(event, QueueErrorEvent):
-                with sessionmaker(bind=db.engine).begin() as session:
-                    err = self.handle_error(event=event, session=session, message_id=self._message_id)
-                yield self.error_to_stream_response(err)
-                break
-            elif isinstance(event, QueueStopEvent | QueueMessageEndEvent):
-                if isinstance(event, QueueMessageEndEvent):
-                    if event.llm_result:
-                        self._task_state.llm_result = event.llm_result
-                else:
-                    self._handle_stop(event)
+            match event:
+                case QueueErrorEvent():
+                    with sessionmaker(bind=db.engine).begin() as session:
+                        err = self.handle_error(event=event, session=session, message_id=self._message_id)
+                    yield self.error_to_stream_response(err)
+                    break
+                case QueueStopEvent() | QueueMessageEndEvent():
+                    if isinstance(event, QueueMessageEndEvent):
+                        if event.llm_result:
+                            self._task_state.llm_result = event.llm_result
+                    else:
+                        self._handle_stop(event)
 
-                # handle output moderation
-                output_moderation_answer = self.handle_output_moderation_when_task_finished(
-                    self._task_state.llm_result.message.get_text_content()
-                )
-                if output_moderation_answer:
-                    self._task_state.llm_result.message.content = output_moderation_answer
-                    yield self._message_cycle_manager.message_replace_to_stream_response(
-                        answer=output_moderation_answer
+                    # handle output moderation
+                    output_moderation_answer = self.handle_output_moderation_when_task_finished(
+                        self._task_state.llm_result.message.get_text_content()
                     )
-
-                with sessionmaker(bind=db.engine).begin() as session:
-                    # Save message
-                    self._save_message(session=session, trace_manager=trace_manager)
-                message_end_resp = self._message_end_to_stream_response()
-                yield message_end_resp
-            elif isinstance(event, QueueRetrieverResourcesEvent):
-                self._message_cycle_manager.handle_retriever_resources(event)
-            elif isinstance(event, QueueAnnotationReplyEvent):
-                annotation = self._message_cycle_manager.handle_annotation_reply(event)
-                if annotation:
-                    self._task_state.llm_result.message.content = annotation.content
-            elif isinstance(event, QueueAgentThoughtEvent):
-                agent_thought_response = self._agent_thought_to_stream_response(event)
-                if agent_thought_response is not None:
-                    yield agent_thought_response
-            elif isinstance(event, QueueMessageFileEvent):
-                response = self._message_cycle_manager.message_file_to_stream_response(event)
-                if response:
-                    yield response
-            elif isinstance(event, QueueLLMChunkEvent | QueueAgentMessageEvent):
-                chunk = event.chunk
-                delta_text = chunk.delta.message.content
-                if delta_text is None:
-                    continue
-                if isinstance(chunk.delta.message.content, list):
-                    delta_text = ""
-                    for content in chunk.delta.message.content:
-                        logger.debug(
-                            "The content type %s in LLM chunk delta message content.: %r", type(content), content
+                    if output_moderation_answer:
+                        self._task_state.llm_result.message.content = output_moderation_answer
+                        yield self._message_cycle_manager.message_replace_to_stream_response(
+                            answer=output_moderation_answer
                         )
-                        if isinstance(content, TextPromptMessageContent):
-                            delta_text += content.data
-                        elif isinstance(content, str):
-                            delta_text += content  # failback to str
-                        else:
-                            logger.warning(
-                                "Unsupported content type %s in LLM chunk delta message content.: %r",
-                                type(content),
-                                content,
+
+                    with sessionmaker(bind=db.engine).begin() as session:
+                        # Save message
+                        self._save_message(session=session, trace_manager=trace_manager)
+                    message_end_resp = self._message_end_to_stream_response()
+                    yield message_end_resp
+                case QueueRetrieverResourcesEvent():
+                    self._message_cycle_manager.handle_retriever_resources(event)
+                case QueueAnnotationReplyEvent():
+                    annotation = self._message_cycle_manager.handle_annotation_reply(event)
+                    if annotation:
+                        self._task_state.llm_result.message.content = annotation.content
+                case QueueAgentThoughtEvent():
+                    agent_thought_response = self._agent_thought_to_stream_response(event)
+                    if agent_thought_response is not None:
+                        yield agent_thought_response
+                case QueueMessageFileEvent():
+                    response = self._message_cycle_manager.message_file_to_stream_response(event)
+                    if response:
+                        yield response
+                case QueueLLMChunkEvent() | QueueAgentMessageEvent():
+                    chunk = event.chunk
+                    delta_text = chunk.delta.message.content
+                    if delta_text is None:
+                        continue
+                    if isinstance(chunk.delta.message.content, list):
+                        delta_text = ""
+                        for content in chunk.delta.message.content:
+                            logger.debug(
+                                "The content type %s in LLM chunk delta message content.: %r", type(content), content
                             )
-                            continue
+                            match content:
+                                case TextPromptMessageContent():
+                                    delta_text += content.data
+                                case str():
+                                    delta_text += content  # failback to str
+                                case _:
+                                    logger.warning(
+                                        "Unsupported content type %s in LLM chunk delta message content.: %r",
+                                        type(content),
+                                        content,
+                                    )
+                                    continue
 
-                if not self._task_state.llm_result.prompt_messages:
-                    self._task_state.llm_result.prompt_messages = chunk.prompt_messages
+                    if not self._task_state.llm_result.prompt_messages:
+                        self._task_state.llm_result.prompt_messages = chunk.prompt_messages
 
-                # handle output moderation chunk
-                should_direct_answer = self._handle_output_moderation_chunk(cast(str, delta_text))
-                if should_direct_answer:
+                    # handle output moderation chunk
+                    should_direct_answer = self._handle_output_moderation_chunk(cast(str, delta_text))
+                    if should_direct_answer:
+                        continue
+
+                    current_content = cast(str, self._task_state.llm_result.message.content)
+                    current_content += cast(str, delta_text)
+                    self._task_state.llm_result.message.content = current_content
+
+                    match event:
+                        case QueueLLMChunkEvent():
+                            # Determine the event type once, on first LLM chunk, and reuse for subsequent chunks
+                            if not hasattr(self, "_precomputed_event_type") or self._precomputed_event_type is None:
+                                self._precomputed_event_type = self._message_cycle_manager.get_message_event_type(
+                                    message_id=self._message_id
+                                )
+                            yield self._message_cycle_manager.message_to_stream_response(
+                                answer=cast(str, delta_text),
+                                message_id=self._message_id,
+                                event_type=self._precomputed_event_type,
+                            )
+                        case _:
+                            yield self._agent_message_to_stream_response(
+                                answer=cast(str, delta_text),
+                                message_id=self._message_id,
+                            )
+                case QueueMessageReplaceEvent():
+                    yield self._message_cycle_manager.message_replace_to_stream_response(answer=event.text)
+                case QueuePingEvent():
+                    yield self.ping_stream_response()
+                case _:
                     continue
-
-                current_content = cast(str, self._task_state.llm_result.message.content)
-                current_content += cast(str, delta_text)
-                self._task_state.llm_result.message.content = current_content
-
-                if isinstance(event, QueueLLMChunkEvent):
-                    # Determine the event type once, on first LLM chunk, and reuse for subsequent chunks
-                    if not hasattr(self, "_precomputed_event_type") or self._precomputed_event_type is None:
-                        self._precomputed_event_type = self._message_cycle_manager.get_message_event_type(
-                            message_id=self._message_id
-                        )
-                    yield self._message_cycle_manager.message_to_stream_response(
-                        answer=cast(str, delta_text),
-                        message_id=self._message_id,
-                        event_type=self._precomputed_event_type,
-                    )
-                else:
-                    yield self._agent_message_to_stream_response(
-                        answer=cast(str, delta_text),
-                        message_id=self._message_id,
-                    )
-            elif isinstance(event, QueueMessageReplaceEvent):
-                yield self._message_cycle_manager.message_replace_to_stream_response(answer=event.text)
-            elif isinstance(event, QueuePingEvent):
-                yield self.ping_stream_response()
-            else:
-                continue
         if publisher:
             publisher.publish(None)
         if self._conversation_name_generate_thread:

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

@@ -105,52 +105,31 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
         self._node_sequence = 0
 
     def on_event(self, event: GraphEngineEvent) -> None:
-        if isinstance(event, GraphRunStartedEvent):
-            self._handle_graph_run_started()
-            return
-
-        if isinstance(event, GraphRunSucceededEvent):
-            self._handle_graph_run_succeeded(event)
-            return
-
-        if isinstance(event, GraphRunPartialSucceededEvent):
-            self._handle_graph_run_partial_succeeded(event)
-            return
-
-        if isinstance(event, GraphRunFailedEvent):
-            self._handle_graph_run_failed(event)
-            return
-
-        if isinstance(event, GraphRunAbortedEvent):
-            self._handle_graph_run_aborted(event)
-            return
-
-        if isinstance(event, GraphRunPausedEvent):
-            self._handle_graph_run_paused(event)
-            return
-
-        if isinstance(event, NodeRunRetryEvent):
-            self._handle_node_retry(event)
-            return
-
-        if isinstance(event, NodeRunStartedEvent):
-            self._handle_node_started(event)
-            return
-
-        if isinstance(event, NodeRunSucceededEvent):
-            self._handle_node_succeeded(event)
-            return
-
-        if isinstance(event, NodeRunFailedEvent):
-            self._handle_node_failed(event)
-            return
-
-        if isinstance(event, NodeRunExceptionEvent):
-            self._handle_node_exception(event)
-            return
-
-        if isinstance(event, NodeRunPauseRequestedEvent):
-            self._handle_node_pause_requested(event)
+        match event:
+            case GraphRunStartedEvent():
+                self._handle_graph_run_started()
+            case GraphRunSucceededEvent():
+                self._handle_graph_run_succeeded(event)
+            case GraphRunPartialSucceededEvent():
+                self._handle_graph_run_partial_succeeded(event)
+            case GraphRunFailedEvent():
+                self._handle_graph_run_failed(event)
+            case GraphRunAbortedEvent():
+                self._handle_graph_run_aborted(event)
+            case GraphRunPausedEvent():
+                self._handle_graph_run_paused(event)
+            case NodeRunRetryEvent():
+                self._handle_node_retry(event)
+            case NodeRunStartedEvent():
+                self._handle_node_started(event)
+            case NodeRunSucceededEvent():
+                self._handle_node_succeeded(event)
+            case NodeRunFailedEvent():
+                self._handle_node_failed(event)
+            case NodeRunExceptionEvent():
+                self._handle_node_exception(event)
+            case NodeRunPauseRequestedEvent():
+                self._handle_node_pause_requested(event)
 
     def on_graph_end(self, error: Exception | None) -> None:
         return

api/core/base/tts/app_generator_tts_publisher.py

@@ -95,13 +95,14 @@ class AppGeneratorTTSPublisher:
                     message_content = message.event.chunk.delta.message.content
                     if not message_content:
                         continue
-                    if isinstance(message_content, str):
-                        self.msg_text += message_content
-                    elif isinstance(message_content, list):
-                        for content in message_content:
-                            if not isinstance(content, TextPromptMessageContent):
-                                continue
-                            self.msg_text += content.data
+                    match message_content:
+                        case str():
+                            self.msg_text += message_content
+                        case list():
+                            for content in message_content:
+                                if not isinstance(content, TextPromptMessageContent):
+                                    continue
+                                self.msg_text += content.data
                 elif isinstance(message.event, QueueTextChunkEvent):
                     self.msg_text += message.event.text
                 elif isinstance(message.event, QueueNodeSucceededEvent):

api/core/entities/execution_extra_content.py

@@ -5,7 +5,7 @@ from typing import Any, TypeAlias
 
 from pydantic import BaseModel, ConfigDict, Field
 
-from graphon.nodes.human_input.entities import FormInput, UserAction
+from graphon.nodes.human_input.entities import FormInputConfig, UserActionConfig
 from models.execution_extra_content import ExecutionContentType
 
 
@@ -16,8 +16,8 @@ class HumanInputFormDefinition(BaseModel):
     node_id: str
     node_title: str
     form_content: str
-    inputs: Sequence[FormInput] = Field(default_factory=list)
-    actions: Sequence[UserAction] = Field(default_factory=list)
+    inputs: Sequence[FormInputConfig] = Field(default_factory=list)
+    actions: Sequence[UserActionConfig] = Field(default_factory=list)
     display_in_ui: bool = False
     form_token: str | None = None
     resolved_default_values: Mapping[str, Any] = Field(default_factory=dict)

api/core/entities/provider_configuration.py

@@ -171,9 +171,9 @@ class ProviderConfiguration(BaseModel):
                 current_credential_id = self.custom_configuration.provider.current_credential_id
 
             if current_credential_id:
-                from core.helper.credential_utils import check_credential_policy_compliance
+                from core.helper.credential_utils import runtime_check_credential_policy_compliance
 
-                check_credential_policy_compliance(
+                runtime_check_credential_policy_compliance(
                     credential_id=current_credential_id,
                     provider=self.provider.provider,
                     credential_type=PluginCredentialType.MODEL,
@@ -182,9 +182,9 @@ class ProviderConfiguration(BaseModel):
                 # no current credential id, check all available credentials
                 if self.custom_configuration.provider:
                     for credential_configuration in self.custom_configuration.provider.available_credentials:
-                        from core.helper.credential_utils import check_credential_policy_compliance
+                        from core.helper.credential_utils import runtime_check_credential_policy_compliance
 
-                        check_credential_policy_compliance(
+                        runtime_check_credential_policy_compliance(
                             credential_id=credential_configuration.credential_id,
                             provider=self.provider.provider,
                             credential_type=PluginCredentialType.MODEL,

api/core/errors/error.py

@@ -1,9 +1,9 @@
 class LLMError(ValueError):
     """Base class for all LLM exceptions."""
 
-    description: str | None = None
+    description: str = ""
 
-    def __init__(self, description: str | None = None):
+    def __init__(self, description: str = ""):
         self.description = description
 
 

api/core/helper/credential_utils.py

@@ -2,6 +2,7 @@
 Credential utility functions for checking credential existence and policy compliance.
 """
 
+from configs import dify_config
 from core.entities import PluginCredentialType
 
 
@@ -39,6 +40,16 @@ def is_credential_exists(credential_id: str, credential_type: "PluginCredentialT
         return False
 
 
+def runtime_check_credential_policy_compliance(
+    credential_id: str, provider: str, credential_type: "PluginCredentialType", check_existence: bool = True
+):
+    if dify_config.ENTERPRISE_DISABLE_RUNTIME_CREDENTIAL_CHECK:
+        return
+    check_credential_policy_compliance(
+        credential_id=credential_id, provider=provider, credential_type=credential_type, check_existence=check_existence
+    )
+
+
 def check_credential_policy_compliance(
     credential_id: str, provider: str, credential_type: "PluginCredentialType", check_existence: bool = True
 ) -> None:

api/core/llm_generator/llm_generator.py

@@ -435,7 +435,7 @@ class LLMGenerator:
             stream=False,
         )
 
-        # Runtime type check since pyright has issues with the overload
+        # Runtime type check for overload narrowing.
         if not isinstance(result, LLMResult):
             raise TypeError("Expected LLMResult when stream=False")
         response = result

api/core/llm_generator/output_parser/structured_output.py

@@ -288,11 +288,13 @@ def _parse_structured_output(result_text: str) -> Mapping[str, Any]:
     except ValidationError:
         # if the result_text is not a valid json, try to repair it
         temp_parsed = json_repair.loads(result_text)
-        if not isinstance(temp_parsed, dict):
-            # handle reasoning model like deepseek-r1 got '<think>\n\n</think>\n' prefix
-            if isinstance(temp_parsed, list):
+        match temp_parsed:
+            case dict():
+                pass
+            case list():
+                # handle reasoning model like deepseek-r1 got '<think>\n\n</think>\n' prefix
                 temp_parsed = next((item for item in temp_parsed if isinstance(item, dict)), {})
-            else:
+            case _:
                 raise OutputParserError(f"Failed to parse structured output: {result_text}")
         structured_output = cast(dict, temp_parsed)
     return structured_output
@@ -341,12 +343,13 @@ def remove_additional_properties(schema: dict[str, Any]) -> None:
 
     # Process nested structures recursively
     for value in schema.values():
-        if isinstance(value, dict):
-            remove_additional_properties(value)
-        elif isinstance(value, list):
-            for item in value:
-                if isinstance(item, dict):
-                    remove_additional_properties(item)
+        match value:
+            case dict():
+                remove_additional_properties(value)
+            case list():
+                for item in value:
+                    if isinstance(item, dict):
+                        remove_additional_properties(item)
 
 
 def convert_boolean_to_string(schema: dict[str, Any]) -> None:
@@ -364,9 +367,10 @@ def convert_boolean_to_string(schema: dict[str, Any]) -> None:
 
     # Process nested dictionaries and lists recursively
     for value in schema.values():
-        if isinstance(value, dict):
-            convert_boolean_to_string(value)
-        elif isinstance(value, list):
-            for item in value:
-                if isinstance(item, dict):
-                    convert_boolean_to_string(item)
+        match value:
+            case dict():
+                convert_boolean_to_string(value)
+            case list():
+                for item in value:
+                    if isinstance(item, dict):
+                        convert_boolean_to_string(item)

api/core/mcp/types.py

@@ -2,7 +2,7 @@ from collections.abc import Callable
 from dataclasses import dataclass
 from typing import Annotated, Any, Literal
 
-from pydantic import BaseModel, ConfigDict, Field, FileUrl, RootModel
+from pydantic import BaseModel, ConfigDict, Field, FileUrl, RootModel, model_validator
 from pydantic.networks import AnyUrl, UrlConstraints
 
 """
@@ -173,7 +173,21 @@ class JSONRPCError(BaseModel):
 
 
 class JSONRPCMessage(RootModel[JSONRPCRequest | JSONRPCNotification | JSONRPCResponse | JSONRPCError]):
-    pass
+    @model_validator(mode="before")
+    @classmethod
+    def _select_message_type(
+        cls, value: Any
+    ) -> JSONRPCRequest | JSONRPCNotification | JSONRPCResponse | JSONRPCError | Any:
+        if isinstance(value, dict):
+            if "result" in value:
+                return JSONRPCResponse.model_validate(value)
+            if "error" in value:
+                return JSONRPCError.model_validate(value)
+            if "method" in value:
+                if "id" in value:
+                    return JSONRPCRequest.model_validate(value)
+                return JSONRPCNotification.model_validate(value)
+        return value
 
 
 class EmptyResult(Result):

api/core/model_manager.py

@@ -391,10 +391,10 @@ class ModelInstance:
 
             # Additional policy compliance check as fallback (in case fetch_next didn't catch it)
             try:
-                from core.helper.credential_utils import check_credential_policy_compliance
+                from core.helper.credential_utils import runtime_check_credential_policy_compliance
 
                 if lb_config.credential_id:
-                    check_credential_policy_compliance(
+                    runtime_check_credential_policy_compliance(
                         credential_id=lb_config.credential_id,
                         provider=self.provider,
                         credential_type=PluginCredentialType.MODEL,
@@ -630,10 +630,10 @@ class LBModelManager:
 
             # Check policy compliance for the selected configuration
             try:
-                from core.helper.credential_utils import check_credential_policy_compliance
+                from core.helper.credential_utils import runtime_check_credential_policy_compliance
 
                 if config.credential_id:
-                    check_credential_policy_compliance(
+                    runtime_check_credential_policy_compliance(
                         credential_id=config.credential_id,
                         provider=self._provider,
                         credential_type=PluginCredentialType.MODEL,

api/core/prompt/advanced_prompt_transform.py

@@ -161,35 +161,39 @@ class AdvancedPromptTransform(PromptTransform):
         prompt_messages: list[PromptMessage] = []
         for prompt_item in prompt_template:
             raw_prompt = prompt_item.text
-
-            if prompt_item.edition_type == "basic" or not prompt_item.edition_type:
-                if self.with_variable_tmpl:
-                    vp = VariablePool.empty()
-                    for k, v in inputs.items():
-                        if k.startswith("#"):
-                            vp.add(k[1:-1].split("."), v)
-                    raw_prompt = raw_prompt.replace("{{#context#}}", context or "")
-                    prompt = vp.convert_template(raw_prompt).text
-                else:
-                    parser = PromptTemplateParser(template=raw_prompt, with_variable_tmpl=self.with_variable_tmpl)
-                    prompt_inputs: Mapping[str, str] = {k: inputs[k] for k in parser.variable_keys if k in inputs}
-                    prompt_inputs = self._set_context_variable(
-                        context=context, parser=parser, prompt_inputs=prompt_inputs
-                    )
-                    prompt = parser.format(prompt_inputs)
-            elif prompt_item.edition_type == "jinja2":
-                prompt = raw_prompt
-                prompt_inputs = inputs
-                prompt = Jinja2Formatter.format(template=prompt, inputs=prompt_inputs)
-            else:
-                raise ValueError(f"Invalid edition type: {prompt_item.edition_type}")
-
-            if prompt_item.role == PromptMessageRole.USER:
-                prompt_messages.append(UserPromptMessage(content=prompt))
-            elif prompt_item.role == PromptMessageRole.SYSTEM and prompt:
-                prompt_messages.append(SystemPromptMessage(content=prompt))
-            elif prompt_item.role == PromptMessageRole.ASSISTANT:
-                prompt_messages.append(AssistantPromptMessage(content=prompt))
+            edition_type = prompt_item.edition_type or "basic"
+            match edition_type:
+                case "basic":
+                    if self.with_variable_tmpl:
+                        vp = VariablePool.empty()
+                        for k, v in inputs.items():
+                            if k.startswith("#"):
+                                vp.add(k[1:-1].split("."), v)
+                        raw_prompt = raw_prompt.replace("{{#context#}}", context or "")
+                        prompt = vp.convert_template(raw_prompt).text
+                    else:
+                        parser = PromptTemplateParser(template=raw_prompt, with_variable_tmpl=self.with_variable_tmpl)
+                        prompt_inputs: Mapping[str, str] = {k: inputs[k] for k in parser.variable_keys if k in inputs}
+                        prompt_inputs = self._set_context_variable(
+                            context=context, parser=parser, prompt_inputs=prompt_inputs
+                        )
+                        prompt = parser.format(prompt_inputs)
+                case "jinja2":
+                    prompt = raw_prompt
+                    prompt_inputs = inputs
+                    prompt = Jinja2Formatter.format(template=prompt, inputs=prompt_inputs)
+                case _:
+                    raise ValueError(f"Invalid edition type: {prompt_item.edition_type}")
+            match prompt_item.role:
+                case PromptMessageRole.USER:
+                    prompt_messages.append(UserPromptMessage(content=prompt))
+                case PromptMessageRole.SYSTEM:
+                    if prompt:
+                        prompt_messages.append(SystemPromptMessage(content=prompt))
+                case PromptMessageRole.ASSISTANT:
+                    prompt_messages.append(AssistantPromptMessage(content=prompt))
+                case PromptMessageRole.TOOL:
+                    pass
 
         if query and memory_config and memory_config.query_prompt_template:
             parser = PromptTemplateParser(

api/core/rag/datasource/retrieval_service.py

@@ -9,6 +9,7 @@ from sqlalchemy import select
 from sqlalchemy.orm import Session, load_only
 
 from configs import dify_config
+from core.app.file_access import grant_upload_file_access
 from core.db.session_factory import session_factory
 from core.model_manager import ModelManager
 from core.rag.data_post_processor.data_post_processor import DataPostProcessor, RerankingModelDict, WeightsDict
@@ -890,6 +891,7 @@ class RetrievalService:
                 .limit(1)
             )
             if attachment_binding:
+                grant_upload_file_access([str(upload_file.id)])
                 attachment_info: AttachmentInfoDict = {
                     "id": upload_file.id,
                     "name": upload_file.name,
@@ -906,6 +908,7 @@ class RetrievalService:
         cls, attachment_ids: list[str], session: Session
     ) -> list[SegmentAttachmentInfoResult]:
         attachment_infos: list[SegmentAttachmentInfoResult] = []
+        granted_upload_file_ids: list[str] = []
         upload_files = session.scalars(select(UploadFile).where(UploadFile.id.in_(attachment_ids))).all()
         if upload_files:
             upload_file_ids = [upload_file.id for upload_file in upload_files]
@@ -926,6 +929,7 @@ class RetrievalService:
                         "size": upload_file.size,
                     }
                     if attachment_binding:
+                        granted_upload_file_ids.append(str(upload_file.id))
                         attachment_infos.append(
                             {
                                 "attachment_id": attachment_binding.attachment_id,
@@ -933,4 +937,5 @@ class RetrievalService:
                                 "segment_id": attachment_binding.segment_id,
                             }
                         )
+        grant_upload_file_access(granted_upload_file_ids)
         return attachment_infos

api/core/rag/extractor/extract_processor.py

@@ -183,34 +183,35 @@ class ExtractProcessor:
             return extractor.extract()
         elif extract_setting.datasource_type == DatasourceType.WEBSITE:
             assert extract_setting.website_info is not None, "website_info is required"
-            if extract_setting.website_info.provider == "firecrawl":
-                extractor = FirecrawlWebExtractor(
-                    url=extract_setting.website_info.url,
-                    job_id=extract_setting.website_info.job_id,
-                    tenant_id=extract_setting.website_info.tenant_id,
-                    mode=extract_setting.website_info.mode,
-                    only_main_content=extract_setting.website_info.only_main_content,
-                )
-                return extractor.extract()
-            elif extract_setting.website_info.provider == "watercrawl":
-                extractor = WaterCrawlWebExtractor(
-                    url=extract_setting.website_info.url,
-                    job_id=extract_setting.website_info.job_id,
-                    tenant_id=extract_setting.website_info.tenant_id,
-                    mode=extract_setting.website_info.mode,
-                    only_main_content=extract_setting.website_info.only_main_content,
-                )
-                return extractor.extract()
-            elif extract_setting.website_info.provider == "jinareader":
-                extractor = JinaReaderWebExtractor(
-                    url=extract_setting.website_info.url,
-                    job_id=extract_setting.website_info.job_id,
-                    tenant_id=extract_setting.website_info.tenant_id,
-                    mode=extract_setting.website_info.mode,
-                    only_main_content=extract_setting.website_info.only_main_content,
-                )
-                return extractor.extract()
-            else:
-                raise ValueError(f"Unsupported website provider: {extract_setting.website_info.provider}")
+            match extract_setting.website_info.provider:
+                case "firecrawl":
+                    extractor = FirecrawlWebExtractor(
+                        url=extract_setting.website_info.url,
+                        job_id=extract_setting.website_info.job_id,
+                        tenant_id=extract_setting.website_info.tenant_id,
+                        mode=extract_setting.website_info.mode,
+                        only_main_content=extract_setting.website_info.only_main_content,
+                    )
+                    return extractor.extract()
+                case "watercrawl":
+                    extractor = WaterCrawlWebExtractor(
+                        url=extract_setting.website_info.url,
+                        job_id=extract_setting.website_info.job_id,
+                        tenant_id=extract_setting.website_info.tenant_id,
+                        mode=extract_setting.website_info.mode,
+                        only_main_content=extract_setting.website_info.only_main_content,
+                    )
+                    return extractor.extract()
+                case "jinareader":
+                    extractor = JinaReaderWebExtractor(
+                        url=extract_setting.website_info.url,
+                        job_id=extract_setting.website_info.job_id,
+                        tenant_id=extract_setting.website_info.tenant_id,
+                        mode=extract_setting.website_info.mode,
+                        only_main_content=extract_setting.website_info.only_main_content,
+                    )
+                    return extractor.extract()
+                case _:
+                    raise ValueError(f"Unsupported website provider: {extract_setting.website_info.provider}")
         else:
             raise ValueError(f"Unsupported datasource type: {extract_setting.datasource_type}")

api/core/rag/extractor/unstructured/unstructured_doc_extractor.py

@@ -19,7 +19,7 @@ class UnstructuredWordExtractor(BaseExtractor):
 
     def extract(self) -> list[Document]:
         from unstructured.__version__ import __version__ as __unstructured_version__
-        from unstructured.file_utils.filetype import (  # pyright: ignore[reportPrivateImportUsage]
+        from unstructured.file_utils.filetype import (
             FileType,
             detect_filetype,
         )
@@ -27,7 +27,7 @@ class UnstructuredWordExtractor(BaseExtractor):
         unstructured_version = tuple(int(x) for x in __unstructured_version__.split("."))
         # check the file extension
         try:
-            import magic  # noqa: F401  # pyright: ignore[reportUnusedImport]
+            import magic  # noqa: F401
 
             is_doc = detect_filetype(self._file_path) == FileType.DOC
         except ImportError:

api/core/rag/retrieval/dataset_retrieval.py

@@ -19,6 +19,7 @@ from core.app.app_config.entities import (
     ModelConfig,
 )
 from core.app.entities.app_invoke_entities import InvokeFrom, ModelConfigWithCredentialsEntity
+from core.app.file_access import grant_retriever_segment_access, grant_upload_file_access
 from core.callback_handler.index_tool_callback_handler import DatasetIndexToolCallbackHandler
 from core.db.session_factory import session_factory
 from core.entities.agent_entities import PlanningStrategy
@@ -326,6 +327,7 @@ class DatasetRetrieval:
                         if record.summary:
                             source.summary = record.summary
 
+                        grant_retriever_segment_access([str(segment.id)])
                         retrieval_resource_list.append(source)
 
         if retrieval_resource_list:
@@ -515,6 +517,9 @@ class DatasetRetrieval:
                             )
                         ).all()
                         if attachments_with_bindings:
+                            grant_upload_file_access(
+                                str(upload_file.id) for _, upload_file in attachments_with_bindings
+                            )
                             for _, upload_file in attachments_with_bindings:
                                 attachment_info = File(
                                     file_id=upload_file.id,

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

@@ -28,10 +28,10 @@ class FunctionCallMultiDatasetRouter:
                 SystemPromptMessage(content="You are a helpful AI assistant."),
                 UserPromptMessage(content=query),
             ]
-            result: LLMResult = model_instance.invoke_llm(  # pyright: ignore[reportCallIssue, reportArgumentType]
+            result: LLMResult = model_instance.invoke_llm(  # pyrefly: ignore[no-matching-overload]
                 prompt_messages=prompt_messages,
                 tools=dataset_tools,
-                stream=False,  # pyright: ignore[reportArgumentType]
+                stream=False,
                 model_parameters={"temperature": 0.2, "top_p": 0.3, "max_tokens": 1500},
             )
             usage = result.usage or LLMUsage.empty_usage()

api/core/tools/mcp_tool/tool.py

@@ -67,23 +67,27 @@ class MCPTool(Tool):
 
         # handle dify tool output
         for content in result.content:
-            if isinstance(content, TextContent):
-                yield from self._process_text_content(content)
-            elif isinstance(content, ImageContent | AudioContent):
-                yield self.create_blob_message(
-                    blob=base64.b64decode(content.data), meta={"mime_type": content.mimeType}
-                )
-            elif isinstance(content, EmbeddedResource):
-                resource = content.resource
-                if isinstance(resource, TextResourceContents):
-                    yield self.create_text_message(resource.text)
-                elif isinstance(resource, BlobResourceContents):
-                    mime_type = resource.mimeType or "application/octet-stream"
-                    yield self.create_blob_message(blob=base64.b64decode(resource.blob), meta={"mime_type": mime_type})
-                else:
-                    raise ToolInvokeError(f"Unsupported embedded resource type: {type(resource)}")
-            else:
-                logger.warning("Unsupported content type=%s", type(content))
+            match content:
+                case TextContent():
+                    yield from self._process_text_content(content)
+                case ImageContent() | AudioContent():
+                    yield self.create_blob_message(
+                        blob=base64.b64decode(content.data), meta={"mime_type": content.mimeType}
+                    )
+                case EmbeddedResource():
+                    resource = content.resource
+                    match resource:
+                        case TextResourceContents():
+                            yield self.create_text_message(resource.text)
+                        case BlobResourceContents():
+                            mime_type = resource.mimeType or "application/octet-stream"
+                            yield self.create_blob_message(
+                                blob=base64.b64decode(resource.blob), meta={"mime_type": mime_type}
+                            )
+                        case _:
+                            raise ToolInvokeError(f"Unsupported embedded resource type: {type(resource)}")
+                case _:
+                    logger.warning("Unsupported content type=%s", type(content))
 
         # handle MCP structured output
         if self.entity.output_schema and result.structuredContent:

api/core/tools/tool_manager.py

@@ -264,9 +264,9 @@ class ToolManager:
                     if builtin_provider is None:
                         raise ToolProviderNotFoundError(f"builtin provider {provider_id} not found")
 
-                from core.helper.credential_utils import check_credential_policy_compliance
+                from core.helper.credential_utils import runtime_check_credential_policy_compliance
 
-                check_credential_policy_compliance(
+                runtime_check_credential_policy_compliance(
                     credential_id=builtin_provider.id,
                     provider=provider_id,
                     credential_type=PluginCredentialType.TOOL,

api/core/workflow/node_factory.py

@@ -1,16 +1,16 @@
 import importlib
 import pkgutil
-from collections.abc import Callable, Iterator, Mapping, MutableMapping
+from collections.abc import Callable, Iterator, Mapping, MutableMapping, Sequence
 from dataclasses import dataclass
 from functools import lru_cache
 from typing import TYPE_CHECKING, Any, cast, final, override
 
 from sqlalchemy import select
-from sqlalchemy.orm import Session
 
 from configs import dify_config
 from core.app.entities.app_invoke_entities import DIFY_RUN_CONTEXT_KEY, DifyRunContext
 from core.app.llm.model_access import build_dify_model_access, fetch_model_config
+from core.db.session_factory import session_factory
 from core.helper.code_executor.code_executor import (
     CodeExecutionError,
     CodeExecutor,
@@ -39,7 +39,6 @@ from core.workflow.nodes.agent.plugin_strategy_adapter import (
 from core.workflow.nodes.agent.runtime_support import AgentRuntimeSupport
 from core.workflow.system_variables import SystemVariableKey, get_system_text, system_variable_selector
 from core.workflow.template_rendering import CodeExecutorJinja2TemplateRenderer
-from extensions.ext_database import db
 from graphon.entities.base_node_data import BaseNodeData
 from graphon.entities.graph_config import NodeConfigDict, NodeConfigDictAdapter
 from graphon.enums import BuiltinNodeTypes, NodeType
@@ -48,7 +47,7 @@ from graphon.graph.graph import NodeFactory
 from graphon.model_runtime.memory import PromptMessageMemory
 from graphon.model_runtime.model_providers.base.large_language_model import LargeLanguageModel
 from graphon.nodes.base.node import Node
-from graphon.nodes.code.code_node import WorkflowCodeExecutor
+from graphon.nodes.code.code_node import CodeExecutorProtocol
 from graphon.nodes.code.entities import CodeLanguage
 from graphon.nodes.code.limits import CodeNodeLimits
 from graphon.nodes.document_extractor import UnstructuredApiConfig
@@ -56,6 +55,7 @@ from graphon.nodes.http_request import build_http_request_config
 from graphon.nodes.llm.entities import LLMNodeData
 from graphon.nodes.parameter_extractor.entities import ParameterExtractorNodeData
 from graphon.nodes.question_classifier.entities import QuestionClassifierNodeData
+from graphon.variables.segments import ArrayObjectSegment
 from models.model import Conversation
 
 if TYPE_CHECKING:
@@ -228,10 +228,14 @@ def fetch_memory(
     node_data_memory: MemoryConfig | None,
     model_instance: ModelInstance,
 ) -> TokenBufferMemory | None:
+    """Build prompt memory for node construction without requiring Flask-local state."""
     if not node_data_memory or not conversation_id:
         return None
 
-    with Session(db.engine, expire_on_commit=False) as session:
+    # Node construction can happen in graph initialization paths where Flask's
+    # app context is not active. Use the app-configured session factory instead
+    # of resolving db.engine through Flask-SQLAlchemy's current_app proxy.
+    with session_factory.create_session() as session:
         stmt = select(Conversation).where(Conversation.app_id == app_id, Conversation.id == conversation_id)
         conversation = session.scalar(stmt)
         if not conversation:
@@ -285,7 +289,7 @@ class DifyNodeFactory(NodeFactory):
         self.graph_init_params = graph_init_params
         self.graph_runtime_state = graph_runtime_state
         self._dify_context = self._resolve_dify_context(graph_init_params.run_context)
-        self._code_executor: WorkflowCodeExecutor = DefaultWorkflowCodeExecutor()
+        self._code_executor: CodeExecutorProtocol = DefaultWorkflowCodeExecutor()
         self._code_limits = CodeNodeLimits(
             max_string_length=dify_config.CODE_MAX_STRING_LENGTH,
             max_number=dify_config.CODE_MAX_NUMBER,
@@ -393,6 +397,7 @@ class DifyNodeFactory(NodeFactory):
             },
             BuiltinNodeTypes.HUMAN_INPUT: lambda: {
                 "runtime": self._human_input_runtime,
+                "file_reference_factory": self._file_reference_factory,
                 "form_repository": self._human_input_runtime.build_form_repository(),
             },
             BuiltinNodeTypes.LLM: lambda: self._build_llm_compatible_node_init_kwargs(
@@ -430,7 +435,7 @@ class DifyNodeFactory(NodeFactory):
                 include_jinja2_template_renderer=False,
             ),
             BuiltinNodeTypes.TOOL: lambda: {
-                "tool_file_manager_factory": self._bound_tool_file_manager_factory(),
+                "tool_file_manager": self._bound_tool_file_manager_factory(),
                 "runtime": self._tool_runtime,
             },
             BuiltinNodeTypes.AGENT: lambda: {
@@ -496,13 +501,47 @@ class DifyNodeFactory(NodeFactory):
         if include_prompt_message_serializer:
             node_init_kwargs["prompt_message_serializer"] = self._prompt_message_serializer
         if include_retriever_attachment_loader:
-            node_init_kwargs["retriever_attachment_loader"] = self._retriever_attachment_loader
+            node_init_kwargs["retriever_attachment_loader"] = self._build_retriever_attachment_loader(
+                cast(LLMNodeData, validated_node_data)
+            )
         if include_jinja2_template_renderer:
             node_init_kwargs["jinja2_template_renderer"] = self._jinja2_template_renderer
         if validated_node_data.type == BuiltinNodeTypes.LLM:
             node_init_kwargs["default_query_selector"] = system_variable_selector(SystemVariableKey.QUERY)
         return node_init_kwargs
 
+    def _build_retriever_attachment_loader(self, node_data: LLMNodeData) -> DifyRetrieverAttachmentLoader:
+        return DifyRetrieverAttachmentLoader(
+            file_reference_factory=self._file_reference_factory,
+            segment_access_checker=self._build_retriever_segment_access_checker(
+                node_data.context.variable_selector if node_data.context.enabled else None
+            ),
+        )
+
+    def _build_retriever_segment_access_checker(
+        self,
+        context_variable_selector: Sequence[str] | None,
+    ) -> Callable[[str], bool]:
+        def checker(segment_id: str) -> bool:
+            if not context_variable_selector:
+                return False
+
+            context_value = self.graph_runtime_state.variable_pool.get(context_variable_selector)
+            if not isinstance(context_value, ArrayObjectSegment):
+                return False
+
+            for item in context_value.value:
+                if not isinstance(item, Mapping):
+                    continue
+                metadata = item.get("metadata")
+                if not isinstance(metadata, Mapping):
+                    continue
+                if metadata.get("_source") == "knowledge" and str(metadata.get("segment_id")) == str(segment_id):
+                    return True
+            return False
+
+        return checker
+
     def _build_model_instance_for_llm_node(self, node_data: LLMCompatibleNodeData) -> ModelInstance:
         node_data_model = node_data.model
         model_instance, _ = fetch_model_config(

api/core/workflow/node_runtime.py

@@ -8,7 +8,11 @@ from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from core.app.entities.app_invoke_entities import DIFY_RUN_CONTEXT_KEY, DifyRunContext
-from core.app.file_access import DatabaseFileAccessController
+from core.app.file_access import (
+    DatabaseFileAccessController,
+    grant_upload_file_access,
+    is_retriever_segment_access_granted,
+)
 from core.callback_handler.workflow_tool_callback_handler import DifyWorkflowCallbackHandler
 from core.helper.trace_id_helper import ParentTraceContext
 from core.llm_generator.output_parser.errors import OutputParserError
@@ -45,7 +49,7 @@ from graphon.model_runtime.entities.model_entities import AIModelEntity
 from graphon.model_runtime.model_providers.base.large_language_model import LargeLanguageModel
 from graphon.nodes.human_input.entities import HumanInputNodeData
 from graphon.nodes.llm.runtime_protocols import (
-    PreparedLLMProtocol,
+    LLMProtocol,
     PromptMessageSerializerProtocol,
     RetrieverAttachmentLoaderProtocol,
 )
@@ -136,7 +140,7 @@ class DifyFileReferenceFactory(FileReferenceFactoryProtocol):
         )
 
 
-class DifyPreparedLLM(PreparedLLMProtocol):
+class DifyPreparedLLM(LLMProtocol):
     """Workflow-layer adapter that hides the full `ModelInstance` API from `graphon` nodes."""
 
     def __init__(self, model_instance: ModelInstance) -> None:
@@ -275,10 +279,23 @@ class DifyPromptMessageSerializer(PromptMessageSerializerProtocol):
 class DifyRetrieverAttachmentLoader(RetrieverAttachmentLoaderProtocol):
     """Resolve retriever attachments through Dify persistence and return graph file references."""
 
-    def __init__(self, *, file_reference_factory: FileReferenceFactoryProtocol) -> None:
+    _segment_access_checker: Callable[[str], bool] | None
+
+    def __init__(
+        self,
+        *,
+        file_reference_factory: FileReferenceFactoryProtocol,
+        segment_access_checker: Callable[[str], bool] | None = None,
+    ) -> None:
         self._file_reference_factory = file_reference_factory
+        self._segment_access_checker = segment_access_checker
 
     def load(self, *, segment_id: str) -> Sequence[File]:
+        if not is_retriever_segment_access_granted(segment_id):
+            return []
+        if self._segment_access_checker is not None and not self._segment_access_checker(segment_id):
+            return []
+
         with Session(db.engine, expire_on_commit=False) as session:
             attachments_with_bindings = session.execute(
                 select(SegmentAttachmentBinding, UploadFile)
@@ -286,6 +303,7 @@ class DifyRetrieverAttachmentLoader(RetrieverAttachmentLoaderProtocol):
                 .where(SegmentAttachmentBinding.segment_id == segment_id)
             ).all()
 
+        grant_upload_file_access(str(upload_file.id) for _, upload_file in attachments_with_bindings)
         return [
             self._file_reference_factory.build_from_mapping(
                 mapping={
@@ -590,64 +608,67 @@ class DifyToolNodeRuntime(ToolNodeRuntimeProtocol):
 
         from core.tools.entities.tool_entities import ToolInvokeMessage as CoreToolInvokeMessage
 
-        if isinstance(message, CoreToolInvokeMessage.TextMessage):
-            return ToolRuntimeMessage.TextMessage(text=message.text)
-        if isinstance(message, CoreToolInvokeMessage.JsonMessage):
-            return ToolRuntimeMessage.JsonMessage(
-                json_object=message.json_object,
-                suppress_output=message.suppress_output,
-            )
-        if isinstance(message, CoreToolInvokeMessage.BlobMessage):
-            return ToolRuntimeMessage.BlobMessage(blob=message.blob)
-        if isinstance(message, CoreToolInvokeMessage.BlobChunkMessage):
-            return ToolRuntimeMessage.BlobChunkMessage(
-                id=message.id,
-                sequence=message.sequence,
-                total_length=message.total_length,
-                blob=message.blob,
-                end=message.end,
-            )
-        if isinstance(message, CoreToolInvokeMessage.FileMessage):
-            return ToolRuntimeMessage.FileMessage(file_marker=message.file_marker)
-        if isinstance(message, CoreToolInvokeMessage.VariableMessage):
-            return ToolRuntimeMessage.VariableMessage(
-                variable_name=message.variable_name,
-                variable_value=message.variable_value,
-                stream=message.stream,
-            )
-        if isinstance(message, CoreToolInvokeMessage.LogMessage):
-            return ToolRuntimeMessage.LogMessage(
-                id=message.id,
-                label=message.label,
-                parent_id=message.parent_id,
-                error=message.error,
-                status=ToolRuntimeMessage.LogMessage.LogStatus(message.status.value),
-                data=dict(message.data),
-                metadata=dict(message.metadata),
-            )
-        if isinstance(message, CoreToolInvokeMessage.RetrieverResourceMessage):
-            retriever_resources = [
-                resource.model_dump() if hasattr(resource, "model_dump") else dict(resource)
-                for resource in message.retriever_resources
-            ]
-            return ToolRuntimeMessage.RetrieverResourceMessage(
-                retriever_resources=retriever_resources,
-                context=message.context,
-            )
-
-        raise TypeError(f"unsupported tool message payload: {type(message).__name__}")
+        match message:
+            case CoreToolInvokeMessage.TextMessage():
+                return ToolRuntimeMessage.TextMessage(text=message.text)
+            case CoreToolInvokeMessage.JsonMessage():
+                return ToolRuntimeMessage.JsonMessage(
+                    json_object=message.json_object,
+                    suppress_output=message.suppress_output,
+                )
+            case CoreToolInvokeMessage.BlobMessage():
+                return ToolRuntimeMessage.BlobMessage(blob=message.blob)
+            case CoreToolInvokeMessage.BlobChunkMessage():
+                return ToolRuntimeMessage.BlobChunkMessage(
+                    id=message.id,
+                    sequence=message.sequence,
+                    total_length=message.total_length,
+                    blob=message.blob,
+                    end=message.end,
+                )
+            case CoreToolInvokeMessage.FileMessage():
+                return ToolRuntimeMessage.FileMessage(file_marker=message.file_marker)
+            case CoreToolInvokeMessage.VariableMessage():
+                return ToolRuntimeMessage.VariableMessage(
+                    variable_name=message.variable_name,
+                    variable_value=message.variable_value,
+                    stream=message.stream,
+                )
+            case CoreToolInvokeMessage.LogMessage():
+                return ToolRuntimeMessage.LogMessage(
+                    id=message.id,
+                    label=message.label,
+                    parent_id=message.parent_id,
+                    error=message.error,
+                    status=ToolRuntimeMessage.LogMessage.LogStatus(message.status.value),
+                    data=dict(message.data),
+                    metadata=dict(message.metadata),
+                )
+            case CoreToolInvokeMessage.RetrieverResourceMessage():
+                retriever_resources = [
+                    resource.model_dump() if hasattr(resource, "model_dump") else dict(resource)
+                    for resource in message.retriever_resources
+                ]
+                return ToolRuntimeMessage.RetrieverResourceMessage(
+                    retriever_resources=retriever_resources,
+                    context=message.context,
+                )
+            case _:
+                raise TypeError(f"unsupported tool message payload: {type(message).__name__}")
 
     @staticmethod
     def _map_invocation_exception(exc: Exception, *, provider_name: str) -> ToolNodeError:
-        if isinstance(exc, ToolNodeError):
-            return exc
-        if isinstance(exc, PluginInvokeError):
-            return ToolRuntimeInvocationError(exc.to_user_friendly_error(plugin_name=provider_name))
-        if isinstance(exc, PluginDaemonClientSideError):
-            return ToolRuntimeInvocationError(f"Failed to invoke tool, error: {exc.description}")
-        if isinstance(exc, ToolInvokeError):
-            return ToolRuntimeInvocationError(f"Failed to invoke tool {provider_name}: {exc}")
-        return ToolRuntimeInvocationError(str(exc))
+        match exc:
+            case ToolNodeError():
+                return exc
+            case PluginInvokeError():
+                return ToolRuntimeInvocationError(exc.to_user_friendly_error(plugin_name=provider_name))
+            case PluginDaemonClientSideError():
+                return ToolRuntimeInvocationError(f"Failed to invoke tool, error: {exc.description}")
+            case ToolInvokeError():
+                return ToolRuntimeInvocationError(f"Failed to invoke tool {provider_name}: {exc}")
+            case _:
+                return ToolRuntimeInvocationError(str(exc))
 
 
 class DifyHumanInputNodeRuntime(HumanInputNodeRuntimeProtocol):

api/dev/generate_swagger_markdown_docs.py

@@ -8,6 +8,7 @@ OpenAPI output early.
 from __future__ import annotations
 
 import argparse
+import json
 import logging
 import subprocess
 import sys
@@ -28,6 +29,125 @@ CONSOLE_SWAGGER_FILENAME = "console-swagger.json"
 STALE_COMBINED_MARKDOWN_FILENAME = "api-reference.md"
 
 
+def _definition_ref_name(schema: object) -> str | None:
+    if not isinstance(schema, dict):
+        return None
+
+    ref = schema.get("$ref")
+    if not isinstance(ref, str) or not ref.startswith("#/definitions/"):
+        return None
+
+    return ref.removeprefix("#/definitions/")
+
+
+def _markdown_anchor(name: str) -> str:
+    return name.lower()
+
+
+def _schema_markdown_type(schema: object) -> str:
+    if not isinstance(schema, dict):
+        return ""
+
+    ref_name = _definition_ref_name(schema)
+    if ref_name is not None:
+        return f"[{ref_name}](#{_markdown_anchor(ref_name)})"
+
+    for union_key in ("oneOf", "anyOf"):
+        variants = schema.get(union_key)
+        if not isinstance(variants, list):
+            continue
+
+        variant_types = [
+            variant_type
+            for variant in variants
+            if not (isinstance(variant, dict) and variant.get("type") == "null")
+            for variant_type in [_schema_markdown_type(variant)]
+            if variant_type
+        ]
+        if len(variant_types) == 1:
+            return variant_types[0]
+        if variant_types:
+            return "<br>".join(variant_types)
+
+    schema_type = schema.get("type")
+    if schema_type == "array":
+        item_type = _schema_markdown_type(schema.get("items"))
+        return f"[ {item_type or 'object'} ]"
+    if isinstance(schema_type, str):
+        return schema_type
+
+    return ""
+
+
+def _replace_schema_table_type(markdown: str, definition_name: str, row_name: str, type_markdown: str) -> str:
+    if not type_markdown:
+        return markdown
+
+    lines = markdown.splitlines()
+    section_header = f"#### {definition_name}"
+    in_section = False
+
+    for index, line in enumerate(lines):
+        if line == section_header:
+            in_section = True
+            continue
+        if in_section and line.startswith("#### "):
+            break
+        if not in_section or not line.startswith(f"| {row_name} |"):
+            continue
+
+        cells = line.split("|")
+        if len(cells) < 5:
+            continue
+        cells[2] = f" {type_markdown} "
+        lines[index] = "|".join(cells)
+        break
+
+    return "\n".join(lines)
+
+
+def _patch_union_schema_markdown(markdown: str, spec_path: Path) -> str:
+    """Fill Swagger Markdown table cells that `swagger-markdown` leaves blank for union schemas."""
+
+    spec = json.loads(spec_path.read_text(encoding="utf-8"))
+    definitions = spec.get("definitions")
+    if not isinstance(definitions, dict):
+        return markdown
+
+    for definition_name, schema in definitions.items():
+        if not isinstance(definition_name, str) or not isinstance(schema, dict):
+            continue
+        one_of = schema.get("oneOf")
+        if not isinstance(one_of, list):
+            continue
+
+        markdown = _replace_schema_table_type(
+            markdown,
+            definition_name,
+            definition_name,
+            _schema_markdown_type(schema),
+        )
+
+        for variant in one_of:
+            variant_name = _definition_ref_name(variant)
+            variant_schema = definitions.get(variant_name) if variant_name is not None else None
+            if not isinstance(variant_name, str) or not isinstance(variant_schema, dict):
+                continue
+            properties = variant_schema.get("properties")
+            if not isinstance(properties, dict):
+                continue
+            for property_name, property_schema in properties.items():
+                if isinstance(property_name, str):
+                    markdown = _replace_schema_table_type(
+                        markdown,
+                        variant_name,
+                        property_name,
+                        _schema_markdown_type(property_schema),
+                    )
+
+    return markdown
+
+
 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:
@@ -57,7 +177,10 @@ def _convert_spec_to_markdown(spec_path: Path, markdown_path: Path) -> None:
             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")
+        converted_markdown = _patch_union_schema_markdown(
+            temp_markdown_path.read_text(encoding="utf-8"),
+            spec_path,
+        )
         if not converted_markdown.strip():
             raise RuntimeError(f"swagger-markdown wrote an empty document for {markdown_path}")
 

api/enterprise/telemetry/enterprise_trace.py

@@ -77,28 +77,29 @@ class EnterpriseOtelTrace:
         self._exporter = exporter
 
     def trace(self, trace_info: BaseTraceInfo) -> None:
-        if isinstance(trace_info, WorkflowTraceInfo):
-            self._workflow_trace(trace_info)
-        elif isinstance(trace_info, MessageTraceInfo):
-            self._message_trace(trace_info)
-        elif isinstance(trace_info, ToolTraceInfo):
-            self._tool_trace(trace_info)
-        elif isinstance(trace_info, DraftNodeExecutionTrace):
-            self._draft_node_execution_trace(trace_info)
-        elif isinstance(trace_info, WorkflowNodeTraceInfo):
-            self._node_execution_trace(trace_info)
-        elif isinstance(trace_info, ModerationTraceInfo):
-            self._moderation_trace(trace_info)
-        elif isinstance(trace_info, SuggestedQuestionTraceInfo):
-            self._suggested_question_trace(trace_info)
-        elif isinstance(trace_info, DatasetRetrievalTraceInfo):
-            self._dataset_retrieval_trace(trace_info)
-        elif isinstance(trace_info, GenerateNameTraceInfo):
-            self._generate_name_trace(trace_info)
-        elif isinstance(trace_info, PromptGenerationTraceInfo):
-            self._prompt_generation_trace(trace_info)
-        else:
-            raise AssertionError("this statment should be unreachable")
+        match trace_info:
+            case WorkflowTraceInfo():
+                self._workflow_trace(trace_info)
+            case MessageTraceInfo():
+                self._message_trace(trace_info)
+            case ToolTraceInfo():
+                self._tool_trace(trace_info)
+            case DraftNodeExecutionTrace():
+                self._draft_node_execution_trace(trace_info)
+            case WorkflowNodeTraceInfo():
+                self._node_execution_trace(trace_info)
+            case ModerationTraceInfo():
+                self._moderation_trace(trace_info)
+            case SuggestedQuestionTraceInfo():
+                self._suggested_question_trace(trace_info)
+            case DatasetRetrievalTraceInfo():
+                self._dataset_retrieval_trace(trace_info)
+            case GenerateNameTraceInfo():
+                self._generate_name_trace(trace_info)
+            case PromptGenerationTraceInfo():
+                self._prompt_generation_trace(trace_info)
+            case _:
+                raise AssertionError("this statment should be unreachable")
 
     def _common_attrs(self, trace_info: BaseTraceInfo) -> dict[str, Any]:
         metadata = self._metadata(trace_info)

api/extensions/ext_app_metrics.py

@@ -11,14 +11,14 @@ from dify_app import DifyApp
 
 def init_app(app: DifyApp):
     @app.after_request
-    def after_request(response):  # pyright: ignore[reportUnusedFunction]
+    def after_request(response):
         """Add Version headers to the response."""
         response.headers.add("X-Version", dify_config.project.version)
         response.headers.add("X-Env", dify_config.DEPLOY_ENV)
         return response
 
     @app.route("/health")
-    def health():  # pyright: ignore[reportUnusedFunction]
+    def health():
         return Response(
             json.dumps({"pid": os.getpid(), "status": "ok", "version": dify_config.project.version}),
             status=200,
@@ -27,7 +27,7 @@ def init_app(app: DifyApp):
 
     @app.route("/threads")
     @admin_required
-    def threads():  # pyright: ignore[reportUnusedFunction]
+    def threads():
         num_threads = threading.active_count()
         threads = threading.enumerate()
 
@@ -53,7 +53,7 @@ def init_app(app: DifyApp):
 
     @app.route("/db-pool-stat")
     @admin_required
-    def pool_stat():  # pyright: ignore[reportUnusedFunction]
+    def pool_stat():
         from extensions.ext_database import db
 
         engine = db.engine

api/extensions/ext_database.py

@@ -33,7 +33,7 @@ def _setup_gevent_compatibility():
         return
 
     @event.listens_for(Pool, "reset")
-    def _safe_reset(dbapi_connection, connection_record, reset_state):  # pyright: ignore[reportUnusedFunction]
+    def _safe_reset(dbapi_connection, connection_record, reset_state):
         if reset_state.terminate_only:
             return
 

api/extensions/ext_import_modules.py

@@ -2,4 +2,4 @@ from dify_app import DifyApp
 
 
 def init_app(app: DifyApp):
-    from events import event_handlers  # noqa: F401 # pyright: ignore[reportUnusedImport]
+    from events import event_handlers  # noqa: F401

api/extensions/ext_mail.py

@@ -67,7 +67,7 @@ class Mail:
             case _:
                 raise ValueError(f"Unsupported mail type {mail_type}")
 
-    def send(self, to: str, subject: str, html: str, from_: str | None = None):
+    def send(self, to: str, subject: str, html: str, from_: str = ""):
         if not self._client:
             raise ValueError("Mail client is not initialized")
 

api/extensions/ext_socketio.py

@@ -1,5 +1,9 @@
+from typing import Any, cast
+
 import socketio  # type: ignore[reportMissingTypeStubs]
 
 from configs import dify_config
 
-sio = socketio.Server(async_mode="gevent", cors_allowed_origins=dify_config.CONSOLE_CORS_ALLOW_ORIGINS)
+# TODO: FIXME(chariri) - Casting to any because app_factory attaches the
+# current app as the `app` attribute on this - Bad.
+sio = cast(Any, socketio.Server(async_mode="gevent", cors_allowed_origins=dify_config.CONSOLE_CORS_ALLOW_ORIGINS))

api/extensions/logstore/aliyun_logstore.py

@@ -198,7 +198,7 @@ class AliyunLogStore:
         )
 
         # Append Dify identification to the existing user agent
-        original_user_agent = self.client._user_agent  # pyright: ignore[reportPrivateUsage]
+        original_user_agent = self.client._user_agent
         dify_version = dify_config.project.version
         enhanced_user_agent = f"Dify,Dify-{dify_version},{original_user_agent}"
         self.client.set_user_agent(enhanced_user_agent)

api/fields/annotation_fields.py

@@ -5,12 +5,7 @@ from datetime import datetime
 from pydantic import Field, field_validator
 
 from fields.base import ResponseModel
-
-
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
+from libs.helper import to_timestamp
 
 
 class Annotation(ResponseModel):
@@ -23,7 +18,7 @@ class Annotation(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class AnnotationList(ResponseModel):
@@ -50,7 +45,7 @@ class AnnotationHitHistory(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class AnnotationHitHistoryList(ResponseModel):

api/fields/conversation_fields.py

@@ -7,6 +7,7 @@ from pydantic import Field, field_validator, model_validator
 
 from fields.base import ResponseModel
 from graphon.file import File
+from libs.helper import to_timestamp
 
 type JSONValue = Any
 
@@ -47,9 +48,7 @@ class SimpleConversation(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return to_timestamp(value)
-        return value
+        return to_timestamp(value)
 
 
 class ConversationInfiniteScrollPagination(ResponseModel):
@@ -90,9 +89,7 @@ class ConversationAnnotation(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return to_timestamp(value)
-        return value
+        return to_timestamp(value)
 
 
 class ConversationAnnotationHitHistory(ResponseModel):
@@ -103,9 +100,7 @@ class ConversationAnnotationHitHistory(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return to_timestamp(value)
-        return value
+        return to_timestamp(value)
 
 
 class AgentThought(ResponseModel):
@@ -125,9 +120,7 @@ class AgentThought(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return to_timestamp(value)
-        return value
+        return to_timestamp(value)
 
     @model_validator(mode="after")
     def _fallback_chain_id(self):
@@ -169,9 +162,7 @@ class MessageDetail(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return to_timestamp(value)
-        return value
+        return to_timestamp(value)
 
 
 class FeedbackStat(ResponseModel):
@@ -237,9 +228,7 @@ class Conversation(ResponseModel):
     @field_validator("read_at", "created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return to_timestamp(value)
-        return value
+        return to_timestamp(value)
 
 
 class ConversationPagination(ResponseModel):
@@ -263,9 +252,7 @@ class ConversationMessageDetail(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return to_timestamp(value)
-        return value
+        return to_timestamp(value)
 
 
 class ConversationWithSummary(ResponseModel):
@@ -291,9 +278,7 @@ class ConversationWithSummary(ResponseModel):
     @field_validator("read_at", "created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return to_timestamp(value)
-        return value
+        return to_timestamp(value)
 
 
 class ConversationWithSummaryPagination(ResponseModel):
@@ -322,15 +307,7 @@ class ConversationDetail(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return to_timestamp(value)
-        return value
-
-
-def to_timestamp(value: datetime | None) -> int | None:
-    if value is None:
-        return None
-    return int(value.timestamp())
+        return to_timestamp(value)
 
 
 def format_files_contained(value: JSONValue) -> JSONValue:

api/fields/conversation_variable_fields.py

@@ -8,7 +8,7 @@ from pydantic import field_validator
 
 from fields.base import ResponseModel
 from graphon.variables.types import SegmentType
-from libs.helper import TimestampField
+from libs.helper import TimestampField, to_timestamp
 
 from ._value_type_serializer import serialize_value_type
 
@@ -37,12 +37,6 @@ conversation_variable_infinite_scroll_pagination_fields = {
 }
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 class ConversationVariableResponse(ResponseModel):
     id: str
     name: str
@@ -88,7 +82,7 @@ class ConversationVariableResponse(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class PaginatedConversationVariableResponse(ResponseModel):

api/fields/file_fields.py

@@ -5,12 +5,7 @@ from datetime import datetime
 from pydantic import field_validator
 
 from fields.base import ResponseModel
-
-
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
+from libs.helper import to_timestamp
 
 
 class UploadConfig(ResponseModel):
@@ -45,7 +40,7 @@ class FileResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class RemoteFileInfo(ResponseModel):
@@ -66,7 +61,7 @@ class FileWithSignedUrl(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 __all__ = [

api/fields/member_fields.py

@@ -6,7 +6,7 @@ from flask_restx import fields
 from pydantic import computed_field, field_validator
 
 from fields.base import ResponseModel
-from graphon.file import helpers as file_helpers
+from libs.helper import build_avatar_url, to_timestamp
 
 simple_account_fields = {
     "id": fields.String,
@@ -15,20 +15,6 @@ simple_account_fields = {
 }
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
-def _build_avatar_url(avatar: str | None) -> str | None:
-    if avatar is None:
-        return None
-    if avatar.startswith(("http://", "https://")):
-        return avatar
-    return file_helpers.get_signed_file_url(avatar)
-
-
 class SimpleAccount(ResponseModel):
     id: str
     name: str
@@ -41,7 +27,7 @@ class _AccountAvatar(ResponseModel):
     @computed_field(return_type=str | None)  # type: ignore[prop-decorator]
     @property
     def avatar_url(self) -> str | None:
-        return _build_avatar_url(self.avatar)
+        return build_avatar_url(self.avatar)
 
 
 class Account(_AccountAvatar):
@@ -59,7 +45,7 @@ class Account(_AccountAvatar):
     @field_validator("last_login_at", "created_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class AccountWithRole(_AccountAvatar):
@@ -75,7 +61,7 @@ class AccountWithRole(_AccountAvatar):
     @field_validator("last_login_at", "last_active_at", "created_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class AccountWithRoleList(ResponseModel):

api/fields/message_fields.py

@@ -9,6 +9,7 @@ from core.entities.execution_extra_content import ExecutionExtraContentDomainMod
 from fields.base import ResponseModel
 from fields.conversation_fields import AgentThought, JSONValue, MessageFile
 from graphon.file import File
+from libs.helper import to_timestamp
 
 type JSONValueType = JSONValue
 
@@ -39,9 +40,7 @@ class RetrieverResource(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return to_timestamp(value)
-        return value
+        return to_timestamp(value)
 
 
 class MessageListItem(ResponseModel):
@@ -68,9 +67,7 @@ class MessageListItem(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return to_timestamp(value)
-        return value
+        return to_timestamp(value)
 
 
 class WebMessageListItem(MessageListItem):
@@ -106,9 +103,7 @@ class SavedMessageItem(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        if isinstance(value, datetime):
-            return to_timestamp(value)
-        return value
+        return to_timestamp(value)
 
 
 class SavedMessageInfiniteScrollPagination(ResponseModel):
@@ -121,12 +116,6 @@ class SuggestedQuestionsResponse(ResponseModel):
     data: list[str]
 
 
-def to_timestamp(value: datetime | None) -> int | None:
-    if value is None:
-        return None
-    return int(value.timestamp())
-
-
 def format_files_contained(value: JSONValueType) -> JSONValueType:
     if isinstance(value, File):
         # Response payloads must preserve legacy file keys like `related_id`/`url`

api/fields/online_user_fields.py

@@ -1,16 +0,0 @@
-from flask_restx import fields
-
-online_user_partial_fields = {
-    "user_id": fields.String,
-    "username": fields.String,
-    "avatar": fields.String,
-}
-
-workflow_online_users_fields = {
-    "app_id": fields.String,
-    "users": fields.List(fields.Nested(online_user_partial_fields)),
-}
-
-online_user_list_fields = {
-    "data": fields.List(fields.Nested(workflow_online_users_fields)),
-}

api/fields/workflow_app_log_fields.py

@@ -17,7 +17,7 @@ from fields.workflow_run_fields import (
     workflow_run_for_archived_log_fields,
     workflow_run_for_log_fields,
 )
-from libs.helper import TimestampField
+from libs.helper import TimestampField, to_timestamp
 
 workflow_app_log_partial_fields = {
     "id": fields.String,
@@ -96,12 +96,6 @@ def build_workflow_archived_log_pagination_model(api_or_ns: Namespace):
     return api_or_ns.model("WorkflowArchivedLogPagination", copied_fields)
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 class WorkflowAppLogPartialResponse(ResponseModel):
     id: str
     workflow_run: WorkflowRunForLogResponse | None = None
@@ -115,7 +109,7 @@ class WorkflowAppLogPartialResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class WorkflowArchivedLogPartialResponse(ResponseModel):
@@ -129,7 +123,7 @@ class WorkflowArchivedLogPartialResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return _to_timestamp(value)
+        return to_timestamp(value)
 
 
 class WorkflowAppLogPaginationResponse(ResponseModel):

api/fields/workflow_comment_fields.py

@@ -1,96 +0,0 @@
-from flask_restx import fields
-
-from libs.helper import AvatarUrlField, TimestampField
-
-# basic account fields for comments
-account_fields = {
-    "id": fields.String,
-    "name": fields.String,
-    "email": fields.String,
-    "avatar_url": AvatarUrlField,
-}
-
-# Comment mention fields
-workflow_comment_mention_fields = {
-    "mentioned_user_id": fields.String,
-    "mentioned_user_account": fields.Nested(account_fields, allow_null=True),
-    "reply_id": fields.String,
-}
-
-# Comment reply fields
-workflow_comment_reply_fields = {
-    "id": fields.String,
-    "content": fields.String,
-    "created_by": fields.String,
-    "created_by_account": fields.Nested(account_fields, allow_null=True),
-    "created_at": TimestampField,
-}
-
-# Basic comment fields (for list views)
-workflow_comment_basic_fields = {
-    "id": fields.String,
-    "position_x": fields.Float,
-    "position_y": fields.Float,
-    "content": fields.String,
-    "created_by": fields.String,
-    "created_by_account": fields.Nested(account_fields, allow_null=True),
-    "created_at": TimestampField,
-    "updated_at": TimestampField,
-    "resolved": fields.Boolean,
-    "resolved_at": TimestampField,
-    "resolved_by": fields.String,
-    "resolved_by_account": fields.Nested(account_fields, allow_null=True),
-    "reply_count": fields.Integer,
-    "mention_count": fields.Integer,
-    "participants": fields.List(fields.Nested(account_fields)),
-}
-
-# Detailed comment fields (for single comment view)
-workflow_comment_detail_fields = {
-    "id": fields.String,
-    "position_x": fields.Float,
-    "position_y": fields.Float,
-    "content": fields.String,
-    "created_by": fields.String,
-    "created_by_account": fields.Nested(account_fields, allow_null=True),
-    "created_at": TimestampField,
-    "updated_at": TimestampField,
-    "resolved": fields.Boolean,
-    "resolved_at": TimestampField,
-    "resolved_by": fields.String,
-    "resolved_by_account": fields.Nested(account_fields, allow_null=True),
-    "replies": fields.List(fields.Nested(workflow_comment_reply_fields)),
-    "mentions": fields.List(fields.Nested(workflow_comment_mention_fields)),
-}
-
-# Comment creation response fields (simplified)
-workflow_comment_create_fields = {
-    "id": fields.String,
-    "created_at": TimestampField,
-}
-
-# Comment update response fields (simplified)
-workflow_comment_update_fields = {
-    "id": fields.String,
-    "updated_at": TimestampField,
-}
-
-# Comment resolve response fields
-workflow_comment_resolve_fields = {
-    "id": fields.String,
-    "resolved": fields.Boolean,
-    "resolved_at": TimestampField,
-    "resolved_by": fields.String,
-}
-
-# Reply creation response fields (simplified)
-workflow_comment_reply_create_fields = {
-    "id": fields.String,
-    "created_at": TimestampField,
-}
-
-# Reply update response fields
-workflow_comment_reply_update_fields = {
-    "id": fields.String,
-    "updated_at": TimestampField,
-}