chore(api): polishhelp output for legacy-model-types migration script (#36707)

Signed-off-by: vuko <[email protected]>

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

@@ -1,13 +1,8 @@
 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,11 +120,7 @@ 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 ../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
+          uv run dev/generate_swagger_markdown_docs.py --swagger-dir openapi --markdown-dir openapi/markdown
 
       - name: ESLint autofix
         if: github.event_name != 'merge_group' && steps.web-changes.outputs.any_changed == 'true'

.gitignore

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

AGENTS.md

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

Makefile

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

api/.env.example

@@ -557,7 +557,7 @@ MAX_VARIABLE_SIZE=204800
 
 # GraphEngine Worker Pool Configuration
 # Minimum number of workers per GraphEngine instance (default: 1)
-GRAPH_ENGINE_MIN_WORKERS=3
+GRAPH_ENGINE_MIN_WORKERS=1
 # 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 response DTOs with `dump_response(...)`,
+  DTOs with `register_response_schema_models(...)`, serialize with `ResponseModel.model_validate(...).model_dump(...)`,
   and avoid adding new legacy `ns.model(...)`, `@marshal_with(...)`, or GET `@ns.expect(...)` patterns.
 
 ### Miscellaneous

api/Dockerfile

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

api/README.md

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

api/app_factory.py

@@ -117,7 +117,7 @@ def create_flask_app_with_configs() -> DifyApp:
             logger.warning("Failed to add trace headers to response", exc_info=True)
         return response
 
-    # Capture the decorator return values so static checkers do not treat the hooks as unused.
+    # Capture the decorator's return value to avoid pyright reportUnusedFunction
     _ = before_request
     _ = add_trace_headers
 
@@ -221,10 +221,11 @@ def initialize_extensions(app: DifyApp):
 
 def create_migrations_app() -> DifyApp:
     app = create_flask_app_with_configs()
-    from extensions import ext_database, ext_migrate
+    from extensions import ext_commands, ext_database, ext_migrate
 
     # Initialize only required extensions
     ext_database.init_app(app)
     ext_migrate.init_app(app)
+    ext_commands.init_app(app)
 
     return app

api/commands/__init__.py

@@ -3,6 +3,7 @@ CLI command modules extracted from `commands.py`.
 """
 
 from .account import create_tenant, reset_email, reset_password
+from .data_migrate import data_migrate, legacy_model_types
 from .plugin import (
     extract_plugins,
     extract_unique_plugins,
@@ -44,6 +45,7 @@ __all__ = [
     "clear_orphaned_file_records",
     "convert_to_agent_apps",
     "create_tenant",
+    "data_migrate",
     "delete_archived_workflow_runs",
     "export_app_messages",
     "extract_plugins",
@@ -52,6 +54,7 @@ __all__ = [
     "fix_app_site_missing",
     "install_plugins",
     "install_rag_pipeline_plugins",
+    "legacy_model_types",
     "migrate_annotation_vector_database",
     "migrate_data_for_plugin",
     "migrate_knowledge_vector_database",

api/commands/data_migrate.py

@@ -0,0 +1,179 @@
+import io
+import os
+import sys
+from contextlib import AbstractContextManager, nullcontext
+from pathlib import Path
+from typing import cast
+
+import click
+
+from extensions.ext_database import db
+from graphon.model_runtime.entities.model_entities import ModelType
+from services.legacy_model_type_migration import (
+    VALID_TABLE_NAMES,
+    LegacyModelTypeMigrationService,
+    load_tenant_ids_from_file,
+)
+
+_SUPPORTED_MODEL_TYPE_CHOICES = (
+    ModelType.LLM.value,
+    ModelType.TEXT_EMBEDDING.value,
+    ModelType.RERANK.value,
+)
+_DEFAULT_CONCURRENCY = os.cpu_count() or 1
+
+
+def _normalize_multi_value_option(
+    values: tuple[str, ...],
+    *,
+    valid_values: tuple[str, ...],
+    option_name: str,
+) -> tuple[str, ...]:
+    normalized_values: list[str] = []
+    seen_values: set[str] = set()
+
+    for value in values:
+        for item in value.split(","):
+            normalized_item = item.strip()
+            if not normalized_item:
+                continue
+            if normalized_item not in valid_values:
+                raise click.BadParameter(
+                    f"invalid value '{normalized_item}'. valid values: {', '.join(valid_values)}",
+                    param_hint=option_name,
+                )
+            if normalized_item in seen_values:
+                continue
+            seen_values.add(normalized_item)
+            normalized_values.append(normalized_item)
+
+    return tuple(normalized_values)
+
+
+@click.group(
+    "data-migrate",
+    help="Online data migration commands.",
+)
+def data_migrate() -> None:
+    """Namespace for production data migration commands."""
+
+
+@click.command(
+    "legacy-model-types",
+    help=(
+        "Migrate legacy provider model_type values to canonical values. "
+        "Default is dry-run and emits JSON lines only. "
+        "If --tables includes provider_model_credentials, the command may also update "
+        "provider_models and load_balancing_model_configs references so merged credentials stay reachable."
+    ),
+)
+@click.option(
+    "--apply",
+    is_flag=True,
+    default=False,
+    help="Apply the migration. Default is dry-run.",
+)
+@click.option(
+    "--tables",
+    "tables",
+    multiple=True,
+    type=str,
+    help=(
+        "Limit migration to specific tables. Accepts comma-separated values or repeated flags.\n"
+        "\n"
+        "Options: load_balancing_model_configs, provider_model_credentials, "
+        "provider_model_settings, provider_models, tenant_default_models.\n\n"
+        "When provider_model_credentials is selected, provider_models and "
+        "load_balancing_model_configs may also be updated for credential reference rewrites.\n"
+        "\n"
+        "If unspecified, all relevant tables are migrated."
+    ),
+)
+@click.option(
+    "--model-types",
+    "model_types",
+    multiple=True,
+    type=str,
+    help=(
+        "Canonical model types to migrate. Accepts comma-separated values or repeated flags.\n"
+        "\n"
+        "Options: llm,text-embedding,rerank\n"
+        "\n"
+        "If unspecified, all relevant legacy model types are migrated."
+    ),
+)
+@click.option(
+    "--tenant-id-file",
+    type=click.Path(exists=True, dir_okay=False, readable=True, resolve_path=True),
+    help="Optional file containing tenant ids, one per line.",
+)
+@click.option(
+    "--output",
+    type=click.Path(dir_okay=False, resolve_path=True, path_type=Path),
+    help=(
+        "Optional file path for JSON lines event logs. Defaults to stdout.\n"
+        "It's highly recommended to save the event logs to a file and preserve it for a period of time."
+    ),
+)
+@click.option(
+    "--concurrency",
+    type=click.IntRange(min=1),
+    default=_DEFAULT_CONCURRENCY,
+    show_default=True,
+    help="Number of tenant-level worker threads to run in parallel.",
+)
+def legacy_model_types(
+    apply: bool,
+    tables: tuple[str, ...],
+    model_types: tuple[str, ...],
+    tenant_id_file: str | None,
+    output: Path | None,
+    concurrency: int = _DEFAULT_CONCURRENCY,
+) -> None:
+    """
+    Migrate legacy provider-related model_type values and emit JSON lines events.
+    """
+
+    normalized_tables = _normalize_multi_value_option(
+        tables,
+        valid_values=VALID_TABLE_NAMES,
+        option_name="--tables",
+    )
+    normalized_model_types = _normalize_multi_value_option(
+        model_types,
+        valid_values=_SUPPORTED_MODEL_TYPE_CHOICES,
+        option_name="--model-types",
+    )
+    selected_model_types = (
+        tuple(ModelType.value_of(model_type) for model_type in normalized_model_types)
+        if normalized_model_types
+        else (
+            ModelType.LLM,
+            ModelType.TEXT_EMBEDDING,
+            ModelType.RERANK,
+        )
+    )
+    tenant_ids = load_tenant_ids_from_file(tenant_id_file) if tenant_id_file else None
+
+    output_context: AbstractContextManager[io.TextIOBase]
+    if output is None:
+        output_context = nullcontext(cast(io.TextIOBase, sys.stdout))
+    else:
+        try:
+            output_context = output.open("w", encoding="utf-8")
+        except OSError as exc:
+            raise click.ClickException(f"failed to open output file '{output}': {exc.strerror or exc}") from exc
+
+    with output_context as output_stream:
+        LegacyModelTypeMigrationService(
+            engine=db.engine,
+            apply=apply,
+            concurrency=concurrency,
+            output=cast(io.TextIOBase, output_stream),
+            tables=normalized_tables or None,
+            model_types=selected_model_types,
+            tenant_ids=tenant_ids,
+        ).migrate()
+
+
+data_migrate.add_command(legacy_model_types)

api/commands/plugin.py

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

api/commands/system.py

@@ -14,7 +14,6 @@ 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__)
 
@@ -24,16 +23,13 @@ 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 and tool provider credentials "
-    "(builtin / API / MCP) will be purged, requiring re-entry. "
+    "After the reset, all LLM credentials will become invalid, "
+    "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 will also purge builtin / API / MCP tool provider records for every tenant. "
-        "This operation cannot be rolled back!",
-        fg="red",
+        "Are you sure you want to reset encrypt key pair? This operation cannot be rolled back!", fg="red"
     )
 )
 def reset_encrypt_key_pair():
@@ -57,13 +53,6 @@ def reset_encrypt_key_pair():
             session.execute(delete(Provider).where(Provider.provider_type == "custom", Provider.tenant_id == tenant.id))
             session.execute(delete(ProviderModel).where(ProviderModel.tenant_id == tenant.id))
 
-            # Purge tool provider records that hold credentials encrypted under the
-            # tenant key. Leaving them in place causes /console/api/workspaces/current/
-            # tool-providers to 500 because decryption fails on stale ciphertext (#35396).
-            session.execute(delete(BuiltinToolProvider).where(BuiltinToolProvider.tenant_id == tenant.id))
-            session.execute(delete(ApiToolProvider).where(ApiToolProvider.tenant_id == tenant.id))
-            session.execute(delete(MCPToolProvider).where(MCPToolProvider.tenant_id == tenant.id))
-
             click.echo(
                 click.style(
                     f"Congratulations! The asymmetric key pair of workspace {tenant.id} has been reset.",

api/configs/enterprise/__init__.py

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

api/configs/feature/__init__.py

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

api/configs/middleware/__init__.py

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

api/controllers/API_SCHEMA_GUIDE.md

@@ -34,7 +34,6 @@ 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(...)`:
@@ -83,7 +82,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 dump_response(WorkflowRunNodeExecutionResponse, result)
+    return WorkflowRunNodeExecutionResponse.model_validate(result, from_attributes=True).model_dump(mode="json")
 ```
 
 ## Query Parameters
@@ -106,7 +105,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 dump_response(WorkflowRunPaginationResponse, result)
+    return WorkflowRunPaginationResponse.model_validate(result, from_attributes=True).model_dump(mode="json")
 ```
 
 Do not do this for GET query parameters:
@@ -146,25 +145,10 @@ def post(...):
 Serialize explicitly:
 
 ```python
-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,
-    },
-)
+return WorkflowRunNodeExecutionResponse.model_validate(
+    workflow_node_execution,
+    from_attributes=True,
+).model_dump(mode="json")
 ```
 
 If the service can return `None`, translate that into the expected HTTP error before validation:
@@ -174,12 +158,9 @@ workflow_run = service.get_workflow_run(...)
 if workflow_run is None:
     raise NotFound("Workflow run not found")
 
-return dump_response(WorkflowRunDetailResponse, workflow_run)
+return WorkflowRunDetailResponse.model_validate(workflow_run, from_attributes=True).model_dump(mode="json")
 ```
 
-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:
@@ -209,3 +190,4 @@ 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,21 +1,6 @@
-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,7 +39,6 @@ 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):
@@ -66,35 +65,6 @@ def _register_schema_model(namespace: Namespace, model: type[BaseModel], *, mode
     )
 
 
-def _swagger_2_compatible_schema(value: Any) -> Any:
-    if isinstance(value, list):
-        return [_swagger_2_compatible_schema(item) for item in value]
-
-    if not isinstance(value, dict):
-        return value
-
-    converted = {key: _swagger_2_compatible_schema(child) for key, child in value.items()}
-    any_of = value.get("anyOf")
-    if not isinstance(any_of, list):
-        return converted
-
-    non_null_candidates = [
-        candidate for candidate in any_of if isinstance(candidate, Mapping) and candidate.get("type") != "null"
-    ]
-    has_null_candidate = any(isinstance(candidate, Mapping) and candidate.get("type") == "null" for candidate in any_of)
-    if not has_null_candidate or len(non_null_candidates) != 1:
-        return converted
-
-    non_null_schema = _swagger_2_compatible_schema(dict(non_null_candidates[0]))
-    if not isinstance(non_null_schema, dict):
-        return converted
-
-    converted.pop("anyOf", None)
-    converted.update(non_null_schema)
-    converted["x-nullable"] = True
-    return converted
-
-
 def register_schema_model(namespace: Namespace, model: type[BaseModel]) -> None:
     """Register a BaseModel and its nested schema definitions for Swagger documentation."""
 

api/controllers/console/__init__.py

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

api/controllers/console/admin.py

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

api/controllers/console/apikey.py

@@ -11,7 +11,6 @@ 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
@@ -22,6 +21,12 @@ 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
@@ -32,7 +37,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,6 +3,7 @@ import re
 import uuid
 from datetime import datetime
 from typing import Any, Literal
+from uuid import UUID
 
 from flask import request
 from flask_restx import Resource
@@ -33,7 +34,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, to_timestamp
+from libs.helper import build_icon_url
 from libs.login import current_account_with_tenant, login_required
 from models import App, DatasetPermissionEnum, Workflow
 from models.model import IconType
@@ -177,6 +178,12 @@ 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
@@ -193,7 +200,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):
@@ -207,7 +214,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):
@@ -268,7 +275,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):
@@ -311,7 +318,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):
@@ -354,7 +361,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):
@@ -384,7 +391,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):
@@ -849,11 +856,10 @@ class AppTraceApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @get_app_model
-    def get(self, app_model):
+    def get(self, app_id: UUID):
         """Get app trace"""
         with session_factory.create_session() as session:
-            app_trace_config = OpsTraceManager.get_app_tracing_config(app_model.id, session)
+            app_trace_config = OpsTraceManager.get_app_tracing_config(str(app_id), session)
 
         return app_trace_config
 
@@ -867,13 +873,12 @@ class AppTraceApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
-    @get_app_model
-    def post(self, app_model):
+    def post(self, app_id: UUID):
         # add app trace
         args = AppTracePayload.model_validate(console_ns.payload)
 
         OpsTraceManager.update_app_tracing_config(
-            app_id=app_model.id,
+            app_id=str(app_id),
             enabled=args.enabled,
             tracing_provider=args.tracing_provider,
         )

api/controllers/console/app/conversation_variables.py

@@ -16,7 +16,6 @@ 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
@@ -26,6 +25,12 @@ 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
@@ -60,7 +65,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,7 +13,6 @@ 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
@@ -31,6 +30,12 @@ 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
@@ -54,7 +59,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,9 +37,10 @@ from fields.conversation_fields import (
     JSONValue,
     MessageFile,
     format_files_contained,
+    to_timestamp,
 )
 from graphon.model_runtime.errors.invoke import InvokeError
-from libs.helper import to_timestamp, uuid_value
+from libs.helper import 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
@@ -143,7 +144,9 @@ class MessageDetailResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        return to_timestamp(value)
+        if isinstance(value, datetime):
+            return to_timestamp(value)
+        return value
 
 
 class MessageInfiniteScrollPaginationResponse(ResponseModel):

api/controllers/console/app/ops_trace.py

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

api/controllers/console/app/workflow.py

@@ -1,22 +1,17 @@
 import json
 import logging
 from collections.abc import Sequence
-from datetime import datetime
-from typing import Any, NotRequired, TypedDict
+from typing import Any
 
 from flask import abort, request
-from flask_restx import Resource, fields
-from pydantic import AliasChoices, BaseModel, Field, ValidationError, field_validator
+from flask_restx import Resource, fields, marshal, marshal_with
+from pydantic import 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_response_schema_models,
-    register_schema_models,
-)
+from controllers.common.schema import register_response_schema_model, register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.error import ConversationCompletedError, DraftWorkflowNotExist, DraftWorkflowNotSync
 from controllers.console.app.wraps import get_app_model
@@ -27,7 +22,6 @@ 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
@@ -40,18 +34,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.base import ResponseModel
-from fields.member_fields import SimpleAccount
+from fields.member_fields import simple_account_fields
+from fields.online_user_fields import online_user_list_fields
+from fields.workflow_fields import workflow_fields, workflow_pagination_fields
 from fields.workflow_run_fields import WorkflowRunNodeExecutionResponse
 from graphon.enums import NodeType
 from graphon.file import File
 from graphon.file import helpers as file_helpers
 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, dump_response, to_timestamp, uuid_value
+from libs.helper import TimestampField, uuid_value
 from libs.login import current_account_with_tenant, login_required
 from models import App
 from models.model import AppMode
@@ -70,15 +64,42 @@ 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
 
-class EnvironmentVariableResponseDict(TypedDict):
-    value_type: str
-    id: NotRequired[str]
-    name: NotRequired[str]
-    value: NotRequired[Any]
-    description: NotRequired[str | None]
+# 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 SyncDraftWorkflowPayload(BaseModel):
@@ -149,110 +170,6 @@ 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
 
@@ -280,17 +197,6 @@ 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
@@ -312,56 +218,18 @@ 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",
-        console_ns.models[WorkflowResponse.__name__],
-    )
+    @console_ns.response(200, "Draft workflow retrieved successfully", workflow_model)
     @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):
         """
@@ -374,8 +242,8 @@ class DraftWorkflowApi(Resource):
         if not workflow:
             raise DraftWorkflowNotExist()
 
-        # return workflow, if not found, return 404
-        return dump_response(WorkflowResponse, workflow)
+        # return workflow, if not found, return None (initiate graph by frontend)
+        return workflow
 
     @setup_required
     @login_required
@@ -949,15 +817,13 @@ 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, or null if not found",
-        console_ns.models[WorkflowResponse.__name__],
-    )
+    @console_ns.response(200, "Published workflow retrieved successfully", workflow_model)
+    @console_ns.response(404, "Published 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):
         """
@@ -968,10 +834,7 @@ class PublishedWorkflowApi(Resource):
         workflow = workflow_service.get_published_workflow(app_model=app_model)
 
         # return workflow, if not found, return None
-        if workflow is None:
-            return None
-
-        return dump_response(WorkflowResponse, workflow)
+        return workflow
 
     @console_ns.expect(console_ns.models[PublishWorkflowPayload.__name__])
     @setup_required
@@ -1130,11 +993,7 @@ 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",
-        console_ns.models[WorkflowPaginationResponse.__name__],
-    )
+    @console_ns.response(200, "Published workflows retrieved successfully", workflow_pagination_model)
     @setup_required
     @login_required
     @account_initialization_required
@@ -1166,14 +1025,14 @@ class PublishedAllWorkflowApi(Resource):
                 user_id=user_id,
                 named_only=named_only,
             )
-            return WorkflowPaginationResponse.model_validate(
-                {
-                    "items": workflows,
-                    "page": page,
-                    "limit": limit,
-                    "has_more": has_more,
-                }
-            ).model_dump(mode="json")
+            serialized_workflows = marshal(workflows, workflow_fields_copy)
+
+            return {
+                "items": serialized_workflows,
+                "page": page,
+                "limit": limit,
+                "has_more": has_more,
+            }
 
 
 @console_ns.route("/apps/<uuid:app_id>/workflows/<string:workflow_id>/restore")
@@ -1219,13 +1078,14 @@ 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", console_ns.models[WorkflowResponse.__name__])
+    @console_ns.response(200, "Workflow updated successfully", workflow_model)
     @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):
         """
@@ -1259,7 +1119,7 @@ class WorkflowByIdApi(Resource):
             if not workflow:
                 raise NotFound("Workflow not found")
 
-        return dump_response(WorkflowResponse, workflow)
+        return workflow
 
     @setup_required
     @login_required
@@ -1544,16 +1404,12 @@ 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 {})
 
