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'

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/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/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/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/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/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/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/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/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/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/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/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/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/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/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/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,
-}

api/fields/workflow_run_fields.py

@@ -16,7 +16,7 @@ from pydantic import AliasChoices, Field, field_validator
 from fields.base import ResponseModel
 from fields.end_user_fields import SimpleEndUser
 from fields.member_fields import SimpleAccount
-from libs.helper import TimestampField
+from libs.helper import TimestampField, to_timestamp
 
 workflow_run_for_log_fields = {
     "id": fields.String,
@@ -50,12 +50,6 @@ def build_workflow_run_for_archived_log_model(api_or_ns: Namespace):
     return api_or_ns.model("WorkflowRunForArchivedLog", workflow_run_for_archived_log_fields)
 
 
-def _to_timestamp(value: datetime | int | None) -> int | None:
-    if isinstance(value, datetime):
-        return int(value.timestamp())
-    return value
-
-
 class WorkflowRunForLogResponse(ResponseModel):
     id: str
     version: str | None = None
@@ -79,7 +73,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 WorkflowRunForArchivedLogResponse(ResponseModel):
@@ -120,7 +114,7 @@ class WorkflowRunForListResponse(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 AdvancedChatWorkflowRunForListResponse(WorkflowRunForListResponse):
@@ -180,7 +174,7 @@ class WorkflowRunDetailResponse(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 WorkflowRunNodeExecutionResponse(ResponseModel):
@@ -217,7 +211,7 @@ class WorkflowRunNodeExecutionResponse(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 WorkflowRunNodeExecutionListResponse(ResponseModel):

api/libs/helper.py

@@ -10,7 +10,7 @@ import uuid
 from collections.abc import Callable, Generator, Mapping
 from datetime import datetime
 from hashlib import sha256
-from typing import TYPE_CHECKING, Annotated, Any, Protocol, cast
+from typing import TYPE_CHECKING, Annotated, Any, Protocol, cast, overload
 from uuid import UUID
 from zoneinfo import available_timezones
 
@@ -39,6 +39,7 @@ class _TokenData(TypedDict, total=False):
     token_type: str
     code: str
     old_email: str
+    phase: str
 
 
 _token_data_adapter: TypeAdapter[_TokenData] = TypeAdapter(_TokenData)
@@ -97,12 +98,13 @@ def extract_tenant_id(user: "Account | EndUser") -> str | None:
     from models import Account
     from models.model import EndUser
 
-    if isinstance(user, Account):
-        return user.current_tenant_id
-    elif isinstance(user, EndUser):
-        return user.tenant_id
-    else:
-        raise ValueError(f"Invalid user type: {type(user)}. Expected Account or EndUser.")
+    match user:
+        case Account():
+            return user.current_tenant_id
+        case EndUser():
+            return user.tenant_id
+        case _:
+            raise ValueError(f"Invalid user type: {type(user)}. Expected Account or EndUser.")
 
 
 def run(script):
@@ -136,6 +138,14 @@ def build_icon_url(icon_type: Any, icon: str | None) -> str | None:
     return file_helpers.get_signed_file_url(icon)
 
 
+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 AvatarUrlField(fields.Raw):
     def output(self, key, obj, **kwargs):
         if obj is None:
@@ -144,9 +154,7 @@ class AvatarUrlField(fields.Raw):
         from models import Account
 
         if isinstance(obj, Account) and obj.avatar is not None:
-            if obj.avatar.startswith(("http://", "https://")):
-                return obj.avatar
-            return file_helpers.get_signed_file_url(obj.avatar)
+            return build_avatar_url(obj.avatar)
         return None
 
 
@@ -162,6 +170,35 @@ class OptionalTimestampField(fields.Raw):
         return int(value.timestamp())
 
 
+@overload
+def to_timestamp(value: datetime) -> int: ...
+
+
+@overload
+def to_timestamp(value: int) -> int: ...
+
+
+@overload
+def to_timestamp(value: None) -> None: ...
+
+
+def to_timestamp(value: datetime | int | None) -> int | None:
+    """Normalize API response timestamp values to epoch seconds."""
+    if isinstance(value, datetime):
+        return int(value.timestamp())
+    return value
+
+
+def dump_response(model: type[BaseModel], data: Any) -> dict[str, Any]:
+    """Serialize a Pydantic response model to JSON-compatible dict output."""
+    return model.model_validate(data, from_attributes=True).model_dump(mode="json")
+
+
+def current_timestamp() -> int:
+    """Return the current Unix timestamp in seconds."""
+    return int(time.time())
+
+
 def email(email):
     # Define a regex pattern for email addresses
     pattern = r"^[\w\.!#$%&'*+\-/=?^_`{|}~]+@([\w-]+\.)+[\w-]{2,}$"
@@ -386,18 +423,19 @@ def length_prefixed_response(
         # | Magic Number 1byte | Reserved 1byte | Header Length 2bytes | Data Length 4bytes | Reserved 6bytes | Data
         return struct.pack("<BBHI", magic_number, 0, header_length, data_length) + b"\x00" * 6 + response
 
-    if isinstance(response, Mapping):
-        return Response(
-            response=pack_response_with_length_prefix(json.dumps(jsonable_encoder(response)).encode("utf-8")),
-            status=200,
-            mimetype="application/json",
-        )
-    elif isinstance(response, BaseModel):
-        return Response(
-            response=pack_response_with_length_prefix(response.model_dump_json().encode("utf-8")),
-            status=200,
-            mimetype="application/json",
-        )
+    match response:
+        case Mapping():
+            return Response(
+                response=pack_response_with_length_prefix(json.dumps(jsonable_encoder(response)).encode("utf-8")),
+                status=200,
+                mimetype="application/json",
+            )
+        case BaseModel():
+            return Response(
+                response=pack_response_with_length_prefix(response.model_dump_json().encode("utf-8")),
+                status=200,
+                mimetype="application/json",
+            )
 
     stream_response = response
 

api/libs/oauth.py

@@ -1,3 +1,6 @@
+import base64
+import binascii
+import json
 import logging
 import urllib.parse
 from dataclasses import dataclass
@@ -27,6 +30,12 @@ class AccessTokenResponse(TypedDict, total=False):
     access_token: str
 
 
+class OAuthState(TypedDict, total=False):
+    invite_token: str
+    timezone: str
+    language: str
+
+
 class GitHubEmailRecord(TypedDict, total=False):
     email: str
     primary: bool
@@ -46,6 +55,7 @@ class GoogleRawUserInfo(TypedDict):
 
 
 ACCESS_TOKEN_RESPONSE_ADAPTER = TypeAdapter(AccessTokenResponse)
+OAUTH_STATE_ADAPTER = TypeAdapter(OAuthState)
 GITHUB_RAW_USER_INFO_ADAPTER = TypeAdapter(GitHubRawUserInfo)
 GITHUB_EMAIL_RECORDS_ADAPTER = TypeAdapter(list[GitHubEmailRecord])
 GOOGLE_RAW_USER_INFO_ADAPTER = TypeAdapter(GoogleRawUserInfo)
@@ -58,6 +68,37 @@ class OAuthUserInfo:
     email: str
 
 
+def encode_oauth_state(
+    invite_token: str | None = None,
+    timezone: str | None = None,
+    language: str | None = None,
+) -> str | None:
+    state: OAuthState = {}
+    if invite_token:
+        state["invite_token"] = invite_token
+    if timezone:
+        state["timezone"] = timezone
+    if language:
+        state["language"] = language
+    if not state:
+        return None
+
+    raw_state = json.dumps(state, separators=(",", ":")).encode("utf-8")
+    return base64.urlsafe_b64encode(raw_state).decode("ascii").rstrip("=")
+
+
+def decode_oauth_state(state: str | None) -> OAuthState:
+    if not state:
+        return {}
+
+    try:
+        padded_state = state + "=" * (-len(state) % 4)
+        raw_state = base64.urlsafe_b64decode(padded_state.encode("ascii")).decode("utf-8")
+        return OAUTH_STATE_ADAPTER.validate_python(json.loads(raw_state))
+    except (binascii.Error, ValueError, UnicodeDecodeError, json.JSONDecodeError, ValidationError):
+        return {}
+
+
 def _json_object(response: httpx.Response) -> JsonObject:
     return JSON_OBJECT_ADAPTER.validate_python(response.json())
 
@@ -76,7 +117,12 @@ class OAuth:
         self.client_secret = client_secret
         self.redirect_uri = redirect_uri
 
-    def get_authorization_url(self, invite_token: str | None = None) -> str:
+    def get_authorization_url(
+        self,
+        invite_token: str | None = None,
+        timezone: str | None = None,
+        language: str | None = None,
+    ) -> str:
         raise NotImplementedError()
 
     def get_access_token(self, code: str) -> str:
@@ -99,14 +145,20 @@ class GitHubOAuth(OAuth):
     _USER_INFO_URL = "https://api.github.com/user"
     _EMAIL_INFO_URL = "https://api.github.com/user/emails"
 
-    def get_authorization_url(self, invite_token: str | None = None) -> str:
+    def get_authorization_url(
+        self,
+        invite_token: str | None = None,
+        timezone: str | None = None,
+        language: str | None = None,
+    ) -> str:
         params = {
             "client_id": self.client_id,
             "redirect_uri": self.redirect_uri,
             "scope": "user:email",  # Request only basic user information
         }
-        if invite_token:
-            params["state"] = invite_token
+        state = encode_oauth_state(invite_token=invite_token, timezone=timezone, language=language)
+        if state:
+            params["state"] = state
         return f"{self._AUTH_URL}?{urllib.parse.urlencode(params)}"
 
     def get_access_token(self, code: str) -> str:
@@ -186,15 +238,21 @@ class GoogleOAuth(OAuth):
     _TOKEN_URL = "https://oauth2.googleapis.com/token"
     _USER_INFO_URL = "https://www.googleapis.com/oauth2/v3/userinfo"
 
-    def get_authorization_url(self, invite_token: str | None = None) -> str:
+    def get_authorization_url(
+        self,
+        invite_token: str | None = None,
+        timezone: str | None = None,
+        language: str | None = None,
+    ) -> str:
         params = {
             "client_id": self.client_id,
             "response_type": "code",
             "redirect_uri": self.redirect_uri,
             "scope": "openid email",
         }
-        if invite_token:
-            params["state"] = invite_token
+        state = encode_oauth_state(invite_token=invite_token, timezone=timezone, language=language)
+        if state:
+            params["state"] = state
         return f"{self._AUTH_URL}?{urllib.parse.urlencode(params)}"
 
     def get_access_token(self, code: str) -> str:

api/models/utils/file_input_compat.py

@@ -48,20 +48,23 @@ def build_file_from_mapping_without_lookup(*, file_mapping: Mapping[str, Any]) -
     """Build a graph `File` directly from serialized metadata."""
 
     def _coerce_file_type(value: Any) -> FileType:
-        if isinstance(value, FileType):
-            return value
-        if isinstance(value, str):
-            return FileType.value_of(value)
-        raise ValueError("file type is required in file mapping")
+        match value:
+            case FileType():
+                return value
+            case str():
+                return FileType.value_of(value)
+            case _:
+                raise ValueError("file type is required in file mapping")
 
     mapping = dict(file_mapping)
     transfer_method_value = mapping.get("transfer_method")
-    if isinstance(transfer_method_value, FileTransferMethod):
-        transfer_method = transfer_method_value
-    elif isinstance(transfer_method_value, str):
-        transfer_method = FileTransferMethod.value_of(transfer_method_value)
-    else:
-        raise ValueError("transfer_method is required in file mapping")
+    match transfer_method_value:
+        case FileTransferMethod():
+            transfer_method = transfer_method_value
+        case str():
+            transfer_method = FileTransferMethod.value_of(transfer_method_value)
+        case _:
+            raise ValueError("transfer_method is required in file mapping")
 
     file_id = mapping.get("file_id")
     if not isinstance(file_id, str) or not file_id:
@@ -151,15 +154,15 @@ def rebuild_serialized_graph_files_without_lookup(value: Any) -> Any:
     so historical JSON blobs remain readable without reintroducing global graph
     patches or test-local coercion.
     """
-    if isinstance(value, list):
-        return [rebuild_serialized_graph_files_without_lookup(item) for item in value]
-
-    if isinstance(value, dict):
-        if maybe_file_object(value):
-            return build_file_from_mapping_without_lookup(file_mapping=value)
-        return {key: rebuild_serialized_graph_files_without_lookup(item) for key, item in value.items()}
-
-    return value
+    match value:
+        case list():
+            return [rebuild_serialized_graph_files_without_lookup(item) for item in value]
+        case dict():
+            if maybe_file_object(value):
+                return build_file_from_mapping_without_lookup(file_mapping=value)
+            return {key: rebuild_serialized_graph_files_without_lookup(item) for key, item in value.items()}
+        case _:
+            return value
 
 
 def build_file_from_stored_mapping(

api/openapi/markdown/service-swagger.md

@@ -2136,11 +2136,11 @@ Returns a list of available models for the specified model type.
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| content |  |  | No |
-| created_at |  |  | No |
-| hit_count |  |  | No |
+| content | string |  | No |
+| created_at | integer |  | No |
+| hit_count | integer |  | No |
 | id | string |  | Yes |
-| question |  |  | No |
+| question | string |  | No |
 
 #### AnnotationCreatePayload
 
@@ -2172,13 +2172,13 @@ Returns a list of available models for the specified model type.
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | auto_generate_name | boolean | Auto generate conversation name | No |
-| conversation_id |  | Conversation UUID | No |
-| files |  |  | No |
+| conversation_id | string | Conversation UUID | No |
+| files | [ object ] |  | No |
 | inputs | object |  | Yes |
 | query | string |  | Yes |
-| response_mode |  |  | No |
+| response_mode | string | *Enum:* `"blocking"`, `"streaming"` | No |
 | retriever_from | string |  | No |
-| workflow_id |  | Workflow ID for advanced chat | No |
+| workflow_id | string | Workflow ID for advanced chat | No |
 
 #### ChildChunkCreatePayload
 
@@ -2190,7 +2190,7 @@ Returns a list of available models for the specified model type.
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| keyword |  |  | No |
+| keyword | string |  | No |
 | limit | integer |  | No |
 | page | integer |  | No |
 
@@ -2204,10 +2204,10 @@ Returns a list of available models for the specified model type.
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| files |  |  | No |
+| files | [ object ] |  | No |
 | inputs | object |  | Yes |
 | query | string |  | No |
-| response_mode |  |  | No |
+| response_mode | string | *Enum:* `"blocking"`, `"streaming"` | No |
 | retriever_from | string |  | No |
 
 #### Condition
@@ -2224,7 +2224,7 @@ Condition detail
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| last_id |  | Last conversation ID for pagination | No |
+| last_id | string | Last conversation ID for pagination | No |
 | limit | integer | Number of conversations to return | No |
 | sort_by | string | Sort order for conversations<br>*Enum:* `"-created_at"`, `"-updated_at"`, `"created_at"`, `"updated_at"` | No |
 
@@ -2233,7 +2233,7 @@ Condition detail
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | auto_generate | boolean |  | No |
-| name |  |  | No |
+| name | string |  | No |
 
 #### ConversationVariableInfiniteScrollPaginationResponse
 
@@ -2247,12 +2247,12 @@ Condition detail
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| created_at |  |  | No |
-| description |  |  | No |
+| created_at | integer |  | No |
+| description | string |  | No |
 | id | string |  | Yes |
 | name | string |  | Yes |
-| updated_at |  |  | No |
-| value |  |  | No |
+| updated_at | integer |  | No |
+| value | string |  | No |
 | value_type | string |  | Yes |
 
 #### ConversationVariableUpdatePayload
@@ -2265,15 +2265,15 @@ Condition detail
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| last_id |  | Last variable ID for pagination | No |
+| last_id | string | Last variable ID for pagination | No |
 | limit | integer | Number of variables to return | No |
-| variable_name |  | Filter variables by name | No |
+| variable_name | string | Filter variables by name | No |
 
 #### DataSetTag
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| binding_count |  |  | No |
+| binding_count | string |  | No |
 | id | string |  | Yes |
 | name | string |  | Yes |
 | type | string |  | Yes |
@@ -2283,23 +2283,23 @@ Condition detail
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | description | string | Dataset description (max 400 chars) | No |
-| embedding_model |  |  | No |
-| embedding_model_provider |  |  | No |
-| external_knowledge_api_id |  |  | No |
-| external_knowledge_id |  |  | No |
-| indexing_technique |  |  | No |
+| embedding_model | string |  | No |
+| embedding_model_provider | string |  | No |
+| external_knowledge_api_id | string |  | No |
+| external_knowledge_id | string |  | No |
+| indexing_technique | string | *Enum:* `"economy"`, `"high_quality"` | No |
 | name | string |  | Yes |
-| permission |  |  | No |
+| permission | [DatasetPermissionEnum](#datasetpermissionenum) |  | No |
 | provider | string |  | No |
-| retrieval_model |  |  | No |
-| summary_index_setting |  |  | No |
+| retrieval_model | [RetrievalModel](#retrievalmodel) |  | No |
+| summary_index_setting | object |  | No |
 
 #### DatasetListQuery
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | include_all | boolean | Include all datasets | No |
-| keyword |  | Search keyword | No |
+| keyword | string | Search keyword | No |
 | limit | integer | Number of items per page | No |
 | page | integer | Page number | No |
 | tag_ids | [ string ] | Filter by tag IDs | No |
@@ -2314,23 +2314,23 @@ Condition detail
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| description |  | Dataset description (max 400 chars) | No |
-| embedding_model |  |  | No |
-| embedding_model_provider |  |  | No |
-| external_knowledge_api_id |  |  | No |
-| external_knowledge_id |  |  | No |
-| external_retrieval_model |  |  | No |
-| indexing_technique |  |  | No |
-| name |  |  | No |
-| partial_member_list |  |  | No |
-| permission |  |  | No |
-| retrieval_model |  |  | No |
+| description | string | Dataset description (max 400 chars) | No |
+| embedding_model | string |  | No |
+| embedding_model_provider | string |  | No |
+| external_knowledge_api_id | string |  | No |
+| external_knowledge_id | string |  | No |
+| external_retrieval_model | object |  | No |
+| indexing_technique | string | *Enum:* `"economy"`, `"high_quality"` | No |
+| name | string |  | No |
+| partial_member_list | [ object ] |  | No |
+| permission | [DatasetPermissionEnum](#datasetpermissionenum) |  | No |
+| retrieval_model | [RetrievalModel](#retrievalmodel) |  | No |
 
 #### DatasourceNodeRunPayload
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| credential_id |  |  | No |
+| credential_id | string |  | No |
 | datasource_type | string |  | Yes |
 | inputs | object |  | Yes |
 | is_published | boolean |  | Yes |
@@ -2347,10 +2347,10 @@ Request payload for bulk downloading documents as a zip archive.
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| keyword |  | Search keyword | No |
+| keyword | string | Search keyword | No |
 | limit | integer | Number of items per page | No |
 | page | integer | Page number | No |
-| status |  | Document status filter | No |
+| status | string | Document status filter | No |
 
 #### DocumentMetadataOperation
 
@@ -2366,13 +2366,13 @@ Request payload for bulk downloading documents as a zip archive.
 | ---- | ---- | ----------- | -------- |
 | doc_form | string |  | No |
 | doc_language | string |  | No |
-| embedding_model |  |  | No |
-| embedding_model_provider |  |  | No |
-| indexing_technique |  |  | No |
+| embedding_model | string |  | No |
+| embedding_model_provider | string |  | No |
+| indexing_technique | string |  | No |
 | name | string |  | Yes |
-| original_document_id |  |  | No |
-| process_rule |  |  | No |
-| retrieval_model |  |  | No |
+| original_document_id | string |  | No |
+| process_rule | [ProcessRule](#processrule) |  | No |
+| retrieval_model | [RetrievalModel](#retrievalmodel) |  | No |
 | text | string |  | Yes |
 
 #### DocumentTextUpdate
@@ -2381,10 +2381,10 @@ Request payload for bulk downloading documents as a zip archive.
 | ---- | ---- | ----------- | -------- |
 | doc_form | string |  | No |
 | doc_language | string |  | No |
-| name |  |  | No |
-| process_rule |  |  | No |
-| retrieval_model |  |  | No |
-| text |  |  | No |
+| name | string |  | No |
+| process_rule | [ProcessRule](#processrule) |  | No |
+| retrieval_model | [RetrievalModel](#retrievalmodel) |  | No |
+| text | string |  | No |
 
 #### FeedbackListQuery
 
@@ -2403,29 +2403,29 @@ Request payload for bulk downloading documents as a zip archive.
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| conversation_id |  |  | No |
-| created_at |  |  | No |
-| created_by |  |  | No |
-| extension |  |  | No |
-| file_key |  |  | No |
+| conversation_id | string |  | No |
+| created_at | integer |  | No |
+| created_by | string |  | No |
+| extension | string |  | No |
+| file_key | string |  | No |
 | id | string |  | Yes |
-| mime_type |  |  | No |
+| mime_type | string |  | No |
 | name | string |  | Yes |
-| original_url |  |  | No |
-| preview_url |  |  | No |
+| original_url | string |  | No |
+| preview_url | string |  | No |
 | size | integer |  | Yes |
-| source_url |  |  | No |
-| tenant_id |  |  | No |
-| user_id |  |  | No |
+| source_url | string |  | No |
+| tenant_id | string |  | No |
+| user_id | string |  | No |
 
 #### HitTestingPayload
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| attachment_ids |  |  | No |
-| external_retrieval_model |  |  | No |
+| attachment_ids | [ string ] |  | No |
+| external_retrieval_model | object |  | No |
 | query | string |  | Yes |
-| retrieval_model |  |  | No |
+| retrieval_model | [RetrievalModel](#retrievalmodel) |  | No |
 
 #### HumanInputFormSubmitPayload
 
@@ -2444,15 +2444,15 @@ Request payload for bulk downloading documents as a zip archive.
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| content |  |  | No |
-| rating |  |  | No |
+| content | string |  | No |
+| rating | string | *Enum:* `"dislike"`, `"like"` | No |
 
 #### MessageListQuery
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | conversation_id | string | Conversation UUID | Yes |
-| first_id |  | First message ID for pagination | No |
+| first_id | string | First message ID for pagination | No |
 | limit | integer | Number of messages to return (1-100) | No |
 
 #### MetadataArgs
@@ -2476,8 +2476,8 @@ Metadata Filtering Condition.
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| conditions |  |  | No |
-| logical_operator |  |  | No |
+| conditions | [ [Condition](#condition) ] |  | No |
+| logical_operator | string | *Enum:* `"and"`, `"or"` | No |
 
 #### MetadataOperationData
 
@@ -2516,14 +2516,14 @@ Metadata operation data
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | mode | string | *Enum:* `"automatic"`, `"custom"`, `"hierarchical"` | Yes |
-| rules |  |  | No |
+| rules | [Rule](#rule) |  | No |
 
 #### RerankingModel
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| reranking_model_name |  |  | No |
-| reranking_provider_name |  |  | No |
+| reranking_model_name | string |  | No |
+| reranking_provider_name | string |  | No |
 
 #### RetrievalMethod
 
@@ -2535,49 +2535,49 @@ Metadata operation data
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| metadata_filtering_conditions |  |  | No |
+| metadata_filtering_conditions | [MetadataFilteringCondition](#metadatafilteringcondition) |  | No |
 | reranking_enable | boolean |  | Yes |
-| reranking_mode |  |  | No |
-| reranking_model |  |  | No |
-| score_threshold |  |  | No |
+| reranking_mode | string |  | No |
+| reranking_model | [RerankingModel](#rerankingmodel) |  | No |
+| score_threshold | number |  | No |
 | score_threshold_enabled | boolean |  | Yes |
 | search_method | [RetrievalMethod](#retrievalmethod) |  | Yes |
 | top_k | integer |  | Yes |
-| weights |  |  | No |
+| weights | [WeightModel](#weightmodel) |  | No |
 
 #### Rule
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| parent_mode |  |  | No |
-| pre_processing_rules |  |  | No |
-| segmentation |  |  | No |
-| subchunk_segmentation |  |  | No |
+| parent_mode | string | *Enum:* `"full-doc"`, `"paragraph"` | No |
+| pre_processing_rules | [ [PreProcessingRule](#preprocessingrule) ] |  | No |
+| segmentation | [Segmentation](#segmentation) |  | No |
+| subchunk_segmentation | [Segmentation](#segmentation) |  | No |
 
 #### SegmentCreatePayload
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| segments |  |  | No |
+| segments | [ object ] |  | No |
 
 #### SegmentListQuery
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| keyword |  |  | No |
+| keyword | string |  | No |
 | status | [ string ] |  | No |
 
 #### SegmentUpdateArgs
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| answer |  |  | No |
-| attachment_ids |  |  | No |
-| content |  |  | No |
-| enabled |  |  | No |
-| keywords |  |  | No |
+| answer | string |  | No |
+| attachment_ids | [ string ] |  | No |
+| content | string |  | No |
+| enabled | boolean |  | No |
+| keywords | [ string ] |  | No |
 | regenerate_child_chunks | boolean |  | No |
-| summary |  |  | No |
+| summary | string |  | No |
 
 #### SegmentUpdatePayload
 
@@ -2607,7 +2607,7 @@ Metadata operation data
 | ---- | ---- | ----------- | -------- |
 | id | string |  | Yes |
 | is_anonymous | boolean |  | Yes |
-| session_id |  |  | No |
+| session_id | string |  | No |
 | type | string |  | Yes |
 
 #### TagBindingPayload
@@ -2635,7 +2635,7 @@ Accept the legacy single-tag Service API payload while exposing a normalized tag
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| tag_id |  |  | No |
+| tag_id | string |  | No |
 | tag_ids | [ string ] |  | No |
 | target_id | string |  | Yes |
 
@@ -2650,10 +2650,10 @@ Accept the legacy single-tag Service API payload while exposing a normalized tag
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| message_id |  | Message ID | No |
-| streaming |  | Enable streaming response | No |
-| text |  | Text to convert to audio | No |
-| voice |  | Voice to use for TTS | No |
+| message_id | string | Message ID | No |
+| streaming | boolean | Enable streaming response | No |
+| text | string | Text to convert to audio | No |
+| voice | string | Voice to use for TTS | No |
 
 #### WeightKeywordSetting
 
@@ -2665,9 +2665,9 @@ Accept the legacy single-tag Service API payload while exposing a normalized tag
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| keyword_setting |  |  | No |
-| vector_setting |  |  | No |
-| weight_type |  |  | No |
+| keyword_setting | [WeightKeywordSetting](#weightkeywordsetting) |  | No |
+| vector_setting | [WeightVectorSetting](#weightvectorsetting) |  | No |
+| weight_type | string | *Enum:* `"customized"`, `"keyword_first"`, `"semantic_first"` | No |
 
 #### WeightVectorSetting
 
@@ -2691,64 +2691,64 @@ Accept the legacy single-tag Service API payload while exposing a normalized tag
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| created_at |  |  | No |
-| created_by_account |  |  | No |
-| created_by_end_user |  |  | No |
-| created_by_role |  |  | No |
-| created_from |  |  | No |
+| created_at | integer |  | No |
+| created_by_account | [SimpleAccount](#simpleaccount) |  | No |
+| created_by_end_user | [SimpleEndUser](#simpleenduser) |  | No |
+| created_by_role | string |  | No |
+| created_from | string |  | No |
 | details |  |  | No |
 | id | string |  | Yes |
-| workflow_run |  |  | No |
+| workflow_run | [WorkflowRunForLogResponse](#workflowrunforlogresponse) |  | No |
 
 #### WorkflowLogQuery
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| created_at__after |  |  | No |
-| created_at__before |  |  | No |
-| created_by_account |  |  | No |
-| created_by_end_user_session_id |  |  | No |
-| keyword |  |  | No |
+| created_at__after | string |  | No |
+| created_at__before | string |  | No |
+| created_by_account | string |  | No |
+| created_by_end_user_session_id | string |  | No |
+| keyword | string |  | No |
 | limit | integer |  | No |
 | page | integer |  | No |
-| status |  |  | No |
+| status | string | *Enum:* `"failed"`, `"stopped"`, `"succeeded"` | No |
 
 #### WorkflowRunForLogResponse
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| created_at |  |  | No |
+| created_at | integer |  | No |
 | elapsed_time |  |  | No |
-| error |  |  | No |
-| exceptions_count |  |  | No |
-| finished_at |  |  | No |
+| error | string |  | No |
+| exceptions_count | integer |  | No |
+| finished_at | integer |  | No |
 | id | string |  | Yes |
-| status |  |  | No |
-| total_steps |  |  | No |
-| total_tokens |  |  | No |
-| triggered_from |  |  | No |
-| version |  |  | No |
+| status | string |  | No |
+| total_steps | integer |  | No |
+| total_tokens | integer |  | No |
+| triggered_from | string |  | No |
+| version | string |  | No |
 
 #### WorkflowRunPayload
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| files |  |  | No |
+| files | [ object ] |  | No |
 | inputs | object |  | Yes |
-| response_mode |  |  | No |
+| response_mode | string | *Enum:* `"blocking"`, `"streaming"` | No |
 
 #### WorkflowRunResponse
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| created_at |  |  | No |
+| created_at | integer |  | No |
 | elapsed_time |  |  | No |
-| error |  |  | No |
-| finished_at |  |  | No |
+| error | string |  | No |
+| finished_at | integer |  | No |
 | id | string |  | Yes |
 | inputs |  |  | No |
 | outputs | object |  | No |
 | status | string |  | Yes |
-| total_steps |  |  | No |
-| total_tokens |  |  | No |
+| total_steps | integer |  | No |
+| total_tokens | integer |  | No |
 | workflow_id | string |  | Yes |

api/openapi/markdown/web-swagger.md

@@ -1042,38 +1042,38 @@ Returns Server-Sent Events stream.
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| appCode |  | Application code | No |
-| appId |  | Application ID | No |
+| appCode | string | Application code | No |
+| appId | string | Application ID | No |
 
 #### ChatMessagePayload
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| conversation_id |  | Conversation ID | No |
-| files |  | Files to be processed | No |
+| conversation_id | string | Conversation ID | No |
+| files | [ object ] | Files to be processed | No |
 | inputs | object | Input variables for the chat | Yes |
-| parent_message_id |  | Parent message ID | No |
+| parent_message_id | string | Parent message ID | No |
 | query | string | User query/message | Yes |
-| response_mode |  | Response mode: blocking or streaming | No |
+| response_mode | string | Response mode: blocking or streaming<br>*Enum:* `"blocking"`, `"streaming"` | No |
 | retriever_from | string | Source of retriever | No |
 
 #### CompletionMessagePayload
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| files |  | Files to be processed | No |
+| files | [ object ] | Files to be processed | No |
 | inputs | object | Input variables for the completion | Yes |
 | query | string | Query text for completion | No |
-| response_mode |  | Response mode: blocking or streaming | No |
+| response_mode | string | Response mode: blocking or streaming<br>*Enum:* `"blocking"`, `"streaming"` | No |
 | retriever_from | string | Source of retriever | No |
 
 #### ConversationListQuery
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| last_id |  |  | No |
+| last_id | string |  | No |
 | limit | integer |  | No |
-| pinned |  |  | No |
+| pinned | boolean |  | No |
 | sort_by | string | *Enum:* `"-created_at"`, `"-updated_at"`, `"created_at"`, `"updated_at"` | No |
 
 #### ConversationRenamePayload
@@ -1081,14 +1081,14 @@ Returns Server-Sent Events stream.
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | auto_generate | boolean |  | No |
-| name |  |  | No |
+| name | string |  | No |
 
 #### EmailCodeLoginSendPayload
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | email | string |  | Yes |
-| language |  |  | No |
+| language | string |  | No |
 
 #### EmailCodeLoginVerifyPayload
 
@@ -1102,33 +1102,33 @@ Returns Server-Sent Events stream.
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| conversation_id |  |  | No |
-| created_at |  |  | No |
-| created_by |  |  | No |
-| extension |  |  | No |
-| file_key |  |  | No |
+| conversation_id | string |  | No |
+| created_at | integer |  | No |
+| created_by | string |  | No |
+| extension | string |  | No |
+| file_key | string |  | No |
 | id | string |  | Yes |
-| mime_type |  |  | No |
+| mime_type | string |  | No |
 | name | string |  | Yes |
-| original_url |  |  | No |
-| preview_url |  |  | No |
+| original_url | string |  | No |
+| preview_url | string |  | No |
 | size | integer |  | Yes |
-| source_url |  |  | No |
-| tenant_id |  |  | No |
-| user_id |  |  | No |
+| source_url | string |  | No |
+| tenant_id | string |  | No |
+| user_id | string |  | No |
 
 #### FileWithSignedUrl
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| created_at |  |  | No |
-| created_by |  |  | No |
-| extension |  |  | No |
+| created_at | integer |  | No |
+| created_by | string |  | No |
+| extension | string |  | No |
 | id | string |  | Yes |
-| mime_type |  |  | No |
+| mime_type | string |  | No |
 | name | string |  | Yes |
 | size | integer |  | Yes |
-| url |  |  | No |
+| url | string |  | No |
 
 #### ForgotPasswordCheckPayload
 
@@ -1151,7 +1151,7 @@ Returns Server-Sent Events stream.
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | email | string |  | Yes |
-| language |  |  | No |
+| language | string |  | No |
 
 #### LoginPayload
 
@@ -1164,15 +1164,15 @@ Returns Server-Sent Events stream.
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| content |  |  | No |
-| rating |  |  | No |
+| content | string |  | No |
+| rating | string | *Enum:* `"dislike"`, `"like"` | No |
 
 #### MessageListQuery
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | conversation_id | string | Conversation UUID | Yes |
-| first_id |  | First message ID for pagination | No |
+| first_id | string | First message ID for pagination | No |
 | limit | integer | Number of messages to return (1-100) | No |
 
 #### MessageMoreLikeThisQuery
@@ -1204,21 +1204,21 @@ Returns Server-Sent Events stream.
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| last_id |  |  | No |
+| last_id | string |  | No |
 | limit | integer |  | No |
 
 #### TextToAudioPayload
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| message_id |  | Message ID | No |
-| streaming |  | Enable streaming response | No |
-| text |  | Text to convert to audio | No |
-| voice |  | Voice to use for TTS | No |
+| message_id | string | Message ID | No |
+| streaming | boolean | Enable streaming response | No |
+| text | string | Text to convert to audio | No |
+| voice | string | Voice to use for TTS | No |
 
 #### WorkflowRunPayload
 
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| files |  |  | No |
+| files | [ object ] |  | No |
 | inputs | object |  | Yes |

api/providers/trace/trace-arize-phoenix/src/dify_trace_arize_phoenix/arize_phoenix_trace.py

@@ -15,6 +15,7 @@ from openinference.semconv.trace import (
     SpanAttributes,
     ToolCallAttributes,
 )
+from opentelemetry.context import Context
 from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter as GrpcOTLPSpanExporter
 from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter as HttpOTLPSpanExporter
 from opentelemetry.sdk import trace as trace_sdk
@@ -45,8 +46,8 @@ from dify_trace_arize_phoenix.config import ArizeConfig, PhoenixConfig
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from graphon.enums import WorkflowNodeExecutionStatus
-from models.model import EndUser, MessageFile
-from models.workflow import WorkflowNodeExecutionTriggeredFrom
+from models.model import App, EndUser, MessageFile
+from models.workflow import WorkflowNodeExecutionTriggeredFrom, WorkflowRun
 
 logger = logging.getLogger(__name__)
 
@@ -139,6 +140,48 @@ def _resolve_published_parent_span_context(parent_node_execution_id: str) -> dic
     return normalized_carrier
 
 
+def _app_uses_phoenix_provider(app_tracing_config: Mapping[str, Any] | None) -> bool:
+    if not app_tracing_config or not app_tracing_config.get("enabled"):
+        return False
+    return app_tracing_config.get("tracing_provider") in {"arize", "phoenix"}
+
+
+def _parent_workflow_can_publish_span_context(parent_workflow_run_id: str) -> bool:
+    parent_run = db.session.query(WorkflowRun).where(WorkflowRun.id == parent_workflow_run_id).first()
+    if parent_run is None:
+        return True
+
+    parent_app = db.session.query(App).where(App.id == parent_run.app_id).first()
+    if parent_app is None or not parent_app.tracing:
+        return False
+
+    try:
+        app_tracing_config = json.loads(parent_app.tracing)
+    except (TypeError, json.JSONDecodeError):
+        return False
+    if not isinstance(app_tracing_config, Mapping):
+        return False
+
+    return _app_uses_phoenix_provider(app_tracing_config)
+
+
+def _resolve_workflow_parent_carrier(
+    parent_node_execution_id: str, parent_workflow_run_id: str | None
+) -> dict[str, str] | None:
+    try:
+        return _resolve_published_parent_span_context(parent_node_execution_id)
+    except PendingTraceParentContextError:
+        if parent_workflow_run_id and not _parent_workflow_can_publish_span_context(parent_workflow_run_id):
+            logger.info(
+                "[Arize/Phoenix] Parent workflow cannot publish Phoenix span context; falling back to root span: "
+                "parent_workflow_run_id=%s parent_node_execution_id=%s",
+                parent_workflow_run_id,
+                parent_node_execution_id,
+            )
+            return None
+        raise
+
+
 def setup_tracer(arize_phoenix_config: ArizeConfig | PhoenixConfig) -> tuple[trace_sdk.Tracer, SimpleSpanProcessor]:
     """Configure OpenTelemetry tracer with OTLP exporter for Arize/Phoenix."""
     try:
@@ -581,9 +624,11 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
             workflow_session_id,
         )
 
+        workflow_parent_carrier: dict[str, str] | None = None
         if parent_node_execution_id:
-            workflow_parent_carrier = _resolve_published_parent_span_context(parent_node_execution_id)
-        else:
+            workflow_parent_carrier = _resolve_workflow_parent_carrier(parent_node_execution_id, parent_workflow_run_id)
+
+        if workflow_parent_carrier is None:
             root_trace_id = _resolve_workflow_root_trace_id(trace_info)
             workflow_root_span_name: str | None = trace_info.workflow_run_id
             if not isinstance(workflow_root_span_name, str) or not workflow_root_span_name.strip():
@@ -1176,7 +1221,7 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
             if root_span_attributes:
                 root_span_attributes_dict.update(root_span_attributes)
 
-            root_span = self.tracer.start_span(name=span_name, attributes=root_span_attributes_dict)
+            root_span = self.tracer.start_span(name=span_name, attributes=root_span_attributes_dict, context=Context())
 
             with use_span(root_span, end_on_exit=False):
                 self.propagator.inject(carrier=carrier)

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

@@ -1,3 +1,5 @@
+import json
+from collections.abc import Sequence
 from datetime import UTC, datetime, timedelta
 from types import SimpleNamespace
 from typing import Any, cast
@@ -8,8 +10,10 @@ import pytest
 from dify_trace_arize_phoenix.arize_phoenix_trace import (
     _NODE_TYPE_TO_SPAN_KIND,
     ArizePhoenixDataTrace,
+    _app_uses_phoenix_provider,
     _build_graph_parent_index,
     _get_node_span_kind,
+    _parent_workflow_can_publish_span_context,
     _phoenix_parent_span_redis_key,
     _resolve_node_parent,
     _resolve_published_parent_span_context,
@@ -25,9 +29,13 @@ from dify_trace_arize_phoenix.arize_phoenix_trace import (
 )
 from dify_trace_arize_phoenix.config import ArizeConfig, PhoenixConfig
 from openinference.semconv.trace import OpenInferenceSpanKindValues, SpanAttributes
-from opentelemetry.sdk.trace import Tracer
+from opentelemetry.context import Context
+from opentelemetry.sdk import trace as trace_sdk
+from opentelemetry.sdk.trace import ReadableSpan, Tracer
+from opentelemetry.sdk.trace.export import SimpleSpanProcessor, SpanExporter, SpanExportResult
 from opentelemetry.semconv.trace import SpanAttributes as OTELSpanAttributes
-from opentelemetry.trace import StatusCode
+from opentelemetry.trace import NonRecordingSpan, SpanContext, StatusCode, TraceFlags, TraceState, use_span
+from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
 
 from core.ops.entities.trace_entity import (
     DatasetRetrievalTraceInfo,
@@ -96,6 +104,32 @@ def _get_start_span_call(start_span_mock, *, span_name: str):
     raise AssertionError(f"Could not find start_span call with name={span_name!r}")
 
 
+class _FakeQuery:
+    def __init__(self, result):
+        self._result = result
+
+    def filter(self, *args, **kwargs):
+        return self
+
+    def where(self, *args, **kwargs):
+        return self
+
+    def first(self):
+        return self._result
+
+
+class _CollectingSpanExporter(SpanExporter):
+    def __init__(self):
+        self.spans: list[ReadableSpan] = []
+
+    def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
+        self.spans.extend(spans)
+        return SpanExportResult.SUCCESS
+
+    def shutdown(self) -> None:
+        return None
+
+
 def _make_node_execution(**kwargs):
     defaults = {
         "node_type": "tool",
@@ -233,6 +267,45 @@ def test_wrap_span_metadata():
     assert res == {"a": 1, "b": 2, "created_from": "Dify"}
 
 
+def test_app_uses_phoenix_provider_only_for_enabled_arize_or_phoenix():
+    assert _app_uses_phoenix_provider({"enabled": True, "tracing_provider": "phoenix"}) is True
+    assert _app_uses_phoenix_provider({"enabled": True, "tracing_provider": "arize"}) is True
+    assert _app_uses_phoenix_provider({"enabled": False, "tracing_provider": "phoenix"}) is False
+    assert _app_uses_phoenix_provider({"enabled": True, "tracing_provider": "langfuse"}) is False
+    assert _app_uses_phoenix_provider(None) is False
+
+
+def test_parent_workflow_can_publish_span_context_keeps_unknown_parent_retryable(monkeypatch):
+    monkeypatch.setattr(
+        "dify_trace_arize_phoenix.arize_phoenix_trace.db.session.query",
+        lambda model: _FakeQuery(None),
+    )
+
+    assert _parent_workflow_can_publish_span_context("missing-run") is True
+
+
+def test_parent_workflow_can_publish_span_context_checks_parent_app_tracing(monkeypatch):
+    parent_run = SimpleNamespace(app_id="parent-app")
+    parent_app = SimpleNamespace(tracing=json.dumps({"enabled": True, "tracing_provider": "phoenix"}))
+
+    def fake_query(model):
+        if getattr(model, "__tablename__", None) == "workflow_runs":
+            return _FakeQuery(parent_run)
+        if getattr(model, "__tablename__", None) == "apps":
+            return _FakeQuery(parent_app)
+        raise AssertionError(f"Unexpected model query: {model}")
+
+    monkeypatch.setattr("dify_trace_arize_phoenix.arize_phoenix_trace.db.session.query", fake_query)
+
+    assert _parent_workflow_can_publish_span_context("parent-run") is True
+
+    parent_app.tracing = json.dumps({"enabled": False, "tracing_provider": "phoenix"})
+    assert _parent_workflow_can_publish_span_context("parent-run") is False
+
+    parent_app.tracing = json.dumps({"enabled": True, "tracing_provider": "langfuse"})
+    assert _parent_workflow_can_publish_span_context("parent-run") is False
+
+
 class TestGetNodeSpanKind:
     def test_all_node_types_are_mapped_correctly(self):
         special_mappings = {
@@ -839,6 +912,10 @@ def test_workflow_trace_raises_pending_parent_error_when_parent_node_context_is_
 
     with (
         patch.object(trace_instance, "get_service_account_with_tenant", return_value=MagicMock()),
+        patch(
+            "dify_trace_arize_phoenix.arize_phoenix_trace._parent_workflow_can_publish_span_context",
+            return_value=True,
+        ),
         patch.object(trace_instance, "ensure_root_span") as mock_ensure_root_span,
         pytest.raises(PendingTraceParentContextError) as exc_info,
     ):
@@ -851,6 +928,102 @@ def test_workflow_trace_raises_pending_parent_error_when_parent_node_context_is_
     mock_ensure_root_span.assert_not_called()
 
 
+@patch("dify_trace_arize_phoenix.arize_phoenix_trace.db")
+@patch("dify_trace_arize_phoenix.arize_phoenix_trace.DifyCoreRepositoryFactory")
+@patch("dify_trace_arize_phoenix.arize_phoenix_trace.sessionmaker")
+def test_workflow_trace_falls_back_when_parent_app_tracing_cannot_publish_parent_context(
+    mock_sessionmaker,
+    mock_repo_factory,
+    mock_db,
+    trace_instance,
+):
+    mock_db.engine = MagicMock()
+    info = _make_workflow_info(
+        message_id="message-1",
+        workflow_run_id="workflow-run-1",
+        metadata={
+            "app_id": "app1",
+            "parent_trace_context": {
+                "parent_workflow_run_id": "outer-workflow-run-1",
+                "parent_node_execution_id": "outer-node-execution-1",
+            },
+        },
+    )
+    repo = MagicMock()
+    repo.get_by_workflow_execution.return_value = []
+    mock_repo_factory.create_workflow_node_execution_repository.return_value = repo
+    trace_instance._mock_redis_client.get.return_value = None
+
+    parent_carrier = {}
+    parent_context = object()
+
+    with (
+        patch.object(trace_instance, "get_service_account_with_tenant", return_value=MagicMock()),
+        patch(
+            "dify_trace_arize_phoenix.arize_phoenix_trace._parent_workflow_can_publish_span_context",
+            return_value=False,
+        ),
+        patch.object(trace_instance, "ensure_root_span", return_value=parent_carrier) as mock_ensure_root_span,
+        patch.object(trace_instance.propagator, "extract", return_value=parent_context) as mock_extract,
+    ):
+        trace_instance.workflow_trace(info)
+
+    mock_ensure_root_span.assert_called_once_with(
+        "outer-workflow-run-1",
+        root_span_name="workflow-run-1",
+        root_span_attributes={
+            SpanAttributes.INPUT_VALUE: safe_json_dumps(info.workflow_run_inputs),
+            SpanAttributes.INPUT_MIME_TYPE: "application/json",
+            SpanAttributes.OUTPUT_VALUE: safe_json_dumps(info.workflow_run_outputs),
+            SpanAttributes.OUTPUT_MIME_TYPE: "application/json",
+        },
+    )
+    mock_extract.assert_called_once_with(carrier=parent_carrier)
+    workflow_span_call = _get_start_span_call(trace_instance.tracer.start_span, span_name="workflow_workflow-run-1")
+    assert workflow_span_call.kwargs["context"] is parent_context
+
+
+@patch("dify_trace_arize_phoenix.arize_phoenix_trace.db")
+@patch("dify_trace_arize_phoenix.arize_phoenix_trace.DifyCoreRepositoryFactory")
+@patch("dify_trace_arize_phoenix.arize_phoenix_trace.sessionmaker")
+def test_workflow_trace_still_retries_when_parent_app_can_publish_parent_context(
+    mock_sessionmaker,
+    mock_repo_factory,
+    mock_db,
+    trace_instance,
+):
+    mock_db.engine = MagicMock()
+    info = _make_workflow_info(
+        message_id="message-1",
+        workflow_run_id="workflow-run-1",
+        metadata={
+            "app_id": "app1",
+            "parent_trace_context": {
+                "parent_workflow_run_id": "outer-workflow-run-1",
+                "parent_node_execution_id": "outer-node-execution-1",
+            },
+        },
+    )
+    repo = MagicMock()
+    repo.get_by_workflow_execution.return_value = []
+    mock_repo_factory.create_workflow_node_execution_repository.return_value = repo
+    trace_instance._mock_redis_client.get.return_value = None
+
+    with (
+        patch.object(trace_instance, "get_service_account_with_tenant", return_value=MagicMock()),
+        patch(
+            "dify_trace_arize_phoenix.arize_phoenix_trace._parent_workflow_can_publish_span_context",
+            return_value=True,
+        ),
+        patch.object(trace_instance, "ensure_root_span") as mock_ensure_root_span,
+        pytest.raises(PendingTraceParentContextError) as exc_info,
+    ):
+        trace_instance.workflow_trace(info)
+
+    assert exc_info.value.parent_node_execution_id == "outer-node-execution-1"
+    mock_ensure_root_span.assert_not_called()
+
+
 @patch("dify_trace_arize_phoenix.arize_phoenix_trace.db")
 @patch("dify_trace_arize_phoenix.arize_phoenix_trace.DifyCoreRepositoryFactory")
 @patch("dify_trace_arize_phoenix.arize_phoenix_trace.sessionmaker")
@@ -1544,6 +1717,38 @@ def test_ensure_root_span_basic(trace_instance):
     assert "tid" in trace_instance.dify_trace_ids
 
 
+def test_ensure_root_span_ignores_unsampled_ambient_otel_parent():
+    exporter = _CollectingSpanExporter()
+    provider = trace_sdk.TracerProvider()
+    provider.add_span_processor(SimpleSpanProcessor(exporter))
+    trace_instance = ArizePhoenixDataTrace.__new__(ArizePhoenixDataTrace)
+    trace_instance.tracer = cast(Tracer, provider.get_tracer("test-phoenix-root-span"))
+    trace_instance.propagator = TraceContextTextMapPropagator()
+    trace_instance.project = "p"
+    trace_instance.dify_trace_ids = set()
+    trace_instance.root_span_carriers = {}
+    trace_instance.carrier = {}
+
+    ambient_span_context = SpanContext(
+        trace_id=0x11111111111111111111111111111111,
+        span_id=0x2222222222222222,
+        is_remote=True,
+        trace_flags=TraceFlags(0),
+        trace_state=TraceState(),
+    )
+
+    with use_span(NonRecordingSpan(ambient_span_context), end_on_exit=False):
+        carrier = trace_instance.ensure_root_span("tid")
+
+    assert len(exporter.spans) == 1
+    root_span = exporter.spans[0]
+    root_span_context = root_span.get_span_context()
+    assert root_span_context is not None
+    assert root_span.parent is None
+    assert root_span_context.trace_id != ambient_span_context.trace_id
+    assert carrier["traceparent"].split("-")[1] == f"{root_span_context.trace_id:032x}"
+
+
 def test_ensure_root_span_uses_custom_name_and_attributes(trace_instance):
     root_attributes = {
         SpanAttributes.INPUT_VALUE: '{"input":"value"}',
@@ -1561,6 +1766,7 @@ def test_ensure_root_span_uses_custom_name_and_attributes(trace_instance):
             SpanAttributes.INPUT_VALUE: '{"input":"value"}',
             SpanAttributes.OUTPUT_VALUE: '{"output":"value"}',
         },
+        context=Context(),
     )
 
 

api/providers/trace/trace-mlflow/src/dify_trace_mlflow/mlflow_trace.py

@@ -89,20 +89,21 @@ class MLflowDataTrace(BaseTraceInstance):
     def trace(self, trace_info: BaseTraceInfo):
         """Simple dispatch to trace methods"""
         try:
-            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, ModerationTraceInfo):
-                self.moderation_trace(trace_info)
-            elif isinstance(trace_info, DatasetRetrievalTraceInfo):
-                self.dataset_retrieval_trace(trace_info)
-            elif isinstance(trace_info, SuggestedQuestionTraceInfo):
-                self.suggested_question_trace(trace_info)
-            elif isinstance(trace_info, GenerateNameTraceInfo):
-                self.generate_name_trace(trace_info)
+            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 ModerationTraceInfo():
+                    self.moderation_trace(trace_info)
+                case DatasetRetrievalTraceInfo():
+                    self.dataset_retrieval_trace(trace_info)
+                case SuggestedQuestionTraceInfo():
+                    self.suggested_question_trace(trace_info)
+                case GenerateNameTraceInfo():
+                    self.generate_name_trace(trace_info)
         except Exception:
             logger.exception("[MLflow] Trace error")
             raise
@@ -480,14 +481,15 @@ class MLflowDataTrace(BaseTraceInstance):
 
     def _parse_prompts(self, prompts):
         """Postprocess prompts format to be standard chat messages"""
-        if isinstance(prompts, str):
-            return prompts
-        elif isinstance(prompts, dict):
-            return self._parse_single_message(prompts)
-        elif isinstance(prompts, list):
-            messages = [self._parse_single_message(item) for item in prompts]
-            messages = self._resolve_tool_call_ids(messages)
-            return messages
+        match prompts:
+            case str():
+                return prompts
+            case dict():
+                return self._parse_single_message(prompts)
+            case list():
+                messages = [self._parse_single_message(item) for item in prompts]
+                messages = self._resolve_tool_call_ids(messages)
+                return messages
         return prompts  # Fallback to original format
 
     def _parse_single_message(self, item: dict[str, Any]):

api/providers/trace/trace-tencent/src/dify_trace_tencent/tencent_trace.py

@@ -58,20 +58,21 @@ class TencentDataTrace(BaseTraceInstance):
 
     def trace(self, trace_info: BaseTraceInfo) -> None:
         """Main tracing entry point - coordinates different trace types."""
-        if isinstance(trace_info, WorkflowTraceInfo):
-            self.workflow_trace(trace_info)
-        elif isinstance(trace_info, MessageTraceInfo):
-            self.message_trace(trace_info)
-        elif isinstance(trace_info, ModerationTraceInfo):
-            pass
-        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, ToolTraceInfo):
-            self.tool_trace(trace_info)
-        elif isinstance(trace_info, GenerateNameTraceInfo):
-            pass
+        match trace_info:
+            case WorkflowTraceInfo():
+                self.workflow_trace(trace_info)
+            case MessageTraceInfo():
+                self.message_trace(trace_info)
+            case ModerationTraceInfo():
+                pass
+            case SuggestedQuestionTraceInfo():
+                self.suggested_question_trace(trace_info)
+            case DatasetRetrievalTraceInfo():
+                self.dataset_retrieval_trace(trace_info)
+            case ToolTraceInfo():
+                self.tool_trace(trace_info)
+            case GenerateNameTraceInfo():
+                pass
 
     def api_check(self) -> bool:
         return self.trace_client.api_check()

api/pyproject.toml

@@ -42,9 +42,8 @@ dependencies = [
     "readabilipy>=0.3.0,<1.0.0",
     "resend>=2.27.0,<3.0.0",
     # Emerging: newer and fast-moving, use compatible pins
-    "dify-agent",
     "fastopenapi[flask]~=0.7.0",
-    "graphon~=0.3.1",
+    "graphon~=0.4.0",
     "httpx-sse~=0.4.0",
     "json-repair~=0.59.4",
 ]
@@ -115,6 +114,7 @@ override-dependencies = [
 ############################################################
 dev = [
     "coverage>=7.13.4",
+    "dify-agent",
     "dotenv-linter>=0.7.0",
     "faker>=40.15.0",
     "lxml-stubs>=0.5.1",

api/schedule/trigger_provider_refresh_task.py

@@ -1,6 +1,5 @@
 import logging
 import math
-import time
 from collections.abc import Iterable, Sequence
 
 from celery import group
@@ -13,16 +12,13 @@ from configs import dify_config
 from core.trigger.utils.locks import build_trigger_refresh_lock_keys
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
+from libs.helper import current_timestamp
 from models.trigger import TriggerSubscription
 from tasks.trigger_subscription_refresh_tasks import trigger_subscription_refresh
 
 logger = logging.getLogger(__name__)
 
 
-def _now_ts() -> int:
-    return int(time.time())
-
-
 def _build_due_filter(now_ts: int):
     """Build SQLAlchemy filter for due credential or subscription refresh."""
     credential_due: ColumnElement[bool] = and_(
@@ -54,7 +50,7 @@ def trigger_provider_refresh() -> None:
     """
     Scan due trigger subscriptions and enqueue refresh tasks with in-flight locks.
     """
-    now: int = _now_ts()
+    now: int = current_timestamp()
 
     batch_size: int = int(dify_config.TRIGGER_PROVIDER_REFRESH_BATCH_SIZE)
     lock_ttl: int = max(300, int(dify_config.TRIGGER_PROVIDER_SUBSCRIPTION_THRESHOLD_SECONDS))

api/services/account_service.py

@@ -29,6 +29,7 @@ from extensions.ext_database import db
 from extensions.ext_redis import redis_client, redis_fallback
 from libs.datetime_utils import naive_utc_now
 from libs.helper import RateLimiter, TokenManager
+from libs.helper import timezone as validate_timezone
 from libs.passport import PassportService
 from libs.password import compare_password, hash_password, valid_password
 from libs.rsa import generate_key_pair
@@ -271,8 +272,9 @@ class AccountService:
         password: str | None = None,
         interface_theme: str = "light",
         is_setup: bool | None = False,
+        timezone: str | None = None,
     ) -> Account:
-        """create account"""
+        """Create an account, preferring explicit user timezone over language-derived defaults."""
         if not FeatureService.get_system_features().is_allow_register and not is_setup:
             from controllers.console.error import AccountNotFound
 
@@ -302,6 +304,10 @@ class AccountService:
             password_to_set = base64_password_hashed
             salt_to_set = base64_salt
 
+        resolved_timezone = language_timezone_mapping.get(interface_language, "UTC")
+        if timezone is not None:
+            resolved_timezone = validate_timezone(timezone)
+
         account = Account(
             name=name,
             email=email,
@@ -309,7 +315,7 @@ class AccountService:
             password_salt=salt_to_set,
             interface_language=interface_language,
             interface_theme=interface_theme,
-            timezone=language_timezone_mapping.get(interface_language, "UTC"),
+            timezone=resolved_timezone,
         )
 
         db.session.add(account)
@@ -318,11 +324,15 @@ class AccountService:
 
     @staticmethod
     def create_account_and_tenant(
-        email: str, name: str, interface_language: str, password: str | None = None
+        email: str, name: str, interface_language: str, password: str | None = None, timezone: str | None = None
     ) -> Account:
-        """create account"""
+        """Create an account and owner workspace."""
         account = AccountService.create_account(
-            email=email, name=name, interface_language=interface_language, password=password
+            email=email,
+            name=name,
+            interface_language=interface_language,
+            password=password,
+            timezone=timezone,
         )
 
         try:
@@ -1474,8 +1484,8 @@ class RegisterService:
     @classmethod
     def register(
         cls,
-        email,
-        name,
+        email: str,
+        name: str,
         password: str | None = None,
         open_id: str | None = None,
         provider: str | None = None,
@@ -1483,16 +1493,19 @@ class RegisterService:
         status: AccountStatus | None = None,
         is_setup: bool | None = False,
         create_workspace_required: bool | None = True,
+        timezone: str | None = None,
     ) -> Account:
-        db.session.begin_nested()
         """Register account"""
+        db.session.begin_nested()
         try:
+            interface_language = get_valid_language(language)
             account = AccountService.create_account(
                 email=email,
                 name=name,
-                interface_language=get_valid_language(language),
+                interface_language=interface_language,
                 password=password,
                 is_setup=is_setup,
+                timezone=timezone,
             )
             account.status = status or AccountStatus.ACTIVE
             account.initialized_at = naive_utc_now()

api/services/app_dsl_service.py

@@ -10,7 +10,6 @@ from uuid import uuid4
 import yaml
 from Crypto.Cipher import AES
 from Crypto.Util.Padding import pad, unpad
-from packaging import version
 from packaging.version import parse as parse_version
 from pydantic import BaseModel
 from sqlalchemy import select
@@ -40,6 +39,7 @@ from libs.datetime_utils import naive_utc_now
 from models import Account, App, AppMode
 from models.model import AppModelConfig, AppModelConfigDict, IconType
 from models.workflow import Workflow
+from services.dsl_version import check_version_compatibility
 from services.entities.dsl_entities import CheckDependenciesResult, ImportMode, ImportStatus
 from services.plugin.dependencies_analysis import DependenciesAnalysisService
 from services.workflow_draft_variable_service import WorkflowDraftVariableService
@@ -64,30 +64,6 @@ class Import(BaseModel):
     error: str = ""
 
 
-def _check_version_compatibility(imported_version: str) -> ImportStatus:
-    """Determine import status based on version comparison"""
-    try:
-        current_ver = version.parse(CURRENT_DSL_VERSION)
-        imported_ver = version.parse(imported_version)
-    except version.InvalidVersion:
-        return ImportStatus.FAILED
-
-    # If imported version is newer than current, always return PENDING
-    if imported_ver > current_ver:
-        return ImportStatus.PENDING
-
-    # If imported version is older than current's major, return PENDING
-    if imported_ver.major < current_ver.major:
-        return ImportStatus.PENDING
-
-    # If imported version is older than current's minor, return COMPLETED_WITH_WARNINGS
-    if imported_ver.minor < current_ver.minor:
-        return ImportStatus.COMPLETED_WITH_WARNINGS
-
-    # If imported version equals or is older than current's micro, return COMPLETED
-    return ImportStatus.COMPLETED
-
-
 class PendingData(BaseModel):
     import_mode: str
     yaml_content: str
@@ -203,7 +179,7 @@ class AppDslService:
             # check if imported_version is a float-like string
             if not isinstance(imported_version, str):
                 raise ValueError(f"Invalid version type, expected str, got {type(imported_version)}")
-            status = _check_version_compatibility(imported_version)
+            status = check_version_compatibility(imported_version, CURRENT_DSL_VERSION)
 
             # Extract app data
             app_data = data.get("app")

api/services/attachment_service.py

@@ -22,10 +22,11 @@ class AttachmentService:
             raise AssertionError("must be a sessionmaker or an Engine.")
 
     def get_file_base64(self, file_id: str) -> str:
-        upload_file = self._session_maker(expire_on_commit=False).scalar(
-            select(UploadFile).where(UploadFile.id == file_id).limit(1)
-        )
-        if not upload_file:
-            raise NotFound("File not found")
-        blob = storage.load_once(upload_file.key)
+        with self._session_maker(expire_on_commit=False) as session:
+            upload_file = session.scalar(select(UploadFile).where(UploadFile.id == file_id).limit(1))
+            if not upload_file:
+                raise NotFound("File not found")
+            upload_file_key = upload_file.key
+
+        blob = storage.load_once(upload_file_key)
         return base64.b64encode(blob).decode()

api/services/dsl_version.py

@@ -0,0 +1,20 @@
+from packaging import version
+
+from services.entities.dsl_entities import ImportStatus
+
+
+def check_version_compatibility(imported_version: str, current_version: str) -> ImportStatus:
+    """Determine DSL import status based on imported and current versions."""
+    try:
+        current_ver = version.parse(current_version)
+        imported_ver = version.parse(imported_version)
+    except version.InvalidVersion:
+        return ImportStatus.FAILED
+
+    if imported_ver > current_ver:
+        return ImportStatus.PENDING
+    if imported_ver.major < current_ver.major:
+        return ImportStatus.PENDING
+    if imported_ver.minor < current_ver.minor:
+        return ImportStatus.COMPLETED_WITH_WARNINGS
+    return ImportStatus.COMPLETED

api/services/file_service.py

@@ -131,12 +131,13 @@ class FileService:
         return file_size <= file_size_limit
 
     def get_file_base64(self, file_id: str) -> str:
-        upload_file = self._session_maker(expire_on_commit=False).scalar(
-            select(UploadFile).where(UploadFile.id == file_id).limit(1)
-        )
-        if not upload_file:
-            raise NotFound("File not found")
-        blob = storage.load_once(upload_file.key)
+        with self._session_maker(expire_on_commit=False) as session:
+            upload_file = session.scalar(select(UploadFile).where(UploadFile.id == file_id).limit(1))
+            if not upload_file:
+                raise NotFound("File not found")
+            upload_file_key = upload_file.key
+
+        blob = storage.load_once(upload_file_key)
         return base64.b64encode(blob).decode()
 
     def upload_text(self, text: str, text_name: str, user_id: str, tenant_id: str) -> UploadFile:
@@ -172,12 +173,14 @@ class FileService:
 
         return upload_file
 
-    def get_file_preview(self, file_id: str):
+    def get_file_preview(self, file_id: str, tenant_id: str):
         """
         Return a short text preview extracted from a document file.
         """
         with self._session_maker(expire_on_commit=False) as session:
-            upload_file = session.scalar(select(UploadFile).where(UploadFile.id == file_id).limit(1))
+            upload_file = session.scalar(
+                select(UploadFile).where(UploadFile.id == file_id, UploadFile.tenant_id == tenant_id).limit(1)
+            )
 
         if not upload_file:
             raise NotFound("File not found")

api/services/rag_pipeline/rag_pipeline_dsl_service.py

@@ -13,7 +13,6 @@ import yaml  # type: ignore
 from Crypto.Cipher import AES
 from Crypto.Util.Padding import pad, unpad
 from flask_login import current_user
-from packaging import version
 from pydantic import BaseModel
 from sqlalchemy import select
 from sqlalchemy.orm import Session
@@ -37,6 +36,7 @@ from models import Account
 from models.dataset import Dataset, DatasetCollectionBinding, Pipeline
 from models.enums import CollectionBindingType, DatasetRuntimeMode
 from models.workflow import Workflow, WorkflowType
+from services.dsl_version import check_version_compatibility
 from services.entities.dsl_entities import CheckDependenciesResult, ImportMode, ImportStatus
 from services.entities.knowledge_entities.rag_pipeline_entities import (
     IconInfo,
@@ -64,30 +64,6 @@ class RagPipelineImportInfo(BaseModel):
     dataset_id: str | None = None
 
 
-def _check_version_compatibility(imported_version: str) -> ImportStatus:
-    """Determine import status based on version comparison"""
-    try:
-        current_ver = version.parse(CURRENT_DSL_VERSION)
-        imported_ver = version.parse(imported_version)
-    except version.InvalidVersion:
-        return ImportStatus.FAILED
-
-    # If imported version is newer than current, always return PENDING
-    if imported_ver > current_ver:
-        return ImportStatus.PENDING
-
-    # If imported version is older than current's major, return PENDING
-    if imported_ver.major < current_ver.major:
-        return ImportStatus.PENDING
-
-    # If imported version is older than current's minor, return COMPLETED_WITH_WARNINGS
-    if imported_ver.minor < current_ver.minor:
-        return ImportStatus.COMPLETED_WITH_WARNINGS
-
-    # If imported version equals or is older than current's micro, return COMPLETED
-    return ImportStatus.COMPLETED
-
-
 class RagPipelinePendingData(BaseModel):
     import_mode: str
     yaml_content: str
@@ -202,7 +178,7 @@ class RagPipelineDslService:
             # check if imported_version is a float-like string
             if not isinstance(imported_version, str):
                 raise ValueError(f"Invalid version type, expected str, got {type(imported_version)}")
-            status = _check_version_compatibility(imported_version)
+            status = check_version_compatibility(imported_version, CURRENT_DSL_VERSION)
 
             # Extract app data
             pipeline_data = data.get("rag_pipeline")

api/services/tag_service.py

@@ -21,7 +21,6 @@ class SaveTagPayload(BaseModel):
 
 class UpdateTagPayload(BaseModel):
     name: str = Field(min_length=1, max_length=50)
-    type: TagType
 
 
 class TagBindingCreatePayload(BaseModel):

api/services/workflow_service.py

@@ -29,7 +29,11 @@ from core.workflow.node_factory import (
     get_node_type_classes_mapping,
     is_start_node_type,
 )
-from core.workflow.node_runtime import DifyHumanInputNodeRuntime, apply_dify_debug_email_recipient
+from core.workflow.node_runtime import (
+    DifyFileReferenceFactory,
+    DifyHumanInputNodeRuntime,
+    apply_dify_debug_email_recipient,
+)
 from core.workflow.system_variables import build_bootstrap_variables, build_system_variables, default_system_variables
 from core.workflow.variable_pool_initializer import add_node_inputs_to_pool, add_variables_to_pool
 from core.workflow.workflow_entry import WorkflowEntry
@@ -1259,6 +1263,7 @@ class WorkflowService:
             data=node_data,
             graph_init_params=graph_init_params,
             graph_runtime_state=graph_runtime_state,
+            file_reference_factory=DifyFileReferenceFactory(graph_init_params.run_context),
             runtime=DifyHumanInputNodeRuntime(run_context),
         )
         return node

api/tasks/document_indexing_update_task.py

@@ -7,10 +7,12 @@ from sqlalchemy import delete, select
 
 from core.db.session_factory import session_factory
 from core.indexing_runner import DocumentIsPausedError, IndexingRunner
+from core.rag.index_processor.constant.index_type import IndexStructureType, IndexTechniqueType
 from core.rag.index_processor.index_processor_factory import IndexProcessorFactory
 from libs.datetime_utils import naive_utc_now
 from models.dataset import Dataset, Document, DocumentSegment
 from models.enums import IndexingStatus
+from tasks.generate_summary_index_task import generate_summary_index_task
 
 logger = logging.getLogger(__name__)
 
@@ -70,6 +72,7 @@ def document_indexing_update_task(dataset_id: str, document_id: str):
             segment_delete_stmt = delete(DocumentSegment).where(DocumentSegment.document_id == document_id)
             session.execute(segment_delete_stmt)
 
+    has_error = False
     try:
         indexing_runner = IndexingRunner()
         indexing_runner.run([document])
@@ -77,5 +80,45 @@ def document_indexing_update_task(dataset_id: str, document_id: str):
         logger.info(click.style(f"update document: {document.id} latency: {end_at - start_at}", fg="green"))
     except DocumentIsPausedError as ex:
         logger.info(click.style(str(ex), fg="yellow"))
+        has_error = True
     except Exception:
         logger.exception("document_indexing_update_task failed, document_id: %s", document_id)
+        has_error = True
+
+    if has_error:
+        return
+
+    # Trigger summary index generation for the updated document if enabled.
+    # Only generate for high_quality indexing technique and when summary_index_setting is enabled.
+    with session_factory.create_session() as session:
+        dataset = session.scalar(select(Dataset).where(Dataset.id == dataset_id).limit(1))
+        if not dataset:
+            logger.warning("Dataset %s not found after update indexing", dataset_id)
+            return
+
+        if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
+            summary_index_setting = dataset.summary_index_setting
+            if summary_index_setting and summary_index_setting.get("enable"):
+                session.expire_all()
+                document = session.scalar(
+                    select(Document).where(Document.id == document_id, Document.dataset_id == dataset_id).limit(1)
+                )
+                if (
+                    document
+                    and document.indexing_status == IndexingStatus.COMPLETED
+                    and document.doc_form != IndexStructureType.QA_INDEX
+                    and document.need_summary is True
+                ):
+                    try:
+                        generate_summary_index_task.delay(dataset.id, document.id, None)
+                        logger.info(
+                            "Queued summary index generation task for document %s in dataset %s "
+                            "after update indexing completed",
+                            document.id,
+                            dataset.id,
+                        )
+                    except Exception:
+                        logger.exception(
+                            "Failed to queue summary index generation task for document %s after update",
+                            document.id,
+                        )

api/tasks/trigger_subscription_refresh_tasks.py

@@ -1,5 +1,4 @@
 import logging
-import time
 from collections.abc import Mapping
 from typing import Any
 
@@ -12,16 +11,13 @@ from core.db.session_factory import session_factory
 from core.plugin.entities.plugin_daemon import CredentialType
 from core.trigger.utils.locks import build_trigger_refresh_lock_key
 from extensions.ext_redis import redis_client
+from libs.helper import current_timestamp
 from models.trigger import TriggerSubscription
 from services.trigger.trigger_provider_service import TriggerProviderService
 
 logger = logging.getLogger(__name__)
 
 
-def _now_ts() -> int:
-    return int(time.time())
-
-
 def _load_subscription(session: Session, tenant_id: str, subscription_id: str) -> TriggerSubscription | None:
     return session.scalar(
         select(TriggerSubscription)
@@ -96,7 +92,7 @@ def trigger_subscription_refresh(tenant_id: str, subscription_id: str) -> None:
 
     logger.info("Begin subscription refresh: tenant=%s id=%s", tenant_id, subscription_id)
     try:
-        now: int = _now_ts()
+        now: int = current_timestamp()
         with session_factory.create_session() as session:
             subscription: TriggerSubscription | None = _load_subscription(session, tenant_id, subscription_id)
 

api/tests/integration_tests/workflow/nodes/test_tool.py

@@ -60,14 +60,14 @@ def init_tool_node(config: dict):
 
     graph = Graph.init(graph_config=graph_config, node_factory=node_factory, root_node_id="start")
 
-    tool_file_manager_factory = MagicMock(spec=ToolFileManagerProtocol)
+    tool_file_manager = MagicMock(spec=ToolFileManagerProtocol)
 
     node = ToolNode(
         node_id=str(uuid.uuid4()),
         data=ToolNodeData.model_validate(config["data"]),
         graph_init_params=init_params,
         graph_runtime_state=graph_runtime_state,
-        tool_file_manager_factory=tool_file_manager_factory,
+        tool_file_manager=tool_file_manager,
         runtime=DifyToolNodeRuntime(init_params.run_context),
     )
     return node

api/tests/test_containers_integration_tests/conftest.py

@@ -22,7 +22,7 @@ from sqlalchemy import Engine, text
 from sqlalchemy.orm import Session
 from testcontainers.core.container import DockerContainer
 from testcontainers.core.network import Network
-from testcontainers.core.waiting_utils import wait_for_logs
+from testcontainers.core.wait_strategies import LogMessageWaitStrategy
 from testcontainers.postgres import PostgresContainer
 from testcontainers.redis import RedisContainer
 
@@ -54,6 +54,10 @@ def _auto_close[T: _CloserProtocol](closer: T) -> Generator[T, None, None]:
     closer.close()
 
 
+def _wait_for_log_message(message: str, timeout: int) -> LogMessageWaitStrategy:
+    return LogMessageWaitStrategy(message).with_startup_timeout(timeout)
+
+
 class DifyTestContainers:
     """
     Manages all test containers required for Dify integration tests.
@@ -99,6 +103,7 @@ class DifyTestContainers:
         self.postgres = PostgresContainer(
             image="postgres:14-alpine",
         ).with_network(self.network)
+        self.postgres.waiting_for(_wait_for_log_message("is ready to accept connections", 30))
         self.postgres.start()
         db_host = self.postgres.get_container_host_ip()
         db_port = self.postgres.get_exposed_port(5432)
@@ -115,9 +120,6 @@ class DifyTestContainers:
             self.postgres.dbname,
         )
 
-        # Wait for PostgreSQL to be ready
-        logger.info("Waiting for PostgreSQL to be ready to accept connections...")
-        wait_for_logs(self.postgres, "is ready to accept connections", timeout=30)
         logger.info("PostgreSQL container is ready and accepting connections")
 
         conn = psycopg2.connect(
@@ -152,6 +154,7 @@ class DifyTestContainers:
         # Redis is used for storing session data, cache entries, and temporary data
         logger.info("Initializing Redis container...")
         self.redis = RedisContainer(image="redis:6-alpine", port=6379).with_network(self.network)
+        self.redis.waiting_for(_wait_for_log_message("Ready to accept connections", 30))
         self.redis.start()
         redis_host = self.redis.get_container_host_ip()
         redis_port = self.redis.get_exposed_port(6379)
@@ -159,9 +162,6 @@ class DifyTestContainers:
         os.environ["REDIS_PORT"] = str(redis_port)
         logger.info("Redis container started successfully - Host: %s, Port: %s", redis_host, redis_port)
 
-        # Wait for Redis to be ready
-        logger.info("Waiting for Redis to be ready to accept connections...")
-        wait_for_logs(self.redis, "Ready to accept connections", timeout=30)
         logger.info("Redis container is ready and accepting connections")
 
         # Start Dify Sandbox container for code execution environment.
@@ -170,6 +170,7 @@ class DifyTestContainers:
         sandbox_image = os.getenv(SANDBOX_TEST_IMAGE_ENV, DEFAULT_SANDBOX_TEST_IMAGE)
         self.dify_sandbox = DockerContainer(image=sandbox_image).with_network(self.network)
         self.dify_sandbox.with_exposed_ports(8194)
+        self.dify_sandbox.waiting_for(_wait_for_log_message("config init success", 60))
         self.dify_sandbox.env = {
             "API_KEY": "test_api_key",
         }
@@ -185,9 +186,6 @@ class DifyTestContainers:
             sandbox_port,
         )
 
-        # Wait for Dify Sandbox to be ready
-        logger.info("Waiting for Dify Sandbox to be ready to accept connections...")
-        wait_for_logs(self.dify_sandbox, "config init success", timeout=60)
         logger.info("Dify Sandbox container is ready and accepting connections")
 
         # Start Dify Plugin Daemon container for plugin management
@@ -197,6 +195,7 @@ class DifyTestContainers:
             self.network
         )
         self.dify_plugin_daemon.with_exposed_ports(5002)
+        self.dify_plugin_daemon.waiting_for(_wait_for_log_message("start plugin manager daemon", 60))
         # Get container internal network addresses
         postgres_container_name = self.postgres.get_wrapped_container().name
         redis_container_name = self.redis.get_wrapped_container().name
@@ -243,9 +242,6 @@ class DifyTestContainers:
                 plugin_daemon_port,
             )
 
-            # Wait for Dify Plugin Daemon to be ready
-            logger.info("Waiting for Dify Plugin Daemon to be ready to accept connections...")
-            wait_for_logs(self.dify_plugin_daemon, "start plugin manager daemon", timeout=60)
             logger.info("Dify Plugin Daemon container is ready and accepting connections")
         except Exception as e:
             logger.warning("Failed to start Dify Plugin Daemon container: %s", e)
@@ -500,6 +496,51 @@ def db_session_with_containers(flask_app_with_containers: Flask) -> Generator[Se
             logger.debug("Database session closed")
 
 
+def _truncate_container_database(app: Flask) -> None:
+    """
+    Reset application tables after a container integration test.
+
+    Tests in this package share one PostgreSQL container for performance, while
+    application code may commit through db.session, Session(db.engine), or
+    session_factory-created sessions. Truncating after each test gives the suite
+    a central DB isolation contract that does not depend on which session a test used.
+    This only covers SQLAlchemy application tables in db.metadata for now;
+    Redis, object storage, and custom ad hoc metadata still need their own cleanup.
+    """
+    with app.app_context():
+        db.session.remove()
+
+        tables = db.metadata.sorted_tables
+        if not tables:
+            return
+
+        preparer = db.engine.dialect.identifier_preparer
+        table_names = ", ".join(preparer.format_table(table) for table in tables)
+
+        with db.engine.begin() as conn:
+            conn.execute(text("SET LOCAL lock_timeout = '5s'"))
+            conn.execute(text(f"TRUNCATE TABLE {table_names} RESTART IDENTITY CASCADE"))
+
+        db.session.remove()
+
+
+@pytest.fixture(autouse=True)
+def isolate_container_database(request: pytest.FixtureRequest) -> Generator[None, None, None]:
+    """
+    Clean DB state after tests that use the containerized Flask app.
+
+    This fixture intentionally does not depend on flask_app_with_containers so
+    non-DB tests under this package do not start the full app/container stack.
+    """
+    yield
+
+    if "flask_app_with_containers" not in request.fixturenames:
+        return
+
+    app = request.getfixturevalue("flask_app_with_containers")
+    _truncate_container_database(app)
+
+
 @pytest.fixture(scope="package", autouse=True)
 def mock_ssrf_proxy_requests():
     """

api/tests/test_containers_integration_tests/controllers/console/app/test_app_apis.py

@@ -8,7 +8,9 @@ from unittest.mock import MagicMock, patch
 
 import pytest
 from flask import Flask
+from flask.testing import FlaskClient
 from pydantic import ValidationError
+from sqlalchemy.orm import Session
 from werkzeug.exceptions import BadRequest, NotFound
 
 from controllers.console import console_ns
@@ -57,6 +59,12 @@ from controllers.console.app.workflow_app_log import WorkflowAppLogQuery
 from controllers.console.app.workflow_draft_variable import WorkflowDraftVariableUpdatePayload
 from controllers.console.app.workflow_statistic import WorkflowStatisticQuery
 from controllers.console.app.workflow_trigger import Parser, ParserEnable
+from models.model import AppMode
+from tests.test_containers_integration_tests.controllers.console.helpers import (
+    authenticate_console_client,
+    create_console_account_and_tenant,
+    create_console_app,
+)
 
 
 def _unwrap(func):
@@ -270,6 +278,35 @@ class TestOpsTraceEndpoints:
     def app(self, flask_app_with_containers: Flask):
         return flask_app_with_containers
 
+    @pytest.mark.parametrize(
+        "path_template",
+        [
+            "/console/api/apps/{app_id}/trace-config?tracing_provider=langfuse",
+            "/console/api/apps/{app_id}/trace",
+        ],
+    )
+    def test_trace_endpoints_hide_apps_from_other_tenants(
+        self,
+        db_session_with_containers: Session,
+        test_client_with_containers: FlaskClient,
+        path_template: str,
+    ):
+        account, _tenant = create_console_account_and_tenant(db_session_with_containers)
+        foreign_account, foreign_tenant = create_console_account_and_tenant(db_session_with_containers)
+        foreign_app = create_console_app(
+            db_session_with_containers,
+            tenant_id=foreign_tenant.id,
+            account_id=foreign_account.id,
+            mode=AppMode.CHAT,
+        )
+
+        response = test_client_with_containers.get(
+            path_template.format(app_id=foreign_app.id),
+            headers=authenticate_console_client(test_client_with_containers, account),
+        )
+
+        assert response.status_code == 404
+
     def test_ops_trace_query_basic(self):
         query = TraceProviderQuery(tracing_provider="langfuse")
         assert query.tracing_provider == "langfuse"
@@ -289,7 +326,7 @@ class TestOpsTraceEndpoints:
         )
 
         with app.test_request_context("/?tracing_provider=langfuse"):
-            result = method(app_id="app-1")
+            result = method(app_model=MagicMock(id="app-1"))
 
         assert result == {"has_not_configured": True}
 
@@ -308,7 +345,7 @@ class TestOpsTraceEndpoints:
             json={"tracing_provider": "langfuse", "tracing_config": {"api_key": "k"}},
         ):
             with pytest.raises(BadRequest):
-                method(app_id="app-1")
+                method(app_model=MagicMock(id="app-1"))
 
     def test_trace_app_config_delete_not_found(self, app: Flask, monkeypatch: pytest.MonkeyPatch):
         api = ops_trace_module.TraceAppConfigApi()
@@ -322,7 +359,7 @@ class TestOpsTraceEndpoints:
 
         with app.test_request_context("/?tracing_provider=langfuse"):
             with pytest.raises(BadRequest):
-                method(app_id="app-1")
+                method(app_model=MagicMock(id="app-1"))
 
 
 class TestSiteEndpoints:

api/tests/test_containers_integration_tests/controllers/console/app/test_chat_conversation_status_count_api.py

@@ -6,17 +6,17 @@ import uuid
 from flask.testing import FlaskClient
 from sqlalchemy.orm import Session
 
-from configs import dify_config
 from constants import HEADER_NAME_CSRF_TOKEN
 from graphon.enums import WorkflowExecutionStatus
 from libs.datetime_utils import naive_utc_now
 from libs.token import _real_cookie_name, generate_csrf_token
-from models import Account, DifySetup, Tenant, TenantAccountJoin
+from models import Account, Tenant, TenantAccountJoin
 from models.account import AccountStatus, TenantAccountRole, TenantStatus
 from models.enums import ConversationFromSource, CreatorUserRole
 from models.model import App, AppMode, Conversation, Message
 from models.workflow import WorkflowRun
 from services.account_service import AccountService
+from tests.test_containers_integration_tests.controllers.console.helpers import ensure_dify_setup
 
 
 def _create_account_and_tenant(db_session: Session) -> tuple[Account, Tenant]:
@@ -47,9 +47,7 @@ def _create_account_and_tenant(db_session: Session) -> tuple[Account, Tenant]:
     account.timezone = "UTC"
     db_session.commit()
 
-    dify_setup = DifySetup(version=dify_config.project.version)
-    db_session.add(dify_setup)
-    db_session.commit()
+    ensure_dify_setup(db_session)
 
     return account, tenant
 

api/tests/test_containers_integration_tests/controllers/console/auth/test_email_register.py

@@ -143,7 +143,118 @@ class TestEmailRegisterResetApi:
                 response = EmailRegisterResetApi().post()
 
         assert response == {"result": "success", "data": {"access_token": "a", "refresh_token": "r"}}
-        mock_create_account.assert_called_once_with("invitee@example.com", "ValidPass123!")
+        mock_create_account.assert_called_once_with(
+            email="invitee@example.com",
+            password="ValidPass123!",
+            timezone=None,
+            language=None,
+        )
+        mock_reset_login_rate.assert_called_once_with("invitee@example.com")
+        mock_revoke_token.assert_called_once_with("token-123")
+        mock_extract_ip.assert_called_once()
+
+    @patch("controllers.console.auth.email_register.AccountService.reset_login_error_rate_limit")
+    @patch("controllers.console.auth.email_register.AccountService.login")
+    @patch("controllers.console.auth.email_register.EmailRegisterResetApi._create_new_account")
+    @patch("controllers.console.auth.email_register.AccountService.get_account_by_email_with_case_fallback")
+    @patch("controllers.console.auth.email_register.AccountService.revoke_email_register_token")
+    @patch("controllers.console.auth.email_register.AccountService.get_email_register_data")
+    @patch("controllers.console.auth.email_register.extract_remote_ip", return_value="127.0.0.1")
+    def test_reset_passes_timezone_to_new_account(
+        self,
+        mock_extract_ip,
+        mock_get_data,
+        mock_revoke_token,
+        mock_get_account,
+        mock_create_account,
+        mock_login,
+        mock_reset_login_rate,
+        app: Flask,
+    ):
+        mock_get_data.return_value = {"phase": "register", "email": "Invitee@Example.com"}
+        mock_create_account.return_value = MagicMock()
+        token_pair = MagicMock()
+        token_pair.model_dump.return_value = {"access_token": "a", "refresh_token": "r"}
+        mock_login.return_value = token_pair
+        mock_get_account.return_value = None
+
+        feature_flags = SimpleNamespace(enable_email_password_login=True, is_allow_register=True)
+        with (
+            patch("controllers.console.wraps.dify_config", SimpleNamespace(EDITION="CLOUD")),
+            patch("controllers.console.wraps.FeatureService.get_system_features", return_value=feature_flags),
+        ):
+            with app.test_request_context(
+                "/email-register",
+                method="POST",
+                json={
+                    "token": "token-123",
+                    "new_password": "ValidPass123!",
+                    "password_confirm": "ValidPass123!",
+                    "timezone": "Asia/Shanghai",
+                },
+            ):
+                response = EmailRegisterResetApi().post()
+
+        assert response == {"result": "success", "data": {"access_token": "a", "refresh_token": "r"}}
+        mock_create_account.assert_called_once_with(
+            email="invitee@example.com",
+            password="ValidPass123!",
+            timezone="Asia/Shanghai",
+            language=None,
+        )
+        mock_reset_login_rate.assert_called_once_with("invitee@example.com")
+        mock_revoke_token.assert_called_once_with("token-123")
+        mock_extract_ip.assert_called_once()
+
+    @patch("controllers.console.auth.email_register.AccountService.reset_login_error_rate_limit")
+    @patch("controllers.console.auth.email_register.AccountService.login")
+    @patch("controllers.console.auth.email_register.EmailRegisterResetApi._create_new_account")
+    @patch("controllers.console.auth.email_register.AccountService.get_account_by_email_with_case_fallback")
+    @patch("controllers.console.auth.email_register.AccountService.revoke_email_register_token")
+    @patch("controllers.console.auth.email_register.AccountService.get_email_register_data")
+    @patch("controllers.console.auth.email_register.extract_remote_ip", return_value="127.0.0.1")
+    def test_reset_passes_language_to_new_account(
+        self,
+        mock_extract_ip,
+        mock_get_data,
+        mock_revoke_token,
+        mock_get_account,
+        mock_create_account,
+        mock_login,
+        mock_reset_login_rate,
+        app: Flask,
+    ):
+        mock_get_data.return_value = {"phase": "register", "email": "Invitee@Example.com"}
+        mock_create_account.return_value = MagicMock()
+        token_pair = MagicMock()
+        token_pair.model_dump.return_value = {"access_token": "a", "refresh_token": "r"}
+        mock_login.return_value = token_pair
+        mock_get_account.return_value = None
+
+        feature_flags = SimpleNamespace(enable_email_password_login=True, is_allow_register=True)
+        with (
+            patch("controllers.console.wraps.dify_config", SimpleNamespace(EDITION="CLOUD")),
+            patch("controllers.console.wraps.FeatureService.get_system_features", return_value=feature_flags),
+        ):
+            with app.test_request_context(
+                "/email-register",
+                method="POST",
+                json={
+                    "token": "token-123",
+                    "new_password": "ValidPass123!",
+                    "password_confirm": "ValidPass123!",
+                    "language": "zh-Hans",
+                },
+            ):
+                response = EmailRegisterResetApi().post()
+
+        assert response == {"result": "success", "data": {"access_token": "a", "refresh_token": "r"}}
+        mock_create_account.assert_called_once_with(
+            email="invitee@example.com",
+            password="ValidPass123!",
+            timezone=None,
+            language="zh-Hans",
+        )
         mock_reset_login_rate.assert_called_once_with("invitee@example.com")
         mock_revoke_token.assert_called_once_with("token-123")
         mock_extract_ip.assert_called_once()

api/tests/test_containers_integration_tests/controllers/console/auth/test_oauth.py

@@ -14,7 +14,7 @@ from controllers.console.auth.oauth import (
     _get_account_by_openid_or_email,
     get_oauth_providers,
 )
-from libs.oauth import OAuthUserInfo
+from libs.oauth import OAuthUserInfo, encode_oauth_state
 from models.account import AccountStatus
 from services.account_service import AccountService
 from services.errors.account import AccountRegisterError
@@ -101,7 +101,55 @@ class TestOAuthLogin:
         with app.test_request_context(f"/auth/oauth/github?{query_string}"):
             resource.get("github")
 
-        mock_oauth_provider.get_authorization_url.assert_called_once_with(invite_token=expected_token)
+        mock_oauth_provider.get_authorization_url.assert_called_once_with(
+            invite_token=expected_token,
+            timezone=None,
+            language=None,
+        )
+        mock_redirect.assert_called_once_with("https://github.com/login/oauth/authorize?...")
+
+    @patch("controllers.console.auth.oauth.get_oauth_providers")
+    @patch("controllers.console.auth.oauth.redirect")
+    def test_should_pass_timezone_to_oauth_state(
+        self,
+        mock_redirect,
+        mock_get_providers,
+        resource,
+        app: Flask,
+        mock_oauth_provider,
+    ):
+        mock_get_providers.return_value = {"github": mock_oauth_provider, "google": None}
+
+        with app.test_request_context("/auth/oauth/github?timezone=Asia/Shanghai"):
+            resource.get("github")
+
+        mock_oauth_provider.get_authorization_url.assert_called_once_with(
+            invite_token=None,
+            timezone="Asia/Shanghai",
+            language=None,
+        )
+        mock_redirect.assert_called_once_with("https://github.com/login/oauth/authorize?...")
+
+    @patch("controllers.console.auth.oauth.get_oauth_providers")
+    @patch("controllers.console.auth.oauth.redirect")
+    def test_should_pass_language_to_oauth_state(
+        self,
+        mock_redirect,
+        mock_get_providers,
+        resource,
+        app: Flask,
+        mock_oauth_provider,
+    ):
+        mock_get_providers.return_value = {"github": mock_oauth_provider, "google": None}
+
+        with app.test_request_context("/auth/oauth/github?language=zh-Hans"):
+            resource.get("github")
+
+        mock_oauth_provider.get_authorization_url.assert_called_once_with(
+            invite_token=None,
+            timezone=None,
+            language="zh-Hans",
+        )
         mock_redirect.assert_called_once_with("https://github.com/login/oauth/authorize?...")
 
     @pytest.mark.parametrize(
@@ -229,7 +277,8 @@ class TestOAuthCallback:
         mock_register_service.is_valid_invite_token.return_value = True
         mock_register_service.get_invitation_by_token.return_value = {"email": "user@example.com"}
 
-        with app.test_request_context("/auth/oauth/github/callback?code=test_code&state=invite123"):
+        state = encode_oauth_state(invite_token="invite123", timezone="Asia/Shanghai")
+        with app.test_request_context(f"/auth/oauth/github/callback?code=test_code&state={state}"):
             resource.get("github")
 
         mock_register_service.get_invitation_by_token.assert_called_once_with(token="invite123")
@@ -488,7 +537,13 @@ class TestAccountGeneration:
 
                 if should_create:
                     mock_register_service.register.assert_called_once_with(
-                        email="test@example.com", name="Test User", password=None, open_id="123", provider="github"
+                        email="test@example.com",
+                        name="Test User",
+                        password=None,
+                        open_id="123",
+                        provider="github",
+                        language="en-US",
+                        timezone=None,
                     )
                 else:
                     mock_register_service.register.assert_not_called()
@@ -515,7 +570,75 @@ class TestAccountGeneration:
             _generate_account("github", user_info)
 
         mock_register_service.register.assert_called_once_with(
-            email="upper@example.com", name="Test User", password=None, open_id="123", provider="github"
+            email="upper@example.com",
+            name="Test User",
+            password=None,
+            open_id="123",
+            provider="github",
+            language="en-US",
+            timezone=None,
+        )
+
+    @patch("controllers.console.auth.oauth._get_account_by_openid_or_email", return_value=None)
+    @patch("controllers.console.auth.oauth.FeatureService")
+    @patch("controllers.console.auth.oauth.RegisterService")
+    @patch("controllers.console.auth.oauth.AccountService")
+    @patch("controllers.console.auth.oauth.TenantService")
+    def test_should_register_with_browser_timezone(
+        self,
+        mock_tenant_service,
+        mock_account_service,
+        mock_register_service,
+        mock_feature_service,
+        mock_get_account,
+        app: Flask,
+        user_info,
+    ):
+        mock_feature_service.get_system_features.return_value.is_allow_register = True
+        mock_register_service.register.return_value = MagicMock()
+
+        with app.test_request_context(headers={"Accept-Language": "zh-Hans,zh;q=0.9"}):
+            _generate_account("github", user_info, timezone="Asia/Shanghai")
+
+        mock_register_service.register.assert_called_once_with(
+            email="test@example.com",
+            name="Test User",
+            password=None,
+            open_id="123",
+            provider="github",
+            language="zh-Hans",
+            timezone="Asia/Shanghai",
+        )
+
+    @patch("controllers.console.auth.oauth._get_account_by_openid_or_email", return_value=None)
+    @patch("controllers.console.auth.oauth.FeatureService")
+    @patch("controllers.console.auth.oauth.RegisterService")
+    @patch("controllers.console.auth.oauth.AccountService")
+    @patch("controllers.console.auth.oauth.TenantService")
+    def test_should_register_with_state_language(
+        self,
+        mock_tenant_service,
+        mock_account_service,
+        mock_register_service,
+        mock_feature_service,
+        mock_get_account,
+        app: Flask,
+        user_info,
+    ):
+        mock_feature_service.get_system_features.return_value.is_allow_register = True
+        mock_register_service.register.return_value = MagicMock()
+
+        with app.test_request_context(headers={"Accept-Language": "en-US,en;q=0.9"}):
+            _generate_account("github", user_info, language="zh-Hans")
+
+        mock_register_service.register.assert_called_once_with(
+            email="test@example.com",
+            name="Test User",
+            password=None,
+            open_id="123",
+            provider="github",
+            language="zh-Hans",
+            timezone=None,
         )
 
     @patch("controllers.console.auth.oauth._get_account_by_openid_or_email")

api/tests/test_containers_integration_tests/controllers/console/datasets/rag_pipeline/test_rag_pipeline_workflow.py

@@ -2,8 +2,10 @@
 
 from __future__ import annotations
 
+import json
 from datetime import datetime
 from types import SimpleNamespace
+from typing import TypedDict, Unpack
 from unittest.mock import MagicMock, patch
 from uuid import uuid4
 
@@ -35,9 +37,54 @@ from controllers.console.datasets.rag_pipeline.rag_pipeline_workflow import (
 )
 from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
 from libs.datetime_utils import naive_utc_now
+from models.account import Account
+from models.workflow import Workflow
 from services.errors.app import IsDraftWorkflowError, WorkflowHashNotEqualError, WorkflowNotFoundError
 from services.errors.llm import InvokeRateLimitError
 
+DEFAULT_WORKFLOW_TENANT_ID = "00000000-0000-0000-0000-000000000001"
+DEFAULT_WORKFLOW_APP_ID = "00000000-0000-0000-0000-000000000002"
+DEFAULT_WORKFLOW_CREATED_BY = "00000000-0000-0000-0000-000000000003"
+type WorkflowVariablePayload = dict[str, object]
+
+
+class WorkflowFactoryPayload(TypedDict):
+    id: str
+    tenant_id: str
+    app_id: str
+    type: str
+    version: str
+    marked_name: str
+    marked_comment: str
+    graph: str
+    features: str
+    created_by: str
+    created_at: datetime
+    updated_by: str | None
+    updated_at: datetime
+    environment_variables: list[WorkflowVariablePayload]
+    conversation_variables: list[WorkflowVariablePayload]
+    rag_pipeline_variables: list[WorkflowVariablePayload]
+
+
+class WorkflowFactoryOverrides(TypedDict, total=False):
+    id: str
+    tenant_id: str
+    app_id: str
+    type: str
+    version: str
+    marked_name: str
+    marked_comment: str
+    graph: str
+    features: str
+    created_by: str
+    created_at: datetime
+    updated_by: str | None
+    updated_at: datetime
+    environment_variables: list[WorkflowVariablePayload]
+    conversation_variables: list[WorkflowVariablePayload]
+    rag_pipeline_variables: list[WorkflowVariablePayload]
+
 
 def unwrap(func):
     while hasattr(func, "__wrapped__"):
@@ -74,17 +121,52 @@ def make_node_execution(**overrides):
     return SimpleNamespace(**payload)
 
 
+def default_workflow_payload() -> WorkflowFactoryPayload:
+    return {
+        "id": "workflow-1",
+        "tenant_id": DEFAULT_WORKFLOW_TENANT_ID,
+        "app_id": DEFAULT_WORKFLOW_APP_ID,
+        "type": "workflow",
+        "version": "1",
+        "marked_name": "Release 1",
+        "marked_comment": "Initial release",
+        "graph": json.dumps({"nodes": [], "edges": []}),
+        "features": json.dumps({"file_upload": {"enabled": False}}),
+        "created_by": DEFAULT_WORKFLOW_CREATED_BY,
+        "created_at": datetime(2024, 1, 1, 12, 0, 0),
+        "updated_by": None,
+        "updated_at": datetime(2024, 1, 1, 12, 1, 0),
+        "environment_variables": [],
+        "conversation_variables": [],
+        "rag_pipeline_variables": [],
+    }
+
+
+def make_workflow(**overrides: Unpack[WorkflowFactoryOverrides]) -> Workflow:
+    payload = default_workflow_payload()
+    payload.update(overrides)
+    return Workflow(**payload)
+
+
+@pytest.fixture
+def workflow_author(db_session_with_containers: Session) -> Account:
+    account = Account(name="Alice", email=f"alice-{uuid4()}@example.com")
+    db_session_with_containers.add(account)
+    db_session_with_containers.commit()
+    return account
+
+
 class TestDraftWorkflowApi:
     @pytest.fixture
     def app(self, flask_app_with_containers: Flask):
         return flask_app_with_containers
 
-    def test_get_draft_success(self, app: Flask):
+    def test_get_draft_success(self, app: Flask, workflow_author: Account):
         api = DraftRagPipelineApi()
         method = unwrap(api.get)
 
         pipeline = MagicMock()
-        workflow = MagicMock()
+        workflow = make_workflow(created_by=workflow_author.id)
 
         service = MagicMock()
         service.get_draft_workflow.return_value = workflow
@@ -97,7 +179,17 @@ class TestDraftWorkflowApi:
             ),
         ):
             result = method(api, pipeline)
-            assert result == workflow
+
+        assert result["id"] == "workflow-1"
+        assert result["graph"] == {"nodes": [], "edges": []}
+        assert result["features"] == {"file_upload": {"enabled": False}}
+        assert result["hash"] == workflow.unique_hash
+        assert result["created_by"] == {
+            "id": workflow_author.id,
+            "name": workflow_author.name,
+            "email": workflow_author.email,
+        }
+        assert result["updated_by"] is None
 
     def test_get_draft_not_exist(self, app: Flask):
         api = DraftRagPipelineApi()
@@ -642,7 +734,7 @@ class TestPublishedAllRagPipelineApi:
         user = MagicMock(id="u1")
 
         service = MagicMock()
-        service.get_all_published_workflow.return_value = ([{"id": "w1"}], False)
+        service.get_all_published_workflow.return_value = ([make_workflow(id="w1")], False)
 
         with (
             app.test_request_context("/"),
@@ -657,7 +749,8 @@ class TestPublishedAllRagPipelineApi:
         ):
             result = method(api, pipeline)
 
-        assert result["items"] == [{"id": "w1"}]
+        assert result["items"][0]["id"] == "w1"
+        assert result["items"][0]["graph"] == {"nodes": [], "edges": []}
         assert result["has_more"] is False
 
     def test_get_published_workflows_forbidden(self, app: Flask):
@@ -690,7 +783,7 @@ class TestRagPipelineByIdApi:
         pipeline = MagicMock(tenant_id="t1")
         user = MagicMock(id="u1")
 
-        workflow = MagicMock()
+        workflow = make_workflow(id="w1", marked_name="test")
 
         service = MagicMock()
         service.update_workflow.return_value = workflow
@@ -711,7 +804,9 @@ class TestRagPipelineByIdApi:
         ):
             result = method(api, pipeline, "w1")
 
-        assert result == workflow
+        assert result["id"] == "w1"
+        assert result["marked_name"] == "test"
+        assert result["hash"] == workflow.unique_hash
 
     def test_patch_no_fields(self, app: Flask):
         api = RagPipelineByIdApi()

api/tests/test_containers_integration_tests/controllers/console/test_apikey.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 from unittest.mock import MagicMock, patch
 
 import pytest
+from flask import Flask
 from flask.testing import FlaskClient
 from sqlalchemy import delete
 from sqlalchemy.orm import Session
@@ -126,7 +127,7 @@ class TestAppApiKeyResource:
 
     def test_delete_forbidden_for_non_admin(
         self,
-        flask_app_with_containers,
+        flask_app_with_containers: Flask,
     ) -> None:
         """A non-admin member cannot delete API keys via the controller permission check."""
         from werkzeug.exceptions import Forbidden

api/tests/test_containers_integration_tests/controllers/console/workspace/test_trigger_providers.py

@@ -575,7 +575,7 @@ class TestTriggerSubscriptionVerifyApi:
             assert method(api, "github", "s1") == {"ok": True}
 
     @pytest.mark.parametrize("raised_exception", [ValueError("bad"), Exception("boom")])
-    def test_verify_errors(self, app, raised_exception):
+    def test_verify_errors(self, app: Flask, raised_exception):
         api = TriggerSubscriptionVerifyApi()
         method = unwrap(api.post)
 

api/tests/test_containers_integration_tests/controllers/web/test_wraps.py

@@ -277,7 +277,7 @@ class TestDecodeJwtToken:
         mock_extract: MagicMock,
         mock_passport_cls: MagicMock,
         mock_features: MagicMock,
-        app,
+        app: Flask,
     ) -> None:
         non_existent_id = str(uuid4())
         mock_extract.return_value = "jwt-token"