@@ -1596,18 +1452,10 @@ 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:
-                        avatar = file_helpers.get_signed_file_url(avatar)
+                        user_info["avatar"] = file_helpers.get_signed_file_url(avatar)
                     except Exception as exc:
                         logger.warning(
                             "Failed to sign workflow online user avatar; using original value. "
@@ -1617,7 +1465,7 @@ class WorkflowOnlineUsersApi(Resource):
                             exc,
                         )
 
-                users.append({"user_id": user_id, "username": username, "avatar": avatar})
+                users.append(user_info)
             results.append({"app_id": app_id, "users": users})
 
-        return WorkflowOnlineUsersResponse.model_validate({"data": results}).model_dump(mode="json")
+        return {"data": results}

api/controllers/console/app/workflow_app_log.py

@@ -16,7 +16,6 @@ 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
@@ -83,7 +82,9 @@ 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)
+        if isinstance(value, datetime):
+            return int(value.timestamp())
+        return value
 
 
 class WorkflowRunForArchivedLogResponse(ResponseModel):
@@ -116,7 +117,9 @@ class WorkflowAppLogPartialResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return to_timestamp(value)
+        if isinstance(value, datetime):
+            return int(value.timestamp())
+        return value
 
 
 class WorkflowArchivedLogPartialResponse(ResponseModel):
@@ -130,7 +133,9 @@ class WorkflowArchivedLogPartialResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return to_timestamp(value)
+        if isinstance(value, datetime):
+            return int(value.timestamp())
+        return value
 
 
 class WorkflowAppLogPaginationResponse(ResponseModel):

api/controllers/console/app/workflow_comment.py

@@ -1,16 +1,22 @@
 import logging
-from datetime import datetime
 
-from flask_restx import Resource
-from pydantic import BaseModel, Field, TypeAdapter, computed_field, field_validator
+from flask_restx import Resource, marshal_with
+from pydantic import BaseModel, Field, TypeAdapter
 
-from controllers.common.schema import register_response_schema_models, register_schema_models
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
-from fields.base import ResponseModel
 from fields.member_fields import AccountWithRole
-from libs.helper import build_avatar_url, dump_response, to_timestamp
+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.login import current_user, login_required
 from models import App
 from services.account_service import TenantService
@@ -45,138 +51,6 @@ 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,
@@ -185,19 +59,17 @@ register_schema_models(
     WorkflowCommentUpdatePayload,
     WorkflowCommentReplyPayload,
 )
-register_response_schema_models(
-    console_ns,
-    WorkflowCommentAccount,
-    WorkflowCommentReply,
-    WorkflowCommentMention,
-    WorkflowCommentBasic,
-    WorkflowCommentBasicList,
-    WorkflowCommentDetail,
-    WorkflowCommentCreate,
-    WorkflowCommentUpdate,
-    WorkflowCommentResolve,
-    WorkflowCommentReplyCreate,
-    WorkflowCommentReplyUpdate,
+
+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
 )
 
 
@@ -208,26 +80,28 @@ 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", console_ns.models[WorkflowCommentBasicList.__name__])
+    @console_ns.response(200, "Comments retrieved successfully", workflow_comment_basic_model)
     @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 WorkflowCommentBasicList.model_validate({"data": comments}).model_dump(mode="json")
+        return comments
 
     @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", console_ns.models[WorkflowCommentCreate.__name__])
+    @console_ns.response(201, "Comment created successfully", workflow_comment_create_model)
     @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."""
@@ -243,7 +117,7 @@ class WorkflowCommentListApi(Resource):
             mentioned_user_ids=payload.mentioned_user_ids,
         )
 
-        return dump_response(WorkflowCommentCreate, result), 201
+        return result, 201
 
 
 @console_ns.route("/apps/<uuid:app_id>/workflow/comments/<string:comment_id>")
@@ -253,28 +127,30 @@ 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", console_ns.models[WorkflowCommentDetail.__name__])
+    @console_ns.response(200, "Comment retrieved successfully", workflow_comment_detail_model)
     @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 dump_response(WorkflowCommentDetail, comment)
+        return 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", console_ns.models[WorkflowCommentUpdate.__name__])
+    @console_ns.response(200, "Comment updated successfully", workflow_comment_update_model)
     @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."""
@@ -291,7 +167,7 @@ class WorkflowCommentDetailApi(Resource):
             mentioned_user_ids=payload.mentioned_user_ids,
         )
 
-        return dump_response(WorkflowCommentUpdate, result)
+        return result
 
     @console_ns.doc("delete_workflow_comment")
     @console_ns.doc(description="Delete a workflow comment")
@@ -321,11 +197,12 @@ 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", console_ns.models[WorkflowCommentResolve.__name__])
+    @console_ns.response(200, "Comment resolved successfully", workflow_comment_resolve_model)
     @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."""
@@ -336,7 +213,7 @@ class WorkflowCommentResolveApi(Resource):
             user_id=current_user.id,
         )
 
-        return dump_response(WorkflowCommentResolve, comment)
+        return comment
 
 
 @console_ns.route("/apps/<uuid:app_id>/workflow/comments/<string:comment_id>/replies")
@@ -347,11 +224,12 @@ 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", console_ns.models[WorkflowCommentReplyCreate.__name__])
+    @console_ns.response(201, "Reply created successfully", workflow_comment_reply_create_model)
     @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."""
@@ -369,7 +247,7 @@ class WorkflowCommentReplyApi(Resource):
             mentioned_user_ids=payload.mentioned_user_ids,
         )
 
-        return dump_response(WorkflowCommentReplyCreate, result), 201
+        return result, 201
 
 
 @console_ns.route("/apps/<uuid:app_id>/workflow/comments/<string:comment_id>/replies/<string:reply_id>")
@@ -380,11 +258,12 @@ 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", console_ns.models[WorkflowCommentReplyUpdate.__name__])
+    @console_ns.response(200, "Reply updated successfully", workflow_comment_reply_update_model)
     @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."""
@@ -405,7 +284,7 @@ class WorkflowCommentReplyDetailApi(Resource):
             mentioned_user_ids=payload.mentioned_user_ids,
         )
 
-        return dump_response(WorkflowCommentReplyUpdate, reply)
+        return 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 get_valid_language, languages
+from constants.languages import languages
 from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.auth.error import (
@@ -15,12 +15,11 @@ 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 AccountRegisterError
+from services.errors.account import AccountNotFoundError, AccountRegisterError
 
 from ..error import AccountInFreezeError, EmailSendIpLimitError
 from ..wraps import email_password_login_enabled, email_register_enabled, setup_required
@@ -41,21 +40,12 @@ 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)
 
@@ -154,32 +144,26 @@ class EmailRegisterResetApi(Resource):
 
         if account:
             raise EmailAlreadyInUseError()
-
-        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)
+        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)
 
         return {"result": "success", "data": token_pair.model_dump()}
 
-    def _create_new_account(
-        self,
-        email: str,
-        password: str,
-        timezone: str | None = None,
-        language: str | None = None,
-    ) -> Account:
+    def _create_new_account(self, email: str, password: str) -> Account | None:
+        # Create new account if allowed
+        account = None
         try:
-            return AccountService.create_account_and_tenant(
+            account = AccountService.create_account_and_tenant(
                 email=email,
                 name=email,
                 password=password,
-                interface_language=get_valid_language(language),
-                timezone=timezone,
+                interface_language=languages[0],
             )
         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, field_validator
+from pydantic import BaseModel, Field
 from werkzeug.exceptions import Unauthorized
 
 import services
@@ -34,7 +34,6 @@ 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,
@@ -70,14 +69,6 @@ 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)
@@ -297,7 +288,6 @@ 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,8 +12,7 @@ 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.helper import timezone as validate_timezone_string
-from libs.oauth import GitHubOAuth, GoogleOAuth, OAuthUserInfo, decode_oauth_state
+from libs.oauth import GitHubOAuth, GoogleOAuth, OAuthUserInfo
 from libs.token import (
     set_access_token_to_cookie,
     set_csrf_token_to_cookie,
@@ -54,31 +53,6 @@ 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")
@@ -90,19 +64,13 @@ 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,
-            timezone=timezone,
-            language=language,
-        )
+        auth_url = oauth_provider.get_authorization_url(invite_token=invite_token)
         return redirect(auth_url)
 
 
@@ -128,10 +96,9 @@ class OAuthCallback(Resource):
 
         code = request.args.get("code")
         state = request.args.get("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"))
+        invite_token = None
+        if state:
+            invite_token = state
 
         if not code:
             return {"error": "Authorization code is required"}, 400
@@ -162,7 +129,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, timezone=timezone, language=language)
+            account, oauth_new_user = _generate_account(provider, user_info)
         except AccountNotFoundError:
             return redirect(f"{dify_config.CONSOLE_WEB_URL}/signin?message=Account not found.")
         except (WorkSpaceNotFoundError, WorkSpaceNotAllowedCreateError):
@@ -217,12 +184,7 @@ def _get_account_by_openid_or_email(provider: str, user_info: OAuthUserInfo) ->
     return account
 
 
-def _generate_account(
-    provider: str,
-    user_info: OAuthUserInfo,
-    timezone: str | None = None,
-    language: str | None = None,
-) -> tuple[Account, bool]:
+def _generate_account(provider: str, user_info: OAuthUserInfo) -> tuple[Account, bool]:
     # Get account by openid or email.
     account = _get_account_by_openid_or_email(provider, user_info)
     oauth_new_user = False
@@ -249,19 +211,26 @@ def _generate_account(
                         "30 days and is temporarily unavailable for new account registration"
                     )
                 )
-            raise AccountRegisterError(description=("Invalid email or password"))
+            else:
+                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,7 +39,6 @@ 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
@@ -72,6 +71,12 @@ 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
@@ -96,7 +101,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):
@@ -147,7 +152,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,7 +8,6 @@ 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
@@ -20,6 +19,12 @@ 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
@@ -56,7 +61,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
+from flask_restx import Resource, marshal_with  # type: ignore
 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,
-    WorkflowPaginationResponse,
-    WorkflowResponse,
+    workflow_model,
+    workflow_pagination_model,
 )
 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, dump_response
+from libs.helper import TimestampField, UUIDStrOrEmpty
 from libs.login import current_account_with_tenant, current_user, login_required
 from models import Account
 from models.dataset import Pipeline
@@ -142,17 +142,12 @@ 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
@@ -164,8 +159,8 @@ class DraftRagPipelineApi(Resource):
         if not workflow:
             raise DraftWorkflowNotExist()
 
-        # return workflow, if not found, return 404
-        return dump_response(WorkflowResponse, workflow)
+        # return workflow, if not found, return None (initiate graph by frontend)
+        return workflow
 
     @setup_required
     @login_required
@@ -481,16 +476,12 @@ 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
@@ -503,10 +494,7 @@ class PublishedRagPipelineApi(Resource):
         workflow = rag_pipeline_service.get_published_workflow(pipeline=pipeline)
 
         # return workflow, if not found, return None
-        if workflow is None:
-            return None
-
-        return dump_response(WorkflowResponse, workflow)
+        return workflow
 
     @setup_required
     @login_required
@@ -579,17 +567,12 @@ 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
@@ -618,14 +601,12 @@ class PublishedAllRagPipelineApi(Resource):
                 named_only=named_only,
             )
 
-            return WorkflowPaginationResponse.model_validate(
-                {
-                    "items": workflows,
-                    "page": page,
-                    "limit": limit,
-                    "has_more": has_more,
-                }
-            ).model_dump(mode="json")
+            return {
+                "items": workflows,
+                "page": page,
+                "limit": limit,
+                "has_more": has_more,
+            }
 
 
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/<string:workflow_id>/restore")
@@ -660,15 +641,12 @@ 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
@@ -697,7 +675,7 @@ class RagPipelineByIdApi(Resource):
             if not workflow:
                 raise NotFound("Workflow not found")
 
-            return dump_response(WorkflowResponse, workflow)
+            return workflow
 
     @setup_required
     @login_required

api/controllers/console/explore/installed_app.py

@@ -16,7 +16,6 @@ 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
@@ -106,7 +105,9 @@ class InstalledAppResponse(ResponseModel):
     @field_validator("last_used_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return to_timestamp(value)
+        if isinstance(value, datetime):
+            return int(value.timestamp())
+        return value
 
 
 class InstalledAppListResponse(ResponseModel):

api/controllers/console/extension.py

@@ -7,7 +7,6 @@ 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
@@ -41,6 +40,12 @@ 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
@@ -56,7 +61,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,8 +105,7 @@ class FilePreviewApi(Resource):
     @account_initialization_required
     def get(self, file_id):
         file_id = str(file_id)
-        _, tenant_id = current_account_with_tenant()
-        text = FileService(db.engine).get_file_preview(file_id, tenant_id)
+        text = FileService(db.engine).get_file_preview(file_id)
         return {"content": text}
 
 

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, to_timestamp
+from libs.helper import EmailStr, extract_remote_ip, timezone
 from libs.login import current_account_with_tenant, login_required
 from models import AccountIntegrate, InvitationCode
 from models.account import AccountStatus, InvitationCodeStatus
@@ -50,6 +50,12 @@ from models.enums import CreatorUserRole
 from models.model import UploadFile
 from services.account_service import AccountService
 from services.billing_service import BillingService
+from services.entities.auth_entities import (
+    ChangeEmailNewEmailToken,
+    ChangeEmailNewEmailVerifiedToken,
+    ChangeEmailOldEmailToken,
+    ChangeEmailOldEmailVerifiedToken,
+)
 from services.errors.account import CurrentPasswordIncorrectError as ServiceCurrentPasswordIncorrectError
 
 
@@ -185,6 +191,12 @@ 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
@@ -194,7 +206,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):
@@ -214,7 +226,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):
@@ -601,8 +613,8 @@ class ChangeEmailSendEmailApi(Resource):
             language = "zh-Hans"
         else:
             language = "en-US"
-        account = None
-        user_email = None
+        account = current_user
+        user_email = current_user.email
         email_for_sending = args.email.lower()
         # Default to the initial phase; any legacy/unexpected client input is
         # coerced back to `old_email` so we never trust the caller to declare
@@ -617,24 +629,18 @@ class ChangeEmailSendEmailApi(Resource):
             if reset_data is None:
                 raise InvalidTokenError()
 
-            # The token used to request a new-email code must come from the
-            # old-email verification step. This prevents the bypass described
-            # in GHSA-4q3w-q5mc-45rq where the phase-1 token was reused here.
-            token_phase = reset_data.get(AccountService.CHANGE_EMAIL_TOKEN_PHASE_KEY)
-            if token_phase != AccountService.CHANGE_EMAIL_PHASE_OLD_VERIFIED:
+            if not isinstance(reset_data, ChangeEmailOldEmailVerifiedToken):
                 raise InvalidTokenError()
-            user_email = reset_data.get("email", "")
+            if not reset_data.is_bound_to_account(current_user.id):
+                raise InvalidTokenError()
+            user_email = reset_data.email
 
             if user_email.lower() != current_user.email.lower():
                 raise InvalidEmailError()
-
-            user_email = current_user.email
         else:
-            account = AccountService.get_account_by_email_with_case_fallback(args.email)
-            if account is None:
-                raise AccountNotFound()
-            email_for_sending = account.email
-            user_email = account.email
+            if email_for_sending != current_user.email.lower():
+                raise InvalidEmailError()
+            email_for_sending = current_user.email
 
         token = AccountService.send_change_email_email(
             account=account,
@@ -654,6 +660,7 @@ class ChangeEmailCheckApi(Resource):
     @login_required
     @account_initialization_required
     def post(self):
+        current_user, _ = current_account_with_tenant()
         payload = console_ns.payload or {}
         args = ChangeEmailValidityPayload.model_validate(payload)
 
@@ -666,42 +673,26 @@ class ChangeEmailCheckApi(Resource):
         token_data = AccountService.get_change_email_data(args.token)
         if token_data is None:
             raise InvalidTokenError()
+        if not token_data.is_bound_to_account(current_user.id):
+            raise InvalidTokenError()
 
-        token_email = token_data.get("email")
-        normalized_token_email = token_email.lower() if isinstance(token_email, str) else token_email
+        normalized_token_email = token_data.email.lower()
         if user_email != normalized_token_email:
             raise InvalidEmailError()
 
-        if args.code != token_data.get("code"):
+        if args.code != token_data.code:
             AccountService.add_change_email_error_rate_limit(user_email)
             raise EmailCodeError()
 
-        # Only advance tokens that were minted by the matching send-code step;
-        # refuse tokens that have already progressed or lack a phase marker so
-        # the chain `old_email -> old_email_verified -> new_email -> new_email_verified`
-        # is strictly enforced.
-        phase_transitions = {
-            AccountService.CHANGE_EMAIL_PHASE_OLD: AccountService.CHANGE_EMAIL_PHASE_OLD_VERIFIED,
-            AccountService.CHANGE_EMAIL_PHASE_NEW: AccountService.CHANGE_EMAIL_PHASE_NEW_VERIFIED,
-        }
-        token_phase = token_data.get(AccountService.CHANGE_EMAIL_TOKEN_PHASE_KEY)
-        if not isinstance(token_phase, str):
-            raise InvalidTokenError()
-        refreshed_phase = phase_transitions.get(token_phase)
-        if refreshed_phase is None:
+        if isinstance(token_data, ChangeEmailOldEmailToken | ChangeEmailNewEmailToken):
+            refreshed_token_data = token_data.promote()
+        else:
             raise InvalidTokenError()
 
         # Verified, revoke the first token
         AccountService.revoke_change_email_token(args.token)
 
-        # Refresh token data by generating a new token that carries the
-        # upgraded phase so later steps can check it.
-        _, new_token = AccountService.generate_change_email_token(
-            user_email,
-            code=args.code,
-            old_email=token_data.get("old_email"),
-            additional_data={AccountService.CHANGE_EMAIL_TOKEN_PHASE_KEY: refreshed_phase},
-        )
+        new_token = AccountService.generate_change_email_token(refreshed_token_data, current_user)
 
         AccountService.reset_change_email_error_rate_limit(user_email)
         return {"is_valid": True, "email": normalized_token_email, "token": new_token}
@@ -726,27 +717,22 @@ class ChangeEmailResetApi(Resource):
         if not AccountService.check_email_unique(normalized_new_email):
             raise EmailAlreadyInUseError()
 
+        current_user, _ = current_account_with_tenant()
         reset_data = AccountService.get_change_email_data(args.token)
         if not reset_data:
             raise InvalidTokenError()
+        if not reset_data.is_bound_to_account(current_user.id):
+            raise InvalidTokenError()
 
-        # Only tokens that completed both verification phases may be used to
-        # change the email. This closes GHSA-4q3w-q5mc-45rq where a token from
-        # the initial send-code step could be replayed directly here.
-        token_phase = reset_data.get(AccountService.CHANGE_EMAIL_TOKEN_PHASE_KEY)
-        if token_phase != AccountService.CHANGE_EMAIL_PHASE_NEW_VERIFIED:
+        if not isinstance(reset_data, ChangeEmailNewEmailVerifiedToken):
             raise InvalidTokenError()
 
         # Bind the new email to the token that was mailed and verified, so a
         # verified token cannot be reused with a different `new_email` value.
-        token_email = reset_data.get("email")
-        normalized_token_email = token_email.lower() if isinstance(token_email, str) else token_email
-        if normalized_token_email != normalized_new_email:
+        if reset_data.email.lower() != normalized_new_email:
             raise InvalidTokenError()
 
-        old_email = reset_data.get("old_email", "")
-        current_user, _ = current_account_with_tenant()
-        if current_user.email.lower() != old_email.lower():
+        if current_user.email.lower() != reset_data.old_email.lower():
             raise AccountNotFound()
 
         # Revoke only after all checks pass so failed attempts don't burn a

api/controllers/console/workspace/members.py

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

api/controllers/console/workspace/workspace.py

@@ -29,7 +29,7 @@ from controllers.console.wraps import (
 from enums.cloud_plan import CloudPlan
 from extensions.ext_database import db
 from fields.base import ResponseModel
-from libs.helper import TimestampField, to_timestamp
+from libs.helper import TimestampField
 from libs.login import current_account_with_tenant, login_required
 from models.account import Tenant, TenantCustomConfigDict, TenantStatus
 from services.account_service import TenantService
@@ -86,7 +86,9 @@ class TenantInfoResponse(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None):
-        return to_timestamp(value)
+        if isinstance(value, datetime):
+            return int(value.timestamp())
+        return 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, to_timestamp
+from libs.helper import UUIDStrOrEmpty
 from models.model import App, AppMode, EndUser
 from services.conversation_service import ConversationService
 
@@ -115,7 +115,9 @@ 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)
+        if isinstance(value, datetime):
+            return int(value.timestamp())
+        return 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, stringify_form_default_values
+from controllers.common.human_input import HumanInputFormSubmitPayload
 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,14 +28,30 @@ 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_form_default_values(definition_payload["default_values"]),
+        "resolved_default_values": _stringify_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,7 +39,6 @@ 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
@@ -69,6 +68,12 @@ 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)
 
@@ -104,7 +109,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):
@@ -128,7 +133,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):
@@ -149,7 +154,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/web/human_input_form.py

@@ -4,6 +4,7 @@ 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
@@ -12,12 +13,12 @@ from sqlalchemy import select
 from werkzeug.exceptions import Forbidden
 
 from configs import dify_config
-from controllers.common.human_input import HumanInputFormSubmitPayload, stringify_form_default_values
+from controllers.common.human_input import HumanInputFormSubmitPayload
 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, to_timestamp
+from libs.helper import RateLimiter, extract_remote_ip
 from models.account import TenantStatus
 from models.model import App, Site
 from services.human_input_service import Form, FormNotFoundError, HumanInputService
@@ -37,6 +38,22 @@ _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
@@ -52,9 +69,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_form_default_values(definition_payload["default_values"]),
+        "resolved_default_values": _stringify_default_values(definition_payload["default_values"]),
         "user_actions": definition_payload["user_actions"],
-        "expiration_time": to_timestamp(form.expiration_time),
+        "expiration_time": _to_timestamp(form.expiration_time),
     }
     if site_payload is not None:
         payload["site"] = site_payload

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

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

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

@@ -9,7 +9,7 @@ from datetime import datetime
 from threading import Thread
 from typing import Any, Union
 
-from sqlalchemy import select, update
+from sqlalchemy import select
 from sqlalchemy.orm import Session, sessionmaker
 
 from constants.tts_auto_play_timeout import TTS_AUTO_PLAY_TIMEOUT, TTS_AUTO_PLAY_YIELD_CPU_TIME
@@ -245,50 +245,49 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         """
         human_input_responses: list[HumanInputRequiredResponse] = []
         for stream_response in generator:
-            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
+            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
 
-                    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
+                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
 
         if human_input_responses:
             return self._build_paused_blocking_response_from_human_input(human_input_responses)
@@ -426,7 +425,11 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         self._workflow_run_id = run_id
 
         with self._database_session() as session:
-            session.execute(update(Message).where(Message.id == self._message_id).values(workflow_run_id=run_id))
+            message = self._get_message(session=session)
+            if not message:
+                raise ValueError(f"Message not found: {self._message_id}")
+
+            message.workflow_run_id = run_id
 
         workflow_start_resp = self._workflow_response_converter.workflow_start_to_stream_response(
             task_id=self._application_generate_entity.task_id,

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

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

api/core/app/apps/base_app_generator.py

@@ -195,23 +195,22 @@ class BaseAppGenerator:
             )
 
         if variable_entity.type == VariableEntityType.NUMBER:
-            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}")
+            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 variable_entity.type:
             case VariableEntityType.SELECT:
@@ -242,18 +241,17 @@ class BaseAppGenerator:
                         f"{variable_entity.variable} in input form must be less than {variable_entity.max_length} files"
                     )
             case VariableEntityType.CHECKBOX:
-                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
+                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
             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,24 +45,20 @@ class WorkflowAppGenerateResponseConverter(AppGenerateResponseConverter[Workflow
             chunk = cast(WorkflowAppStreamResponse, chunk)
             sub_stream_response = chunk.stream_response
 
-            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())
+            if isinstance(sub_stream_response, PingStreamResponse):
+                yield "ping"
+                continue
 
+            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
@@ -78,28 +74,20 @@ class WorkflowAppGenerateResponseConverter(AppGenerateResponseConverter[Workflow
             chunk = cast(WorkflowAppStreamResponse, chunk)
             sub_stream_response = chunk.stream_response
 
-            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())
+            if isinstance(sub_stream_response, PingStreamResponse):
+                yield "ping"
+                continue
 
+            response_chunk = {
+                "event": sub_stream_response.event.value,
+                "workflow_run_id": chunk.workflow_run_id,
+            }
+
+            if isinstance(sub_stream_response, ErrorStreamResponse):
+                data = cls._error_to_stream_response(sub_stream_response.err)
+                response_chunk.update(cast(dict, data))
+            elif isinstance(sub_stream_response, NodeStartStreamResponse | NodeFinishStreamResponse):
+                response_chunk.update(cast(dict, sub_stream_response.to_ignore_detail_dict()))
+            else:
+                response_chunk.update(sub_stream_response.model_dump())
             yield response_chunk

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

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

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

@@ -52,24 +52,20 @@ class WorkflowAppGenerateResponseConverter(
             chunk = cast(WorkflowAppStreamResponse, chunk)
             sub_stream_response = chunk.stream_response
 
-            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"))
+            if isinstance(sub_stream_response, PingStreamResponse):
+                yield "ping"
+                continue
 
+            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
@@ -85,28 +81,20 @@ class WorkflowAppGenerateResponseConverter(
             chunk = cast(WorkflowAppStreamResponse, chunk)
             sub_stream_response = chunk.stream_response
 
-            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"))
+            if isinstance(sub_stream_response, PingStreamResponse):
+                yield "ping"
+                continue
 
+            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,51 +145,50 @@ class WorkflowAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         """
         human_input_responses: list[HumanInputRequiredResponse] = []
         for stream_response in generator:
-            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 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
 
         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,281 +399,278 @@ class WorkflowBasedAppRunner:
         :param workflow_entry: workflow entry
         :param event: event
         """
-        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)
+        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,
                 )
-            case GraphRunFailedEvent():
-                self._publish_event(
-                    QueueWorkflowFailedEvent(error=event.error, exceptions_count=event.exceptions_count)
-                )
-            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,
-                    )
-                )
-            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(
+            )
+        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,
+                )
+            )
+        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,
+                )
+            )
+        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,
+                    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,
-                    outputs=node_run_result.outputs,
+                    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,
                 )
-                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,
-                    )
-                )
-            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(
+            )
+        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,
+                )
+            )
+        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,
+                    node_type=event.node_type,
+                    start_at=event.start_at,
+                    finished_at=event.finished_at,
                     inputs=inputs,
-                    outputs=node_run_result.outputs,
+                    process_data=process_data,
+                    outputs=outputs,
+                    execution_metadata=execution_metadata,
+                    in_iteration_id=event.in_iteration_id,
+                    in_loop_id=event.in_loop_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(
+            )
+        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,
                     node_type=event.node_type,
+                    start_at=event.start_at,
+                    finished_at=event.finished_at,
                     inputs=event.node_run_result.inputs,
-                    outputs=event.node_run_result.outputs,
+                    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,
                 )
-                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(
+            )
+        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,
                     node_type=event.node_type,
+                    start_at=event.start_at,
+                    finished_at=event.finished_at,
                     inputs=event.node_run_result.inputs,
-                    outputs=event.node_run_result.outputs,
+                    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,
                 )
-                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, 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,
                 )
-            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, 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 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, 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 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, 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 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, 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 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, (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 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, 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 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, 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 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,
-                    )
+            )
+        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,
                 )
+            )
 
     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 FormInputConfig, UserActionConfig
+from graphon.nodes.human_input.entities import FormInput, UserAction
 
 
 class AnnotationReplyAccount(BaseModel):
@@ -284,8 +284,8 @@ class HumanInputRequiredResponse(StreamResponse):
         node_id: str
         node_title: str
         form_content: str
-        inputs: Sequence[FormInputConfig] = Field(default_factory=list)
-        actions: Sequence[UserActionConfig] = Field(default_factory=list)
+        inputs: Sequence[FormInput] = Field(default_factory=list)
+        actions: Sequence[UserAction] = 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[FormInputConfig] = Field(default_factory=list)
-    actions: Sequence[UserActionConfig] = Field(default_factory=list)
+    inputs: Sequence[FormInput] = Field(default_factory=list)
+    actions: Sequence[UserAction] = 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/task_pipeline/easy_ui_based_generate_task_pipeline.py

@@ -140,43 +140,42 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
         :return:
         """
         for stream_response in generator:
-            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,
-                            ),
-                        )
+            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,
+                        ),
+                    )
 
-                    return response
-                case _:
-                    continue
+                return response
+            else:
+                continue
 
         raise RuntimeError("queue listening stopped unexpectedly.")
 
@@ -266,107 +265,104 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
                 publisher.publish(message)
             event = message.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)
+            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)
 
-                    # handle output moderation
-                    output_moderation_answer = self.handle_output_moderation_when_task_finished(
-                        self._task_state.llm_result.message.get_text_content()
+                # 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
                     )
-                    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
-                        )
 
-                    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
-                            )
-                            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
-
-                    # 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 _:
+                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 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,
+                            )
+                            continue
+
+                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:
+                    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,31 +105,52 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
         self._node_sequence = 0
 
     def on_event(self, event: GraphEngineEvent) -> None:
-        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)
+        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)
 
     def on_graph_end(self, error: Exception | None) -> None:
         return

api/core/base/tts/app_generator_tts_publisher.py

@@ -95,14 +95,13 @@ class AppGeneratorTTSPublisher:
                     message_content = message.event.chunk.delta.message.content
                     if not message_content:
                         continue
-                    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
+                    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
                 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 FormInputConfig, UserActionConfig
+from graphon.nodes.human_input.entities import FormInput, UserAction
 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[FormInputConfig] = Field(default_factory=list)
-    actions: Sequence[UserActionConfig] = Field(default_factory=list)
+    inputs: Sequence[FormInput] = Field(default_factory=list)
+    actions: Sequence[UserAction] = Field(default_factory=list)
     display_in_ui: bool = False
     form_token: str | None = None
     resolved_default_values: Mapping[str, Any] = Field(default_factory=dict)

api/core/entities/provider_configuration.py

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

api/core/errors/error.py

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

api/core/helper/credential_utils.py

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

api/core/llm_generator/llm_generator.py

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

api/core/llm_generator/output_parser/structured_output.py

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

api/core/mcp/types.py

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

api/core/model_manager.py

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

api/core/prompt/advanced_prompt_transform.py

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

api/core/rag/extractor/extract_processor.py

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

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

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

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

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

api/core/tools/mcp_tool/tool.py

@@ -67,27 +67,23 @@ class MCPTool(Tool):
 
         # handle dify tool output
         for content in result.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))
+            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))
 
         # handle MCP structured output
         if self.entity.output_schema and result.structuredContent:

api/core/tools/tool_manager.py

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

api/core/workflow/node_factory.py

@@ -6,11 +6,11 @@ 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,6 +39,7 @@ 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
@@ -47,7 +48,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 CodeExecutorProtocol
+from graphon.nodes.code.code_node import WorkflowCodeExecutor
 from graphon.nodes.code.entities import CodeLanguage
 from graphon.nodes.code.limits import CodeNodeLimits
 from graphon.nodes.document_extractor import UnstructuredApiConfig
@@ -228,14 +229,10 @@ 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
 
-    # 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:
+    with Session(db.engine, expire_on_commit=False) as session:
         stmt = select(Conversation).where(Conversation.app_id == app_id, Conversation.id == conversation_id)
         conversation = session.scalar(stmt)
         if not conversation:
@@ -289,7 +286,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: CodeExecutorProtocol = DefaultWorkflowCodeExecutor()
+        self._code_executor: WorkflowCodeExecutor = DefaultWorkflowCodeExecutor()
         self._code_limits = CodeNodeLimits(
             max_string_length=dify_config.CODE_MAX_STRING_LENGTH,
             max_number=dify_config.CODE_MAX_NUMBER,
@@ -397,7 +394,6 @@ 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(
@@ -435,7 +431,7 @@ class DifyNodeFactory(NodeFactory):
                 include_jinja2_template_renderer=False,
             ),
             BuiltinNodeTypes.TOOL: lambda: {
-                "tool_file_manager": self._bound_tool_file_manager_factory(),
+                "tool_file_manager_factory": self._bound_tool_file_manager_factory(),
                 "runtime": self._tool_runtime,
             },
             BuiltinNodeTypes.AGENT: lambda: {

api/core/workflow/node_runtime.py

@@ -49,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 (
-    LLMProtocol,
+    PreparedLLMProtocol,
     PromptMessageSerializerProtocol,
     RetrieverAttachmentLoaderProtocol,
 )
@@ -140,7 +140,7 @@ class DifyFileReferenceFactory(FileReferenceFactoryProtocol):
         )
 
 
-class DifyPreparedLLM(LLMProtocol):
+class DifyPreparedLLM(PreparedLLMProtocol):
     """Workflow-layer adapter that hides the full `ModelInstance` API from `graphon` nodes."""
 
     def __init__(self, model_instance: ModelInstance) -> None:
@@ -608,67 +608,64 @@ class DifyToolNodeRuntime(ToolNodeRuntimeProtocol):
 
         from core.tools.entities.tool_entities import ToolInvokeMessage as CoreToolInvokeMessage
 
-        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__}")
+        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__}")
 
     @staticmethod
     def _map_invocation_exception(exc: Exception, *, provider_name: str) -> ToolNodeError:
-        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))
+        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))
 
 
 class DifyHumanInputNodeRuntime(HumanInputNodeRuntimeProtocol):

api/dev/generate_swagger_markdown_docs.py

@@ -8,7 +8,6 @@ OpenAPI output early.
 from __future__ import annotations
 
 import argparse
-import json
 import logging
 import subprocess
 import sys
@@ -29,125 +28,6 @@ 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:
@@ -177,10 +57,7 @@ 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 = _patch_union_schema_markdown(
-            temp_markdown_path.read_text(encoding="utf-8"),
-            spec_path,
-        )
+        converted_markdown = temp_markdown_path.read_text(encoding="utf-8")
         if not converted_markdown.strip():
             raise RuntimeError(f"swagger-markdown wrote an empty document for {markdown_path}")
 

api/enterprise/telemetry/enterprise_trace.py

@@ -77,29 +77,28 @@ class EnterpriseOtelTrace:
         self._exporter = exporter
 
     def trace(self, trace_info: BaseTraceInfo) -> None:
-        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")
+        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")
 
     def _common_attrs(self, trace_info: BaseTraceInfo) -> dict[str, Any]:
         metadata = self._metadata(trace_info)

api/extensions/ext_app_metrics.py

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

api/extensions/ext_commands.py

@@ -12,6 +12,7 @@ def init_app(app: DifyApp):
         clear_orphaned_file_records,
         convert_to_agent_apps,
         create_tenant,
+        data_migrate,
         delete_archived_workflow_runs,
         export_app_messages,
         extract_plugins,
@@ -44,6 +45,7 @@ def init_app(app: DifyApp):
         convert_to_agent_apps,
         add_qdrant_index,
         create_tenant,
+        data_migrate,
         upgrade_db,
         fix_app_site_missing,
         migrate_data_for_plugin,

api/extensions/ext_database.py

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

api/extensions/ext_import_modules.py

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

api/extensions/ext_mail.py

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

api/extensions/ext_socketio.py

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

api/extensions/logstore/aliyun_logstore.py

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

api/fields/annotation_fields.py

@@ -5,7 +5,12 @@ from datetime import datetime
 from pydantic import Field, field_validator
 
 from fields.base import ResponseModel
-from libs.helper import to_timestamp
+
+
+def _to_timestamp(value: datetime | int | None) -> int | None:
+    if isinstance(value, datetime):
+        return int(value.timestamp())
+    return value
 
 
 class Annotation(ResponseModel):
@@ -18,7 +23,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):
@@ -45,7 +50,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,7 +7,6 @@ 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
 
@@ -48,7 +47,9 @@ class SimpleConversation(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return to_timestamp(value)
+        if isinstance(value, datetime):
+            return to_timestamp(value)
+        return value
 
 
 class ConversationInfiniteScrollPagination(ResponseModel):
@@ -89,7 +90,9 @@ class ConversationAnnotation(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        return to_timestamp(value)
+        if isinstance(value, datetime):
+            return to_timestamp(value)
+        return value
 
 
 class ConversationAnnotationHitHistory(ResponseModel):
@@ -100,7 +103,9 @@ class ConversationAnnotationHitHistory(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        return to_timestamp(value)
+        if isinstance(value, datetime):
+            return to_timestamp(value)
+        return value
 
 
 class AgentThought(ResponseModel):
@@ -120,7 +125,9 @@ class AgentThought(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        return to_timestamp(value)
+        if isinstance(value, datetime):
+            return to_timestamp(value)
+        return value
 
     @model_validator(mode="after")
     def _fallback_chain_id(self):
@@ -162,7 +169,9 @@ class MessageDetail(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        return to_timestamp(value)
+        if isinstance(value, datetime):
+            return to_timestamp(value)
+        return value
 
 
 class FeedbackStat(ResponseModel):
@@ -228,7 +237,9 @@ class Conversation(ResponseModel):
     @field_validator("read_at", "created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return to_timestamp(value)
+        if isinstance(value, datetime):
+            return to_timestamp(value)
+        return value
 
 
 class ConversationPagination(ResponseModel):
@@ -252,7 +263,9 @@ class ConversationMessageDetail(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        return to_timestamp(value)
+        if isinstance(value, datetime):
+            return to_timestamp(value)
+        return value
 
 
 class ConversationWithSummary(ResponseModel):
@@ -278,7 +291,9 @@ class ConversationWithSummary(ResponseModel):
     @field_validator("read_at", "created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return to_timestamp(value)
+        if isinstance(value, datetime):
+            return to_timestamp(value)
+        return value
 
 
 class ConversationWithSummaryPagination(ResponseModel):
@@ -307,7 +322,15 @@ class ConversationDetail(ResponseModel):
     @field_validator("created_at", "updated_at", mode="before")
     @classmethod
     def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
-        return to_timestamp(value)
+        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())
 
 
 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, to_timestamp
+from libs.helper import TimestampField
 
 from ._value_type_serializer import serialize_value_type
 
@@ -37,6 +37,12 @@ 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
@@ -82,7 +88,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,7 +5,12 @@ from datetime import datetime
 from pydantic import field_validator
 
 from fields.base import ResponseModel
-from libs.helper import to_timestamp
+
+
+def _to_timestamp(value: datetime | int | None) -> int | None:
+    if isinstance(value, datetime):
+        return int(value.timestamp())
+    return value
 
 
 class UploadConfig(ResponseModel):
@@ -40,7 +45,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):
@@ -61,7 +66,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 libs.helper import build_avatar_url, to_timestamp
+from graphon.file import helpers as file_helpers
 
 simple_account_fields = {
     "id": fields.String,
@@ -15,6 +15,20 @@ 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
@@ -27,7 +41,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):
@@ -45,7 +59,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):
@@ -61,7 +75,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,7 +9,6 @@ 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
 
@@ -40,7 +39,9 @@ class RetrieverResource(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        return to_timestamp(value)
+        if isinstance(value, datetime):
+            return to_timestamp(value)
+        return value
 
 
 class MessageListItem(ResponseModel):
@@ -67,7 +68,9 @@ class MessageListItem(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        return to_timestamp(value)
+        if isinstance(value, datetime):
+            return to_timestamp(value)
+        return value
 
 
 class WebMessageListItem(MessageListItem):
@@ -103,7 +106,9 @@ class SavedMessageItem(ResponseModel):
     @field_validator("created_at", mode="before")
     @classmethod
     def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
-        return to_timestamp(value)
+        if isinstance(value, datetime):
+            return to_timestamp(value)
+        return value
 
 
 class SavedMessageInfiniteScrollPagination(ResponseModel):
@@ -116,6 +121,12 @@ 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

@@ -0,0 +1,16 @@
+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, to_timestamp
+from libs.helper import TimestampField
 
 workflow_app_log_partial_fields = {
     "id": fields.String,
@@ -96,6 +96,12 @@ 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
@@ -109,7 +115,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):
@@ -123,7 +129,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

@@ -0,0 +1,96 @@
+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, to_timestamp
+from libs.helper import TimestampField
 
 workflow_run_for_log_fields = {
     "id": fields.String,
@@ -50,6 +50,12 @@ 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
@@ -73,7 +79,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):
@@ -114,7 +120,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):
@@ -174,7 +180,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):
@@ -211,7 +217,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,13 +10,13 @@ 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, overload
+from typing import TYPE_CHECKING, Annotated, Any, Protocol, cast
 from uuid import UUID
 from zoneinfo import available_timezones
 
 from flask import Response, stream_with_context
 from flask_restx import fields
-from pydantic import BaseModel, TypeAdapter
+from pydantic import BaseModel, ConfigDict, TypeAdapter, with_config
 from pydantic.functional_validators import AfterValidator
 from typing_extensions import TypedDict
 
@@ -33,13 +33,29 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
+@with_config(ConfigDict(extra="allow"))
 class _TokenData(TypedDict, total=False):
+    """Shared baseline token payload.
+
+    `extra='allow'` keeps TokenManager from silently stripping business-
+    specific metadata keys while still validating the common auth fields.
+    Business flows that need stronger guarantees should validate again at
+    their own boundary with a dedicated Pydantic model.
+
+    For the change-email flow specifically, `email_change_phase` is the
+    discriminator used by `services.entities.auth_entities.ChangeEmailTokenData`.
+    It is declared here so the shared token adapter can still provide baseline
+    validation for the state-machine key without taking over the full business
+    model.
+    """
+
     account_id: str | None
     email: str
     token_type: str
     code: str
     old_email: str
     phase: str
+    email_change_phase: str
 
 
 _token_data_adapter: TypeAdapter[_TokenData] = TypeAdapter(_TokenData)
@@ -98,13 +114,12 @@ def extract_tenant_id(user: "Account | EndUser") -> str | None:
     from models import Account
     from models.model import 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.")
+    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.")
 
 
 def run(script):
@@ -138,14 +153,6 @@ 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:
@@ -154,7 +161,9 @@ class AvatarUrlField(fields.Raw):
         from models import Account
 
         if isinstance(obj, Account) and obj.avatar is not None:
-            return build_avatar_url(obj.avatar)
+            if obj.avatar.startswith(("http://", "https://")):
+                return obj.avatar
+            return file_helpers.get_signed_file_url(obj.avatar)
         return None
 
 
@@ -170,35 +179,6 @@ 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,}$"
@@ -423,19 +403,18 @@ 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
 
-    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",
-            )
+    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",
+        )
 
     stream_response = response
 
@@ -466,7 +445,7 @@ class TokenManager:
             raise ValueError("Account or email must be provided")
 
         account_id = account.id if account else None
-        account_email = account.email if account else email
+        account_email = email if email is not None else account.email if account else None
 
         if account_id:
             old_token = cls._get_current_token_for_account(account_id, token_type)
@@ -508,8 +487,7 @@ class TokenManager:
         if token_data_json is None:
             logger.warning("%s token %s not found with key %s", token_type, token, key)
             return None
-        token_data = dict(_token_data_adapter.validate_json(token_data_json))
-        return token_data
+        return dict(_token_data_adapter.validate_json(token_data_json))
 
     @classmethod
     def _get_current_token_for_account(cls, account_id: str, token_type: str) -> str | None:

api/libs/oauth.py

@@ -1,6 +1,3 @@
-import base64
-import binascii
-import json
 import logging
 import urllib.parse
 from dataclasses import dataclass
@@ -30,12 +27,6 @@ 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
@@ -55,7 +46,6 @@ 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)
@@ -68,37 +58,6 @@ 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())
 
@@ -117,12 +76,7 @@ class OAuth:
         self.client_secret = client_secret
         self.redirect_uri = redirect_uri
 
-    def get_authorization_url(
-        self,
-        invite_token: str | None = None,
-        timezone: str | None = None,
-        language: str | None = None,
-    ) -> str:
+    def get_authorization_url(self, invite_token: str | None = None) -> str:
         raise NotImplementedError()
 
     def get_access_token(self, code: str) -> str:
@@ -145,20 +99,14 @@ 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,
-        timezone: str | None = None,
-        language: str | None = None,
-    ) -> str:
+    def get_authorization_url(self, invite_token: str | None = None) -> str:
         params = {
             "client_id": self.client_id,
             "redirect_uri": self.redirect_uri,
             "scope": "user:email",  # Request only basic user information
         }
-        state = encode_oauth_state(invite_token=invite_token, timezone=timezone, language=language)
-        if state:
-            params["state"] = state
+        if invite_token:
+            params["state"] = invite_token
         return f"{self._AUTH_URL}?{urllib.parse.urlencode(params)}"
 
     def get_access_token(self, code: str) -> str:
@@ -238,21 +186,15 @@ 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,
-        timezone: str | None = None,
-        language: str | None = None,
-    ) -> str:
+    def get_authorization_url(self, invite_token: str | None = None) -> str:
         params = {
             "client_id": self.client_id,
             "response_type": "code",
             "redirect_uri": self.redirect_uri,
             "scope": "openid email",
         }
-        state = encode_oauth_state(invite_token=invite_token, timezone=timezone, language=language)
-        if state:
-            params["state"] = state
+        if invite_token:
+            params["state"] = invite_token
         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,23 +48,20 @@ 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:
-        match value:
-            case FileType():
-                return value
-            case str():
-                return FileType.value_of(value)
-            case _:
-                raise ValueError("file type is required in file mapping")
+        if isinstance(value, FileType):
+            return value
+        if isinstance(value, str):
+            return FileType.value_of(value)
+        raise ValueError("file type is required in file mapping")
 
     mapping = dict(file_mapping)
     transfer_method_value = mapping.get("transfer_method")
-    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")
+    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")
 
     file_id = mapping.get("file_id")
     if not isinstance(file_id, str) or not file_id:
@@ -154,15 +151,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.
     """
-    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
+    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
 
 
 def build_file_from_stored_mapping(