test

.github/CODEOWNERS

@@ -9,9 +9,6 @@
 # CODEOWNERS file
 /.github/CODEOWNERS @laipz8200 @crazywoola
 
-# Agents
-/.agents/skills/ @hyoban
-
 # Docs
 /docs/ @crazywoola
 

.github/workflows/build-push.yml

@@ -8,6 +8,7 @@ on:
       - "build/**"
       - "release/e-*"
       - "hotfix/**"
+      - "feat/hitl-backend"
     tags:
       - "*"
 

api/.env.example

@@ -717,3 +717,28 @@ SANDBOX_EXPIRED_RECORDS_CLEAN_GRACEFUL_PERIOD=21
 SANDBOX_EXPIRED_RECORDS_CLEAN_BATCH_SIZE=1000
 SANDBOX_EXPIRED_RECORDS_RETENTION_DAYS=30
 SANDBOX_EXPIRED_RECORDS_CLEAN_TASK_LOCK_TTL=90000
+
+
+# Redis URL used for PubSub between API and
+# celery worker
+# defaults to url constructed from `REDIS_*`
+# configurations
+PUBSUB_REDIS_URL=
+# Pub/sub channel type for streaming events.
+# valid options are:
+#
+# - pubsub: for normal Pub/Sub
+# - sharded: for sharded Pub/Sub
+#
+# It's highly recommended to use sharded Pub/Sub AND redis cluster
+# for large deployments.
+PUBSUB_REDIS_CHANNEL_TYPE=pubsub
+# Whether to use Redis cluster mode while running
+# PubSub.
+#  It's highly recommended to enable this for large deployments.
+PUBSUB_REDIS_USE_CLUSTERS=false
+
+# Whether to Enable human input timeout check task
+ENABLE_HUMAN_INPUT_TIMEOUT_TASK=true
+# Human input timeout check interval in minutes
+HUMAN_INPUT_TIMEOUT_TASK_INTERVAL=1

api/.importlinter

@@ -36,6 +36,8 @@ ignore_imports =
     core.workflow.nodes.loop.loop_node -> core.workflow.graph_engine
     core.workflow.nodes.loop.loop_node -> core.workflow.graph
     core.workflow.nodes.loop.loop_node -> core.workflow.graph_engine.command_channels
+    # TODO(QuantumGhost): fix the import violation later
+    core.workflow.entities.pause_reason -> core.workflow.nodes.human_input.entities
 
 [importlinter:contract:workflow-infrastructure-dependencies]
 name = Workflow Infrastructure Dependencies
@@ -58,6 +60,8 @@ ignore_imports =
     core.workflow.graph_engine.command_channels.redis_channel -> extensions.ext_redis
     core.workflow.graph_engine.manager -> extensions.ext_redis
     core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> extensions.ext_redis
+    # TODO(QuantumGhost): use DI to avoid depending on global DB.
+    core.workflow.nodes.human_input.human_input_node -> extensions.ext_database
 
 [importlinter:contract:workflow-external-imports]
 name = Workflow External Imports
@@ -145,6 +149,7 @@ ignore_imports =
     core.workflow.nodes.agent.agent_node -> core.agent.entities
     core.workflow.nodes.agent.agent_node -> core.agent.plugin_entities
     core.workflow.nodes.base.node -> core.app.entities.app_invoke_entities
+    core.workflow.nodes.human_input.human_input_node -> core.app.entities.app_invoke_entities
     core.workflow.nodes.knowledge_index.knowledge_index_node -> core.app.entities.app_invoke_entities
     core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.app.app_config.entities
     core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.app.entities.app_invoke_entities
@@ -248,6 +253,7 @@ ignore_imports =
     core.workflow.nodes.document_extractor.node -> core.variables.segments
     core.workflow.nodes.http_request.executor -> core.variables.segments
     core.workflow.nodes.http_request.node -> core.variables.segments
+    core.workflow.nodes.human_input.entities -> core.variables.consts
     core.workflow.nodes.iteration.iteration_node -> core.variables
     core.workflow.nodes.iteration.iteration_node -> core.variables.segments
     core.workflow.nodes.iteration.iteration_node -> core.variables.variables
@@ -294,6 +300,8 @@ ignore_imports =
     core.workflow.nodes.llm.llm_utils -> extensions.ext_database
     core.workflow.nodes.llm.node -> extensions.ext_database
     core.workflow.nodes.tool.tool_node -> extensions.ext_database
+    core.workflow.nodes.human_input.human_input_node -> extensions.ext_database
+    core.workflow.nodes.human_input.human_input_node -> core.repositories.human_input_repository
     core.workflow.workflow_entry -> extensions.otel.runtime
     core.workflow.nodes.agent.agent_node -> models
     core.workflow.nodes.base.node -> models.enums

api/.ruff.toml

@@ -53,7 +53,6 @@ select = [
     "S301", # suspicious-pickle-usage, disallow use of `pickle` and its wrappers.
     "S302", # suspicious-marshal-usage, disallow use of `marshal` module
     "S311", # suspicious-non-cryptographic-random-usage,
-    "TID",   # flake8-tidy-imports
 
 ]
 
@@ -89,7 +88,6 @@ ignore = [
     "SIM113",  # enumerate-for-loop
     "SIM117",  # multiple-with-statements
     "SIM210",  # if-expr-with-true-false
-    "TID252",  # allow relative imports from parent modules
 ]
 
 [lint.per-file-ignores]
@@ -111,20 +109,10 @@ ignore = [
     "S110", # allow ignoring exceptions in tests code (currently)
 
 ]
-"controllers/console/explore/trial.py" = ["TID251"]
-"controllers/console/human_input_form.py" = ["TID251"]
-"controllers/web/human_input_form.py" = ["TID251"]
 
 [lint.pyflakes]
 allowed-unused-imports = [
+    "_pytest.monkeypatch",
     "tests.integration_tests",
     "tests.unit_tests",
 ]
-
-[lint.flake8-tidy-imports]
-
-[lint.flake8-tidy-imports.banned-api."flask_restx.reqparse"]
-msg = "Use Pydantic payload/query models instead of reqparse."
-
-[lint.flake8-tidy-imports.banned-api."flask_restx.reqparse.RequestParser"]
-msg = "Use Pydantic payload/query models instead of reqparse."

api/commands.py

@@ -1450,58 +1450,54 @@ def clear_orphaned_file_records(force: bool):
         all_ids_in_tables = []
         for ids_table in ids_tables:
             query = ""
-            match ids_table["type"]:
-                case "uuid":
-                    click.echo(
-                        click.style(
-                            f"- Listing file ids in column {ids_table['column']} in table {ids_table['table']}",
-                            fg="white",
-                        )
+            if ids_table["type"] == "uuid":
+                click.echo(
+                    click.style(
+                        f"- Listing file ids in column {ids_table['column']} in table {ids_table['table']}", fg="white"
                     )
-                    c = ids_table["column"]
-                    query = f"SELECT {c} FROM {ids_table['table']} WHERE {c} IS NOT NULL"
-                    with db.engine.begin() as conn:
-                        rs = conn.execute(sa.text(query))
-                    for i in rs:
-                        all_ids_in_tables.append({"table": ids_table["table"], "id": str(i[0])})
-                case "text":
-                    t = ids_table["table"]
-                    click.echo(
-                        click.style(
-                            f"- Listing file-id-like strings in column {ids_table['column']} in table {t}",
-                            fg="white",
-                        )
+                )
+                query = (
+                    f"SELECT {ids_table['column']} FROM {ids_table['table']} WHERE {ids_table['column']} IS NOT NULL"
+                )
+                with db.engine.begin() as conn:
+                    rs = conn.execute(sa.text(query))
+                for i in rs:
+                    all_ids_in_tables.append({"table": ids_table["table"], "id": str(i[0])})
+            elif ids_table["type"] == "text":
+                click.echo(
+                    click.style(
+                        f"- Listing file-id-like strings in column {ids_table['column']} in table {ids_table['table']}",
+                        fg="white",
                     )
-                    query = (
-                        f"SELECT regexp_matches({ids_table['column']}, '{guid_regexp}', 'g') AS extracted_id "
-                        f"FROM {ids_table['table']}"
+                )
+                query = (
+                    f"SELECT regexp_matches({ids_table['column']}, '{guid_regexp}', 'g') AS extracted_id "
+                    f"FROM {ids_table['table']}"
+                )
+                with db.engine.begin() as conn:
+                    rs = conn.execute(sa.text(query))
+                for i in rs:
+                    for j in i[0]:
+                        all_ids_in_tables.append({"table": ids_table["table"], "id": j})
+            elif ids_table["type"] == "json":
+                click.echo(
+                    click.style(
+                        (
+                            f"- Listing file-id-like JSON string in column {ids_table['column']} "
+                            f"in table {ids_table['table']}"
+                        ),
+                        fg="white",
                     )
-                    with db.engine.begin() as conn:
-                        rs = conn.execute(sa.text(query))
-                    for i in rs:
-                        for j in i[0]:
-                            all_ids_in_tables.append({"table": ids_table["table"], "id": j})
-                case "json":
-                    click.echo(
-                        click.style(
-                            (
-                                f"- Listing file-id-like JSON string in column {ids_table['column']} "
-                                f"in table {ids_table['table']}"
-                            ),
-                            fg="white",
-                        )
-                    )
-                    query = (
-                        f"SELECT regexp_matches({ids_table['column']}::text, '{guid_regexp}', 'g') AS extracted_id "
-                        f"FROM {ids_table['table']}"
-                    )
-                    with db.engine.begin() as conn:
-                        rs = conn.execute(sa.text(query))
-                    for i in rs:
-                        for j in i[0]:
-                            all_ids_in_tables.append({"table": ids_table["table"], "id": j})
-                case _:
-                    pass
+                )
+                query = (
+                    f"SELECT regexp_matches({ids_table['column']}::text, '{guid_regexp}', 'g') AS extracted_id "
+                    f"FROM {ids_table['table']}"
+                )
+                with db.engine.begin() as conn:
+                    rs = conn.execute(sa.text(query))
+                for i in rs:
+                    for j in i[0]:
+                        all_ids_in_tables.append({"table": ids_table["table"], "id": j})
         click.echo(click.style(f"Found {len(all_ids_in_tables)} file ids in tables.", fg="white"))
 
     except Exception as e:
@@ -1741,18 +1737,59 @@ def file_usage(
                 if src_filter != src:
                     continue
 
-        match ids_table["type"]:
-            case "uuid":
-                # Direct UUID match
-                query = (
-                    f"SELECT {ids_table['pk_column']}, {ids_table['column']} "
-                    f"FROM {ids_table['table']} WHERE {ids_table['column']} IS NOT NULL"
-                )
-                with db.engine.begin() as conn:
-                    rs = conn.execute(sa.text(query))
-                    for row in rs:
-                        record_id = str(row[0])
-                        ref_file_id = str(row[1])
+        if ids_table["type"] == "uuid":
+            # Direct UUID match
+            query = (
+                f"SELECT {ids_table['pk_column']}, {ids_table['column']} "
+                f"FROM {ids_table['table']} WHERE {ids_table['column']} IS NOT NULL"
+            )
+            with db.engine.begin() as conn:
+                rs = conn.execute(sa.text(query))
+                for row in rs:
+                    record_id = str(row[0])
+                    ref_file_id = str(row[1])
+                    if ref_file_id not in file_key_map:
+                        continue
+                    storage_key = file_key_map[ref_file_id]
+
+                    # Apply filters
+                    if file_id and ref_file_id != file_id:
+                        continue
+                    if key and not storage_key.endswith(key):
+                        continue
+
+                    # Only collect items within the requested page range
+                    if offset <= total_count < offset + limit:
+                        paginated_usages.append(
+                            {
+                                "src": f"{ids_table['table']}.{ids_table['column']}",
+                                "record_id": record_id,
+                                "file_id": ref_file_id,
+                                "key": storage_key,
+                            }
+                        )
+                    total_count += 1
+
+        elif ids_table["type"] in ("text", "json"):
+            # Extract UUIDs from text/json content
+            column_cast = f"{ids_table['column']}::text" if ids_table["type"] == "json" else ids_table["column"]
+            query = (
+                f"SELECT {ids_table['pk_column']}, {column_cast} "
+                f"FROM {ids_table['table']} WHERE {ids_table['column']} IS NOT NULL"
+            )
+            with db.engine.begin() as conn:
+                rs = conn.execute(sa.text(query))
+                for row in rs:
+                    record_id = str(row[0])
+                    content = str(row[1])
+
+                    # Find all UUIDs in the content
+                    import re
+
+                    uuid_pattern = re.compile(guid_regexp, re.IGNORECASE)
+                    matches = uuid_pattern.findall(content)
+
+                    for ref_file_id in matches:
                         if ref_file_id not in file_key_map:
                             continue
                         storage_key = file_key_map[ref_file_id]
@@ -1775,50 +1812,6 @@ def file_usage(
                             )
                         total_count += 1
 
-            case "text" | "json":
-                # Extract UUIDs from text/json content
-                column_cast = f"{ids_table['column']}::text" if ids_table["type"] == "json" else ids_table["column"]
-                query = (
-                    f"SELECT {ids_table['pk_column']}, {column_cast} "
-                    f"FROM {ids_table['table']} WHERE {ids_table['column']} IS NOT NULL"
-                )
-                with db.engine.begin() as conn:
-                    rs = conn.execute(sa.text(query))
-                    for row in rs:
-                        record_id = str(row[0])
-                        content = str(row[1])
-
-                        # Find all UUIDs in the content
-                        import re
-
-                        uuid_pattern = re.compile(guid_regexp, re.IGNORECASE)
-                        matches = uuid_pattern.findall(content)
-
-                        for ref_file_id in matches:
-                            if ref_file_id not in file_key_map:
-                                continue
-                            storage_key = file_key_map[ref_file_id]
-
-                            # Apply filters
-                            if file_id and ref_file_id != file_id:
-                                continue
-                            if key and not storage_key.endswith(key):
-                                continue
-
-                            # Only collect items within the requested page range
-                            if offset <= total_count < offset + limit:
-                                paginated_usages.append(
-                                    {
-                                        "src": f"{ids_table['table']}.{ids_table['column']}",
-                                        "record_id": record_id,
-                                        "file_id": ref_file_id,
-                                        "key": storage_key,
-                                    }
-                                )
-                            total_count += 1
-            case _:
-                pass
-
     # Output results
     if output_json:
         result = {

api/configs/feature/__init__.py

@@ -1,3 +1,4 @@
+from datetime import timedelta
 from enum import StrEnum
 from typing import Literal
 
@@ -48,6 +49,16 @@ class SecurityConfig(BaseSettings):
         default=5,
     )
 
+    WEB_FORM_SUBMIT_RATE_LIMIT_MAX_ATTEMPTS: PositiveInt = Field(
+        description="Maximum number of web form submissions allowed per IP within the rate limit window",
+        default=30,
+    )
+
+    WEB_FORM_SUBMIT_RATE_LIMIT_WINDOW_SECONDS: PositiveInt = Field(
+        description="Time window in seconds for web form submission rate limiting",
+        default=60,
+    )
+
     LOGIN_DISABLED: bool = Field(
         description="Whether to disable login checks",
         default=False,
@@ -82,6 +93,12 @@ class AppExecutionConfig(BaseSettings):
         default=0,
     )
 
+    HITL_GLOBAL_TIMEOUT_SECONDS: PositiveInt = Field(
+        description="Maximum seconds a workflow run can stay paused waiting for human input before global timeout.",
+        default=int(timedelta(days=3).total_seconds()),
+        ge=1,
+    )
+
 
 class CodeExecutionSandboxConfig(BaseSettings):
     """
@@ -1134,6 +1151,14 @@ class CeleryScheduleTasksConfig(BaseSettings):
         description="Enable queue monitor task",
         default=False,
     )
+    ENABLE_HUMAN_INPUT_TIMEOUT_TASK: bool = Field(
+        description="Enable human input timeout check task",
+        default=True,
+    )
+    HUMAN_INPUT_TIMEOUT_TASK_INTERVAL: PositiveInt = Field(
+        description="Human input timeout check interval in minutes",
+        default=1,
+    )
     ENABLE_CHECK_UPGRADABLE_PLUGIN_TASK: bool = Field(
         description="Enable check upgradable plugin task",
         default=True,

api/configs/middleware/__init__.py

@@ -6,6 +6,7 @@ from pydantic import Field, NonNegativeFloat, NonNegativeInt, PositiveFloat, Pos
 from pydantic_settings import BaseSettings
 
 from .cache.redis_config import RedisConfig
+from .cache.redis_pubsub_config import RedisPubSubConfig
 from .storage.aliyun_oss_storage_config import AliyunOSSStorageConfig
 from .storage.amazon_s3_storage_config import S3StorageConfig
 from .storage.azure_blob_storage_config import AzureBlobStorageConfig
@@ -317,6 +318,7 @@ class MiddlewareConfig(
     CeleryConfig,  # Note: CeleryConfig already inherits from DatabaseConfig
     KeywordStoreConfig,
     RedisConfig,
+    RedisPubSubConfig,
     # configs of storage and storage providers
     StorageConfig,
     AliyunOSSStorageConfig,

api/configs/middleware/cache/redis_pubsub_config.py

@@ -0,0 +1,96 @@
+from typing import Literal, Protocol
+from urllib.parse import quote_plus, urlunparse
+
+from pydantic import Field
+from pydantic_settings import BaseSettings
+
+
+class RedisConfigDefaults(Protocol):
+    REDIS_HOST: str
+    REDIS_PORT: int
+    REDIS_USERNAME: str | None
+    REDIS_PASSWORD: str | None
+    REDIS_DB: int
+    REDIS_USE_SSL: bool
+    REDIS_USE_SENTINEL: bool | None
+    REDIS_USE_CLUSTERS: bool
+
+
+class RedisConfigDefaultsMixin:
+    def _redis_defaults(self: RedisConfigDefaults) -> RedisConfigDefaults:
+        return self
+
+
+class RedisPubSubConfig(BaseSettings, RedisConfigDefaultsMixin):
+    """
+    Configuration settings for Redis pub/sub streaming.
+    """
+
+    PUBSUB_REDIS_URL: str | None = Field(
+        alias="PUBSUB_REDIS_URL",
+        description=(
+            "Redis connection URL for pub/sub streaming events between API "
+            "and celery worker, defaults to url constructed from "
+            "`REDIS_*` configurations"
+        ),
+        default=None,
+    )
+
+    PUBSUB_REDIS_USE_CLUSTERS: bool = Field(
+        description=(
+            "Enable Redis Cluster mode for pub/sub streaming. It's highly "
+            "recommended to enable this for large deployments."
+        ),
+        default=False,
+    )
+
+    PUBSUB_REDIS_CHANNEL_TYPE: Literal["pubsub", "sharded"] = Field(
+        description=(
+            "Pub/sub channel type for streaming events. "
+            "Valid options are:\n"
+            "\n"
+            " - pubsub: for normal Pub/Sub\n"
+            " - sharded: for sharded Pub/Sub\n"
+            "\n"
+            "It's highly recommended to use sharded Pub/Sub AND redis cluster "
+            "for large deployments."
+        ),
+        default="pubsub",
+    )
+
+    def _build_default_pubsub_url(self) -> str:
+        defaults = self._redis_defaults()
+        if not defaults.REDIS_HOST or not defaults.REDIS_PORT:
+            raise ValueError("PUBSUB_REDIS_URL must be set when default Redis URL cannot be constructed")
+
+        scheme = "rediss" if defaults.REDIS_USE_SSL else "redis"
+        username = defaults.REDIS_USERNAME or None
+        password = defaults.REDIS_PASSWORD or None
+
+        userinfo = ""
+        if username:
+            userinfo = quote_plus(username)
+        if password:
+            password_part = quote_plus(password)
+            userinfo = f"{userinfo}:{password_part}" if userinfo else f":{password_part}"
+        if userinfo:
+            userinfo = f"{userinfo}@"
+
+        host = defaults.REDIS_HOST
+        port = defaults.REDIS_PORT
+        db = defaults.REDIS_DB
+
+        netloc = f"{userinfo}{host}:{port}"
+        return urlunparse((scheme, netloc, f"/{db}", "", "", ""))
+
+    @property
+    def normalized_pubsub_redis_url(self) -> str:
+        pubsub_redis_url = self.PUBSUB_REDIS_URL
+        if pubsub_redis_url:
+            cleaned = pubsub_redis_url.strip()
+            pubsub_redis_url = cleaned or None
+
+        if pubsub_redis_url:
+            return pubsub_redis_url
+
+        return self._build_default_pubsub_url()

api/controllers/console/__init__.py

@@ -37,6 +37,7 @@ from . import (
     apikey,
     extension,
     feature,
+    human_input_form,
     init_validate,
     ping,
     setup,
@@ -171,6 +172,7 @@ __all__ = [
     "forgot_password",
     "generator",
     "hit_testing",
+    "human_input_form",
     "init_validate",
     "installed_app",
     "load_balancing_config",

api/controllers/console/app/annotation.py

@@ -1,11 +1,10 @@
 from typing import Any, Literal
 
 from flask import abort, make_response, request
-from flask_restx import Resource
-from pydantic import BaseModel, Field, TypeAdapter, field_validator
+from flask_restx import Resource, fields, marshal, marshal_with
+from pydantic import BaseModel, Field, field_validator
 
 from controllers.common.errors import NoFileUploadedError, TooManyFilesError
-from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import (
     account_initialization_required,
@@ -17,11 +16,9 @@ from controllers.console.wraps import (
 )
 from extensions.ext_redis import redis_client
 from fields.annotation_fields import (
-    Annotation,
-    AnnotationExportList,
-    AnnotationHitHistory,
-    AnnotationHitHistoryList,
-    AnnotationList,
+    annotation_fields,
+    annotation_hit_history_fields,
+    build_annotation_model,
 )
 from libs.helper import uuid_value
 from libs.login import login_required
@@ -92,14 +89,6 @@ reg(CreateAnnotationPayload)
 reg(UpdateAnnotationPayload)
 reg(AnnotationReplyStatusQuery)
 reg(AnnotationFilePayload)
-register_schema_models(
-    console_ns,
-    Annotation,
-    AnnotationList,
-    AnnotationExportList,
-    AnnotationHitHistory,
-    AnnotationHitHistoryList,
-)
 
 
 @console_ns.route("/apps/<uuid:app_id>/annotation-reply/<string:action>")
@@ -118,11 +107,10 @@ class AnnotationReplyActionApi(Resource):
     def post(self, app_id, action: Literal["enable", "disable"]):
         app_id = str(app_id)
         args = AnnotationReplyPayload.model_validate(console_ns.payload)
-        match action:
-            case "enable":
-                result = AppAnnotationService.enable_app_annotation(args.model_dump(), app_id)
-            case "disable":
-                result = AppAnnotationService.disable_app_annotation(app_id)
+        if action == "enable":
+            result = AppAnnotationService.enable_app_annotation(args.model_dump(), app_id)
+        elif action == "disable":
+            result = AppAnnotationService.disable_app_annotation(app_id)
         return result, 200
 
 
@@ -213,33 +201,33 @@ class AnnotationApi(Resource):
 
         app_id = str(app_id)
         annotation_list, total = AppAnnotationService.get_annotation_list_by_app_id(app_id, page, limit, keyword)
-        annotation_models = TypeAdapter(list[Annotation]).validate_python(annotation_list, from_attributes=True)
-        response = AnnotationList(
-            data=annotation_models,
-            has_more=len(annotation_list) == limit,
-            limit=limit,
-            total=total,
-            page=page,
-        )
-        return response.model_dump(mode="json"), 200
+        response = {
+            "data": marshal(annotation_list, annotation_fields),
+            "has_more": len(annotation_list) == limit,
+            "limit": limit,
+            "total": total,
+            "page": page,
+        }
+        return response, 200
 
     @console_ns.doc("create_annotation")
     @console_ns.doc(description="Create a new annotation for an app")
     @console_ns.doc(params={"app_id": "Application ID"})
     @console_ns.expect(console_ns.models[CreateAnnotationPayload.__name__])
-    @console_ns.response(201, "Annotation created successfully", console_ns.models[Annotation.__name__])
+    @console_ns.response(201, "Annotation created successfully", build_annotation_model(console_ns))
     @console_ns.response(403, "Insufficient permissions")
     @setup_required
     @login_required
     @account_initialization_required
     @cloud_edition_billing_resource_check("annotation")
+    @marshal_with(annotation_fields)
     @edit_permission_required
     def post(self, app_id):
         app_id = str(app_id)
         args = CreateAnnotationPayload.model_validate(console_ns.payload)
         data = args.model_dump(exclude_none=True)
         annotation = AppAnnotationService.up_insert_app_annotation_from_message(data, app_id)
-        return Annotation.model_validate(annotation, from_attributes=True).model_dump(mode="json")
+        return annotation
 
     @setup_required
     @login_required
@@ -276,7 +264,7 @@ class AnnotationExportApi(Resource):
     @console_ns.response(
         200,
         "Annotations exported successfully",
-        console_ns.models[AnnotationExportList.__name__],
+        console_ns.model("AnnotationList", {"data": fields.List(fields.Nested(build_annotation_model(console_ns)))}),
     )
     @console_ns.response(403, "Insufficient permissions")
     @setup_required
@@ -286,8 +274,7 @@ class AnnotationExportApi(Resource):
     def get(self, app_id):
         app_id = str(app_id)
         annotation_list = AppAnnotationService.export_annotation_list_by_app_id(app_id)
-        annotation_models = TypeAdapter(list[Annotation]).validate_python(annotation_list, from_attributes=True)
-        response_data = AnnotationExportList(data=annotation_models).model_dump(mode="json")
+        response_data = {"data": marshal(annotation_list, annotation_fields)}
 
         # Create response with secure headers for CSV export
         response = make_response(response_data, 200)
@@ -302,7 +289,7 @@ class AnnotationUpdateDeleteApi(Resource):
     @console_ns.doc("update_delete_annotation")
     @console_ns.doc(description="Update or delete an annotation")
     @console_ns.doc(params={"app_id": "Application ID", "annotation_id": "Annotation ID"})
-    @console_ns.response(200, "Annotation updated successfully", console_ns.models[Annotation.__name__])
+    @console_ns.response(200, "Annotation updated successfully", build_annotation_model(console_ns))
     @console_ns.response(204, "Annotation deleted successfully")
     @console_ns.response(403, "Insufficient permissions")
     @console_ns.expect(console_ns.models[UpdateAnnotationPayload.__name__])
@@ -311,6 +298,7 @@ class AnnotationUpdateDeleteApi(Resource):
     @account_initialization_required
     @cloud_edition_billing_resource_check("annotation")
     @edit_permission_required
+    @marshal_with(annotation_fields)
     def post(self, app_id, annotation_id):
         app_id = str(app_id)
         annotation_id = str(annotation_id)
@@ -318,7 +306,7 @@ class AnnotationUpdateDeleteApi(Resource):
         annotation = AppAnnotationService.update_app_annotation_directly(
             args.model_dump(exclude_none=True), app_id, annotation_id
         )
-        return Annotation.model_validate(annotation, from_attributes=True).model_dump(mode="json")
+        return annotation
 
     @setup_required
     @login_required
@@ -426,7 +414,14 @@ class AnnotationHitHistoryListApi(Resource):
     @console_ns.response(
         200,
         "Hit histories retrieved successfully",
-        console_ns.models[AnnotationHitHistoryList.__name__],
+        console_ns.model(
+            "AnnotationHitHistoryList",
+            {
+                "data": fields.List(
+                    fields.Nested(console_ns.model("AnnotationHitHistoryItem", annotation_hit_history_fields))
+                )
+            },
+        ),
     )
     @console_ns.response(403, "Insufficient permissions")
     @setup_required
@@ -441,14 +436,11 @@ class AnnotationHitHistoryListApi(Resource):
         annotation_hit_history_list, total = AppAnnotationService.get_annotation_hit_histories(
             app_id, annotation_id, page, limit
         )
-        history_models = TypeAdapter(list[AnnotationHitHistory]).validate_python(
-            annotation_hit_history_list, from_attributes=True
-        )
-        response = AnnotationHitHistoryList(
-            data=history_models,
-            has_more=len(annotation_hit_history_list) == limit,
-            limit=limit,
-            total=total,
-            page=page,
-        )
-        return response.model_dump(mode="json")
+        response = {
+            "data": marshal(annotation_hit_history_list, annotation_hit_history_fields),
+            "has_more": len(annotation_hit_history_list) == limit,
+            "limit": limit,
+            "total": total,
+            "page": page,
+        }
+        return response

api/controllers/console/app/audio.py

@@ -6,7 +6,6 @@ from pydantic import BaseModel, Field
 from werkzeug.exceptions import InternalServerError
 
 import services
-from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.error import (
     AppUnavailableError,
@@ -34,6 +33,7 @@ from services.errors.audio import (
 )
 
 logger = logging.getLogger(__name__)
+DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
 
 class TextToSpeechPayload(BaseModel):
@@ -47,11 +47,13 @@ class TextToSpeechVoiceQuery(BaseModel):
     language: str = Field(..., description="Language code")
 
 
-class AudioTranscriptResponse(BaseModel):
-    text: str = Field(description="Transcribed text from audio")
-
-
-register_schema_models(console_ns, AudioTranscriptResponse, TextToSpeechPayload, TextToSpeechVoiceQuery)
+console_ns.schema_model(
+    TextToSpeechPayload.__name__, TextToSpeechPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
+console_ns.schema_model(
+    TextToSpeechVoiceQuery.__name__,
+    TextToSpeechVoiceQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
 
 
 @console_ns.route("/apps/<uuid:app_id>/audio-to-text")
@@ -62,7 +64,7 @@ class ChatMessageAudioApi(Resource):
     @console_ns.response(
         200,
         "Audio transcription successful",
-        console_ns.models[AudioTranscriptResponse.__name__],
+        console_ns.model("AudioTranscriptResponse", {"text": fields.String(description="Transcribed text from audio")}),
     )
     @console_ns.response(400, "Bad request - No audio uploaded or unsupported type")
     @console_ns.response(413, "Audio file too large")

api/controllers/console/app/conversation.py

@@ -89,6 +89,7 @@ status_count_model = console_ns.model(
         "success": fields.Integer,
         "failed": fields.Integer,
         "partial_success": fields.Integer,
+        "paused": fields.Integer,
     },
 )
 
@@ -508,19 +509,16 @@ class ChatConversationApi(Resource):
                 case "created_at" | "-created_at" | _:
                     query = query.where(Conversation.created_at <= end_datetime_utc)
 
-        match args.annotation_status:
-            case "annotated":
-                query = query.options(joinedload(Conversation.message_annotations)).join(  # type: ignore
-                    MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id
-                )
-            case "not_annotated":
-                query = (
-                    query.outerjoin(MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id)
-                    .group_by(Conversation.id)
-                    .having(func.count(MessageAnnotation.id) == 0)
-                )
-            case "all":
-                pass
+        if args.annotation_status == "annotated":
+            query = query.options(joinedload(Conversation.message_annotations)).join(  # type: ignore
+                MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id
+            )
+        elif args.annotation_status == "not_annotated":
+            query = (
+                query.outerjoin(MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id)
+                .group_by(Conversation.id)
+                .having(func.count(MessageAnnotation.id) == 0)
+            )
 
         if app_model.mode == AppMode.ADVANCED_CHAT:
             query = query.where(Conversation.invoke_from != InvokeFrom.DEBUGGER)

api/controllers/console/app/generator.py

@@ -1,4 +1,5 @@
 from collections.abc import Sequence
+from typing import Any
 
 from flask_restx import Resource
 from pydantic import BaseModel, Field
@@ -11,12 +12,10 @@ from controllers.console.app.error import (
     ProviderQuotaExceededError,
 )
 from controllers.console.wraps import account_initialization_required, setup_required
-from core.app.app_config.entities import ModelConfig
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
 from core.helper.code_executor.code_node_provider import CodeNodeProvider
 from core.helper.code_executor.javascript.javascript_code_provider import JavascriptCodeProvider
 from core.helper.code_executor.python3.python3_code_provider import Python3CodeProvider
-from core.llm_generator.entities import RuleCodeGeneratePayload, RuleGeneratePayload, RuleStructuredOutputPayload
 from core.llm_generator.llm_generator import LLMGenerator
 from core.model_runtime.errors.invoke import InvokeError
 from extensions.ext_database import db
@@ -27,13 +26,28 @@ from services.workflow_service import WorkflowService
 DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
 
+class RuleGeneratePayload(BaseModel):
+    instruction: str = Field(..., description="Rule generation instruction")
+    model_config_data: dict[str, Any] = Field(..., alias="model_config", description="Model configuration")
+    no_variable: bool = Field(default=False, description="Whether to exclude variables")
+
+
+class RuleCodeGeneratePayload(RuleGeneratePayload):
+    code_language: str = Field(default="javascript", description="Programming language for code generation")
+
+
+class RuleStructuredOutputPayload(BaseModel):
+    instruction: str = Field(..., description="Structured output generation instruction")
+    model_config_data: dict[str, Any] = Field(..., alias="model_config", description="Model configuration")
+
+
 class InstructionGeneratePayload(BaseModel):
     flow_id: str = Field(..., description="Workflow/Flow ID")
     node_id: str = Field(default="", description="Node ID for workflow context")
     current: str = Field(default="", description="Current instruction text")
     language: str = Field(default="javascript", description="Programming language (javascript/python)")
     instruction: str = Field(..., description="Instruction for generation")
-    model_config_data: ModelConfig = Field(..., alias="model_config", description="Model configuration")
+    model_config_data: dict[str, Any] = Field(..., alias="model_config", description="Model configuration")
     ideal_output: str = Field(default="", description="Expected ideal output")
 
 
@@ -50,7 +64,6 @@ reg(RuleCodeGeneratePayload)
 reg(RuleStructuredOutputPayload)
 reg(InstructionGeneratePayload)
 reg(InstructionTemplatePayload)
-reg(ModelConfig)
 
 
 @console_ns.route("/rule-generate")
@@ -69,7 +82,12 @@ class RuleGenerateApi(Resource):
         _, current_tenant_id = current_account_with_tenant()
 
         try:
-            rules = LLMGenerator.generate_rule_config(tenant_id=current_tenant_id, args=args)
+            rules = LLMGenerator.generate_rule_config(
+                tenant_id=current_tenant_id,
+                instruction=args.instruction,
+                model_config=args.model_config_data,
+                no_variable=args.no_variable,
+            )
         except ProviderTokenNotInitError as ex:
             raise ProviderNotInitializeError(ex.description)
         except QuotaExceededError:
@@ -100,7 +118,9 @@ class RuleCodeGenerateApi(Resource):
         try:
             code_result = LLMGenerator.generate_code(
                 tenant_id=current_tenant_id,
-                args=args,
+                instruction=args.instruction,
+                model_config=args.model_config_data,
+                code_language=args.code_language,
             )
         except ProviderTokenNotInitError as ex:
             raise ProviderNotInitializeError(ex.description)
@@ -132,7 +152,8 @@ class RuleStructuredOutputGenerateApi(Resource):
         try:
             structured_output = LLMGenerator.generate_structured_output(
                 tenant_id=current_tenant_id,
-                args=args,
+                instruction=args.instruction,
+                model_config=args.model_config_data,
             )
         except ProviderTokenNotInitError as ex:
             raise ProviderNotInitializeError(ex.description)
@@ -183,29 +204,23 @@ class InstructionGenerateApi(Resource):
                     case "llm":
                         return LLMGenerator.generate_rule_config(
                             current_tenant_id,
-                            args=RuleGeneratePayload(
-                                instruction=args.instruction,
-                                model_config=args.model_config_data,
-                                no_variable=True,
-                            ),
+                            instruction=args.instruction,
+                            model_config=args.model_config_data,
+                            no_variable=True,
                         )
                     case "agent":
                         return LLMGenerator.generate_rule_config(
                             current_tenant_id,
-                            args=RuleGeneratePayload(
-                                instruction=args.instruction,
-                                model_config=args.model_config_data,
-                                no_variable=True,
-                            ),
+                            instruction=args.instruction,
+                            model_config=args.model_config_data,
+                            no_variable=True,
                         )
                     case "code":
                         return LLMGenerator.generate_code(
                             tenant_id=current_tenant_id,
-                            args=RuleCodeGeneratePayload(
-                                instruction=args.instruction,
-                                model_config=args.model_config_data,
-                                code_language=args.language,
-                            ),
+                            instruction=args.instruction,
+                            model_config=args.model_config_data,
+                            code_language=args.language,
                         )
                     case _:
                         return {"error": f"invalid node type: {node_type}"}

api/controllers/console/app/message.py

@@ -7,7 +7,6 @@ from pydantic import BaseModel, Field, field_validator
 from sqlalchemy import exists, select
 from werkzeug.exceptions import InternalServerError, NotFound
 
-from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.error import (
     CompletionRequestError,
@@ -33,9 +32,10 @@ from libs.login import current_account_with_tenant, login_required
 from models.model import AppMode, Conversation, Message, MessageAnnotation, MessageFeedback
 from services.errors.conversation import ConversationNotExistsError
 from services.errors.message import MessageNotExistsError, SuggestedQuestionsAfterAnswerDisabledError
-from services.message_service import MessageService
+from services.message_service import MessageService, attach_message_extra_contents
 
 logger = logging.getLogger(__name__)
+DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
 
 class ChatMessagesQuery(BaseModel):
@@ -90,22 +90,13 @@ class FeedbackExportQuery(BaseModel):
         raise ValueError("has_comment must be a boolean value")
 
 
-class AnnotationCountResponse(BaseModel):
-    count: int = Field(description="Number of annotations")
+def reg(cls: type[BaseModel]):
+    console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
 
 
-class SuggestedQuestionsResponse(BaseModel):
-    data: list[str] = Field(description="Suggested question")
-
-
-register_schema_models(
-    console_ns,
-    ChatMessagesQuery,
-    MessageFeedbackPayload,
-    FeedbackExportQuery,
-    AnnotationCountResponse,
-    SuggestedQuestionsResponse,
-)
+reg(ChatMessagesQuery)
+reg(MessageFeedbackPayload)
+reg(FeedbackExportQuery)
 
 # Register models for flask_restx to avoid dict type issues in Swagger
 # Register in dependency order: base models first, then dependent models
@@ -207,6 +198,7 @@ message_detail_model = console_ns.model(
         "created_at": TimestampField,
         "agent_thoughts": fields.List(fields.Nested(agent_thought_model)),
         "message_files": fields.List(fields.Nested(message_file_model)),
+        "extra_contents": fields.List(fields.Raw),
         "metadata": fields.Raw(attribute="message_metadata_dict"),
         "status": fields.String,
         "error": fields.String,
@@ -240,7 +232,7 @@ class ChatMessageListApi(Resource):
     @marshal_with(message_infinite_scroll_pagination_model)
     @edit_permission_required
     def get(self, app_model):
-        args = ChatMessagesQuery.model_validate(request.args.to_dict())
+        args = ChatMessagesQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
 
         conversation = (
             db.session.query(Conversation)
@@ -299,6 +291,7 @@ class ChatMessageListApi(Resource):
             has_more = False
 
         history_messages = list(reversed(history_messages))
+        attach_message_extra_contents(history_messages)
 
         return InfiniteScrollPagination(data=history_messages, limit=args.limit, has_more=has_more)
 
@@ -365,7 +358,7 @@ class MessageAnnotationCountApi(Resource):
     @console_ns.response(
         200,
         "Annotation count retrieved successfully",
-        console_ns.models[AnnotationCountResponse.__name__],
+        console_ns.model("AnnotationCountResponse", {"count": fields.Integer(description="Number of annotations")}),
     )
     @get_app_model
     @setup_required
@@ -385,7 +378,9 @@ class MessageSuggestedQuestionApi(Resource):
     @console_ns.response(
         200,
         "Suggested questions retrieved successfully",
-        console_ns.models[SuggestedQuestionsResponse.__name__],
+        console_ns.model(
+            "SuggestedQuestionsResponse", {"data": fields.List(fields.String(description="Suggested question"))}
+        ),
     )
     @console_ns.response(404, "Message or conversation not found")
     @setup_required
@@ -435,7 +430,7 @@ class MessageFeedbackExportApi(Resource):
     @login_required
     @account_initialization_required
     def get(self, app_model):
-        args = FeedbackExportQuery.model_validate(request.args.to_dict())
+        args = FeedbackExportQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
 
         # Import the service function
         from services.feedback_service import FeedbackService
@@ -481,4 +476,5 @@ class MessageApi(Resource):
         if not message:
             raise NotFound("Message Not Exists.")
 
+        attach_message_extra_contents([message])
         return message

api/controllers/console/app/workflow.py

@@ -507,6 +507,179 @@ class WorkflowDraftRunLoopNodeApi(Resource):
             raise InternalServerError()
 
 
+class HumanInputFormPreviewPayload(BaseModel):
+    inputs: dict[str, Any] = Field(
+        default_factory=dict,
+        description="Values used to fill missing upstream variables referenced in form_content",
+    )
+
+
+class HumanInputFormSubmitPayload(BaseModel):
+    form_inputs: dict[str, Any] = Field(..., description="Values the user provides for the form's own fields")
+    inputs: dict[str, Any] = Field(
+        ...,
+        description="Values used to fill missing upstream variables referenced in form_content",
+    )
+    action: str = Field(..., description="Selected action ID")
+
+
+class HumanInputDeliveryTestPayload(BaseModel):
+    delivery_method_id: str = Field(..., description="Delivery method ID")
+    inputs: dict[str, Any] = Field(
+        default_factory=dict,
+        description="Values used to fill missing upstream variables referenced in form_content",
+    )
+
+
+reg(HumanInputFormPreviewPayload)
+reg(HumanInputFormSubmitPayload)
+reg(HumanInputDeliveryTestPayload)
+
+
+@console_ns.route("/apps/<uuid:app_id>/advanced-chat/workflows/draft/human-input/nodes/<string:node_id>/form/preview")
+class AdvancedChatDraftHumanInputFormPreviewApi(Resource):
+    @console_ns.doc("get_advanced_chat_draft_human_input_form")
+    @console_ns.doc(description="Get human input form preview for advanced chat workflow")
+    @console_ns.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
+    @console_ns.expect(console_ns.models[HumanInputFormPreviewPayload.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT])
+    @edit_permission_required
+    def post(self, app_model: App, node_id: str):
+        """
+        Preview human input form content and placeholders
+        """
+        current_user, _ = current_account_with_tenant()
+        args = HumanInputFormPreviewPayload.model_validate(console_ns.payload or {})
+        inputs = args.inputs
+
+        workflow_service = WorkflowService()
+        preview = workflow_service.get_human_input_form_preview(
+            app_model=app_model,
+            account=current_user,
+            node_id=node_id,
+            inputs=inputs,
+        )
+        return jsonable_encoder(preview)
+
+
+@console_ns.route("/apps/<uuid:app_id>/advanced-chat/workflows/draft/human-input/nodes/<string:node_id>/form/run")
+class AdvancedChatDraftHumanInputFormRunApi(Resource):
+    @console_ns.doc("submit_advanced_chat_draft_human_input_form")
+    @console_ns.doc(description="Submit human input form preview for advanced chat workflow")
+    @console_ns.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
+    @console_ns.expect(console_ns.models[HumanInputFormSubmitPayload.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT])
+    @edit_permission_required
+    def post(self, app_model: App, node_id: str):
+        """
+        Submit human input form preview
+        """
+        current_user, _ = current_account_with_tenant()
+        args = HumanInputFormSubmitPayload.model_validate(console_ns.payload or {})
+        workflow_service = WorkflowService()
+        result = workflow_service.submit_human_input_form_preview(
+            app_model=app_model,
+            account=current_user,
+            node_id=node_id,
+            form_inputs=args.form_inputs,
+            inputs=args.inputs,
+            action=args.action,
+        )
+        return jsonable_encoder(result)
+
+
+@console_ns.route("/apps/<uuid:app_id>/workflows/draft/human-input/nodes/<string:node_id>/form/preview")
+class WorkflowDraftHumanInputFormPreviewApi(Resource):
+    @console_ns.doc("get_workflow_draft_human_input_form")
+    @console_ns.doc(description="Get human input form preview for workflow")
+    @console_ns.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
+    @console_ns.expect(console_ns.models[HumanInputFormPreviewPayload.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.WORKFLOW])
+    @edit_permission_required
+    def post(self, app_model: App, node_id: str):
+        """
+        Preview human input form content and placeholders
+        """
+        current_user, _ = current_account_with_tenant()
+        args = HumanInputFormPreviewPayload.model_validate(console_ns.payload or {})
+        inputs = args.inputs
+
+        workflow_service = WorkflowService()
+        preview = workflow_service.get_human_input_form_preview(
+            app_model=app_model,
+            account=current_user,
+            node_id=node_id,
+            inputs=inputs,
+        )
+        return jsonable_encoder(preview)
+
+
+@console_ns.route("/apps/<uuid:app_id>/workflows/draft/human-input/nodes/<string:node_id>/form/run")
+class WorkflowDraftHumanInputFormRunApi(Resource):
+    @console_ns.doc("submit_workflow_draft_human_input_form")
+    @console_ns.doc(description="Submit human input form preview for workflow")
+    @console_ns.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
+    @console_ns.expect(console_ns.models[HumanInputFormSubmitPayload.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.WORKFLOW])
+    @edit_permission_required
+    def post(self, app_model: App, node_id: str):
+        """
+        Submit human input form preview
+        """
+        current_user, _ = current_account_with_tenant()
+        workflow_service = WorkflowService()
+        args = HumanInputFormSubmitPayload.model_validate(console_ns.payload or {})
+        result = workflow_service.submit_human_input_form_preview(
+            app_model=app_model,
+            account=current_user,
+            node_id=node_id,
+            form_inputs=args.form_inputs,
+            inputs=args.inputs,
+            action=args.action,
+        )
+        return jsonable_encoder(result)
+
+
+@console_ns.route("/apps/<uuid:app_id>/workflows/draft/human-input/nodes/<string:node_id>/delivery-test")
+class WorkflowDraftHumanInputDeliveryTestApi(Resource):
+    @console_ns.doc("test_workflow_draft_human_input_delivery")
+    @console_ns.doc(description="Test human input delivery for workflow")
+    @console_ns.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
+    @console_ns.expect(console_ns.models[HumanInputDeliveryTestPayload.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.WORKFLOW, AppMode.ADVANCED_CHAT])
+    @edit_permission_required
+    def post(self, app_model: App, node_id: str):
+        """
+        Test human input delivery
+        """
+        current_user, _ = current_account_with_tenant()
+        workflow_service = WorkflowService()
+        args = HumanInputDeliveryTestPayload.model_validate(console_ns.payload or {})
+        workflow_service.test_human_input_delivery(
+            app_model=app_model,
+            account=current_user,
+            node_id=node_id,
+            delivery_method_id=args.delivery_method_id,
+            inputs=args.inputs,
+        )
+        return jsonable_encoder({})
+
+
 @console_ns.route("/apps/<uuid:app_id>/workflows/draft/run")
 class DraftWorkflowRunApi(Resource):
     @console_ns.doc("run_draft_workflow")

api/controllers/console/app/workflow_run.py

@@ -5,10 +5,15 @@ from flask import request
 from flask_restx import Resource, fields, marshal_with
 from pydantic import BaseModel, Field, field_validator
 from sqlalchemy import select
+from sqlalchemy.orm import sessionmaker
 
+from configs import dify_config
 from controllers.console import console_ns
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
+from controllers.web.error import NotFoundError
+from core.workflow.entities.pause_reason import HumanInputRequired
+from core.workflow.enums import WorkflowExecutionStatus
 from extensions.ext_database import db
 from fields.end_user_fields import simple_end_user_fields
 from fields.member_fields import simple_account_fields
@@ -27,9 +32,21 @@ from libs.custom_inputs import time_duration
 from libs.helper import uuid_value
 from libs.login import current_user, login_required
 from models import Account, App, AppMode, EndUser, WorkflowArchiveLog, WorkflowRunTriggeredFrom
+from models.workflow import WorkflowRun
+from repositories.factory import DifyAPIRepositoryFactory
 from services.retention.workflow_run.constants import ARCHIVE_BUNDLE_NAME
 from services.workflow_run_service import WorkflowRunService
 
+
+def _build_backstage_input_url(form_token: str | None) -> str | None:
+    if not form_token:
+        return None
+    base_url = dify_config.APP_WEB_URL
+    if not base_url:
+        return None
+    return f"{base_url.rstrip('/')}/form/{form_token}"
+
+
 # Workflow run status choices for filtering
 WORKFLOW_RUN_STATUS_CHOICES = ["running", "succeeded", "failed", "stopped", "partial-succeeded"]
 EXPORT_SIGNED_URL_EXPIRE_SECONDS = 3600
@@ -440,3 +457,63 @@ class WorkflowRunNodeExecutionListApi(Resource):
         )
 
         return {"data": node_executions}
+
+
+@console_ns.route("/workflow/<string:workflow_run_id>/pause-details")
+class ConsoleWorkflowPauseDetailsApi(Resource):
+    """Console API for getting workflow pause details."""
+
+    @account_initialization_required
+    @login_required
+    def get(self, workflow_run_id: str):
+        """
+        Get workflow pause details.
+
+        GET /console/api/workflow/<workflow_run_id>/pause-details
+
+        Returns information about why and where the workflow is paused.
+        """
+
+        # Query WorkflowRun to determine if workflow is suspended
+        session_maker = sessionmaker(bind=db.engine)
+        workflow_run_repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker=session_maker)
+        workflow_run = db.session.get(WorkflowRun, workflow_run_id)
+        if not workflow_run:
+            raise NotFoundError("Workflow run not found")
+
+        # Check if workflow is suspended
+        is_paused = workflow_run.status == WorkflowExecutionStatus.PAUSED
+        if not is_paused:
+            return {
+                "paused_at": None,
+                "paused_nodes": [],
+            }, 200
+
+        pause_entity = workflow_run_repo.get_workflow_pause(workflow_run_id)
+        pause_reasons = pause_entity.get_pause_reasons() if pause_entity else []
+
+        # Build response
+        paused_at = pause_entity.paused_at if pause_entity else None
+        paused_nodes = []
+        response = {
+            "paused_at": paused_at.isoformat() + "Z" if paused_at else None,
+            "paused_nodes": paused_nodes,
+        }
+
+        for reason in pause_reasons:
+            if isinstance(reason, HumanInputRequired):
+                paused_nodes.append(
+                    {
+                        "node_id": reason.node_id,
+                        "node_title": reason.node_title,
+                        "pause_type": {
+                            "type": "human_input",
+                            "form_id": reason.form_id,
+                            "backstage_input_url": _build_backstage_input_url(reason.form_token),
+                        },
+                    }
+                )
+            else:
+                raise AssertionError("unimplemented.")
+
+        return response, 200

api/controllers/console/auth/data_source_oauth.py

@@ -2,11 +2,9 @@ import logging
 
 import httpx
 from flask import current_app, redirect, request
-from flask_restx import Resource
-from pydantic import BaseModel, Field
+from flask_restx import Resource, fields
 
 from configs import dify_config
-from controllers.common.schema import register_schema_models
 from libs.login import login_required
 from libs.oauth_data_source import NotionOAuth
 
@@ -16,26 +14,6 @@ from ..wraps import account_initialization_required, is_admin_or_owner_required,
 logger = logging.getLogger(__name__)
 
 
-class OAuthDataSourceResponse(BaseModel):
-    data: str = Field(description="Authorization URL or 'internal' for internal setup")
-
-
-class OAuthDataSourceBindingResponse(BaseModel):
-    result: str = Field(description="Operation result")
-
-
-class OAuthDataSourceSyncResponse(BaseModel):
-    result: str = Field(description="Operation result")
-
-
-register_schema_models(
-    console_ns,
-    OAuthDataSourceResponse,
-    OAuthDataSourceBindingResponse,
-    OAuthDataSourceSyncResponse,
-)
-
-
 def get_oauth_providers():
     with current_app.app_context():
         notion_oauth = NotionOAuth(
@@ -56,7 +34,10 @@ class OAuthDataSource(Resource):
     @console_ns.response(
         200,
         "Authorization URL or internal setup success",
-        console_ns.models[OAuthDataSourceResponse.__name__],
+        console_ns.model(
+            "OAuthDataSourceResponse",
+            {"data": fields.Raw(description="Authorization URL or 'internal' for internal setup")},
+        ),
     )
     @console_ns.response(400, "Invalid provider")
     @console_ns.response(403, "Admin privileges required")
@@ -120,7 +101,7 @@ class OAuthDataSourceBinding(Resource):
     @console_ns.response(
         200,
         "Data source binding success",
-        console_ns.models[OAuthDataSourceBindingResponse.__name__],
+        console_ns.model("OAuthDataSourceBindingResponse", {"result": fields.String(description="Operation result")}),
     )
     @console_ns.response(400, "Invalid provider or code")
     def get(self, provider: str):
@@ -152,7 +133,7 @@ class OAuthDataSourceSync(Resource):
     @console_ns.response(
         200,
         "Data source sync success",
-        console_ns.models[OAuthDataSourceSyncResponse.__name__],
+        console_ns.model("OAuthDataSourceSyncResponse", {"result": fields.String(description="Operation result")}),
     )
     @console_ns.response(400, "Invalid provider or sync failed")
     @setup_required

api/controllers/console/auth/forgot_password.py

@@ -2,11 +2,10 @@ import base64
 import secrets
 
 from flask import request
-from flask_restx import Resource
+from flask_restx import Resource, fields
 from pydantic import BaseModel, Field, field_validator
 from sqlalchemy.orm import Session
 
-from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.auth.error import (
     EmailCodeError,
@@ -49,31 +48,8 @@ class ForgotPasswordResetPayload(BaseModel):
         return valid_password(value)
 
 
-class ForgotPasswordEmailResponse(BaseModel):
-    result: str = Field(description="Operation result")
-    data: str | None = Field(default=None, description="Reset token")
-    code: str | None = Field(default=None, description="Error code if account not found")
-
-
-class ForgotPasswordCheckResponse(BaseModel):
-    is_valid: bool = Field(description="Whether code is valid")
-    email: EmailStr = Field(description="Email address")
-    token: str = Field(description="New reset token")
-
-
-class ForgotPasswordResetResponse(BaseModel):
-    result: str = Field(description="Operation result")
-
-
-register_schema_models(
-    console_ns,
-    ForgotPasswordSendPayload,
-    ForgotPasswordCheckPayload,
-    ForgotPasswordResetPayload,
-    ForgotPasswordEmailResponse,
-    ForgotPasswordCheckResponse,
-    ForgotPasswordResetResponse,
-)
+for model in (ForgotPasswordSendPayload, ForgotPasswordCheckPayload, ForgotPasswordResetPayload):
+    console_ns.schema_model(model.__name__, model.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
 
 
 @console_ns.route("/forgot-password")
@@ -84,7 +60,14 @@ class ForgotPasswordSendEmailApi(Resource):
     @console_ns.response(
         200,
         "Email sent successfully",
-        console_ns.models[ForgotPasswordEmailResponse.__name__],
+        console_ns.model(
+            "ForgotPasswordEmailResponse",
+            {
+                "result": fields.String(description="Operation result"),
+                "data": fields.String(description="Reset token"),
+                "code": fields.String(description="Error code if account not found"),
+            },
+        ),
     )
     @console_ns.response(400, "Invalid email or rate limit exceeded")
     @setup_required
@@ -123,7 +106,14 @@ class ForgotPasswordCheckApi(Resource):
     @console_ns.response(
         200,
         "Code verified successfully",
-        console_ns.models[ForgotPasswordCheckResponse.__name__],
+        console_ns.model(
+            "ForgotPasswordCheckResponse",
+            {
+                "is_valid": fields.Boolean(description="Whether code is valid"),
+                "email": fields.String(description="Email address"),
+                "token": fields.String(description="New reset token"),
+            },
+        ),
     )
     @console_ns.response(400, "Invalid code or token")
     @setup_required
@@ -173,7 +163,7 @@ class ForgotPasswordResetApi(Resource):
     @console_ns.response(
         200,
         "Password reset successfully",
-        console_ns.models[ForgotPasswordResetResponse.__name__],
+        console_ns.model("ForgotPasswordResetResponse", {"result": fields.String(description="Operation result")}),
     )
     @console_ns.response(400, "Invalid token or password mismatch")
     @setup_required

api/controllers/console/auth/oauth_server.py

@@ -155,43 +155,43 @@ class OAuthServerUserTokenApi(Resource):
             grant_type = OAuthGrantType(payload.grant_type)
         except ValueError:
             raise BadRequest("invalid grant_type")
-        match grant_type:
-            case OAuthGrantType.AUTHORIZATION_CODE:
-                if not payload.code:
-                    raise BadRequest("code is required")
 
-                if payload.client_secret != oauth_provider_app.client_secret:
-                    raise BadRequest("client_secret is invalid")
+        if grant_type == OAuthGrantType.AUTHORIZATION_CODE:
+            if not payload.code:
+                raise BadRequest("code is required")
 
-                if payload.redirect_uri not in oauth_provider_app.redirect_uris:
-                    raise BadRequest("redirect_uri is invalid")
+            if payload.client_secret != oauth_provider_app.client_secret:
+                raise BadRequest("client_secret is invalid")
 
-                access_token, refresh_token = OAuthServerService.sign_oauth_access_token(
-                    grant_type, code=payload.code, client_id=oauth_provider_app.client_id
-                )
-                return jsonable_encoder(
-                    {
-                        "access_token": access_token,
-                        "token_type": "Bearer",
-                        "expires_in": OAUTH_ACCESS_TOKEN_EXPIRES_IN,
-                        "refresh_token": refresh_token,
-                    }
-                )
-            case OAuthGrantType.REFRESH_TOKEN:
-                if not payload.refresh_token:
-                    raise BadRequest("refresh_token is required")
+            if payload.redirect_uri not in oauth_provider_app.redirect_uris:
+                raise BadRequest("redirect_uri is invalid")
 
-                access_token, refresh_token = OAuthServerService.sign_oauth_access_token(
-                    grant_type, refresh_token=payload.refresh_token, client_id=oauth_provider_app.client_id
-                )
-                return jsonable_encoder(
-                    {
-                        "access_token": access_token,
-                        "token_type": "Bearer",
-                        "expires_in": OAUTH_ACCESS_TOKEN_EXPIRES_IN,
-                        "refresh_token": refresh_token,
-                    }
-                )
+            access_token, refresh_token = OAuthServerService.sign_oauth_access_token(
+                grant_type, code=payload.code, client_id=oauth_provider_app.client_id
+            )
+            return jsonable_encoder(
+                {
+                    "access_token": access_token,
+                    "token_type": "Bearer",
+                    "expires_in": OAUTH_ACCESS_TOKEN_EXPIRES_IN,
+                    "refresh_token": refresh_token,
+                }
+            )
+        elif grant_type == OAuthGrantType.REFRESH_TOKEN:
+            if not payload.refresh_token:
+                raise BadRequest("refresh_token is required")
+
+            access_token, refresh_token = OAuthServerService.sign_oauth_access_token(
+                grant_type, refresh_token=payload.refresh_token, client_id=oauth_provider_app.client_id
+            )
+            return jsonable_encoder(
+                {
+                    "access_token": access_token,
+                    "token_type": "Bearer",
+                    "expires_in": OAUTH_ACCESS_TOKEN_EXPIRES_IN,
+                    "refresh_token": refresh_token,
+                }
+            )
 
 
 @console_ns.route("/oauth/provider/account")

api/controllers/console/datasets/data_source.py

@@ -1,6 +1,6 @@
 import json
 from collections.abc import Generator
-from typing import Any, Literal, cast
+from typing import Any, cast
 
 from flask import request
 from flask_restx import Resource, fields, marshal_with
@@ -157,8 +157,9 @@ class DataSourceApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    def patch(self, binding_id, action: Literal["enable", "disable"]):
+    def patch(self, binding_id, action):
         binding_id = str(binding_id)
+        action = str(action)
         with Session(db.engine) as session:
             data_source_binding = session.execute(
                 select(DataSourceOauthBinding).filter_by(id=binding_id)
@@ -166,24 +167,23 @@ class DataSourceApi(Resource):
         if data_source_binding is None:
             raise NotFound("Data source binding not found.")
         # enable binding
-        match action:
-            case "enable":
-                if data_source_binding.disabled:
-                    data_source_binding.disabled = False
-                    data_source_binding.updated_at = naive_utc_now()
-                    db.session.add(data_source_binding)
-                    db.session.commit()
-                else:
-                    raise ValueError("Data source is not disabled.")
-            # disable binding
-            case "disable":
-                if not data_source_binding.disabled:
-                    data_source_binding.disabled = True
-                    data_source_binding.updated_at = naive_utc_now()
-                    db.session.add(data_source_binding)
-                    db.session.commit()
-                else:
-                    raise ValueError("Data source is disabled.")
+        if action == "enable":
+            if data_source_binding.disabled:
+                data_source_binding.disabled = False
+                data_source_binding.updated_at = naive_utc_now()
+                db.session.add(data_source_binding)
+                db.session.commit()
+            else:
+                raise ValueError("Data source is not disabled.")
+        # disable binding
+        if action == "disable":
+            if not data_source_binding.disabled:
+                data_source_binding.disabled = True
+                data_source_binding.updated_at = naive_utc_now()
+                db.session.add(data_source_binding)
+                db.session.commit()
+            else:
+                raise ValueError("Data source is disabled.")
         return {"result": "success"}, 200
 
 

api/controllers/console/datasets/datasets_document.py

@@ -576,62 +576,63 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
             if document.indexing_status in {"completed", "error"}:
                 raise DocumentAlreadyFinishedError()
             data_source_info = document.data_source_info_dict
-            match document.data_source_type:
-                case "upload_file":
-                    if not data_source_info:
-                        continue
-                    file_id = data_source_info["upload_file_id"]
-                    file_detail = (
-                        db.session.query(UploadFile)
-                        .where(UploadFile.tenant_id == current_tenant_id, UploadFile.id == file_id)
-                        .first()
-                    )
 
-                    if file_detail is None:
-                        raise NotFound("File not found.")
+            if document.data_source_type == "upload_file":
+                if not data_source_info:
+                    continue
+                file_id = data_source_info["upload_file_id"]
+                file_detail = (
+                    db.session.query(UploadFile)
+                    .where(UploadFile.tenant_id == current_tenant_id, UploadFile.id == file_id)
+                    .first()
+                )
 
-                    extract_setting = ExtractSetting(
-                        datasource_type=DatasourceType.FILE, upload_file=file_detail, document_model=document.doc_form
-                    )
-                    extract_settings.append(extract_setting)
-                case "notion_import":
-                    if not data_source_info:
-                        continue
-                    extract_setting = ExtractSetting(
-                        datasource_type=DatasourceType.NOTION,
-                        notion_info=NotionInfo.model_validate(
-                            {
-                                "credential_id": data_source_info.get("credential_id"),
-                                "notion_workspace_id": data_source_info["notion_workspace_id"],
-                                "notion_obj_id": data_source_info["notion_page_id"],
-                                "notion_page_type": data_source_info["type"],
-                                "tenant_id": current_tenant_id,
-                            }
-                        ),
-                        document_model=document.doc_form,
-                    )
-                    extract_settings.append(extract_setting)
-                case "website_crawl":
-                    if not data_source_info:
-                        continue
-                    extract_setting = ExtractSetting(
-                        datasource_type=DatasourceType.WEBSITE,
-                        website_info=WebsiteInfo.model_validate(
-                            {
-                                "provider": data_source_info["provider"],
-                                "job_id": data_source_info["job_id"],
-                                "url": data_source_info["url"],
-                                "tenant_id": current_tenant_id,
-                                "mode": data_source_info["mode"],
-                                "only_main_content": data_source_info["only_main_content"],
-                            }
-                        ),
-                        document_model=document.doc_form,
-                    )
-                    extract_settings.append(extract_setting)
+                if file_detail is None:
+                    raise NotFound("File not found.")
 
-                case _:
-                    raise ValueError("Data source type not support")
+                extract_setting = ExtractSetting(
+                    datasource_type=DatasourceType.FILE, upload_file=file_detail, document_model=document.doc_form
+                )
+                extract_settings.append(extract_setting)
+
+            elif document.data_source_type == "notion_import":
+                if not data_source_info:
+                    continue
+                extract_setting = ExtractSetting(
+                    datasource_type=DatasourceType.NOTION,
+                    notion_info=NotionInfo.model_validate(
+                        {
+                            "credential_id": data_source_info.get("credential_id"),
+                            "notion_workspace_id": data_source_info["notion_workspace_id"],
+                            "notion_obj_id": data_source_info["notion_page_id"],
+                            "notion_page_type": data_source_info["type"],
+                            "tenant_id": current_tenant_id,
+                        }
+                    ),
+                    document_model=document.doc_form,
+                )
+                extract_settings.append(extract_setting)
+            elif document.data_source_type == "website_crawl":
+                if not data_source_info:
+                    continue
+                extract_setting = ExtractSetting(
+                    datasource_type=DatasourceType.WEBSITE,
+                    website_info=WebsiteInfo.model_validate(
+                        {
+                            "provider": data_source_info["provider"],
+                            "job_id": data_source_info["job_id"],
+                            "url": data_source_info["url"],
+                            "tenant_id": current_tenant_id,
+                            "mode": data_source_info["mode"],
+                            "only_main_content": data_source_info["only_main_content"],
+                        }
+                    ),
+                    document_model=document.doc_form,
+                )
+                extract_settings.append(extract_setting)
+
+            else:
+                raise ValueError("Data source type not support")
             indexing_runner = IndexingRunner()
             try:
                 response = indexing_runner.indexing_estimate(
@@ -953,24 +954,23 @@ class DocumentProcessingApi(DocumentResource):
         if not current_user.is_dataset_editor:
             raise Forbidden()
 
-        match action:
-            case "pause":
-                if document.indexing_status != "indexing":
-                    raise InvalidActionError("Document not in indexing state.")
+        if action == "pause":
+            if document.indexing_status != "indexing":
+                raise InvalidActionError("Document not in indexing state.")
 
-                document.paused_by = current_user.id
-                document.paused_at = naive_utc_now()
-                document.is_paused = True
-                db.session.commit()
+            document.paused_by = current_user.id
+            document.paused_at = naive_utc_now()
+            document.is_paused = True
+            db.session.commit()
 
-            case "resume":
-                if document.indexing_status not in {"paused", "error"}:
-                    raise InvalidActionError("Document not in paused or error state.")
+        elif action == "resume":
+            if document.indexing_status not in {"paused", "error"}:
+                raise InvalidActionError("Document not in paused or error state.")
 
-                document.paused_by = None
-                document.paused_at = None
-                document.is_paused = False
-                db.session.commit()
+            document.paused_by = None
+            document.paused_at = None
+            document.is_paused = False
+            db.session.commit()
 
         return {"result": "success"}, 200
 
@@ -1339,18 +1339,6 @@ class DocumentGenerateSummaryApi(Resource):
             missing_ids = set(document_list) - found_ids
             raise NotFound(f"Some documents not found: {list(missing_ids)}")
 
-        # Update need_summary to True for documents that don't have it set
-        # This handles the case where documents were created when summary_index_setting was disabled
-        documents_to_update = [doc for doc in documents if not doc.need_summary and doc.doc_form != "qa_model"]
-
-        if documents_to_update:
-            document_ids_to_update = [str(doc.id) for doc in documents_to_update]
-            DocumentService.update_documents_need_summary(
-                dataset_id=dataset_id,
-                document_ids=document_ids_to_update,
-                need_summary=True,
-            )
-
         # Dispatch async tasks for each document
         for document in documents:
             # Skip qa_model documents as they don't generate summaries

api/controllers/console/datasets/metadata.py

@@ -126,11 +126,10 @@ class DatasetMetadataBuiltInFieldActionApi(Resource):
             raise NotFound("Dataset not found.")
         DatasetService.check_dataset_permission(dataset, current_user)
 
-        match action:
-            case "enable":
-                MetadataService.enable_built_in_field(dataset)
-            case "disable":
-                MetadataService.disable_built_in_field(dataset)
+        if action == "enable":
+            MetadataService.enable_built_in_field(dataset)
+        elif action == "disable":
+            MetadataService.disable_built_in_field(dataset)
         return {"result": "success"}, 200
 
 

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

@@ -1,9 +1,10 @@
 import json
 import logging
 from typing import Any, Literal, cast
+from uuid import UUID
 
 from flask import abort, request
-from flask_restx import Resource, marshal_with  # type: ignore
+from flask_restx import Resource, marshal_with, reqparse  # type: ignore
 from pydantic import BaseModel, Field
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
@@ -37,7 +38,7 @@ from core.model_runtime.utils.encoders import jsonable_encoder
 from extensions.ext_database import db
 from factories import variable_factory
 from libs import helper
-from libs.helper import TimestampField, UUIDStrOrEmpty
+from libs.helper import TimestampField
 from libs.login import current_account_with_tenant, current_user, login_required
 from models import Account
 from models.dataset import Pipeline
@@ -109,7 +110,7 @@ class NodeIdQuery(BaseModel):
 
 
 class WorkflowRunQuery(BaseModel):
-    last_id: UUIDStrOrEmpty | None = None
+    last_id: UUID | None = None
     limit: int = Field(default=20, ge=1, le=100)
 
 
@@ -120,10 +121,6 @@ class DatasourceVariablesPayload(BaseModel):
     start_node_title: str
 
 
-class RagPipelineRecommendedPluginQuery(BaseModel):
-    type: str = "all"
-
-
 register_schema_models(
     console_ns,
     DraftWorkflowSyncPayload,
@@ -138,7 +135,6 @@ register_schema_models(
     NodeIdQuery,
     WorkflowRunQuery,
     DatasourceVariablesPayload,
-    RagPipelineRecommendedPluginQuery,
 )
 
 
@@ -979,8 +975,11 @@ class RagPipelineRecommendedPluginApi(Resource):
     @login_required
     @account_initialization_required
     def get(self):
-        query = RagPipelineRecommendedPluginQuery.model_validate(request.args.to_dict())
+        parser = reqparse.RequestParser()
+        parser.add_argument("type", type=str, location="args", required=False, default="all")
+        args = parser.parse_args()
+        type = args["type"]
 
         rag_pipeline_service = RagPipelineService()
-        recommended_plugins = rag_pipeline_service.get_recommended_plugins(query.type)
+        recommended_plugins = rag_pipeline_service.get_recommended_plugins(type)
         return recommended_plugins

api/controllers/console/explore/trial.py

@@ -1,9 +1,8 @@
 import logging
-from typing import Any, Literal, cast
+from typing import Any, cast
 
 from flask import request
-from flask_restx import Resource, fields, marshal, marshal_with
-from pydantic import BaseModel
+from flask_restx import Resource, fields, marshal, marshal_with, reqparse
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
 
 import services
@@ -52,7 +51,7 @@ from fields.app_fields import (
     tag_fields,
 )
 from fields.dataset_fields import dataset_fields
-from fields.member_fields import simple_account_fields
+from fields.member_fields import build_simple_account_model
 from fields.workflow_fields import (
     conversation_variable_fields,
     pipeline_variable_fields,
@@ -104,7 +103,7 @@ app_detail_fields_with_site_copy["tags"] = fields.List(fields.Nested(tag_model))
 app_detail_fields_with_site_copy["site"] = fields.Nested(site_model)
 app_detail_with_site_model = get_or_create_model("TrialAppDetailWithSite", app_detail_fields_with_site_copy)
 
-simple_account_model = get_or_create_model("SimpleAccount", simple_account_fields)
+simple_account_model = build_simple_account_model(console_ns)
 conversation_variable_model = get_or_create_model("TrialConversationVariable", conversation_variable_fields)
 pipeline_variable_model = get_or_create_model("TrialPipelineVariable", pipeline_variable_fields)
 
@@ -118,56 +117,7 @@ workflow_fields_copy["rag_pipeline_variables"] = fields.List(fields.Nested(pipel
 workflow_model = get_or_create_model("TrialWorkflow", workflow_fields_copy)
 
 
-# Pydantic models for request validation
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
-
-class WorkflowRunRequest(BaseModel):
-    inputs: dict
-    files: list | None = None
-
-
-class ChatRequest(BaseModel):
-    inputs: dict
-    query: str
-    files: list | None = None
-    conversation_id: str | None = None
-    parent_message_id: str | None = None
-    retriever_from: str = "explore_app"
-
-
-class TextToSpeechRequest(BaseModel):
-    message_id: str | None = None
-    voice: str | None = None
-    text: str | None = None
-    streaming: bool | None = None
-
-
-class CompletionRequest(BaseModel):
-    inputs: dict
-    query: str = ""
-    files: list | None = None
-    response_mode: Literal["blocking", "streaming"] | None = None
-    retriever_from: str = "explore_app"
-
-
-# Register schemas for Swagger documentation
-console_ns.schema_model(
-    WorkflowRunRequest.__name__, WorkflowRunRequest.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
-)
-console_ns.schema_model(
-    ChatRequest.__name__, ChatRequest.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
-)
-console_ns.schema_model(
-    TextToSpeechRequest.__name__, TextToSpeechRequest.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
-)
-console_ns.schema_model(
-    CompletionRequest.__name__, CompletionRequest.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
-)
-
-
 class TrialAppWorkflowRunApi(TrialAppResource):
-    @console_ns.expect(console_ns.models[WorkflowRunRequest.__name__])
     def post(self, trial_app):
         """
         Run workflow
@@ -179,8 +129,10 @@ class TrialAppWorkflowRunApi(TrialAppResource):
         if app_mode != AppMode.WORKFLOW:
             raise NotWorkflowAppError()
 
-        request_data = WorkflowRunRequest.model_validate(console_ns.payload)
-        args = request_data.model_dump()
+        parser = reqparse.RequestParser()
+        parser.add_argument("inputs", type=dict, required=True, nullable=False, location="json")
+        parser.add_argument("files", type=list, required=False, location="json")
+        args = parser.parse_args()
         assert current_user is not None
         try:
             app_id = app_model.id
@@ -231,7 +183,6 @@ class TrialAppWorkflowTaskStopApi(TrialAppResource):
 
 
 class TrialChatApi(TrialAppResource):
-    @console_ns.expect(console_ns.models[ChatRequest.__name__])
     @trial_feature_enable
     def post(self, trial_app):
         app_model = trial_app
@@ -239,14 +190,14 @@ class TrialChatApi(TrialAppResource):
         if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
-        request_data = ChatRequest.model_validate(console_ns.payload)
-        args = request_data.model_dump()
-
-        # Validate UUID values if provided
-        if args.get("conversation_id"):
-            args["conversation_id"] = uuid_value(args["conversation_id"])
-        if args.get("parent_message_id"):
-            args["parent_message_id"] = uuid_value(args["parent_message_id"])
+        parser = reqparse.RequestParser()
+        parser.add_argument("inputs", type=dict, required=True, location="json")
+        parser.add_argument("query", type=str, required=True, location="json")
+        parser.add_argument("files", type=list, required=False, location="json")
+        parser.add_argument("conversation_id", type=uuid_value, location="json")
+        parser.add_argument("parent_message_id", type=uuid_value, required=False, location="json")
+        parser.add_argument("retriever_from", type=str, required=False, default="explore_app", location="json")
+        args = parser.parse_args()
 
         args["auto_generate_name"] = False
 
@@ -369,16 +320,20 @@ class TrialChatAudioApi(TrialAppResource):
 
 
 class TrialChatTextApi(TrialAppResource):
-    @console_ns.expect(console_ns.models[TextToSpeechRequest.__name__])
     @trial_feature_enable
     def post(self, trial_app):
         app_model = trial_app
         try:
-            request_data = TextToSpeechRequest.model_validate(console_ns.payload)
+            parser = reqparse.RequestParser()
+            parser.add_argument("message_id", type=str, required=False, location="json")
+            parser.add_argument("voice", type=str, location="json")
+            parser.add_argument("text", type=str, location="json")
+            parser.add_argument("streaming", type=bool, location="json")
+            args = parser.parse_args()
 
-            message_id = request_data.message_id
-            text = request_data.text
-            voice = request_data.voice
+            message_id = args.get("message_id", None)
+            text = args.get("text", None)
+            voice = args.get("voice", None)
             if not isinstance(current_user, Account):
                 raise ValueError("current_user must be an Account instance")
 
@@ -416,15 +371,19 @@ class TrialChatTextApi(TrialAppResource):
 
 
 class TrialCompletionApi(TrialAppResource):
-    @console_ns.expect(console_ns.models[CompletionRequest.__name__])
     @trial_feature_enable
     def post(self, trial_app):
         app_model = trial_app
         if app_model.mode != "completion":
             raise NotCompletionAppError()
 
-        request_data = CompletionRequest.model_validate(console_ns.payload)
-        args = request_data.model_dump()
+        parser = reqparse.RequestParser()
+        parser.add_argument("inputs", type=dict, required=True, location="json")
+        parser.add_argument("query", type=str, location="json", default="")
+        parser.add_argument("files", type=list, required=False, location="json")
+        parser.add_argument("response_mode", type=str, choices=["blocking", "streaming"], location="json")
+        parser.add_argument("retriever_from", type=str, required=False, default="explore_app", location="json")
+        args = parser.parse_args()
 
         streaming = args["response_mode"] == "streaming"
         args["auto_generate_name"] = False

api/controllers/console/human_input_form.py

@@ -0,0 +1,217 @@
+"""
+Console/Studio Human Input Form APIs.
+"""
+
+import json
+import logging
+from collections.abc import Generator
+
+from flask import Response, jsonify, request
+from flask_restx import Resource, reqparse
+from sqlalchemy import select
+from sqlalchemy.orm import Session, sessionmaker
+
+from controllers.console import console_ns
+from controllers.console.wraps import account_initialization_required, setup_required
+from controllers.web.error import InvalidArgumentError, NotFoundError
+from core.app.apps.advanced_chat.app_generator import AdvancedChatAppGenerator
+from core.app.apps.common.workflow_response_converter import WorkflowResponseConverter
+from core.app.apps.message_generator import MessageGenerator
+from core.app.apps.workflow.app_generator import WorkflowAppGenerator
+from extensions.ext_database import db
+from libs.login import current_account_with_tenant, login_required
+from models import App
+from models.enums import CreatorUserRole
+from models.human_input import RecipientType
+from models.model import AppMode
+from models.workflow import WorkflowRun
+from repositories.factory import DifyAPIRepositoryFactory
+from services.human_input_service import Form, HumanInputService
+from services.workflow_event_snapshot_service import build_workflow_event_stream
+
+logger = logging.getLogger(__name__)
+
+
+def _jsonify_form_definition(form: Form) -> Response:
+    payload = form.get_definition().model_dump()
+    payload["expiration_time"] = int(form.expiration_time.timestamp())
+    return Response(json.dumps(payload, ensure_ascii=False), mimetype="application/json")
+
+
+@console_ns.route("/form/human_input/<string:form_token>")
+class ConsoleHumanInputFormApi(Resource):
+    """Console API for getting human input form definition."""
+
+    @staticmethod
+    def _ensure_console_access(form: Form):
+        _, current_tenant_id = current_account_with_tenant()
+
+        if form.tenant_id != current_tenant_id:
+            raise NotFoundError("App not found")
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, form_token: str):
+        """
+        Get human input form definition by form token.
+
+        GET /console/api/form/human_input/<form_token>
+        """
+        service = HumanInputService(db.engine)
+        form = service.get_form_definition_by_token_for_console(form_token)
+        if form is None:
+            raise NotFoundError(f"form not found, token={form_token}")
+
+        self._ensure_console_access(form)
+
+        return _jsonify_form_definition(form)
+
+    @account_initialization_required
+    @login_required
+    def post(self, form_token: str):
+        """
+        Submit human input form by form token.
+
+        POST /console/api/form/human_input/<form_token>
+
+        Request body:
+        {
+            "inputs": {
+                "content": "User input content"
+            },
+            "action": "Approve"
+        }
+        """
+        parser = reqparse.RequestParser()
+        parser.add_argument("inputs", type=dict, required=True, location="json")
+        parser.add_argument("action", type=str, required=True, location="json")
+        args = parser.parse_args()
+        current_user, _ = current_account_with_tenant()
+
+        service = HumanInputService(db.engine)
+        form = service.get_form_by_token(form_token)
+        if form is None:
+            raise NotFoundError(f"form not found, token={form_token}")
+
+        self._ensure_console_access(form)
+
+        recipient_type = form.recipient_type
+        if recipient_type not in {RecipientType.CONSOLE, RecipientType.BACKSTAGE}:
+            raise NotFoundError(f"form not found, token={form_token}")
+        # The type checker is not smart enought to validate the following invariant.
+        # So we need to assert it manually.
+        assert recipient_type is not None, "recipient_type cannot be None here."
+
+        service.submit_form_by_token(
+            recipient_type=recipient_type,
+            form_token=form_token,
+            selected_action_id=args["action"],
+            form_data=args["inputs"],
+            submission_user_id=current_user.id,
+        )
+
+        return jsonify({})
+
+
+@console_ns.route("/workflow/<string:workflow_run_id>/events")
+class ConsoleWorkflowEventsApi(Resource):
+    """Console API for getting workflow execution events after resume."""
+
+    @account_initialization_required
+    @login_required
+    def get(self, workflow_run_id: str):
+        """
+        Get workflow execution events stream after resume.
+
+        GET /console/api/workflow/<workflow_run_id>/events
+
+        Returns Server-Sent Events stream.
+        """
+
+        user, tenant_id = current_account_with_tenant()
+        session_maker = sessionmaker(db.engine)
+        repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker)
+        workflow_run = repo.get_workflow_run_by_id_and_tenant_id(
+            tenant_id=tenant_id,
+            run_id=workflow_run_id,
+        )
+        if workflow_run is None:
+            raise NotFoundError(f"WorkflowRun not found, id={workflow_run_id}")
+
+        if workflow_run.created_by_role != CreatorUserRole.ACCOUNT:
+            raise NotFoundError(f"WorkflowRun not created by account, id={workflow_run_id}")
+
+        if workflow_run.created_by != user.id:
+            raise NotFoundError(f"WorkflowRun not created by the current account, id={workflow_run_id}")
+
+        with Session(expire_on_commit=False, bind=db.engine) as session:
+            app = _retrieve_app_for_workflow_run(session, workflow_run)
+
+        if workflow_run.finished_at is not None:
+            # TODO(QuantumGhost): should we modify the handling for finished workflow run here?
+            response = WorkflowResponseConverter.workflow_run_result_to_finish_response(
+                task_id=workflow_run.id,
+                workflow_run=workflow_run,
+                creator_user=user,
+            )
+
+            payload = response.model_dump(mode="json")
+            payload["event"] = response.event.value
+
+            def _generate_finished_events() -> Generator[str, None, None]:
+                yield f"data: {json.dumps(payload)}\n\n"
+
+            event_generator = _generate_finished_events
+
+        else:
+            msg_generator = MessageGenerator()
+            if app.mode == AppMode.ADVANCED_CHAT:
+                generator = AdvancedChatAppGenerator()
+            elif app.mode == AppMode.WORKFLOW:
+                generator = WorkflowAppGenerator()
+            else:
+                raise InvalidArgumentError(f"cannot subscribe to workflow run, workflow_run_id={workflow_run.id}")
+
+            include_state_snapshot = request.args.get("include_state_snapshot", "false").lower() == "true"
+
+            def _generate_stream_events():
+                if include_state_snapshot:
+                    return generator.convert_to_event_stream(
+                        build_workflow_event_stream(
+                            app_mode=AppMode(app.mode),
+                            workflow_run=workflow_run,
+                            tenant_id=workflow_run.tenant_id,
+                            app_id=workflow_run.app_id,
+                            session_maker=session_maker,
+                        )
+                    )
+                return generator.convert_to_event_stream(
+                    msg_generator.retrieve_events(AppMode(app.mode), workflow_run.id),
+                )
+
+            event_generator = _generate_stream_events
+
+        return Response(
+            event_generator(),
+            mimetype="text/event-stream",
+            headers={
+                "Cache-Control": "no-cache",
+                "Connection": "keep-alive",
+            },
+        )
+
+
+def _retrieve_app_for_workflow_run(session: Session, workflow_run: WorkflowRun):
+    query = select(App).where(
+        App.id == workflow_run.app_id,
+        App.tenant_id == workflow_run.tenant_id,
+    )
+    app = session.scalars(query).first()
+    if app is None:
+        raise AssertionError(
+            f"App not found for WorkflowRun, workflow_run_id={workflow_run.id}, "
+            f"app_id={workflow_run.app_id}, tenant_id={workflow_run.tenant_id}"
+        )
+
+    return app

api/controllers/console/init_validate.py

@@ -1,74 +1,87 @@
 import os
-from typing import Literal
 
 from flask import session
+from flask_restx import Resource, fields
 from pydantic import BaseModel, Field
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from configs import dify_config
-from controllers.fastopenapi import console_router
 from extensions.ext_database import db
 from models.model import DifySetup
 from services.account_service import TenantService
 
+from . import console_ns
 from .error import AlreadySetupError, InitValidateFailedError
 from .wraps import only_edition_self_hosted
 
+DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
+
 
 class InitValidatePayload(BaseModel):
-    password: str = Field(..., max_length=30, description="Initialization password")
+    password: str = Field(..., max_length=30)
 
 
-class InitStatusResponse(BaseModel):
-    status: Literal["finished", "not_started"] = Field(..., description="Initialization status")
-
-
-class InitValidateResponse(BaseModel):
-    result: str = Field(description="Operation result", examples=["success"])
-
-
-@console_router.get(
-    "/init",
-    response_model=InitStatusResponse,
-    tags=["console"],
+console_ns.schema_model(
+    InitValidatePayload.__name__,
+    InitValidatePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
 )
-def get_init_status() -> InitStatusResponse:
-    """Get initialization validation status."""
-    init_status = get_init_validate_status()
-    if init_status:
-        return InitStatusResponse(status="finished")
-    return InitStatusResponse(status="not_started")
 
 
-@console_router.post(
-    "/init",
-    response_model=InitValidateResponse,
-    tags=["console"],
-    status_code=201,
-)
-@only_edition_self_hosted
-def validate_init_password(payload: InitValidatePayload) -> InitValidateResponse:
-    """Validate initialization password."""
-    tenant_count = TenantService.get_tenant_count()
-    if tenant_count > 0:
-        raise AlreadySetupError()
+@console_ns.route("/init")
+class InitValidateAPI(Resource):
+    @console_ns.doc("get_init_status")
+    @console_ns.doc(description="Get initialization validation status")
+    @console_ns.response(
+        200,
+        "Success",
+        model=console_ns.model(
+            "InitStatusResponse",
+            {"status": fields.String(description="Initialization status", enum=["finished", "not_started"])},
+        ),
+    )
+    def get(self):
+        """Get initialization validation status"""
+        init_status = get_init_validate_status()
+        if init_status:
+            return {"status": "finished"}
+        return {"status": "not_started"}
 
-    if payload.password != os.environ.get("INIT_PASSWORD"):
-        session["is_init_validated"] = False
-        raise InitValidateFailedError()
+    @console_ns.doc("validate_init_password")
+    @console_ns.doc(description="Validate initialization password for self-hosted edition")
+    @console_ns.expect(console_ns.models[InitValidatePayload.__name__])
+    @console_ns.response(
+        201,
+        "Success",
+        model=console_ns.model("InitValidateResponse", {"result": fields.String(description="Operation result")}),
+    )
+    @console_ns.response(400, "Already setup or validation failed")
+    @only_edition_self_hosted
+    def post(self):
+        """Validate initialization password"""
+        # is tenant created
+        tenant_count = TenantService.get_tenant_count()
+        if tenant_count > 0:
+            raise AlreadySetupError()
 
-    session["is_init_validated"] = True
-    return InitValidateResponse(result="success")
+        payload = InitValidatePayload.model_validate(console_ns.payload)
+        input_password = payload.password
+
+        if input_password != os.environ.get("INIT_PASSWORD"):
+            session["is_init_validated"] = False
+            raise InitValidateFailedError()
+
+        session["is_init_validated"] = True
+        return {"result": "success"}, 201
 
 
-def get_init_validate_status() -> bool:
+def get_init_validate_status():
     if dify_config.EDITION == "SELF_HOSTED":
         if os.environ.get("INIT_PASSWORD"):
             if session.get("is_init_validated"):
                 return True
 
             with Session(db.engine) as db_session:
-                return db_session.execute(select(DifySetup)).scalar_one_or_none() is not None
+                return db_session.execute(select(DifySetup)).scalar_one_or_none()
 
     return True

api/controllers/console/remote_files.py

@@ -1,6 +1,7 @@
 import urllib.parse
 
 import httpx
+from flask_restx import Resource
 from pydantic import BaseModel, Field
 
 import services
@@ -10,7 +11,7 @@ from controllers.common.errors import (
     RemoteFileUploadError,
     UnsupportedFileTypeError,
 )
-from controllers.fastopenapi import console_router
+from controllers.common.schema import register_schema_models
 from core.file import helpers as file_helpers
 from core.helper import ssrf_proxy
 from extensions.ext_database import db
@@ -18,74 +19,84 @@ from fields.file_fields import FileWithSignedUrl, RemoteFileInfo
 from libs.login import current_account_with_tenant
 from services.file_service import FileService
 
+from . import console_ns
+
+register_schema_models(console_ns, RemoteFileInfo, FileWithSignedUrl)
+
+
+@console_ns.route("/remote-files/<path:url>")
+class RemoteFileInfoApi(Resource):
+    @console_ns.response(200, "Remote file info", console_ns.models[RemoteFileInfo.__name__])
+    def get(self, url):
+        decoded_url = urllib.parse.unquote(url)
+        resp = ssrf_proxy.head(decoded_url)
+        if resp.status_code != httpx.codes.OK:
+            # failed back to get method
+            resp = ssrf_proxy.get(decoded_url, timeout=3)
+        resp.raise_for_status()
+        info = RemoteFileInfo(
+            file_type=resp.headers.get("Content-Type", "application/octet-stream"),
+            file_length=int(resp.headers.get("Content-Length", 0)),
+        )
+        return info.model_dump(mode="json")
+
 
 class RemoteFileUploadPayload(BaseModel):
     url: str = Field(..., description="URL to fetch")
 
 
-@console_router.get(
-    "/remote-files/<path:url>",
-    response_model=RemoteFileInfo,
-    tags=["console"],
+console_ns.schema_model(
+    RemoteFileUploadPayload.__name__,
+    RemoteFileUploadPayload.model_json_schema(ref_template="#/definitions/{model}"),
 )
-def get_remote_file_info(url: str) -> RemoteFileInfo:
-    decoded_url = urllib.parse.unquote(url)
-    resp = ssrf_proxy.head(decoded_url)
-    if resp.status_code != httpx.codes.OK:
-        resp = ssrf_proxy.get(decoded_url, timeout=3)
-    resp.raise_for_status()
-    return RemoteFileInfo(
-        file_type=resp.headers.get("Content-Type", "application/octet-stream"),
-        file_length=int(resp.headers.get("Content-Length", 0)),
-    )
 
 
-@console_router.post(
-    "/remote-files/upload",
-    response_model=FileWithSignedUrl,
-    tags=["console"],
-    status_code=201,
-)
-def upload_remote_file(payload: RemoteFileUploadPayload) -> FileWithSignedUrl:
-    url = payload.url
+@console_ns.route("/remote-files/upload")
+class RemoteFileUploadApi(Resource):
+    @console_ns.expect(console_ns.models[RemoteFileUploadPayload.__name__])
+    @console_ns.response(201, "Remote file uploaded", console_ns.models[FileWithSignedUrl.__name__])
+    def post(self):
+        args = RemoteFileUploadPayload.model_validate(console_ns.payload)
+        url = args.url
 
-    try:
-        resp = ssrf_proxy.head(url=url)
-        if resp.status_code != httpx.codes.OK:
-            resp = ssrf_proxy.get(url=url, timeout=3, follow_redirects=True)
-        if resp.status_code != httpx.codes.OK:
-            raise RemoteFileUploadError(f"Failed to fetch file from {url}: {resp.text}")
-    except httpx.RequestError as e:
-        raise RemoteFileUploadError(f"Failed to fetch file from {url}: {str(e)}")
+        try:
+            resp = ssrf_proxy.head(url=url)
+            if resp.status_code != httpx.codes.OK:
+                resp = ssrf_proxy.get(url=url, timeout=3, follow_redirects=True)
+            if resp.status_code != httpx.codes.OK:
+                raise RemoteFileUploadError(f"Failed to fetch file from {url}: {resp.text}")
+        except httpx.RequestError as e:
+            raise RemoteFileUploadError(f"Failed to fetch file from {url}: {str(e)}")
 
-    file_info = helpers.guess_file_info_from_response(resp)
+        file_info = helpers.guess_file_info_from_response(resp)
 
-    if not FileService.is_file_size_within_limit(extension=file_info.extension, file_size=file_info.size):
-        raise FileTooLargeError
+        if not FileService.is_file_size_within_limit(extension=file_info.extension, file_size=file_info.size):
+            raise FileTooLargeError
 
-    content = resp.content if resp.request.method == "GET" else ssrf_proxy.get(url).content
+        content = resp.content if resp.request.method == "GET" else ssrf_proxy.get(url).content
 
-    try:
-        user, _ = current_account_with_tenant()
-        upload_file = FileService(db.engine).upload_file(
-            filename=file_info.filename,
-            content=content,
-            mimetype=file_info.mimetype,
-            user=user,
-            source_url=url,
+        try:
+            user, _ = current_account_with_tenant()
+            upload_file = FileService(db.engine).upload_file(
+                filename=file_info.filename,
+                content=content,
+                mimetype=file_info.mimetype,
+                user=user,
+                source_url=url,
+            )
+        except services.errors.file.FileTooLargeError as file_too_large_error:
+            raise FileTooLargeError(file_too_large_error.description)
+        except services.errors.file.UnsupportedFileTypeError:
+            raise UnsupportedFileTypeError()
+
+        payload = FileWithSignedUrl(
+            id=upload_file.id,
+            name=upload_file.name,
+            size=upload_file.size,
+            extension=upload_file.extension,
+            url=file_helpers.get_signed_file_url(upload_file_id=upload_file.id),
+            mime_type=upload_file.mime_type,
+            created_by=upload_file.created_by,
+            created_at=int(upload_file.created_at.timestamp()),
         )
-    except services.errors.file.FileTooLargeError as file_too_large_error:
-        raise FileTooLargeError(file_too_large_error.description)
-    except services.errors.file.UnsupportedFileTypeError:
-        raise UnsupportedFileTypeError()
-
-    return FileWithSignedUrl(
-        id=upload_file.id,
-        name=upload_file.name,
-        size=upload_file.size,
-        extension=upload_file.extension,
-        url=file_helpers.get_signed_file_url(upload_file_id=upload_file.id),
-        mime_type=upload_file.mime_type,
-        created_by=upload_file.created_by,
-        created_at=int(upload_file.created_at.timestamp()),
-    )
+        return payload.model_dump(mode="json"), 201

api/controllers/console/tag/tags.py

@@ -1,27 +1,17 @@
 from typing import Literal
 
 from flask import request
-from flask_restx import Namespace, Resource, fields, marshal_with
+from flask_restx import Resource, marshal_with
 from pydantic import BaseModel, Field
 from werkzeug.exceptions import Forbidden
 
 from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
+from fields.tag_fields import dataset_tag_fields
 from libs.login import current_account_with_tenant, login_required
 from services.tag_service import TagService
 
-dataset_tag_fields = {
-    "id": fields.String,
-    "name": fields.String,
-    "type": fields.String,
-    "binding_count": fields.String,
-}
-
-
-def build_dataset_tag_fields(api_or_ns: Namespace):
-    return api_or_ns.model("DataSetTag", dataset_tag_fields)
-
 
 class TagBasePayload(BaseModel):
     name: str = Field(description="Tag name", min_length=1, max_length=50)

api/controllers/console/workspace/account.py

@@ -12,7 +12,6 @@ from sqlalchemy.orm import Session
 
 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.auth.error import (
     EmailAlreadyInUseError,
@@ -38,7 +37,7 @@ from controllers.console.wraps import (
     setup_required,
 )
 from extensions.ext_database import db
-from fields.member_fields import Account as AccountResponse
+from fields.member_fields import account_fields
 from libs.datetime_utils import naive_utc_now
 from libs.helper import EmailStr, TimestampField, extract_remote_ip, timezone
 from libs.login import current_account_with_tenant, login_required
@@ -171,12 +170,6 @@ reg(ChangeEmailSendPayload)
 reg(ChangeEmailValidityPayload)
 reg(ChangeEmailResetPayload)
 reg(CheckEmailUniquePayload)
-register_schema_models(console_ns, AccountResponse)
-
-
-def _serialize_account(account) -> dict:
-    return AccountResponse.model_validate(account, from_attributes=True).model_dump(mode="json")
-
 
 integrate_fields = {
     "provider": fields.String,
@@ -243,11 +236,11 @@ class AccountProfileApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
+    @marshal_with(account_fields)
     @enterprise_license_required
     def get(self):
         current_user, _ = current_account_with_tenant()
-        return _serialize_account(current_user)
+        return current_user
 
 
 @console_ns.route("/account/name")
@@ -256,14 +249,14 @@ class AccountNameApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
+    @marshal_with(account_fields)
     def post(self):
         current_user, _ = current_account_with_tenant()
         payload = console_ns.payload or {}
         args = AccountNamePayload.model_validate(payload)
         updated_account = AccountService.update_account(current_user, name=args.name)
 
-        return _serialize_account(updated_account)
+        return updated_account
 
 
 @console_ns.route("/account/avatar")
@@ -272,7 +265,7 @@ class AccountAvatarApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
+    @marshal_with(account_fields)
     def post(self):
         current_user, _ = current_account_with_tenant()
         payload = console_ns.payload or {}
@@ -280,7 +273,7 @@ class AccountAvatarApi(Resource):
 
         updated_account = AccountService.update_account(current_user, avatar=args.avatar)
 
-        return _serialize_account(updated_account)
+        return updated_account
 
 
 @console_ns.route("/account/interface-language")
@@ -289,7 +282,7 @@ class AccountInterfaceLanguageApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
+    @marshal_with(account_fields)
     def post(self):
         current_user, _ = current_account_with_tenant()
         payload = console_ns.payload or {}
@@ -297,7 +290,7 @@ class AccountInterfaceLanguageApi(Resource):
 
         updated_account = AccountService.update_account(current_user, interface_language=args.interface_language)
 
-        return _serialize_account(updated_account)
+        return updated_account
 
 
 @console_ns.route("/account/interface-theme")
@@ -306,7 +299,7 @@ class AccountInterfaceThemeApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
+    @marshal_with(account_fields)
     def post(self):
         current_user, _ = current_account_with_tenant()
         payload = console_ns.payload or {}
@@ -314,7 +307,7 @@ class AccountInterfaceThemeApi(Resource):
 
         updated_account = AccountService.update_account(current_user, interface_theme=args.interface_theme)
 
-        return _serialize_account(updated_account)
+        return updated_account
 
 
 @console_ns.route("/account/timezone")
@@ -323,7 +316,7 @@ class AccountTimezoneApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
+    @marshal_with(account_fields)
     def post(self):
         current_user, _ = current_account_with_tenant()
         payload = console_ns.payload or {}
@@ -331,7 +324,7 @@ class AccountTimezoneApi(Resource):
 
         updated_account = AccountService.update_account(current_user, timezone=args.timezone)
 
-        return _serialize_account(updated_account)
+        return updated_account
 
 
 @console_ns.route("/account/password")
@@ -340,7 +333,7 @@ class AccountPasswordApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
+    @marshal_with(account_fields)
     def post(self):
         current_user, _ = current_account_with_tenant()
         payload = console_ns.payload or {}
@@ -351,7 +344,7 @@ class AccountPasswordApi(Resource):
         except ServiceCurrentPasswordIncorrectError:
             raise CurrentPasswordIncorrectError()
 
-        return _serialize_account(current_user)
+        return {"result": "success"}
 
 
 @console_ns.route("/account/integrates")
@@ -627,7 +620,7 @@ class ChangeEmailResetApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
+    @marshal_with(account_fields)
     def post(self):
         payload = console_ns.payload or {}
         args = ChangeEmailResetPayload.model_validate(payload)
@@ -656,7 +649,7 @@ class ChangeEmailResetApi(Resource):
             email=normalized_new_email,
         )
 
-        return _serialize_account(updated_account)
+        return updated_account
 
 
 @console_ns.route("/account/change-email/check-email-unique")

api/controllers/console/workspace/endpoint.py

@@ -1,10 +1,9 @@
 from typing import Any
 
 from flask import request
-from flask_restx import Resource
+from flask_restx import Resource, fields
 from pydantic import BaseModel, Field
 
-from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, is_admin_or_owner_required, setup_required
 from core.model_runtime.utils.encoders import jsonable_encoder
@@ -39,53 +38,15 @@ class EndpointListForPluginQuery(EndpointListQuery):
     plugin_id: str
 
 
-class EndpointCreateResponse(BaseModel):
-    success: bool = Field(description="Operation success")
-
-
-class EndpointListResponse(BaseModel):
-    endpoints: list[dict[str, Any]] = Field(description="Endpoint information")
-
-
-class PluginEndpointListResponse(BaseModel):
-    endpoints: list[dict[str, Any]] = Field(description="Endpoint information")
-
-
-class EndpointDeleteResponse(BaseModel):
-    success: bool = Field(description="Operation success")
-
-
-class EndpointUpdateResponse(BaseModel):
-    success: bool = Field(description="Operation success")
-
-
-class EndpointEnableResponse(BaseModel):
-    success: bool = Field(description="Operation success")
-
-
-class EndpointDisableResponse(BaseModel):
-    success: bool = Field(description="Operation success")
-
-
 def reg(cls: type[BaseModel]):
     console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
 
 
-register_schema_models(
-    console_ns,
-    EndpointCreatePayload,
-    EndpointIdPayload,
-    EndpointUpdatePayload,
-    EndpointListQuery,
-    EndpointListForPluginQuery,
-    EndpointCreateResponse,
-    EndpointListResponse,
-    PluginEndpointListResponse,
-    EndpointDeleteResponse,
-    EndpointUpdateResponse,
-    EndpointEnableResponse,
-    EndpointDisableResponse,
-)
+reg(EndpointCreatePayload)
+reg(EndpointIdPayload)
+reg(EndpointUpdatePayload)
+reg(EndpointListQuery)
+reg(EndpointListForPluginQuery)
 
 
 @console_ns.route("/workspaces/current/endpoints/create")
@@ -96,7 +57,7 @@ class EndpointCreateApi(Resource):
     @console_ns.response(
         200,
         "Endpoint created successfully",
-        console_ns.models[EndpointCreateResponse.__name__],
+        console_ns.model("EndpointCreateResponse", {"success": fields.Boolean(description="Operation success")}),
     )
     @console_ns.response(403, "Admin privileges required")
     @setup_required
@@ -130,7 +91,9 @@ class EndpointListApi(Resource):
     @console_ns.response(
         200,
         "Success",
-        console_ns.models[EndpointListResponse.__name__],
+        console_ns.model(
+            "EndpointListResponse", {"endpoints": fields.List(fields.Raw(description="Endpoint information"))}
+        ),
     )
     @setup_required
     @login_required
@@ -163,7 +126,9 @@ class EndpointListForSinglePluginApi(Resource):
     @console_ns.response(
         200,
         "Success",
-        console_ns.models[PluginEndpointListResponse.__name__],
+        console_ns.model(
+            "PluginEndpointListResponse", {"endpoints": fields.List(fields.Raw(description="Endpoint information"))}
+        ),
     )
     @setup_required
     @login_required
@@ -198,7 +163,7 @@ class EndpointDeleteApi(Resource):
     @console_ns.response(
         200,
         "Endpoint deleted successfully",
-        console_ns.models[EndpointDeleteResponse.__name__],
+        console_ns.model("EndpointDeleteResponse", {"success": fields.Boolean(description="Operation success")}),
     )
     @console_ns.response(403, "Admin privileges required")
     @setup_required
@@ -225,7 +190,7 @@ class EndpointUpdateApi(Resource):
     @console_ns.response(
         200,
         "Endpoint updated successfully",
-        console_ns.models[EndpointUpdateResponse.__name__],
+        console_ns.model("EndpointUpdateResponse", {"success": fields.Boolean(description="Operation success")}),
     )
     @console_ns.response(403, "Admin privileges required")
     @setup_required
@@ -256,7 +221,7 @@ class EndpointEnableApi(Resource):
     @console_ns.response(
         200,
         "Endpoint enabled successfully",
-        console_ns.models[EndpointEnableResponse.__name__],
+        console_ns.model("EndpointEnableResponse", {"success": fields.Boolean(description="Operation success")}),
     )
     @console_ns.response(403, "Admin privileges required")
     @setup_required
@@ -283,7 +248,7 @@ class EndpointDisableApi(Resource):
     @console_ns.response(
         200,
         "Endpoint disabled successfully",
-        console_ns.models[EndpointDisableResponse.__name__],
+        console_ns.model("EndpointDisableResponse", {"success": fields.Boolean(description="Operation success")}),
     )
     @console_ns.response(403, "Admin privileges required")
     @setup_required

api/controllers/console/workspace/members.py

@@ -1,12 +1,12 @@
 from urllib import parse
 
 from flask import abort, request
-from flask_restx import Resource
-from pydantic import BaseModel, Field, TypeAdapter
+from flask_restx import Resource, fields, marshal_with
+from pydantic import BaseModel, Field
 
 import services
 from configs import dify_config
-from controllers.common.schema import register_enum_models, register_schema_models
+from controllers.common.schema import get_or_create_model, register_enum_models
 from controllers.console import console_ns
 from controllers.console.auth.error import (
     CannotTransferOwnerToSelfError,
@@ -25,7 +25,7 @@ from controllers.console.wraps import (
     setup_required,
 )
 from extensions.ext_database import db
-from fields.member_fields import AccountWithRole, AccountWithRoleList
+from fields.member_fields import account_with_role_fields, account_with_role_list_fields
 from libs.helper import extract_remote_ip
 from libs.login import current_account_with_tenant, login_required
 from models.account import Account, TenantAccountRole
@@ -69,7 +69,12 @@ reg(OwnerTransferEmailPayload)
 reg(OwnerTransferCheckPayload)
 reg(OwnerTransferPayload)
 register_enum_models(console_ns, TenantAccountRole)
-register_schema_models(console_ns, AccountWithRole, AccountWithRoleList)
+
+account_with_role_model = get_or_create_model("AccountWithRole", account_with_role_fields)
+
+account_with_role_list_fields_copy = account_with_role_list_fields.copy()
+account_with_role_list_fields_copy["accounts"] = fields.List(fields.Nested(account_with_role_model))
+account_with_role_list_model = get_or_create_model("AccountWithRoleList", account_with_role_list_fields_copy)
 
 
 @console_ns.route("/workspaces/current/members")
@@ -79,15 +84,13 @@ class MemberListApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @console_ns.response(200, "Success", console_ns.models[AccountWithRoleList.__name__])
+    @marshal_with(account_with_role_list_model)
     def get(self):
         current_user, _ = current_account_with_tenant()
         if not current_user.current_tenant:
             raise ValueError("No current tenant")
         members = TenantService.get_tenant_members(current_user.current_tenant)
-        member_models = TypeAdapter(list[AccountWithRole]).validate_python(members, from_attributes=True)
-        response = AccountWithRoleList(accounts=member_models)
-        return response.model_dump(mode="json"), 200
+        return {"result": "success", "accounts": members}, 200
 
 
 @console_ns.route("/workspaces/current/members/invite-email")
@@ -232,15 +235,13 @@ class DatasetOperatorMemberListApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @console_ns.response(200, "Success", console_ns.models[AccountWithRoleList.__name__])
+    @marshal_with(account_with_role_list_model)
     def get(self):
         current_user, _ = current_account_with_tenant()
         if not current_user.current_tenant:
             raise ValueError("No current tenant")
         members = TenantService.get_dataset_operator_members(current_user.current_tenant)
-        member_models = TypeAdapter(list[AccountWithRole]).validate_python(members, from_attributes=True)
-        response = AccountWithRoleList(accounts=member_models)
-        return response.model_dump(mode="json"), 200
+        return {"result": "success", "accounts": members}, 200
 
 
 @console_ns.route("/workspaces/current/members/send-owner-transfer-confirm-email")

api/controllers/service_api/app/annotation.py

@@ -1,16 +1,16 @@
 from typing import Literal
 
 from flask import request
-from flask_restx import Resource
+from flask_restx import Namespace, Resource, fields
 from flask_restx.api import HTTPStatus
-from pydantic import BaseModel, Field, TypeAdapter
+from pydantic import BaseModel, Field
 
 from controllers.common.schema import register_schema_models
 from controllers.console.wraps import edit_permission_required
 from controllers.service_api import service_api_ns
 from controllers.service_api.wraps import validate_app_token
 from extensions.ext_redis import redis_client
-from fields.annotation_fields import Annotation, AnnotationList
+from fields.annotation_fields import annotation_fields, build_annotation_model
 from models.model import App
 from services.annotation_service import AppAnnotationService
 
@@ -26,9 +26,7 @@ class AnnotationReplyActionPayload(BaseModel):
     embedding_model_name: str = Field(description="Embedding model name")
 
 
-register_schema_models(
-    service_api_ns, AnnotationCreatePayload, AnnotationReplyActionPayload, Annotation, AnnotationList
-)
+register_schema_models(service_api_ns, AnnotationCreatePayload, AnnotationReplyActionPayload)
 
 
 @service_api_ns.route("/apps/annotation-reply/<string:action>")
@@ -47,11 +45,10 @@ class AnnotationReplyActionApi(Resource):
     def post(self, app_model: App, action: Literal["enable", "disable"]):
         """Enable or disable annotation reply feature."""
         args = AnnotationReplyActionPayload.model_validate(service_api_ns.payload or {}).model_dump()
-        match action:
-            case "enable":
-                result = AppAnnotationService.enable_app_annotation(args, app_model.id)
-            case "disable":
-                result = AppAnnotationService.disable_app_annotation(app_model.id)
+        if action == "enable":
+            result = AppAnnotationService.enable_app_annotation(args, app_model.id)
+        elif action == "disable":
+            result = AppAnnotationService.disable_app_annotation(app_model.id)
         return result, 200
 
 
@@ -85,6 +82,23 @@ class AnnotationReplyActionStatusApi(Resource):
         return {"job_id": job_id, "job_status": job_status, "error_msg": error_msg}, 200
 
 
+# Define annotation list response model
+annotation_list_fields = {
+    "data": fields.List(fields.Nested(annotation_fields)),
+    "has_more": fields.Boolean,
+    "limit": fields.Integer,
+    "total": fields.Integer,
+    "page": fields.Integer,
+}
+
+
+def build_annotation_list_model(api_or_ns: Namespace):
+    """Build the annotation list model for the API or Namespace."""
+    copied_annotation_list_fields = annotation_list_fields.copy()
+    copied_annotation_list_fields["data"] = fields.List(fields.Nested(build_annotation_model(api_or_ns)))
+    return api_or_ns.model("AnnotationList", copied_annotation_list_fields)
+
+
 @service_api_ns.route("/apps/annotations")
 class AnnotationListApi(Resource):
     @service_api_ns.doc("list_annotations")
@@ -95,12 +109,8 @@ class AnnotationListApi(Resource):
             401: "Unauthorized - invalid API token",
         }
     )
-    @service_api_ns.response(
-        200,
-        "Annotations retrieved successfully",
-        service_api_ns.models[AnnotationList.__name__],
-    )
     @validate_app_token
+    @service_api_ns.marshal_with(build_annotation_list_model(service_api_ns))
     def get(self, app_model: App):
         """List annotations for the application."""
         page = request.args.get("page", default=1, type=int)
@@ -108,15 +118,13 @@ class AnnotationListApi(Resource):
         keyword = request.args.get("keyword", default="", type=str)
 
         annotation_list, total = AppAnnotationService.get_annotation_list_by_app_id(app_model.id, page, limit, keyword)
-        annotation_models = TypeAdapter(list[Annotation]).validate_python(annotation_list, from_attributes=True)
-        response = AnnotationList(
-            data=annotation_models,
-            has_more=len(annotation_list) == limit,
-            limit=limit,
-            total=total,
-            page=page,
-        )
-        return response.model_dump(mode="json")
+        return {
+            "data": annotation_list,
+            "has_more": len(annotation_list) == limit,
+            "limit": limit,
+            "total": total,
+            "page": page,
+        }
 
     @service_api_ns.expect(service_api_ns.models[AnnotationCreatePayload.__name__])
     @service_api_ns.doc("create_annotation")
@@ -127,18 +135,13 @@ class AnnotationListApi(Resource):
             401: "Unauthorized - invalid API token",
         }
     )
-    @service_api_ns.response(
-        HTTPStatus.CREATED,
-        "Annotation created successfully",
-        service_api_ns.models[Annotation.__name__],
-    )
     @validate_app_token
+    @service_api_ns.marshal_with(build_annotation_model(service_api_ns), code=HTTPStatus.CREATED)
     def post(self, app_model: App):
         """Create a new annotation."""
         args = AnnotationCreatePayload.model_validate(service_api_ns.payload or {}).model_dump()
         annotation = AppAnnotationService.insert_app_annotation_directly(args, app_model.id)
-        response = Annotation.model_validate(annotation, from_attributes=True)
-        return response.model_dump(mode="json"), HTTPStatus.CREATED
+        return annotation, 201
 
 
 @service_api_ns.route("/apps/annotations/<uuid:annotation_id>")
@@ -155,19 +158,14 @@ class AnnotationUpdateDeleteApi(Resource):
             404: "Annotation not found",
         }
     )
-    @service_api_ns.response(
-        200,
-        "Annotation updated successfully",
-        service_api_ns.models[Annotation.__name__],
-    )
     @validate_app_token
     @edit_permission_required
+    @service_api_ns.marshal_with(build_annotation_model(service_api_ns))
     def put(self, app_model: App, annotation_id: str):
         """Update an existing annotation."""
         args = AnnotationCreatePayload.model_validate(service_api_ns.payload or {}).model_dump()
         annotation = AppAnnotationService.update_app_annotation_directly(args, app_model.id, annotation_id)
-        response = Annotation.model_validate(annotation, from_attributes=True)
-        return response.model_dump(mode="json")
+        return annotation
 
     @service_api_ns.doc("delete_annotation")
     @service_api_ns.doc(description="Delete an annotation")

api/controllers/service_api/app/completion.py

@@ -30,7 +30,6 @@ from core.errors.error import (
 from core.helper.trace_id_helper import get_external_trace_id
 from core.model_runtime.errors.invoke import InvokeError
 from libs import helper
-from libs.helper import UUIDStrOrEmpty
 from models.model import App, AppMode, EndUser
 from services.app_generate_service import AppGenerateService
 from services.app_task_service import AppTaskService
@@ -53,7 +52,7 @@ class ChatRequestPayload(BaseModel):
     query: str
     files: list[dict[str, Any]] | None = None
     response_mode: Literal["blocking", "streaming"] | None = None
-    conversation_id: UUIDStrOrEmpty | None = Field(default=None, description="Conversation UUID")
+    conversation_id: str | None = Field(default=None, description="Conversation UUID")
     retriever_from: str = Field(default="dev")
     auto_generate_name: bool = Field(default=True, description="Auto generate conversation name")
     workflow_id: str | None = Field(default=None, description="Workflow ID for advanced chat")

api/controllers/service_api/app/conversation.py

@@ -1,4 +1,5 @@
 from typing import Any, Literal
+from uuid import UUID
 
 from flask import request
 from flask_restx import Resource
@@ -22,13 +23,12 @@ from fields.conversation_variable_fields import (
     build_conversation_variable_infinite_scroll_pagination_model,
     build_conversation_variable_model,
 )
-from libs.helper import UUIDStrOrEmpty
 from models.model import App, AppMode, EndUser
 from services.conversation_service import ConversationService
 
 
 class ConversationListQuery(BaseModel):
-    last_id: UUIDStrOrEmpty | None = Field(default=None, description="Last conversation ID for pagination")
+    last_id: UUID | None = Field(default=None, description="Last conversation ID for pagination")
     limit: int = Field(default=20, ge=1, le=100, description="Number of conversations to return")
     sort_by: Literal["created_at", "-created_at", "updated_at", "-updated_at"] = Field(
         default="-updated_at", description="Sort order for conversations"
@@ -48,7 +48,7 @@ class ConversationRenamePayload(BaseModel):
 
 
 class ConversationVariablesQuery(BaseModel):
-    last_id: UUIDStrOrEmpty | None = Field(default=None, description="Last variable ID for pagination")
+    last_id: UUID | None = Field(default=None, description="Last variable ID for pagination")
     limit: int = Field(default=20, ge=1, le=100, description="Number of variables to return")
     variable_name: str | None = Field(
         default=None, description="Filter variables by name", min_length=1, max_length=255

api/controllers/service_api/app/message.py

@@ -1,5 +1,6 @@
 import logging
 from typing import Literal
+from uuid import UUID
 
 from flask import request
 from flask_restx import Resource
@@ -14,7 +15,6 @@ from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate
 from core.app.entities.app_invoke_entities import InvokeFrom
 from fields.conversation_fields import ResultResponse
 from fields.message_fields import MessageInfiniteScrollPagination, MessageListItem
-from libs.helper import UUIDStrOrEmpty
 from models.model import App, AppMode, EndUser
 from services.errors.message import (
     FirstMessageNotExistsError,
@@ -27,8 +27,8 @@ logger = logging.getLogger(__name__)
 
 
 class MessageListQuery(BaseModel):
-    conversation_id: UUIDStrOrEmpty
-    first_id: UUIDStrOrEmpty | None = None
+    conversation_id: UUID
+    first_id: UUID | None = None
     limit: int = Field(default=20, ge=1, le=100, description="Number of messages to return")
 
 

api/controllers/service_api/app/workflow.py

@@ -33,8 +33,9 @@ from core.workflow.graph_engine.manager import GraphEngineManager
 from extensions.ext_database import db
 from fields.workflow_app_log_fields import build_workflow_app_log_pagination_model
 from libs import helper
-from libs.helper import TimestampField
+from libs.helper import OptionalTimestampField, TimestampField
 from models.model import App, AppMode, EndUser
+from models.workflow import WorkflowRun
 from repositories.factory import DifyAPIRepositoryFactory
 from services.app_generate_service import AppGenerateService
 from services.errors.app import IsDraftWorkflowError, WorkflowIdFormatError, WorkflowNotFoundError
@@ -63,17 +64,32 @@ class WorkflowLogQuery(BaseModel):
 
 register_schema_models(service_api_ns, WorkflowRunPayload, WorkflowLogQuery)
 
+
+class WorkflowRunStatusField(fields.Raw):
+    def output(self, key, obj: WorkflowRun, **kwargs):
+        return obj.status.value
+
+
+class WorkflowRunOutputsField(fields.Raw):
+    def output(self, key, obj: WorkflowRun, **kwargs):
+        if obj.status == WorkflowExecutionStatus.PAUSED:
+            return {}
+
+        outputs = obj.outputs_dict
+        return outputs or {}
+
+
 workflow_run_fields = {
     "id": fields.String,
     "workflow_id": fields.String,
-    "status": fields.String,
+    "status": WorkflowRunStatusField,
     "inputs": fields.Raw,
-    "outputs": fields.Raw,
+    "outputs": WorkflowRunOutputsField,
     "error": fields.String,
     "total_steps": fields.Integer,
     "total_tokens": fields.Integer,
     "created_at": TimestampField,
-    "finished_at": TimestampField,
+    "finished_at": OptionalTimestampField,
     "elapsed_time": fields.Float,
 }
 

api/controllers/service_api/dataset/dataset.py

@@ -17,7 +17,7 @@ from controllers.service_api.wraps import (
 from core.model_runtime.entities.model_entities import ModelType
 from core.provider_manager import ProviderManager
 from fields.dataset_fields import dataset_detail_fields
-from fields.tag_fields import DataSetTag
+from fields.tag_fields import build_dataset_tag_fields
 from libs.login import current_user
 from models.account import Account
 from models.dataset import DatasetPermissionEnum
@@ -114,7 +114,6 @@ register_schema_models(
     TagBindingPayload,
     TagUnbindingPayload,
     DatasetListQuery,
-    DataSetTag,
 )
 
 
@@ -481,14 +480,15 @@ class DatasetTagsApi(DatasetApiResource):
             401: "Unauthorized - invalid API token",
         }
     )
+    @service_api_ns.marshal_with(build_dataset_tag_fields(service_api_ns))
     def get(self, _):
         """Get all knowledge type tags."""
         assert isinstance(current_user, Account)
         cid = current_user.current_tenant_id
         assert cid is not None
         tags = TagService.get_tags("knowledge", cid)
-        tag_models = TypeAdapter(list[DataSetTag]).validate_python(tags, from_attributes=True)
-        return [tag.model_dump(mode="json") for tag in tag_models], 200
+
+        return tags, 200
 
     @service_api_ns.expect(service_api_ns.models[TagCreatePayload.__name__])
     @service_api_ns.doc("create_dataset_tag")
@@ -500,6 +500,7 @@ class DatasetTagsApi(DatasetApiResource):
             403: "Forbidden - insufficient permissions",
         }
     )
+    @service_api_ns.marshal_with(build_dataset_tag_fields(service_api_ns))
     def post(self, _):
         """Add a knowledge type tag."""
         assert isinstance(current_user, Account)
@@ -509,9 +510,7 @@ class DatasetTagsApi(DatasetApiResource):
         payload = TagCreatePayload.model_validate(service_api_ns.payload or {})
         tag = TagService.save_tags({"name": payload.name, "type": "knowledge"})
 
-        response = DataSetTag.model_validate(
-            {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": 0}
-        ).model_dump(mode="json")
+        response = {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": 0}
         return response, 200
 
     @service_api_ns.expect(service_api_ns.models[TagUpdatePayload.__name__])
@@ -524,6 +523,7 @@ class DatasetTagsApi(DatasetApiResource):
             403: "Forbidden - insufficient permissions",
         }
     )
+    @service_api_ns.marshal_with(build_dataset_tag_fields(service_api_ns))
     def patch(self, _):
         assert isinstance(current_user, Account)
         if not (current_user.has_edit_permission or current_user.is_dataset_editor):
@@ -536,9 +536,8 @@ class DatasetTagsApi(DatasetApiResource):
 
         binding_count = TagService.get_tag_binding_count(tag_id)
 
-        response = DataSetTag.model_validate(
-            {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": binding_count}
-        ).model_dump(mode="json")
+        response = {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": binding_count}
+
         return response, 200
 
     @service_api_ns.expect(service_api_ns.models[TagDeletePayload.__name__])

api/controllers/service_api/dataset/hit_testing.py

@@ -1,10 +1,7 @@
-from controllers.common.schema import register_schema_model
-from controllers.console.datasets.hit_testing_base import DatasetsHitTestingBase, HitTestingPayload
+from controllers.console.datasets.hit_testing_base import DatasetsHitTestingBase
 from controllers.service_api import service_api_ns
 from controllers.service_api.wraps import DatasetApiResource, cloud_edition_billing_rate_limit_check
 
-register_schema_model(service_api_ns, HitTestingPayload)
-
 
 @service_api_ns.route("/datasets/<uuid:dataset_id>/hit-testing", "/datasets/<uuid:dataset_id>/retrieve")
 class HitTestingApi(DatasetApiResource, DatasetsHitTestingBase):
@@ -18,7 +15,6 @@ class HitTestingApi(DatasetApiResource, DatasetsHitTestingBase):
             404: "Dataset not found",
         }
     )
-    @service_api_ns.expect(service_api_ns.models[HitTestingPayload.__name__])
     @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
     def post(self, tenant_id, dataset_id):
         """Perform hit testing on a dataset.

api/controllers/service_api/dataset/metadata.py

@@ -168,11 +168,10 @@ class DatasetMetadataBuiltInFieldActionServiceApi(DatasetApiResource):
             raise NotFound("Dataset not found.")
         DatasetService.check_dataset_permission(dataset, current_user)
 
-        match action:
-            case "enable":
-                MetadataService.enable_built_in_field(dataset)
-            case "disable":
-                MetadataService.disable_built_in_field(dataset)
+        if action == "enable":
+            MetadataService.enable_built_in_field(dataset)
+        elif action == "disable":
+            MetadataService.disable_built_in_field(dataset)
         return {"result": "success"}, 200
 
 

api/controllers/service_api/wraps.py

@@ -73,14 +73,14 @@ def validate_app_token(view: Callable[P, R] | None = None, *, fetch_user_arg: Fe
 
             # If caller needs end-user context, attach EndUser to current_user
             if fetch_user_arg:
-                user_id = None
-                match fetch_user_arg.fetch_from:
-                    case WhereisUserArg.QUERY:
-                        user_id = request.args.get("user")
-                    case WhereisUserArg.JSON:
-                        user_id = request.get_json().get("user")
-                    case WhereisUserArg.FORM:
-                        user_id = request.form.get("user")
+                if fetch_user_arg.fetch_from == WhereisUserArg.QUERY:
+                    user_id = request.args.get("user")
+                elif fetch_user_arg.fetch_from == WhereisUserArg.JSON:
+                    user_id = request.get_json().get("user")
+                elif fetch_user_arg.fetch_from == WhereisUserArg.FORM:
+                    user_id = request.form.get("user")
+                else:
+                    user_id = None
 
                 if not user_id and fetch_user_arg.required:
                     raise ValueError("Arg user must be provided.")

api/controllers/web/__init__.py

@@ -23,6 +23,7 @@ from . import (
     feature,
     files,
     forgot_password,
+    human_input_form,
     login,
     message,
     passport,
@@ -30,6 +31,7 @@ from . import (
     saved_message,
     site,
     workflow,
+    workflow_events,
 )
 
 api.add_namespace(web_ns)
@@ -44,6 +46,7 @@ __all__ = [
     "feature",
     "files",
     "forgot_password",
+    "human_input_form",
     "login",
     "message",
     "passport",
@@ -52,4 +55,5 @@ __all__ = [
     "site",
     "web_ns",
     "workflow",
+    "workflow_events",
 ]

api/controllers/web/error.py

@@ -117,6 +117,12 @@ class InvokeRateLimitError(BaseHTTPException):
     code = 429
 
 
+class WebFormRateLimitExceededError(BaseHTTPException):
+    error_code = "web_form_rate_limit_exceeded"
+    description = "Too many form requests. Please try again later."
+    code = 429
+
+
 class NotFoundError(BaseHTTPException):
     error_code = "not_found"
     code = 404

api/controllers/web/human_input_form.py

@@ -0,0 +1,164 @@
+"""
+Web App Human Input Form APIs.
+"""
+
+import json
+import logging
+from datetime import datetime
+
+from flask import Response, request
+from flask_restx import Resource, reqparse
+from werkzeug.exceptions import Forbidden
+
+from configs import dify_config
+from controllers.web import web_ns
+from controllers.web.error import NotFoundError, WebFormRateLimitExceededError
+from controllers.web.site import serialize_app_site_payload
+from extensions.ext_database import db
+from libs.helper import RateLimiter, extract_remote_ip
+from models.account import TenantStatus
+from models.model import App, Site
+from services.human_input_service import Form, FormNotFoundError, HumanInputService
+
+logger = logging.getLogger(__name__)
+
+_FORM_SUBMIT_RATE_LIMITER = RateLimiter(
+    prefix="web_form_submit_rate_limit",
+    max_attempts=dify_config.WEB_FORM_SUBMIT_RATE_LIMIT_MAX_ATTEMPTS,
+    time_window=dify_config.WEB_FORM_SUBMIT_RATE_LIMIT_WINDOW_SECONDS,
+)
+_FORM_ACCESS_RATE_LIMITER = RateLimiter(
+    prefix="web_form_access_rate_limit",
+    max_attempts=dify_config.WEB_FORM_SUBMIT_RATE_LIMIT_MAX_ATTEMPTS,
+    time_window=dify_config.WEB_FORM_SUBMIT_RATE_LIMIT_WINDOW_SECONDS,
+)
+
+
+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, site_payload: dict | None = None) -> Response:
+    """Return the form payload (optionally with site) as a JSON response."""
+    definition_payload = form.get_definition().model_dump()
+    payload = {
+        "form_content": definition_payload["rendered_content"],
+        "inputs": definition_payload["inputs"],
+        "resolved_default_values": _stringify_default_values(definition_payload["default_values"]),
+        "user_actions": definition_payload["user_actions"],
+        "expiration_time": _to_timestamp(form.expiration_time),
+    }
+    if site_payload is not None:
+        payload["site"] = site_payload
+    return Response(json.dumps(payload, ensure_ascii=False), mimetype="application/json")
+
+
+# TODO(QuantumGhost): disable authorization for web app
+# form api temporarily
+
+
+@web_ns.route("/form/human_input/<string:form_token>")
+# class HumanInputFormApi(WebApiResource):
+class HumanInputFormApi(Resource):
+    """API for getting and submitting human input forms via the web app."""
+
+    # def get(self, _app_model: App, _end_user: EndUser, form_token: str):
+    def get(self, form_token: str):
+        """
+        Get human input form definition by token.
+
+        GET /api/form/human_input/<form_token>
+        """
+        ip_address = extract_remote_ip(request)
+        if _FORM_ACCESS_RATE_LIMITER.is_rate_limited(ip_address):
+            raise WebFormRateLimitExceededError()
+        _FORM_ACCESS_RATE_LIMITER.increment_rate_limit(ip_address)
+
+        service = HumanInputService(db.engine)
+        # TODO(QuantumGhost): forbid submision for form tokens
+        # that are only for console.
+        form = service.get_form_by_token(form_token)
+
+        if form is None:
+            raise NotFoundError("Form not found")
+
+        service.ensure_form_active(form)
+        app_model, site = _get_app_site_from_form(form)
+
+        return _jsonify_form_definition(form, site_payload=serialize_app_site_payload(app_model, site, None))
+
+    # def post(self, _app_model: App, _end_user: EndUser, form_token: str):
+    def post(self, form_token: str):
+        """
+        Submit human input form by token.
+
+        POST /api/form/human_input/<form_token>
+
+        Request body:
+        {
+            "inputs": {
+                "content": "User input content"
+            },
+            "action": "Approve"
+        }
+        """
+        parser = reqparse.RequestParser()
+        parser.add_argument("inputs", type=dict, required=True, location="json")
+        parser.add_argument("action", type=str, required=True, location="json")
+        args = parser.parse_args()
+
+        ip_address = extract_remote_ip(request)
+        if _FORM_SUBMIT_RATE_LIMITER.is_rate_limited(ip_address):
+            raise WebFormRateLimitExceededError()
+        _FORM_SUBMIT_RATE_LIMITER.increment_rate_limit(ip_address)
+
+        service = HumanInputService(db.engine)
+        form = service.get_form_by_token(form_token)
+        if form is None:
+            raise NotFoundError("Form not found")
+
+        if (recipient_type := form.recipient_type) is None:
+            logger.warning("Recipient type is None for form, form_id=%", form.id)
+            raise AssertionError("Recipient type is None")
+
+        try:
+            service.submit_form_by_token(
+                recipient_type=recipient_type,
+                form_token=form_token,
+                selected_action_id=args["action"],
+                form_data=args["inputs"],
+                submission_end_user_id=None,
+                # submission_end_user_id=_end_user.id,
+            )
+        except FormNotFoundError:
+            raise NotFoundError("Form not found")
+
+        return {}, 200
+
+
+def _get_app_site_from_form(form: Form) -> tuple[App, Site]:
+    """Resolve App/Site for the form's app and validate tenant status."""
+    app_model = db.session.query(App).where(App.id == form.app_id).first()
+    if app_model is None or app_model.tenant_id != form.tenant_id:
+        raise NotFoundError("Form not found")
+
+    site = db.session.query(Site).where(Site.app_id == app_model.id).first()
+    if site is None:
+        raise Forbidden()
+
+    if app_model.tenant and app_model.tenant.status == TenantStatus.ARCHIVE:
+        raise Forbidden()
+
+    return app_model, site

api/controllers/web/site.py

@@ -1,4 +1,6 @@
-from flask_restx import fields, marshal_with
+from typing import cast
+
+from flask_restx import fields, marshal, marshal_with
 from werkzeug.exceptions import Forbidden
 
 from configs import dify_config
@@ -7,7 +9,7 @@ from controllers.web.wraps import WebApiResource
 from extensions.ext_database import db
 from libs.helper import AppIconUrlField
 from models.account import TenantStatus
-from models.model import Site
+from models.model import App, Site
 from services.feature_service import FeatureService
 
 
@@ -108,3 +110,14 @@ class AppSiteInfo:
                 "remove_webapp_brand": remove_webapp_brand,
                 "replace_webapp_logo": replace_webapp_logo,
             }
+
+
+def serialize_site(site: Site) -> dict:
+    """Serialize Site model using the same schema as AppSiteApi."""
+    return cast(dict, marshal(site, AppSiteApi.site_fields))
+
+
+def serialize_app_site_payload(app_model: App, site: Site, end_user_id: str | None) -> dict:
+    can_replace_logo = FeatureService.get_features(app_model.tenant_id).can_replace_logo
+    app_site_info = AppSiteInfo(app_model.tenant, app_model, site, end_user_id, can_replace_logo)
+    return cast(dict, marshal(app_site_info, AppSiteApi.app_fields))

api/controllers/web/workflow_events.py

@@ -0,0 +1,112 @@
+"""
+Web App Workflow Resume APIs.
+"""
+
+import json
+from collections.abc import Generator
+
+from flask import Response, request
+from sqlalchemy.orm import sessionmaker
+
+from controllers.web import api
+from controllers.web.error import InvalidArgumentError, NotFoundError
+from controllers.web.wraps import WebApiResource
+from core.app.apps.advanced_chat.app_generator import AdvancedChatAppGenerator
+from core.app.apps.base_app_generator import BaseAppGenerator
+from core.app.apps.common.workflow_response_converter import WorkflowResponseConverter
+from core.app.apps.message_generator import MessageGenerator
+from core.app.apps.workflow.app_generator import WorkflowAppGenerator
+from extensions.ext_database import db
+from models.enums import CreatorUserRole
+from models.model import App, AppMode, EndUser
+from repositories.factory import DifyAPIRepositoryFactory
+from services.workflow_event_snapshot_service import build_workflow_event_stream
+
+
+class WorkflowEventsApi(WebApiResource):
+    """API for getting workflow execution events after resume."""
+
+    def get(self, app_model: App, end_user: EndUser, task_id: str):
+        """
+        Get workflow execution events stream after resume.
+
+        GET /api/workflow/<task_id>/events
+
+        Returns Server-Sent Events stream.
+        """
+        workflow_run_id = task_id
+        session_maker = sessionmaker(db.engine)
+        repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker)
+        workflow_run = repo.get_workflow_run_by_id_and_tenant_id(
+            tenant_id=app_model.tenant_id,
+            run_id=workflow_run_id,
+        )
+
+        if workflow_run is None:
+            raise NotFoundError(f"WorkflowRun not found, id={workflow_run_id}")
+
+        if workflow_run.app_id != app_model.id:
+            raise NotFoundError(f"WorkflowRun not found, id={workflow_run_id}")
+
+        if workflow_run.created_by_role != CreatorUserRole.END_USER:
+            raise NotFoundError(f"WorkflowRun not created by end user, id={workflow_run_id}")
+
+        if workflow_run.created_by != end_user.id:
+            raise NotFoundError(f"WorkflowRun not created by the current end user, id={workflow_run_id}")
+
+        if workflow_run.finished_at is not None:
+            response = WorkflowResponseConverter.workflow_run_result_to_finish_response(
+                task_id=workflow_run.id,
+                workflow_run=workflow_run,
+                creator_user=end_user,
+            )
+
+            payload = response.model_dump(mode="json")
+            payload["event"] = response.event.value
+
+            def _generate_finished_events() -> Generator[str, None, None]:
+                yield f"data: {json.dumps(payload)}\n\n"
+
+            event_generator = _generate_finished_events
+        else:
+            app_mode = AppMode.value_of(app_model.mode)
+            msg_generator = MessageGenerator()
+            generator: BaseAppGenerator
+            if app_mode == AppMode.ADVANCED_CHAT:
+                generator = AdvancedChatAppGenerator()
+            elif app_mode == AppMode.WORKFLOW:
+                generator = WorkflowAppGenerator()
+            else:
+                raise InvalidArgumentError(f"cannot subscribe to workflow run, workflow_run_id={workflow_run.id}")
+
+            include_state_snapshot = request.args.get("include_state_snapshot", "false").lower() == "true"
+
+            def _generate_stream_events():
+                if include_state_snapshot:
+                    return generator.convert_to_event_stream(
+                        build_workflow_event_stream(
+                            app_mode=app_mode,
+                            workflow_run=workflow_run,
+                            tenant_id=app_model.tenant_id,
+                            app_id=app_model.id,
+                            session_maker=session_maker,
+                        )
+                    )
+                return generator.convert_to_event_stream(
+                    msg_generator.retrieve_events(app_mode, workflow_run.id),
+                )
+
+            event_generator = _generate_stream_events
+
+        return Response(
+            event_generator(),
+            mimetype="text/event-stream",
+            headers={
+                "Cache-Control": "no-cache",
+                "Connection": "keep-alive",
+            },
+        )
+
+
+# Register the APIs
+api.add_resource(WorkflowEventsApi, "/workflow/<string:task_id>/events")

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

@@ -14,17 +14,16 @@ class AgentConfigManager:
             agent_dict = config.get("agent_mode", {})
             agent_strategy = agent_dict.get("strategy", "cot")
 
-            match agent_strategy:
-                case "function_call":
+            if agent_strategy == "function_call":
+                strategy = AgentEntity.Strategy.FUNCTION_CALLING
+            elif agent_strategy in {"cot", "react"}:
+                strategy = AgentEntity.Strategy.CHAIN_OF_THOUGHT
+            else:
+                # old configs, try to detect default strategy
+                if config["model"]["provider"] == "openai":
                     strategy = AgentEntity.Strategy.FUNCTION_CALLING
-                case "cot" | "react":
+                else:
                     strategy = AgentEntity.Strategy.CHAIN_OF_THOUGHT
-                case _:
-                    # old configs, try to detect default strategy
-                    if config["model"]["provider"] == "openai":
-                        strategy = AgentEntity.Strategy.FUNCTION_CALLING
-                    else:
-                        strategy = AgentEntity.Strategy.CHAIN_OF_THOUGHT
 
             agent_tools = []
             for tool in agent_dict.get("tools", []):

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

@@ -4,8 +4,8 @@ import contextvars
 import logging
 import threading
 import uuid
-from collections.abc import Generator, Mapping
-from typing import TYPE_CHECKING, Any, Literal, Union, overload
+from collections.abc import Generator, Mapping, Sequence
+from typing import TYPE_CHECKING, Any, Literal, TypeVar, Union, overload
 
 from flask import Flask, current_app
 from pydantic import ValidationError
@@ -29,21 +29,25 @@ from core.app.apps.message_based_app_generator import MessageBasedAppGenerator
 from core.app.apps.message_based_app_queue_manager import MessageBasedAppQueueManager
 from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, InvokeFrom
 from core.app.entities.task_entities import ChatbotAppBlockingResponse, ChatbotAppStreamResponse
+from core.app.layers.pause_state_persist_layer import PauseStateLayerConfig, PauseStatePersistenceLayer
 from core.helper.trace_id_helper import extract_external_trace_id_from_args
 from core.model_runtime.errors.invoke import InvokeAuthorizationError
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.prompt.utils.get_thread_messages_length import get_thread_messages_length
 from core.repositories import DifyCoreRepositoryFactory
+from core.workflow.graph_engine.layers.base import GraphEngineLayer
 from core.workflow.repositories.draft_variable_repository import (
     DraftVariableSaverFactory,
 )
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
+from core.workflow.runtime import GraphRuntimeState
 from core.workflow.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader
 from extensions.ext_database import db
 from factories import file_factory
 from libs.flask_utils import preserve_flask_contexts
 from models import Account, App, Conversation, EndUser, Message, Workflow, WorkflowNodeExecutionTriggeredFrom
+from models.base import Base
 from models.enums import WorkflowRunTriggeredFrom
 from services.conversation_service import ConversationService
 from services.workflow_draft_variable_service import (
@@ -65,7 +69,9 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         user: Union[Account, EndUser],
         args: Mapping[str, Any],
         invoke_from: InvokeFrom,
+        workflow_run_id: str,
         streaming: Literal[False],
+        pause_state_config: PauseStateLayerConfig | None = None,
     ) -> Mapping[str, Any]: ...
 
     @overload
@@ -74,9 +80,11 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         app_model: App,
         workflow: Workflow,
         user: Union[Account, EndUser],
-        args: Mapping,
+        args: Mapping[str, Any],
         invoke_from: InvokeFrom,
+        workflow_run_id: str,
         streaming: Literal[True],
+        pause_state_config: PauseStateLayerConfig | None = None,
     ) -> Generator[Mapping | str, None, None]: ...
 
     @overload
@@ -85,9 +93,11 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         app_model: App,
         workflow: Workflow,
         user: Union[Account, EndUser],
-        args: Mapping,
+        args: Mapping[str, Any],
         invoke_from: InvokeFrom,
+        workflow_run_id: str,
         streaming: bool,
+        pause_state_config: PauseStateLayerConfig | None = None,
     ) -> Mapping[str, Any] | Generator[str | Mapping, None, None]: ...
 
     def generate(
@@ -95,9 +105,11 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         app_model: App,
         workflow: Workflow,
         user: Union[Account, EndUser],
-        args: Mapping,
+        args: Mapping[str, Any],
         invoke_from: InvokeFrom,
+        workflow_run_id: str,
         streaming: bool = True,
+        pause_state_config: PauseStateLayerConfig | None = None,
     ) -> Mapping[str, Any] | Generator[str | Mapping, None, None]:
         """
         Generate App response.
@@ -161,7 +173,6 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
             # always enable retriever resource in debugger mode
             app_config.additional_features.show_retrieve_source = True  # type: ignore
 
-        workflow_run_id = str(uuid.uuid4())
         # init application generate entity
         application_generate_entity = AdvancedChatAppGenerateEntity(
             task_id=str(uuid.uuid4()),
@@ -179,7 +190,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
             invoke_from=invoke_from,
             extras=extras,
             trace_manager=trace_manager,
-            workflow_run_id=workflow_run_id,
+            workflow_run_id=str(workflow_run_id),
         )
         contexts.plugin_tool_providers.set({})
         contexts.plugin_tool_providers_lock.set(threading.Lock())
@@ -216,6 +227,38 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
             workflow_node_execution_repository=workflow_node_execution_repository,
             conversation=conversation,
             stream=streaming,
+            pause_state_config=pause_state_config,
+        )
+
+    def resume(
+        self,
+        *,
+        app_model: App,
+        workflow: Workflow,
+        user: Union[Account, EndUser],
+        conversation: Conversation,
+        message: Message,
+        application_generate_entity: AdvancedChatAppGenerateEntity,
+        workflow_execution_repository: WorkflowExecutionRepository,
+        workflow_node_execution_repository: WorkflowNodeExecutionRepository,
+        graph_runtime_state: GraphRuntimeState,
+        pause_state_config: PauseStateLayerConfig | None = None,
+    ):
+        """
+        Resume a paused advanced chat execution.
+        """
+        return self._generate(
+            workflow=workflow,
+            user=user,
+            invoke_from=application_generate_entity.invoke_from,
+            application_generate_entity=application_generate_entity,
+            workflow_execution_repository=workflow_execution_repository,
+            workflow_node_execution_repository=workflow_node_execution_repository,
+            conversation=conversation,
+            message=message,
+            stream=application_generate_entity.stream,
+            pause_state_config=pause_state_config,
+            graph_runtime_state=graph_runtime_state,
         )
 
     def single_iteration_generate(
@@ -396,8 +439,12 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         workflow_execution_repository: WorkflowExecutionRepository,
         workflow_node_execution_repository: WorkflowNodeExecutionRepository,
         conversation: Conversation | None = None,
+        message: Message | None = None,
         stream: bool = True,
         variable_loader: VariableLoader = DUMMY_VARIABLE_LOADER,
+        pause_state_config: PauseStateLayerConfig | None = None,
+        graph_runtime_state: GraphRuntimeState | None = None,
+        graph_engine_layers: Sequence[GraphEngineLayer] = (),
     ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any], Any, None]:
         """
         Generate App response.
@@ -411,12 +458,12 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         :param conversation: conversation
         :param stream: is stream
         """
-        is_first_conversation = False
-        if not conversation:
-            is_first_conversation = True
+        is_first_conversation = conversation is None
 
-        # init generate records
-        (conversation, message) = self._init_generate_records(application_generate_entity, conversation)
+        if conversation is not None and message is not None:
+            pass
+        else:
+            conversation, message = self._init_generate_records(application_generate_entity, conversation)
 
         if is_first_conversation:
             # update conversation features
@@ -439,6 +486,16 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
             message_id=message.id,
         )
 
+        graph_layers: list[GraphEngineLayer] = list(graph_engine_layers)
+        if pause_state_config is not None:
+            graph_layers.append(
+                PauseStatePersistenceLayer(
+                    session_factory=pause_state_config.session_factory,
+                    generate_entity=application_generate_entity,
+                    state_owner_user_id=pause_state_config.state_owner_user_id,
+                )
+            )
+
         # new thread with request context and contextvars
         context = contextvars.copy_context()
 
@@ -454,14 +511,25 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
                 "variable_loader": variable_loader,
                 "workflow_execution_repository": workflow_execution_repository,
                 "workflow_node_execution_repository": workflow_node_execution_repository,
+                "graph_engine_layers": tuple(graph_layers),
+                "graph_runtime_state": graph_runtime_state,
             },
         )
 
         worker_thread.start()
 
         # release database connection, because the following new thread operations may take a long time
-        db.session.refresh(workflow)
-        db.session.refresh(message)
+        with Session(bind=db.engine, expire_on_commit=False) as session:
+            workflow = _refresh_model(session, workflow)
+            message = _refresh_model(session, message)
+        #     workflow_ = session.get(Workflow, workflow.id)
+        #     assert workflow_ is not None
+        #     workflow = workflow_
+        #     message_ = session.get(Message, message.id)
+        #     assert message_ is not None
+        #     message = message_
+        # db.session.refresh(workflow)
+        # db.session.refresh(message)
         # db.session.refresh(user)
         db.session.close()
 
@@ -490,6 +558,8 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         variable_loader: VariableLoader,
         workflow_execution_repository: WorkflowExecutionRepository,
         workflow_node_execution_repository: WorkflowNodeExecutionRepository,
+        graph_engine_layers: Sequence[GraphEngineLayer] = (),
+        graph_runtime_state: GraphRuntimeState | None = None,
     ):
         """
         Generate worker in a new thread.
@@ -547,6 +617,8 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
                 app=app,
                 workflow_execution_repository=workflow_execution_repository,
                 workflow_node_execution_repository=workflow_node_execution_repository,
+                graph_engine_layers=graph_engine_layers,
+                graph_runtime_state=graph_runtime_state,
             )
 
             try:
@@ -614,3 +686,13 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
             else:
                 logger.exception("Failed to process generate task pipeline, conversation_id: %s", conversation.id)
                 raise e
+
+
+_T = TypeVar("_T", bound=Base)
+
+
+def _refresh_model(session, model: _T) -> _T:
+    with Session(bind=db.engine, expire_on_commit=False) as session:
+        detach_model = session.get(type(model), model.id)
+        assert detach_model is not None
+        return detach_model

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

@@ -66,6 +66,7 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
         workflow_execution_repository: WorkflowExecutionRepository,
         workflow_node_execution_repository: WorkflowNodeExecutionRepository,
         graph_engine_layers: Sequence[GraphEngineLayer] = (),
+        graph_runtime_state: GraphRuntimeState | None = None,
     ):
         super().__init__(
             queue_manager=queue_manager,
@@ -82,6 +83,7 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
         self._app = app
         self._workflow_execution_repository = workflow_execution_repository
         self._workflow_node_execution_repository = workflow_node_execution_repository
+        self._resume_graph_runtime_state = graph_runtime_state
 
     @trace_span(WorkflowAppRunnerHandler)
     def run(self):
@@ -110,7 +112,21 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
             invoke_from = InvokeFrom.DEBUGGER
         user_from = self._resolve_user_from(invoke_from)
 
-        if self.application_generate_entity.single_iteration_run or self.application_generate_entity.single_loop_run:
+        resume_state = self._resume_graph_runtime_state
+
+        if resume_state is not None:
+            graph_runtime_state = resume_state
+            variable_pool = graph_runtime_state.variable_pool
+            graph = self._init_graph(
+                graph_config=self._workflow.graph_dict,
+                graph_runtime_state=graph_runtime_state,
+                workflow_id=self._workflow.id,
+                tenant_id=self._workflow.tenant_id,
+                user_id=self.application_generate_entity.user_id,
+                invoke_from=invoke_from,
+                user_from=user_from,
+            )
+        elif self.application_generate_entity.single_iteration_run or self.application_generate_entity.single_loop_run:
             # Handle single iteration or single loop run
             graph, variable_pool, graph_runtime_state = self._prepare_single_node_execution(
                 workflow=self._workflow,

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

@@ -24,6 +24,8 @@ from core.app.entities.queue_entities import (
     QueueAgentLogEvent,
     QueueAnnotationReplyEvent,
     QueueErrorEvent,
+    QueueHumanInputFormFilledEvent,
+    QueueHumanInputFormTimeoutEvent,
     QueueIterationCompletedEvent,
     QueueIterationNextEvent,
     QueueIterationStartEvent,
@@ -42,6 +44,7 @@ from core.app.entities.queue_entities import (
     QueueTextChunkEvent,
     QueueWorkflowFailedEvent,
     QueueWorkflowPartialSuccessEvent,
+    QueueWorkflowPausedEvent,
     QueueWorkflowStartedEvent,
     QueueWorkflowSucceededEvent,
     WorkflowQueueMessage,
@@ -63,6 +66,8 @@ from core.base.tts import AppGeneratorTTSPublisher, AudioTrunk
 from core.model_runtime.entities.llm_entities import LLMUsage
 from core.model_runtime.utils.encoders import jsonable_encoder
 from core.ops.ops_trace_manager import TraceQueueManager
+from core.repositories.human_input_repository import HumanInputFormRepositoryImpl
+from core.workflow.entities.pause_reason import HumanInputRequired
 from core.workflow.enums import WorkflowExecutionStatus
 from core.workflow.nodes import NodeType
 from core.workflow.repositories.draft_variable_repository import DraftVariableSaverFactory
@@ -71,7 +76,8 @@ from core.workflow.system_variable import SystemVariable
 from extensions.ext_database import db
 from libs.datetime_utils import naive_utc_now
 from models import Account, Conversation, EndUser, Message, MessageFile
-from models.enums import CreatorUserRole
+from models.enums import CreatorUserRole, MessageStatus
+from models.execution_extra_content import HumanInputContent
 from models.workflow import Workflow
 
 logger = logging.getLogger(__name__)
@@ -128,6 +134,7 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         )
 
         self._task_state = WorkflowTaskState()
+        self._seed_task_state_from_message(message)
         self._message_cycle_manager = MessageCycleManager(
             application_generate_entity=application_generate_entity, task_state=self._task_state
         )
@@ -135,6 +142,7 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         self._application_generate_entity = application_generate_entity
         self._workflow_id = workflow.id
         self._workflow_features_dict = workflow.features_dict
+        self._workflow_tenant_id = workflow.tenant_id
         self._conversation_id = conversation.id
         self._conversation_mode = conversation.mode
         self._message_id = message.id
@@ -144,8 +152,13 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         self._workflow_run_id: str = ""
         self._draft_var_saver_factory = draft_var_saver_factory
         self._graph_runtime_state: GraphRuntimeState | None = None
+        self._message_saved_on_pause = False
         self._seed_graph_runtime_state_from_queue_manager()
 
+    def _seed_task_state_from_message(self, message: Message) -> None:
+        if message.status == MessageStatus.PAUSED and message.answer:
+            self._task_state.answer = message.answer
+
     def process(self) -> Union[ChatbotAppBlockingResponse, Generator[ChatbotAppStreamResponse, None, None]]:
         """
         Process generate task pipeline.
@@ -308,6 +321,7 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
             task_id=self._application_generate_entity.task_id,
             workflow_run_id=run_id,
             workflow_id=self._workflow_id,
+            reason=event.reason,
         )
 
         yield workflow_start_resp
@@ -525,6 +539,35 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         )
 
         yield workflow_finish_resp
+
+    def _handle_workflow_paused_event(
+        self,
+        event: QueueWorkflowPausedEvent,
+        **kwargs,
+    ) -> Generator[StreamResponse, None, None]:
+        """Handle workflow paused events."""
+        validated_state = self._ensure_graph_runtime_initialized()
+        responses = self._workflow_response_converter.workflow_pause_to_stream_response(
+            event=event,
+            task_id=self._application_generate_entity.task_id,
+            graph_runtime_state=validated_state,
+        )
+        for reason in event.reasons:
+            if isinstance(reason, HumanInputRequired):
+                self._persist_human_input_extra_content(form_id=reason.form_id, node_id=reason.node_id)
+        yield from responses
+        resolved_state: GraphRuntimeState | None = None
+        try:
+            resolved_state = self._ensure_graph_runtime_initialized()
+        except ValueError:
+            resolved_state = None
+
+        with self._database_session() as session:
+            self._save_message(session=session, graph_runtime_state=resolved_state)
+            message = self._get_message(session=session)
+            if message is not None:
+                message.status = MessageStatus.PAUSED
+            self._message_saved_on_pause = True
         self._base_task_pipeline.queue_manager.publish(QueueAdvancedChatMessageEndEvent(), PublishFrom.TASK_PIPELINE)
 
     def _handle_workflow_failed_event(
@@ -614,9 +657,10 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
                 reason=QueueMessageReplaceEvent.MessageReplaceReason.OUTPUT_MODERATION,
             )
 
-        # Save message
-        with self._database_session() as session:
-            self._save_message(session=session, graph_runtime_state=resolved_state)
+        # Save message unless it has already been persisted on pause.
+        if not self._message_saved_on_pause:
+            with self._database_session() as session:
+                self._save_message(session=session, graph_runtime_state=resolved_state)
 
         yield self._message_end_to_stream_response()
 
@@ -642,6 +686,65 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         """Handle message replace events."""
         yield self._message_cycle_manager.message_replace_to_stream_response(answer=event.text, reason=event.reason)
 
+    def _handle_human_input_form_filled_event(
+        self, event: QueueHumanInputFormFilledEvent, **kwargs
+    ) -> Generator[StreamResponse, None, None]:
+        """Handle human input form filled events."""
+        self._persist_human_input_extra_content(node_id=event.node_id)
+        yield self._workflow_response_converter.human_input_form_filled_to_stream_response(
+            event=event, task_id=self._application_generate_entity.task_id
+        )
+
+    def _handle_human_input_form_timeout_event(
+        self, event: QueueHumanInputFormTimeoutEvent, **kwargs
+    ) -> Generator[StreamResponse, None, None]:
+        """Handle human input form timeout events."""
+        yield self._workflow_response_converter.human_input_form_timeout_to_stream_response(
+            event=event, task_id=self._application_generate_entity.task_id
+        )
+
+    def _persist_human_input_extra_content(self, *, node_id: str | None = None, form_id: str | None = None) -> None:
+        if not self._workflow_run_id or not self._message_id:
+            return
+
+        if form_id is None:
+            if node_id is None:
+                return
+            form_id = self._load_human_input_form_id(node_id=node_id)
+            if form_id is None:
+                logger.warning(
+                    "HumanInput form not found for workflow run %s node %s",
+                    self._workflow_run_id,
+                    node_id,
+                )
+                return
+
+        with self._database_session() as session:
+            exists_stmt = select(HumanInputContent).where(
+                HumanInputContent.workflow_run_id == self._workflow_run_id,
+                HumanInputContent.message_id == self._message_id,
+                HumanInputContent.form_id == form_id,
+            )
+            if session.scalar(exists_stmt) is not None:
+                return
+
+            content = HumanInputContent(
+                workflow_run_id=self._workflow_run_id,
+                message_id=self._message_id,
+                form_id=form_id,
+            )
+            session.add(content)
+
+    def _load_human_input_form_id(self, *, node_id: str) -> str | None:
+        form_repository = HumanInputFormRepositoryImpl(
+            session_factory=db.engine,
+            tenant_id=self._workflow_tenant_id,
+        )
+        form = form_repository.get_form(self._workflow_run_id, node_id)
+        if form is None:
+            return None
+        return form.id
+
     def _handle_agent_log_event(self, event: QueueAgentLogEvent, **kwargs) -> Generator[StreamResponse, None, None]:
         """Handle agent log events."""
         yield self._workflow_response_converter.handle_agent_log(
@@ -659,6 +762,7 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
             QueueWorkflowStartedEvent: self._handle_workflow_started_event,
             QueueWorkflowSucceededEvent: self._handle_workflow_succeeded_event,
             QueueWorkflowPartialSuccessEvent: self._handle_workflow_partial_success_event,
+            QueueWorkflowPausedEvent: self._handle_workflow_paused_event,
             QueueWorkflowFailedEvent: self._handle_workflow_failed_event,
             # Node events
             QueueNodeRetryEvent: self._handle_node_retry_event,
@@ -680,6 +784,8 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
             QueueMessageReplaceEvent: self._handle_message_replace_event,
             QueueAdvancedChatMessageEndEvent: self._handle_advanced_chat_message_end_event,
             QueueAgentLogEvent: self._handle_agent_log_event,
+            QueueHumanInputFormFilledEvent: self._handle_human_input_form_filled_event,
+            QueueHumanInputFormTimeoutEvent: self._handle_human_input_form_timeout_event,
         }
 
     def _dispatch_event(
@@ -747,6 +853,9 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
                 case QueueWorkflowFailedEvent():
                     yield from self._handle_workflow_failed_event(event, trace_manager=trace_manager)
                     break
+                case QueueWorkflowPausedEvent():
+                    yield from self._handle_workflow_paused_event(event)
+                    break
 
                 case QueueStopEvent():
                     yield from self._handle_stop_event(event, graph_runtime_state=None, trace_manager=trace_manager)
@@ -772,6 +881,11 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
 
     def _save_message(self, *, session: Session, graph_runtime_state: GraphRuntimeState | None = None):
         message = self._get_message(session=session)
+        if message is None:
+            return
+
+        if message.status == MessageStatus.PAUSED:
+            message.status = MessageStatus.NORMAL
 
         # If there are assistant files, remove markdown image links from answer
         answer_text = self._task_state.answer

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

@@ -5,9 +5,14 @@ from dataclasses import dataclass
 from datetime import datetime
 from typing import Any, NewType, Union
 
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
 from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, InvokeFrom, WorkflowAppGenerateEntity
 from core.app.entities.queue_entities import (
     QueueAgentLogEvent,
+    QueueHumanInputFormFilledEvent,
+    QueueHumanInputFormTimeoutEvent,
     QueueIterationCompletedEvent,
     QueueIterationNextEvent,
     QueueIterationStartEvent,
@@ -19,9 +24,13 @@ from core.app.entities.queue_entities import (
     QueueNodeRetryEvent,
     QueueNodeStartedEvent,
     QueueNodeSucceededEvent,
+    QueueWorkflowPausedEvent,
 )
 from core.app.entities.task_entities import (
     AgentLogStreamResponse,
+    HumanInputFormFilledResponse,
+    HumanInputFormTimeoutResponse,
+    HumanInputRequiredResponse,
     IterationNodeCompletedStreamResponse,
     IterationNodeNextStreamResponse,
     IterationNodeStartStreamResponse,
@@ -31,7 +40,9 @@ from core.app.entities.task_entities import (
     NodeFinishStreamResponse,
     NodeRetryStreamResponse,
     NodeStartStreamResponse,
+    StreamResponse,
     WorkflowFinishStreamResponse,
+    WorkflowPauseStreamResponse,
     WorkflowStartStreamResponse,
 )
 from core.file import FILE_MODEL_IDENTITY, File
@@ -40,6 +51,8 @@ from core.tools.entities.tool_entities import ToolProviderType
 from core.tools.tool_manager import ToolManager
 from core.trigger.trigger_manager import TriggerManager
 from core.variables.segments import ArrayFileSegment, FileSegment, Segment
+from core.workflow.entities.pause_reason import HumanInputRequired
+from core.workflow.entities.workflow_start_reason import WorkflowStartReason
 from core.workflow.enums import (
     NodeType,
     SystemVariableKey,
@@ -51,8 +64,11 @@ from core.workflow.runtime import GraphRuntimeState
 from core.workflow.system_variable import SystemVariable
 from core.workflow.workflow_entry import WorkflowEntry
 from core.workflow.workflow_type_encoder import WorkflowRuntimeTypeConverter
+from extensions.ext_database import db
 from libs.datetime_utils import naive_utc_now
 from models import Account, EndUser
+from models.human_input import HumanInputForm
+from models.workflow import WorkflowRun
 from services.variable_truncator import BaseTruncator, DummyVariableTruncator, VariableTruncator
 
 NodeExecutionId = NewType("NodeExecutionId", str)
@@ -191,6 +207,7 @@ class WorkflowResponseConverter:
         task_id: str,
         workflow_run_id: str,
         workflow_id: str,
+        reason: WorkflowStartReason,
     ) -> WorkflowStartStreamResponse:
         run_id = self._ensure_workflow_run_id(workflow_run_id)
         started_at = naive_utc_now()
@@ -204,6 +221,7 @@ class WorkflowResponseConverter:
                 workflow_id=workflow_id,
                 inputs=self._workflow_inputs,
                 created_at=int(started_at.timestamp()),
+                reason=reason,
             ),
         )
 
@@ -250,7 +268,7 @@ class WorkflowResponseConverter:
             data=WorkflowFinishStreamResponse.Data(
                 id=run_id,
                 workflow_id=workflow_id,
-                status=status,
+                status=status.value,
                 outputs=encoded_outputs,
                 error=error,
                 elapsed_time=elapsed_time,
@@ -264,6 +282,160 @@ class WorkflowResponseConverter:
             ),
         )
 
+    def workflow_pause_to_stream_response(
+        self,
+        *,
+        event: QueueWorkflowPausedEvent,
+        task_id: str,
+        graph_runtime_state: GraphRuntimeState,
+    ) -> list[StreamResponse]:
+        run_id = self._ensure_workflow_run_id()
+        started_at = self._workflow_started_at
+        if started_at is None:
+            raise ValueError(
+                "workflow_pause_to_stream_response called before workflow_start_to_stream_response",
+            )
+        paused_at = naive_utc_now()
+        elapsed_time = (paused_at - started_at).total_seconds()
+        encoded_outputs = self._encode_outputs(event.outputs) or {}
+        if self._application_generate_entity.invoke_from == InvokeFrom.SERVICE_API:
+            encoded_outputs = {}
+        pause_reasons = [reason.model_dump(mode="json") for reason in event.reasons]
+        human_input_form_ids = [reason.form_id for reason in event.reasons if isinstance(reason, HumanInputRequired)]
+        expiration_times_by_form_id: dict[str, datetime] = {}
+        if human_input_form_ids:
+            stmt = select(HumanInputForm.id, HumanInputForm.expiration_time).where(
+                HumanInputForm.id.in_(human_input_form_ids)
+            )
+            with Session(bind=db.engine) as session:
+                for form_id, expiration_time in session.execute(stmt):
+                    expiration_times_by_form_id[str(form_id)] = expiration_time
+
+        responses: list[StreamResponse] = []
+
+        for reason in event.reasons:
+            if isinstance(reason, HumanInputRequired):
+                expiration_time = expiration_times_by_form_id.get(reason.form_id)
+                if expiration_time is None:
+                    raise ValueError(f"HumanInputForm not found for pause reason, form_id={reason.form_id}")
+                responses.append(
+                    HumanInputRequiredResponse(
+                        task_id=task_id,
+                        workflow_run_id=run_id,
+                        data=HumanInputRequiredResponse.Data(
+                            form_id=reason.form_id,
+                            node_id=reason.node_id,
+                            node_title=reason.node_title,
+                            form_content=reason.form_content,
+                            inputs=reason.inputs,
+                            actions=reason.actions,
+                            display_in_ui=reason.display_in_ui,
+                            form_token=reason.form_token,
+                            resolved_default_values=reason.resolved_default_values,
+                            expiration_time=int(expiration_time.timestamp()),
+                        ),
+                    )
+                )
+
+        responses.append(
+            WorkflowPauseStreamResponse(
+                task_id=task_id,
+                workflow_run_id=run_id,
+                data=WorkflowPauseStreamResponse.Data(
+                    workflow_run_id=run_id,
+                    paused_nodes=list(event.paused_nodes),
+                    outputs=encoded_outputs,
+                    reasons=pause_reasons,
+                    status=WorkflowExecutionStatus.PAUSED.value,
+                    created_at=int(started_at.timestamp()),
+                    elapsed_time=elapsed_time,
+                    total_tokens=graph_runtime_state.total_tokens,
+                    total_steps=graph_runtime_state.node_run_steps,
+                ),
+            )
+        )
+
+        return responses
+
+    def human_input_form_filled_to_stream_response(
+        self, *, event: QueueHumanInputFormFilledEvent, task_id: str
+    ) -> HumanInputFormFilledResponse:
+        run_id = self._ensure_workflow_run_id()
+        return HumanInputFormFilledResponse(
+            task_id=task_id,
+            workflow_run_id=run_id,
+            data=HumanInputFormFilledResponse.Data(
+                node_id=event.node_id,
+                node_title=event.node_title,
+                rendered_content=event.rendered_content,
+                action_id=event.action_id,
+                action_text=event.action_text,
+            ),
+        )
+
+    def human_input_form_timeout_to_stream_response(
+        self, *, event: QueueHumanInputFormTimeoutEvent, task_id: str
+    ) -> HumanInputFormTimeoutResponse:
+        run_id = self._ensure_workflow_run_id()
+        return HumanInputFormTimeoutResponse(
+            task_id=task_id,
+            workflow_run_id=run_id,
+            data=HumanInputFormTimeoutResponse.Data(
+                node_id=event.node_id,
+                node_title=event.node_title,
+                expiration_time=int(event.expiration_time.timestamp()),
+            ),
+        )
+
+    @classmethod
+    def workflow_run_result_to_finish_response(
+        cls,
+        *,
+        task_id: str,
+        workflow_run: WorkflowRun,
+        creator_user: Account | EndUser,
+    ) -> WorkflowFinishStreamResponse:
+        run_id = workflow_run.id
+        elapsed_time = workflow_run.elapsed_time
+
+        encoded_outputs = workflow_run.outputs_dict
+        finished_at = workflow_run.finished_at
+        assert finished_at is not None
+
+        created_by: Mapping[str, object]
+        user = creator_user
+        if isinstance(user, Account):
+            created_by = {
+                "id": user.id,
+                "name": user.name,
+                "email": user.email,
+            }
+        else:
+            created_by = {
+                "id": user.id,
+                "user": user.session_id,
+            }
+
+        return WorkflowFinishStreamResponse(
+            task_id=task_id,
+            workflow_run_id=run_id,
+            data=WorkflowFinishStreamResponse.Data(
+                id=run_id,
+                workflow_id=workflow_run.workflow_id,
+                status=workflow_run.status.value,
+                outputs=encoded_outputs,
+                error=workflow_run.error,
+                elapsed_time=elapsed_time,
+                total_tokens=workflow_run.total_tokens,
+                total_steps=workflow_run.total_steps,
+                created_by=created_by,
+                created_at=int(workflow_run.created_at.timestamp()),
+                finished_at=int(finished_at.timestamp()),
+                files=cls.fetch_files_from_node_outputs(encoded_outputs),
+                exceptions_count=workflow_run.exceptions_count,
+            ),
+        )
+
     def workflow_node_start_to_stream_response(
         self,
         *,
@@ -340,13 +512,13 @@ class WorkflowResponseConverter:
         metadata = self._merge_metadata(event.execution_metadata, snapshot)
 
         if isinstance(event, QueueNodeSucceededEvent):
-            status = WorkflowNodeExecutionStatus.SUCCEEDED
+            status = WorkflowNodeExecutionStatus.SUCCEEDED.value
             error_message = event.error
         elif isinstance(event, QueueNodeFailedEvent):
-            status = WorkflowNodeExecutionStatus.FAILED
+            status = WorkflowNodeExecutionStatus.FAILED.value
             error_message = event.error
         else:
-            status = WorkflowNodeExecutionStatus.EXCEPTION
+            status = WorkflowNodeExecutionStatus.EXCEPTION.value
             error_message = event.error
 
         return NodeFinishStreamResponse(
@@ -413,7 +585,7 @@ class WorkflowResponseConverter:
                 process_data_truncated=process_data_truncated,
                 outputs=outputs,
                 outputs_truncated=outputs_truncated,
-                status=WorkflowNodeExecutionStatus.RETRY,
+                status=WorkflowNodeExecutionStatus.RETRY.value,
                 error=event.error,
                 elapsed_time=elapsed_time,
                 execution_metadata=metadata,
@@ -592,7 +764,8 @@ class WorkflowResponseConverter:
             ),
         )
 
-    def fetch_files_from_node_outputs(self, outputs_dict: Mapping[str, Any] | None) -> Sequence[Mapping[str, Any]]:
+    @classmethod
+    def fetch_files_from_node_outputs(cls, outputs_dict: Mapping[str, Any] | None) -> Sequence[Mapping[str, Any]]:
         """
         Fetch files from node outputs
         :param outputs_dict: node outputs dict
@@ -601,7 +774,7 @@ class WorkflowResponseConverter:
         if not outputs_dict:
             return []
 
-        files = [self._fetch_files_from_variable_value(output_value) for output_value in outputs_dict.values()]
+        files = [cls._fetch_files_from_variable_value(output_value) for output_value in outputs_dict.values()]
         # Remove None
         files = [file for file in files if file]
         # Flatten list

api/core/app/apps/message_based_app_generator.py

@@ -1,6 +1,6 @@
 import json
 import logging
-from collections.abc import Generator
+from collections.abc import Callable, Generator, Mapping
 from typing import Union, cast
 
 from sqlalchemy import select
@@ -10,12 +10,14 @@ from core.app.app_config.entities import EasyUIBasedAppConfig, EasyUIBasedAppMod
 from core.app.apps.base_app_generator import BaseAppGenerator
 from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.apps.exc import GenerateTaskStoppedError
+from core.app.apps.streaming_utils import stream_topic_events
 from core.app.entities.app_invoke_entities import (
     AdvancedChatAppGenerateEntity,
     AgentChatAppGenerateEntity,
     AppGenerateEntity,
     ChatAppGenerateEntity,
     CompletionAppGenerateEntity,
+    ConversationAppGenerateEntity,
     InvokeFrom,
 )
 from core.app.entities.task_entities import (
@@ -27,6 +29,8 @@ from core.app.entities.task_entities import (
 from core.app.task_pipeline.easy_ui_based_generate_task_pipeline import EasyUIBasedGenerateTaskPipeline
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
 from extensions.ext_database import db
+from extensions.ext_redis import get_pubsub_broadcast_channel
+from libs.broadcast_channel.channel import Topic
 from libs.datetime_utils import naive_utc_now
 from models import Account
 from models.enums import CreatorUserRole
@@ -156,6 +160,7 @@ class MessageBasedAppGenerator(BaseAppGenerator):
         query = application_generate_entity.query or "New conversation"
         conversation_name = (query[:20] + "…") if len(query) > 20 else query
 
+        created_new_conversation = conversation is None
         try:
             if not conversation:
                 conversation = Conversation(
@@ -232,6 +237,10 @@ class MessageBasedAppGenerator(BaseAppGenerator):
                 db.session.add_all(message_files)
 
             db.session.commit()
+
+            if isinstance(application_generate_entity, ConversationAppGenerateEntity):
+                application_generate_entity.conversation_id = conversation.id
+                application_generate_entity.is_new_conversation = created_new_conversation
             return conversation, message
         except Exception:
             db.session.rollback()
@@ -284,3 +293,29 @@ class MessageBasedAppGenerator(BaseAppGenerator):
             raise MessageNotExistsError("Message not exists")
 
         return message
+
+    @staticmethod
+    def _make_channel_key(app_mode: AppMode, workflow_run_id: str):
+        return f"channel:{app_mode}:{workflow_run_id}"
+
+    @classmethod
+    def get_response_topic(cls, app_mode: AppMode, workflow_run_id: str) -> Topic:
+        key = cls._make_channel_key(app_mode, workflow_run_id)
+        channel = get_pubsub_broadcast_channel()
+        topic = channel.topic(key)
+        return topic
+
+    @classmethod
+    def retrieve_events(
+        cls,
+        app_mode: AppMode,
+        workflow_run_id: str,
+        idle_timeout=300,
+        on_subscribe: Callable[[], None] | None = None,
+    ) -> Generator[Mapping | str, None, None]:
+        topic = cls.get_response_topic(app_mode, workflow_run_id)
+        return stream_topic_events(
+            topic=topic,
+            idle_timeout=idle_timeout,
+            on_subscribe=on_subscribe,
+        )

api/core/app/apps/message_generator.py

@@ -0,0 +1,36 @@
+from collections.abc import Callable, Generator, Mapping
+
+from core.app.apps.streaming_utils import stream_topic_events
+from extensions.ext_redis import get_pubsub_broadcast_channel
+from libs.broadcast_channel.channel import Topic
+from models.model import AppMode
+
+
+class MessageGenerator:
+    @staticmethod
+    def _make_channel_key(app_mode: AppMode, workflow_run_id: str):
+        return f"channel:{app_mode}:{str(workflow_run_id)}"
+
+    @classmethod
+    def get_response_topic(cls, app_mode: AppMode, workflow_run_id: str) -> Topic:
+        key = cls._make_channel_key(app_mode, workflow_run_id)
+        channel = get_pubsub_broadcast_channel()
+        topic = channel.topic(key)
+        return topic
+
+    @classmethod
+    def retrieve_events(
+        cls,
+        app_mode: AppMode,
+        workflow_run_id: str,
+        idle_timeout=300,
+        ping_interval: float = 10.0,
+        on_subscribe: Callable[[], None] | None = None,
+    ) -> Generator[Mapping | str, None, None]:
+        topic = cls.get_response_topic(app_mode, workflow_run_id)
+        return stream_topic_events(
+            topic=topic,
+            idle_timeout=idle_timeout,
+            ping_interval=ping_interval,
+            on_subscribe=on_subscribe,
+        )

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

@@ -120,7 +120,7 @@ class PipelineGenerator(BaseAppGenerator):
                 raise ValueError("Pipeline dataset is required")
         inputs: Mapping[str, Any] = args["inputs"]
         start_node_id: str = args["start_node_id"]
-        datasource_type = DatasourceProviderType(args["datasource_type"])
+        datasource_type: str = args["datasource_type"]
         datasource_info_list: list[Mapping[str, Any]] = self._format_datasource_info_list(
             datasource_type, args["datasource_info_list"], pipeline, workflow, start_node_id, user
         )
@@ -660,7 +660,7 @@ class PipelineGenerator(BaseAppGenerator):
         tenant_id: str,
         dataset_id: str,
         built_in_field_enabled: bool,
-        datasource_type: DatasourceProviderType,
+        datasource_type: str,
         datasource_info: Mapping[str, Any],
         created_from: str,
         position: int,
@@ -668,17 +668,17 @@ class PipelineGenerator(BaseAppGenerator):
         batch: str,
         document_form: str,
     ):
-        match datasource_type:
-            case DatasourceProviderType.LOCAL_FILE:
-                name = datasource_info.get("name", "untitled")
-            case DatasourceProviderType.ONLINE_DOCUMENT:
-                name = datasource_info.get("page", {}).get("page_name", "untitled")
-            case DatasourceProviderType.WEBSITE_CRAWL:
-                name = datasource_info.get("title", "untitled")
-            case DatasourceProviderType.ONLINE_DRIVE:
-                name = datasource_info.get("name", "untitled")
-            case _:
-                raise ValueError(f"Unsupported datasource type: {datasource_type}")
+        if datasource_type == "local_file":
+            name = datasource_info.get("name", "untitled")
+        elif datasource_type == "online_document":
+            name = datasource_info.get("page", {}).get("page_name", "untitled")
+        elif datasource_type == "website_crawl":
+            name = datasource_info.get("title", "untitled")
+        elif datasource_type == "online_drive":
+            name = datasource_info.get("name", "untitled")
+        else:
+            raise ValueError(f"Unsupported datasource type: {datasource_type}")
+
         document = Document(
             tenant_id=tenant_id,
             dataset_id=dataset_id,
@@ -706,7 +706,7 @@ class PipelineGenerator(BaseAppGenerator):
 
     def _format_datasource_info_list(
         self,
-        datasource_type: DatasourceProviderType,
+        datasource_type: str,
         datasource_info_list: list[Mapping[str, Any]],
         pipeline: Pipeline,
         workflow: Workflow,
@@ -716,7 +716,7 @@ class PipelineGenerator(BaseAppGenerator):
         """
         Format datasource info list.
         """
-        if datasource_type == DatasourceProviderType.ONLINE_DRIVE:
+        if datasource_type == "online_drive":
             all_files: list[Mapping[str, Any]] = []
             datasource_node_data = None
             datasource_nodes = workflow.graph_dict.get("nodes", [])

api/core/app/apps/streaming_utils.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+import json
+import time
+from collections.abc import Callable, Generator, Iterable, Mapping
+from typing import Any
+
+from core.app.entities.task_entities import StreamEvent
+from libs.broadcast_channel.channel import Topic
+from libs.broadcast_channel.exc import SubscriptionClosedError
+
+
+def stream_topic_events(
+    *,
+    topic: Topic,
+    idle_timeout: float,
+    ping_interval: float | None = None,
+    on_subscribe: Callable[[], None] | None = None,
+    terminal_events: Iterable[str | StreamEvent] | None = None,
+) -> Generator[Mapping[str, Any] | str, None, None]:
+    # send a PING event immediately to prevent the connection staying in pending state for a long time.
+    #
+    # This simplify the debugging process as the DevTools in Chrome does not
+    # provide complete curl command for pending connections.
+    yield StreamEvent.PING.value
+
+    terminal_values = _normalize_terminal_events(terminal_events)
+    last_msg_time = time.time()
+    last_ping_time = last_msg_time
+    with topic.subscribe() as sub:
+        # on_subscribe fires only after the Redis subscription is active.
+        # This is used to gate task start and reduce pub/sub race for the first event.
+        if on_subscribe is not None:
+            on_subscribe()
+        while True:
+            try:
+                msg = sub.receive(timeout=0.1)
+            except SubscriptionClosedError:
+                return
+            if msg is None:
+                current_time = time.time()
+                if current_time - last_msg_time > idle_timeout:
+                    return
+                if ping_interval is not None and current_time - last_ping_time >= ping_interval:
+                    yield StreamEvent.PING.value
+                    last_ping_time = current_time
+                continue
+
+            last_msg_time = time.time()
+            last_ping_time = last_msg_time
+            event = json.loads(msg)
+            yield event
+            if not isinstance(event, dict):
+                continue
+
+            event_type = event.get("event")
+            if event_type in terminal_values:
+                return
+
+
+def _normalize_terminal_events(terminal_events: Iterable[str | StreamEvent] | None) -> set[str]:
+    if not terminal_events:
+        return {StreamEvent.WORKFLOW_FINISHED.value, StreamEvent.WORKFLOW_PAUSED.value}
+    values: set[str] = set()
+    for item in terminal_events:
+        if isinstance(item, StreamEvent):
+            values.add(item.value)
+        else:
+            values.add(str(item))
+    return values

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

@@ -25,6 +25,7 @@ from core.app.apps.workflow.generate_response_converter import WorkflowAppGenera
 from core.app.apps.workflow.generate_task_pipeline import WorkflowAppGenerateTaskPipeline
 from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity
 from core.app.entities.task_entities import WorkflowAppBlockingResponse, WorkflowAppStreamResponse
+from core.app.layers.pause_state_persist_layer import PauseStateLayerConfig, PauseStatePersistenceLayer
 from core.db.session_factory import session_factory
 from core.helper.trace_id_helper import extract_external_trace_id_from_args
 from core.model_runtime.errors.invoke import InvokeAuthorizationError
@@ -34,12 +35,15 @@ from core.workflow.graph_engine.layers.base import GraphEngineLayer
 from core.workflow.repositories.draft_variable_repository import DraftVariableSaverFactory
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
+from core.workflow.runtime import GraphRuntimeState
 from core.workflow.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader
 from extensions.ext_database import db
 from factories import file_factory
 from libs.flask_utils import preserve_flask_contexts
-from models import Account, App, EndUser, Workflow, WorkflowNodeExecutionTriggeredFrom
+from models.account import Account
 from models.enums import WorkflowRunTriggeredFrom
+from models.model import App, EndUser
+from models.workflow import Workflow, WorkflowNodeExecutionTriggeredFrom
 from services.workflow_draft_variable_service import DraftVarLoader, WorkflowDraftVariableService
 
 if TYPE_CHECKING:
@@ -66,9 +70,11 @@ class WorkflowAppGenerator(BaseAppGenerator):
         invoke_from: InvokeFrom,
         streaming: Literal[True],
         call_depth: int,
+        workflow_run_id: str | uuid.UUID | None = None,
         triggered_from: WorkflowRunTriggeredFrom | None = None,
         root_node_id: str | None = None,
         graph_engine_layers: Sequence[GraphEngineLayer] = (),
+        pause_state_config: PauseStateLayerConfig | None = None,
     ) -> Generator[Mapping[str, Any] | str, None, None]: ...
 
     @overload
@@ -82,9 +88,11 @@ class WorkflowAppGenerator(BaseAppGenerator):
         invoke_from: InvokeFrom,
         streaming: Literal[False],
         call_depth: int,
+        workflow_run_id: str | uuid.UUID | None = None,
         triggered_from: WorkflowRunTriggeredFrom | None = None,
         root_node_id: str | None = None,
         graph_engine_layers: Sequence[GraphEngineLayer] = (),
+        pause_state_config: PauseStateLayerConfig | None = None,
     ) -> Mapping[str, Any]: ...
 
     @overload
@@ -98,9 +106,11 @@ class WorkflowAppGenerator(BaseAppGenerator):
         invoke_from: InvokeFrom,
         streaming: bool,
         call_depth: int,
+        workflow_run_id: str | uuid.UUID | None = None,
         triggered_from: WorkflowRunTriggeredFrom | None = None,
         root_node_id: str | None = None,
         graph_engine_layers: Sequence[GraphEngineLayer] = (),
+        pause_state_config: PauseStateLayerConfig | None = None,
     ) -> Union[Mapping[str, Any], Generator[Mapping[str, Any] | str, None, None]]: ...
 
     def generate(
@@ -113,9 +123,11 @@ class WorkflowAppGenerator(BaseAppGenerator):
         invoke_from: InvokeFrom,
         streaming: bool = True,
         call_depth: int = 0,
+        workflow_run_id: str | uuid.UUID | None = None,
         triggered_from: WorkflowRunTriggeredFrom | None = None,
         root_node_id: str | None = None,
         graph_engine_layers: Sequence[GraphEngineLayer] = (),
+        pause_state_config: PauseStateLayerConfig | None = None,
     ) -> Union[Mapping[str, Any], Generator[Mapping[str, Any] | str, None, None]]:
         files: Sequence[Mapping[str, Any]] = args.get("files") or []
 
@@ -150,7 +162,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
         extras = {
             **extract_external_trace_id_from_args(args),
         }
-        workflow_run_id = str(uuid.uuid4())
+        workflow_run_id = str(workflow_run_id or uuid.uuid4())
         # FIXME (Yeuoly): we need to remove the SKIP_PREPARE_USER_INPUTS_KEY from the args
         # trigger shouldn't prepare user inputs
         if self._should_prepare_user_inputs(args):
@@ -216,13 +228,40 @@ class WorkflowAppGenerator(BaseAppGenerator):
             streaming=streaming,
             root_node_id=root_node_id,
             graph_engine_layers=graph_engine_layers,
+            pause_state_config=pause_state_config,
         )
 
-    def resume(self, *, workflow_run_id: str) -> None:
+    def resume(
+        self,
+        *,
+        app_model: App,
+        workflow: Workflow,
+        user: Union[Account, EndUser],
+        application_generate_entity: WorkflowAppGenerateEntity,
+        graph_runtime_state: GraphRuntimeState,
+        workflow_execution_repository: WorkflowExecutionRepository,
+        workflow_node_execution_repository: WorkflowNodeExecutionRepository,
+        graph_engine_layers: Sequence[GraphEngineLayer] = (),
+        pause_state_config: PauseStateLayerConfig | None = None,
+        variable_loader: VariableLoader = DUMMY_VARIABLE_LOADER,
+    ) -> Union[Mapping[str, Any], Generator[str | Mapping[str, Any], None, None]]:
         """
-        @TBD
+        Resume a paused workflow execution using the persisted runtime state.
         """
-        pass
+        return self._generate(
+            app_model=app_model,
+            workflow=workflow,
+            user=user,
+            application_generate_entity=application_generate_entity,
+            invoke_from=application_generate_entity.invoke_from,
+            workflow_execution_repository=workflow_execution_repository,
+            workflow_node_execution_repository=workflow_node_execution_repository,
+            streaming=application_generate_entity.stream,
+            variable_loader=variable_loader,
+            graph_engine_layers=graph_engine_layers,
+            graph_runtime_state=graph_runtime_state,
+            pause_state_config=pause_state_config,
+        )
 
     def _generate(
         self,
@@ -238,6 +277,8 @@ class WorkflowAppGenerator(BaseAppGenerator):
         variable_loader: VariableLoader = DUMMY_VARIABLE_LOADER,
         root_node_id: str | None = None,
         graph_engine_layers: Sequence[GraphEngineLayer] = (),
+        graph_runtime_state: GraphRuntimeState | None = None,
+        pause_state_config: PauseStateLayerConfig | None = None,
     ) -> Union[Mapping[str, Any], Generator[str | Mapping[str, Any], None, None]]:
         """
         Generate App response.
@@ -251,6 +292,8 @@ class WorkflowAppGenerator(BaseAppGenerator):
         :param workflow_node_execution_repository: repository for workflow node execution
         :param streaming: is stream
         """
+        graph_layers: list[GraphEngineLayer] = list(graph_engine_layers)
+
         # init queue manager
         queue_manager = WorkflowAppQueueManager(
             task_id=application_generate_entity.task_id,
@@ -259,6 +302,15 @@ class WorkflowAppGenerator(BaseAppGenerator):
             app_mode=app_model.mode,
         )
 
+        if pause_state_config is not None:
+            graph_layers.append(
+                PauseStatePersistenceLayer(
+                    session_factory=pause_state_config.session_factory,
+                    generate_entity=application_generate_entity,
+                    state_owner_user_id=pause_state_config.state_owner_user_id,
+                )
+            )
+
         # new thread with request context and contextvars
         context = contextvars.copy_context()
 
@@ -276,7 +328,8 @@ class WorkflowAppGenerator(BaseAppGenerator):
                 "root_node_id": root_node_id,
                 "workflow_execution_repository": workflow_execution_repository,
                 "workflow_node_execution_repository": workflow_node_execution_repository,
-                "graph_engine_layers": graph_engine_layers,
+                "graph_engine_layers": tuple(graph_layers),
+                "graph_runtime_state": graph_runtime_state,
             },
         )
 
@@ -378,6 +431,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
             workflow_node_execution_repository=workflow_node_execution_repository,
             streaming=streaming,
             variable_loader=var_loader,
+            pause_state_config=None,
         )
 
     def single_loop_generate(
@@ -459,6 +513,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
             workflow_node_execution_repository=workflow_node_execution_repository,
             streaming=streaming,
             variable_loader=var_loader,
+            pause_state_config=None,
         )
 
     def _generate_worker(
@@ -472,6 +527,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
         workflow_node_execution_repository: WorkflowNodeExecutionRepository,
         root_node_id: str | None = None,
         graph_engine_layers: Sequence[GraphEngineLayer] = (),
+        graph_runtime_state: GraphRuntimeState | None = None,
     ) -> None:
         """
         Generate worker in a new thread.
@@ -517,6 +573,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
                 workflow_node_execution_repository=workflow_node_execution_repository,
                 root_node_id=root_node_id,
                 graph_engine_layers=graph_engine_layers,
+                graph_runtime_state=graph_runtime_state,
             )
 
             try:

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

@@ -42,6 +42,7 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
         workflow_execution_repository: WorkflowExecutionRepository,
         workflow_node_execution_repository: WorkflowNodeExecutionRepository,
         graph_engine_layers: Sequence[GraphEngineLayer] = (),
+        graph_runtime_state: GraphRuntimeState | None = None,
     ):
         super().__init__(
             queue_manager=queue_manager,
@@ -55,6 +56,7 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
         self._root_node_id = root_node_id
         self._workflow_execution_repository = workflow_execution_repository
         self._workflow_node_execution_repository = workflow_node_execution_repository
+        self._resume_graph_runtime_state = graph_runtime_state
 
     @trace_span(WorkflowAppRunnerHandler)
     def run(self):
@@ -63,23 +65,28 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
         """
         app_config = self.application_generate_entity.app_config
         app_config = cast(WorkflowAppConfig, app_config)
-
-        system_inputs = SystemVariable(
-            files=self.application_generate_entity.files,
-            user_id=self._sys_user_id,
-            app_id=app_config.app_id,
-            timestamp=int(naive_utc_now().timestamp()),
-            workflow_id=app_config.workflow_id,
-            workflow_execution_id=self.application_generate_entity.workflow_execution_id,
-        )
-
         invoke_from = self.application_generate_entity.invoke_from
         # if only single iteration or single loop run is requested
         if self.application_generate_entity.single_iteration_run or self.application_generate_entity.single_loop_run:
             invoke_from = InvokeFrom.DEBUGGER
         user_from = self._resolve_user_from(invoke_from)
 
-        if self.application_generate_entity.single_iteration_run or self.application_generate_entity.single_loop_run:
+        resume_state = self._resume_graph_runtime_state
+
+        if resume_state is not None:
+            graph_runtime_state = resume_state
+            variable_pool = graph_runtime_state.variable_pool
+            graph = self._init_graph(
+                graph_config=self._workflow.graph_dict,
+                graph_runtime_state=graph_runtime_state,
+                workflow_id=self._workflow.id,
+                tenant_id=self._workflow.tenant_id,
+                user_id=self.application_generate_entity.user_id,
+                user_from=user_from,
+                invoke_from=invoke_from,
+                root_node_id=self._root_node_id,
+            )
+        elif self.application_generate_entity.single_iteration_run or self.application_generate_entity.single_loop_run:
             graph, variable_pool, graph_runtime_state = self._prepare_single_node_execution(
                 workflow=self._workflow,
                 single_iteration_run=self.application_generate_entity.single_iteration_run,
@@ -89,7 +96,14 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
             inputs = self.application_generate_entity.inputs
 
             # Create a variable pool.
-
+            system_inputs = SystemVariable(
+                files=self.application_generate_entity.files,
+                user_id=self._sys_user_id,
+                app_id=app_config.app_id,
+                timestamp=int(naive_utc_now().timestamp()),
+                workflow_id=app_config.workflow_id,
+                workflow_execution_id=self.application_generate_entity.workflow_execution_id,
+            )
             variable_pool = VariablePool(
                 system_variables=system_inputs,
                 user_inputs=inputs,
@@ -98,8 +112,6 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
             )
 
             graph_runtime_state = GraphRuntimeState(variable_pool=variable_pool, start_at=time.perf_counter())
-
-            # init graph
             graph = self._init_graph(
                 graph_config=self._workflow.graph_dict,
                 graph_runtime_state=graph_runtime_state,

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

@@ -0,0 +1,7 @@
+from libs.exception import BaseHTTPException
+
+
+class WorkflowPausedInBlockingModeError(BaseHTTPException):
+    error_code = "workflow_paused_in_blocking_mode"
+    description = "Workflow execution paused for human input; blocking response mode is not supported."
+    code = 400

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

@@ -16,6 +16,8 @@ from core.app.entities.queue_entities import (
     MessageQueueMessage,
     QueueAgentLogEvent,
     QueueErrorEvent,
+    QueueHumanInputFormFilledEvent,
+    QueueHumanInputFormTimeoutEvent,
     QueueIterationCompletedEvent,
     QueueIterationNextEvent,
     QueueIterationStartEvent,
@@ -32,6 +34,7 @@ from core.app.entities.queue_entities import (
     QueueTextChunkEvent,
     QueueWorkflowFailedEvent,
     QueueWorkflowPartialSuccessEvent,
+    QueueWorkflowPausedEvent,
     QueueWorkflowStartedEvent,
     QueueWorkflowSucceededEvent,
     WorkflowQueueMessage,
@@ -46,11 +49,13 @@ from core.app.entities.task_entities import (
     WorkflowAppBlockingResponse,
     WorkflowAppStreamResponse,
     WorkflowFinishStreamResponse,
+    WorkflowPauseStreamResponse,
     WorkflowStartStreamResponse,
 )
 from core.app.task_pipeline.based_generate_task_pipeline import BasedGenerateTaskPipeline
 from core.base.tts import AppGeneratorTTSPublisher, AudioTrunk
 from core.ops.ops_trace_manager import TraceQueueManager
+from core.workflow.entities.workflow_start_reason import WorkflowStartReason
 from core.workflow.enums import WorkflowExecutionStatus
 from core.workflow.repositories.draft_variable_repository import DraftVariableSaverFactory
 from core.workflow.runtime import GraphRuntimeState
@@ -132,6 +137,25 @@ class WorkflowAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         for stream_response in generator:
             if isinstance(stream_response, ErrorStreamResponse):
                 raise stream_response.err
+            elif isinstance(stream_response, WorkflowPauseStreamResponse):
+                response = WorkflowAppBlockingResponse(
+                    task_id=self._application_generate_entity.task_id,
+                    workflow_run_id=stream_response.data.workflow_run_id,
+                    data=WorkflowAppBlockingResponse.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,
+                    ),
+                )
+
+                return response
             elif isinstance(stream_response, WorkflowFinishStreamResponse):
                 response = WorkflowAppBlockingResponse(
                     task_id=self._application_generate_entity.task_id,
@@ -146,7 +170,7 @@ class WorkflowAppGenerateTaskPipeline(GraphRuntimeStateSupport):
                         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),
+                        finished_at=int(stream_response.data.finished_at) if stream_response.data.finished_at else None,
                     ),
                 )
 
@@ -259,13 +283,15 @@ class WorkflowAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         run_id = self._extract_workflow_run_id(runtime_state)
         self._workflow_execution_id = run_id
 
-        with self._database_session() as session:
-            self._save_workflow_app_log(session=session, workflow_run_id=self._workflow_execution_id)
+        if event.reason == WorkflowStartReason.INITIAL:
+            with self._database_session() as session:
+                self._save_workflow_app_log(session=session, workflow_run_id=self._workflow_execution_id)
 
         start_resp = self._workflow_response_converter.workflow_start_to_stream_response(
             task_id=self._application_generate_entity.task_id,
             workflow_run_id=run_id,
             workflow_id=self._workflow.id,
+            reason=event.reason,
         )
         yield start_resp
 
@@ -440,6 +466,21 @@ class WorkflowAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         )
         yield workflow_finish_resp
 
+    def _handle_workflow_paused_event(
+        self,
+        event: QueueWorkflowPausedEvent,
+        **kwargs,
+    ) -> Generator[StreamResponse, None, None]:
+        """Handle workflow paused events."""
+        self._ensure_workflow_initialized()
+        validated_state = self._ensure_graph_runtime_initialized()
+        responses = self._workflow_response_converter.workflow_pause_to_stream_response(
+            event=event,
+            task_id=self._application_generate_entity.task_id,
+            graph_runtime_state=validated_state,
+        )
+        yield from responses
+
     def _handle_workflow_failed_and_stop_events(
         self,
         event: Union[QueueWorkflowFailedEvent, QueueStopEvent],
@@ -495,6 +536,22 @@ class WorkflowAppGenerateTaskPipeline(GraphRuntimeStateSupport):
             task_id=self._application_generate_entity.task_id, event=event
         )
 
+    def _handle_human_input_form_filled_event(
+        self, event: QueueHumanInputFormFilledEvent, **kwargs
+    ) -> Generator[StreamResponse, None, None]:
+        """Handle human input form filled events."""
+        yield self._workflow_response_converter.human_input_form_filled_to_stream_response(
+            event=event, task_id=self._application_generate_entity.task_id
+        )
+
+    def _handle_human_input_form_timeout_event(
+        self, event: QueueHumanInputFormTimeoutEvent, **kwargs
+    ) -> Generator[StreamResponse, None, None]:
+        """Handle human input form timeout events."""
+        yield self._workflow_response_converter.human_input_form_timeout_to_stream_response(
+            event=event, task_id=self._application_generate_entity.task_id
+        )
+
     def _get_event_handlers(self) -> dict[type, Callable]:
         """Get mapping of event types to their handlers using fluent pattern."""
         return {
@@ -506,6 +563,7 @@ class WorkflowAppGenerateTaskPipeline(GraphRuntimeStateSupport):
             QueueWorkflowStartedEvent: self._handle_workflow_started_event,
             QueueWorkflowSucceededEvent: self._handle_workflow_succeeded_event,
             QueueWorkflowPartialSuccessEvent: self._handle_workflow_partial_success_event,
+            QueueWorkflowPausedEvent: self._handle_workflow_paused_event,
             # Node events
             QueueNodeRetryEvent: self._handle_node_retry_event,
             QueueNodeStartedEvent: self._handle_node_started_event,
@@ -520,6 +578,8 @@ class WorkflowAppGenerateTaskPipeline(GraphRuntimeStateSupport):
             QueueLoopCompletedEvent: self._handle_loop_completed_event,
             # Agent events
             QueueAgentLogEvent: self._handle_agent_log_event,
+            QueueHumanInputFormFilledEvent: self._handle_human_input_form_filled_event,
+            QueueHumanInputFormTimeoutEvent: self._handle_human_input_form_timeout_event,
         }
 
     def _dispatch_event(
@@ -602,6 +662,9 @@ class WorkflowAppGenerateTaskPipeline(GraphRuntimeStateSupport):
                 case QueueWorkflowFailedEvent():
                     yield from self._handle_workflow_failed_and_stop_events(event)
                     break
+                case QueueWorkflowPausedEvent():
+                    yield from self._handle_workflow_paused_event(event)
+                    break
 
                 case QueueStopEvent():
                     yield from self._handle_workflow_failed_and_stop_events(event)

api/core/app/apps/workflow_app_runner.py

@@ -1,3 +1,4 @@
+import logging
 import time
 from collections.abc import Mapping, Sequence
 from typing import Any, cast
@@ -7,6 +8,8 @@ from core.app.entities.app_invoke_entities import InvokeFrom
 from core.app.entities.queue_entities import (
     AppQueueEvent,
     QueueAgentLogEvent,
+    QueueHumanInputFormFilledEvent,
+    QueueHumanInputFormTimeoutEvent,
     QueueIterationCompletedEvent,
     QueueIterationNextEvent,
     QueueIterationStartEvent,
@@ -22,22 +25,27 @@ from core.app.entities.queue_entities import (
     QueueTextChunkEvent,
     QueueWorkflowFailedEvent,
     QueueWorkflowPartialSuccessEvent,
+    QueueWorkflowPausedEvent,
     QueueWorkflowStartedEvent,
     QueueWorkflowSucceededEvent,
 )
 from core.app.workflow.node_factory import DifyNodeFactory
 from core.workflow.entities import GraphInitParams
+from core.workflow.entities.pause_reason import HumanInputRequired
 from core.workflow.graph import Graph
 from core.workflow.graph_engine.layers.base import GraphEngineLayer
 from core.workflow.graph_events import (
     GraphEngineEvent,
     GraphRunFailedEvent,
     GraphRunPartialSucceededEvent,
+    GraphRunPausedEvent,
     GraphRunStartedEvent,
     GraphRunSucceededEvent,
     NodeRunAgentLogEvent,
     NodeRunExceptionEvent,
     NodeRunFailedEvent,
+    NodeRunHumanInputFormFilledEvent,
+    NodeRunHumanInputFormTimeoutEvent,
     NodeRunIterationFailedEvent,
     NodeRunIterationNextEvent,
     NodeRunIterationStartedEvent,
@@ -61,6 +69,9 @@ from core.workflow.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader,
 from core.workflow.workflow_entry import WorkflowEntry
 from models.enums import UserFrom
 from models.workflow import Workflow
+from tasks.mail_human_input_delivery_task import dispatch_human_input_email_task
+
+logger = logging.getLogger(__name__)
 
 
 class WorkflowBasedAppRunner:
@@ -327,7 +338,7 @@ class WorkflowBasedAppRunner:
         :param event: event
         """
         if isinstance(event, GraphRunStartedEvent):
-            self._publish_event(QueueWorkflowStartedEvent())
+            self._publish_event(QueueWorkflowStartedEvent(reason=event.reason))
         elif isinstance(event, GraphRunSucceededEvent):
             self._publish_event(QueueWorkflowSucceededEvent(outputs=event.outputs))
         elif isinstance(event, GraphRunPartialSucceededEvent):
@@ -338,6 +349,38 @@ class WorkflowBasedAppRunner:
             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,
+                )
+            )
+        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
@@ -544,5 +587,19 @@ class WorkflowBasedAppRunner:
                 )
             )
 
+    def _enqueue_human_input_notifications(self, reasons: Sequence[object]) -> None:
+        for reason in reasons:
+            if not isinstance(reason, HumanInputRequired):
+                continue
+            if not reason.form_id:
+                continue
+            try:
+                dispatch_human_input_email_task.apply_async(
+                    kwargs={"form_id": reason.form_id, "node_title": reason.node_title},
+                    queue="mail",
+                )
+            except Exception:  # pragma: no cover - defensive logging
+                logger.exception("Failed to enqueue human input email task for form %s", reason.form_id)
+
     def _publish_event(self, event: AppQueueEvent):
         self._queue_manager.publish(event, PublishFrom.APPLICATION_MANAGER)

api/core/app/entities/app_invoke_entities.py

@@ -132,7 +132,7 @@ class AppGenerateEntity(BaseModel):
     extras: dict[str, Any] = Field(default_factory=dict)
 
     # tracing instance
-    trace_manager: Optional["TraceQueueManager"] = None
+    trace_manager: Optional["TraceQueueManager"] = Field(default=None, exclude=True, repr=False)
 
 
 class EasyUIBasedAppGenerateEntity(AppGenerateEntity):
@@ -156,6 +156,7 @@ class ConversationAppGenerateEntity(AppGenerateEntity):
     """
 
     conversation_id: str | None = None
+    is_new_conversation: bool = False
     parent_message_id: str | None = Field(
         default=None,
         description=(

api/core/app/entities/queue_entities.py

@@ -8,6 +8,8 @@ from pydantic import BaseModel, ConfigDict, Field
 from core.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk
 from core.rag.entities.citation_metadata import RetrievalSourceMetadata
 from core.workflow.entities import AgentNodeStrategyInit
+from core.workflow.entities.pause_reason import PauseReason
+from core.workflow.entities.workflow_start_reason import WorkflowStartReason
 from core.workflow.enums import WorkflowNodeExecutionMetadataKey
 from core.workflow.nodes import NodeType
 
@@ -46,6 +48,9 @@ class QueueEvent(StrEnum):
     PING = "ping"
     STOP = "stop"
     RETRY = "retry"
+    PAUSE = "pause"
+    HUMAN_INPUT_FORM_FILLED = "human_input_form_filled"
+    HUMAN_INPUT_FORM_TIMEOUT = "human_input_form_timeout"
 
 
 class AppQueueEvent(BaseModel):
@@ -261,6 +266,8 @@ class QueueWorkflowStartedEvent(AppQueueEvent):
     """QueueWorkflowStartedEvent entity."""
 
     event: QueueEvent = QueueEvent.WORKFLOW_STARTED
+    # Always present; mirrors GraphRunStartedEvent.reason for downstream consumers.
+    reason: WorkflowStartReason = WorkflowStartReason.INITIAL
 
 
 class QueueWorkflowSucceededEvent(AppQueueEvent):
@@ -484,6 +491,35 @@ class QueueStopEvent(AppQueueEvent):
         return reason_mapping.get(self.stopped_by, "Stopped by unknown reason.")
 
 
+class QueueHumanInputFormFilledEvent(AppQueueEvent):
+    """
+    QueueHumanInputFormFilledEvent entity
+    """
+
+    event: QueueEvent = QueueEvent.HUMAN_INPUT_FORM_FILLED
+
+    node_execution_id: str
+    node_id: str
+    node_type: NodeType
+    node_title: str
+    rendered_content: str
+    action_id: str
+    action_text: str
+
+
+class QueueHumanInputFormTimeoutEvent(AppQueueEvent):
+    """
+    QueueHumanInputFormTimeoutEvent entity
+    """
+
+    event: QueueEvent = QueueEvent.HUMAN_INPUT_FORM_TIMEOUT
+
+    node_id: str
+    node_type: NodeType
+    node_title: str
+    expiration_time: datetime
+
+
 class QueueMessage(BaseModel):
     """
     QueueMessage abstract entity
@@ -509,3 +545,14 @@ class WorkflowQueueMessage(QueueMessage):
     """
 
     pass
+
+
+class QueueWorkflowPausedEvent(AppQueueEvent):
+    """
+    QueueWorkflowPausedEvent entity
+    """
+
+    event: QueueEvent = QueueEvent.PAUSE
+    reasons: Sequence[PauseReason] = Field(default_factory=list)
+    outputs: Mapping[str, object] = Field(default_factory=dict)
+    paused_nodes: Sequence[str] = Field(default_factory=list)

api/core/app/entities/task_entities.py

@@ -7,7 +7,9 @@ from pydantic import BaseModel, ConfigDict, Field
 from core.model_runtime.entities.llm_entities import LLMResult, LLMUsage
 from core.rag.entities.citation_metadata import RetrievalSourceMetadata
 from core.workflow.entities import AgentNodeStrategyInit
-from core.workflow.enums import WorkflowExecutionStatus, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
+from core.workflow.entities.workflow_start_reason import WorkflowStartReason
+from core.workflow.enums import WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
+from core.workflow.nodes.human_input.entities import FormInput, UserAction
 
 
 class AnnotationReplyAccount(BaseModel):
@@ -69,6 +71,7 @@ class StreamEvent(StrEnum):
     AGENT_THOUGHT = "agent_thought"
     AGENT_MESSAGE = "agent_message"
     WORKFLOW_STARTED = "workflow_started"
+    WORKFLOW_PAUSED = "workflow_paused"
     WORKFLOW_FINISHED = "workflow_finished"
     NODE_STARTED = "node_started"
     NODE_FINISHED = "node_finished"
@@ -82,6 +85,9 @@ class StreamEvent(StrEnum):
     TEXT_CHUNK = "text_chunk"
     TEXT_REPLACE = "text_replace"
     AGENT_LOG = "agent_log"
+    HUMAN_INPUT_REQUIRED = "human_input_required"
+    HUMAN_INPUT_FORM_FILLED = "human_input_form_filled"
+    HUMAN_INPUT_FORM_TIMEOUT = "human_input_form_timeout"
 
 
 class StreamResponse(BaseModel):
@@ -205,6 +211,8 @@ class WorkflowStartStreamResponse(StreamResponse):
         workflow_id: str
         inputs: Mapping[str, Any]
         created_at: int
+        # Always present; mirrors QueueWorkflowStartedEvent.reason for SSE clients.
+        reason: WorkflowStartReason = WorkflowStartReason.INITIAL
 
     event: StreamEvent = StreamEvent.WORKFLOW_STARTED
     workflow_run_id: str
@@ -223,7 +231,7 @@ class WorkflowFinishStreamResponse(StreamResponse):
 
         id: str
         workflow_id: str
-        status: WorkflowExecutionStatus
+        status: str
         outputs: Mapping[str, Any] | None = None
         error: str | None = None
         elapsed_time: float
@@ -231,7 +239,7 @@ class WorkflowFinishStreamResponse(StreamResponse):
         total_steps: int
         created_by: Mapping[str, object] = Field(default_factory=dict)
         created_at: int
-        finished_at: int
+        finished_at: int | None
         exceptions_count: int | None = 0
         files: Sequence[Mapping[str, Any]] | None = []
 
@@ -240,6 +248,85 @@ class WorkflowFinishStreamResponse(StreamResponse):
     data: Data
 
 
+class WorkflowPauseStreamResponse(StreamResponse):
+    """
+    WorkflowPauseStreamResponse entity
+    """
+
+    class Data(BaseModel):
+        """
+        Data entity
+        """
+
+        workflow_run_id: str
+        paused_nodes: Sequence[str] = Field(default_factory=list)
+        outputs: Mapping[str, Any] = Field(default_factory=dict)
+        reasons: Sequence[Mapping[str, Any]] = Field(default_factory=list)
+        status: str
+        created_at: int
+        elapsed_time: float
+        total_tokens: int
+        total_steps: int
+
+    event: StreamEvent = StreamEvent.WORKFLOW_PAUSED
+    workflow_run_id: str
+    data: Data
+
+
+class HumanInputRequiredResponse(StreamResponse):
+    class Data(BaseModel):
+        """
+        Data entity
+        """
+
+        form_id: str
+        node_id: str
+        node_title: str
+        form_content: str
+        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)
+        expiration_time: int = Field(..., description="Unix timestamp in seconds")
+
+    event: StreamEvent = StreamEvent.HUMAN_INPUT_REQUIRED
+    workflow_run_id: str
+    data: Data
+
+
+class HumanInputFormFilledResponse(StreamResponse):
+    class Data(BaseModel):
+        """
+        Data entity
+        """
+
+        node_id: str
+        node_title: str
+        rendered_content: str
+        action_id: str
+        action_text: str
+
+    event: StreamEvent = StreamEvent.HUMAN_INPUT_FORM_FILLED
+    workflow_run_id: str
+    data: Data
+
+
+class HumanInputFormTimeoutResponse(StreamResponse):
+    class Data(BaseModel):
+        """
+        Data entity
+        """
+
+        node_id: str
+        node_title: str
+        expiration_time: int
+
+    event: StreamEvent = StreamEvent.HUMAN_INPUT_FORM_TIMEOUT
+    workflow_run_id: str
+    data: Data
+
+
 class NodeStartStreamResponse(StreamResponse):
     """
     NodeStartStreamResponse entity
@@ -311,7 +398,7 @@ class NodeFinishStreamResponse(StreamResponse):
         process_data_truncated: bool = False
         outputs: Mapping[str, Any] | None = None
         outputs_truncated: bool = True
-        status: WorkflowNodeExecutionStatus
+        status: str
         error: str | None = None
         elapsed_time: float
         execution_metadata: Mapping[WorkflowNodeExecutionMetadataKey, Any] | None = None
@@ -375,7 +462,7 @@ class NodeRetryStreamResponse(StreamResponse):
         process_data_truncated: bool = False
         outputs: Mapping[str, Any] | None = None
         outputs_truncated: bool = False
-        status: WorkflowNodeExecutionStatus
+        status: str
         error: str | None = None
         elapsed_time: float
         execution_metadata: Mapping[WorkflowNodeExecutionMetadataKey, Any] | None = None
@@ -719,14 +806,14 @@ class WorkflowAppBlockingResponse(AppBlockingResponse):
 
         id: str
         workflow_id: str
-        status: WorkflowExecutionStatus
+        status: str
         outputs: Mapping[str, Any] | None = None
         error: str | None = None
         elapsed_time: float
         total_tokens: int
         total_steps: int
         created_at: int
-        finished_at: int
+        finished_at: int | None
 
     workflow_run_id: str
     data: Data

api/core/app/features/rate_limiting/rate_limit.py

@@ -1,3 +1,4 @@
+import contextlib
 import logging
 import time
 import uuid
@@ -103,6 +104,14 @@ class RateLimit:
             )
 
 
+@contextlib.contextmanager
+def rate_limit_context(rate_limit: RateLimit, request_id: str | None):
+    request_id = rate_limit.enter(request_id)
+    yield
+    if request_id is not None:
+        rate_limit.exit(request_id)
+
+
 class RateLimitGenerator:
     def __init__(self, rate_limit: RateLimit, generator: Generator[str, None, None], request_id: str):
         self.rate_limit = rate_limit

api/core/app/layers/pause_state_persist_layer.py

@@ -1,3 +1,4 @@
+from dataclasses import dataclass
 from typing import Annotated, Literal, Self, TypeAlias
 
 from pydantic import BaseModel, Field
@@ -52,6 +53,14 @@ class WorkflowResumptionContext(BaseModel):
         return self.generate_entity.entity
 
 
+@dataclass(frozen=True)
+class PauseStateLayerConfig:
+    """Configuration container for instantiating pause persistence layers."""
+
+    session_factory: Engine | sessionmaker[Session]
+    state_owner_user_id: str
+
+
 class PauseStatePersistenceLayer(GraphEngineLayer):
     def __init__(
         self,

api/core/app/task_pipeline/message_cycle_manager.py

@@ -82,10 +82,11 @@ class MessageCycleManager:
         if isinstance(self._application_generate_entity, CompletionAppGenerateEntity):
             return None
 
-        is_first_message = self._application_generate_entity.conversation_id is None
+        is_first_message = self._application_generate_entity.is_new_conversation
         extras = self._application_generate_entity.extras
         auto_generate_conversation_name = extras.get("auto_generate_conversation_name", True)
 
+        thread: Thread | None = None
         if auto_generate_conversation_name and is_first_message:
             # start generate thread
             # time.sleep not block other logic
@@ -101,9 +102,10 @@ class MessageCycleManager:
             thread.daemon = True
             thread.start()
 
-            return thread
+        if is_first_message:
+            self._application_generate_entity.is_new_conversation = False
 
-        return None
+        return thread
 
     def _generate_conversation_name_worker(self, flask_app: Flask, conversation_id: str, query: str):
         with flask_app.app_context():

api/core/app/workflow/node_factory.py

@@ -4,14 +4,13 @@ from typing import TYPE_CHECKING, final
 from typing_extensions import override
 
 from configs import dify_config
-from core.file.file_manager import file_manager
+from core.file import file_manager
+from core.helper import ssrf_proxy
 from core.helper.code_executor.code_executor import CodeExecutor
 from core.helper.code_executor.code_node_provider import CodeNodeProvider
-from core.helper.ssrf_proxy import ssrf_proxy
 from core.tools.tool_file_manager import ToolFileManager
-from core.workflow.entities.graph_config import NodeConfigDict
 from core.workflow.enums import NodeType
-from core.workflow.graph.graph import NodeFactory
+from core.workflow.graph import NodeFactory
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.code.code_node import CodeNode
 from core.workflow.nodes.code.limits import CodeNodeLimits
@@ -23,6 +22,7 @@ from core.workflow.nodes.template_transform.template_renderer import (
     Jinja2TemplateRenderer,
 )
 from core.workflow.nodes.template_transform.template_transform_node import TemplateTransformNode
+from libs.typing import is_str, is_str_dict
 
 if TYPE_CHECKING:
     from core.workflow.entities import GraphInitParams
@@ -47,9 +47,9 @@ class DifyNodeFactory(NodeFactory):
         code_providers: Sequence[type[CodeNodeProvider]] | None = None,
         code_limits: CodeNodeLimits | None = None,
         template_renderer: Jinja2TemplateRenderer | None = None,
-        http_request_http_client: HttpClientProtocol | None = None,
+        http_request_http_client: HttpClientProtocol = ssrf_proxy,
         http_request_tool_file_manager_factory: Callable[[], ToolFileManager] = ToolFileManager,
-        http_request_file_manager: FileManagerProtocol | None = None,
+        http_request_file_manager: FileManagerProtocol = file_manager,
     ) -> None:
         self.graph_init_params = graph_init_params
         self.graph_runtime_state = graph_runtime_state
@@ -68,12 +68,12 @@ class DifyNodeFactory(NodeFactory):
             max_object_array_length=dify_config.CODE_MAX_OBJECT_ARRAY_LENGTH,
         )
         self._template_renderer = template_renderer or CodeExecutorJinja2TemplateRenderer()
-        self._http_request_http_client = http_request_http_client or ssrf_proxy
+        self._http_request_http_client = http_request_http_client
         self._http_request_tool_file_manager_factory = http_request_tool_file_manager_factory
-        self._http_request_file_manager = http_request_file_manager or file_manager
+        self._http_request_file_manager = http_request_file_manager
 
     @override
-    def create_node(self, node_config: NodeConfigDict) -> Node:
+    def create_node(self, node_config: dict[str, object]) -> Node:
         """
         Create a Node instance from node configuration data using the traditional mapping.
 
@@ -82,14 +82,23 @@ class DifyNodeFactory(NodeFactory):
         :raises ValueError: if node type is unknown or configuration is invalid
         """
         # Get node_id from config
-        node_id = node_config["id"]
+        node_id = node_config.get("id")
+        if not is_str(node_id):
+            raise ValueError("Node config missing id")
 
         # Get node type from config
-        node_data = node_config["data"]
+        node_data = node_config.get("data", {})
+        if not is_str_dict(node_data):
+            raise ValueError(f"Node {node_id} missing data information")
+
+        node_type_str = node_data.get("type")
+        if not is_str(node_type_str):
+            raise ValueError(f"Node {node_id} missing or invalid type information")
+
         try:
-            node_type = NodeType(node_data["type"])
+            node_type = NodeType(node_type_str)
         except ValueError:
-            raise ValueError(f"Unknown node type: {node_data['type']}")
+            raise ValueError(f"Unknown node type: {node_type_str}")
 
         # Get node class
         node_mapping = NODE_TYPE_CLASSES_MAPPING.get(node_type)

api/core/entities/execution_extra_content.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from typing import Any, TypeAlias
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from core.workflow.nodes.human_input.entities import FormInput, UserAction
+from models.execution_extra_content import ExecutionContentType
+
+
+class HumanInputFormDefinition(BaseModel):
+    model_config = ConfigDict(frozen=True)
+
+    form_id: str
+    node_id: str
+    node_title: str
+    form_content: str
+    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)
+    expiration_time: int
+
+
+class HumanInputFormSubmissionData(BaseModel):
+    model_config = ConfigDict(frozen=True)
+
+    node_id: str
+    node_title: str
+    rendered_content: str
+    action_id: str
+    action_text: str
+
+
+class HumanInputContent(BaseModel):
+    model_config = ConfigDict(frozen=True)
+
+    workflow_run_id: str
+    submitted: bool
+    form_definition: HumanInputFormDefinition | None = None
+    form_submission_data: HumanInputFormSubmissionData | None = None
+    type: ExecutionContentType = Field(default=ExecutionContentType.HUMAN_INPUT)
+
+
+ExecutionExtraContentDomainModel: TypeAlias = HumanInputContent
+
+__all__ = [
+    "ExecutionExtraContentDomainModel",
+    "HumanInputContent",
+    "HumanInputFormDefinition",
+    "HumanInputFormSubmissionData",
+]

api/core/entities/provider_configuration.py

@@ -28,8 +28,8 @@ from core.model_runtime.entities.provider_entities import (
 )
 from core.model_runtime.model_providers.__base.ai_model import AIModel
 from core.model_runtime.model_providers.model_provider_factory import ModelProviderFactory
-from extensions.ext_database import db
 from libs.datetime_utils import naive_utc_now
+from models.engine import db
 from models.provider import (
     LoadBalancingModelConfig,
     Provider,

api/core/file/file_manager.py

@@ -168,18 +168,3 @@ def _to_url(f: File, /):
         return sign_tool_file(tool_file_id=f.related_id, extension=f.extension)
     else:
         raise ValueError(f"Unsupported transfer method: {f.transfer_method}")
-
-
-class FileManager:
-    """
-    Adapter exposing file manager helpers behind FileManagerProtocol.
-
-    This is intentionally a thin wrapper over the existing module-level functions so callers can inject it
-    where a protocol-typed file manager is expected.
-    """
-
-    def download(self, f: File, /) -> bytes:
-        return download(f)
-
-
-file_manager = FileManager()

api/core/helper/code_executor/code_node_provider.py

@@ -47,16 +47,15 @@ class CodeNodeProvider(BaseModel, ABC):
 
     @classmethod
     def get_default_config(cls) -> DefaultConfig:
-        variables: list[VariableConfig] = [
-            {"variable": "arg1", "value_selector": []},
-            {"variable": "arg2", "value_selector": []},
-        ]
-        outputs: dict[str, OutputConfig] = {"result": {"type": "string", "children": None}}
-
-        config: CodeConfig = {
-            "variables": variables,
-            "code_language": cls.get_language(),
-            "code": cls.get_default_code(),
-            "outputs": outputs,
+        return {
+            "type": "code",
+            "config": {
+                "variables": [
+                    {"variable": "arg1", "value_selector": []},
+                    {"variable": "arg2", "value_selector": []},
+                ],
+                "code_language": cls.get_language(),
+                "code": cls.get_default_code(),
+                "outputs": {"result": {"type": "string", "children": None}},
+            },
         }
-        return {"type": "code", "config": config}

api/core/helper/ssrf_proxy.py

@@ -230,41 +230,3 @@ def delete(url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any)
 
 def head(url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
     return make_request("HEAD", url, max_retries=max_retries, **kwargs)
-
-
-class SSRFProxy:
-    """
-    Adapter exposing SSRF-protected HTTP helpers behind HttpClientProtocol.
-
-    This is intentionally a thin wrapper over the existing module-level functions so callers can inject it
-    where a protocol-typed HTTP client is expected.
-    """
-
-    @property
-    def max_retries_exceeded_error(self) -> type[Exception]:
-        return max_retries_exceeded_error
-
-    @property
-    def request_error(self) -> type[Exception]:
-        return request_error
-
-    def get(self, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
-        return get(url=url, max_retries=max_retries, **kwargs)
-
-    def head(self, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
-        return head(url=url, max_retries=max_retries, **kwargs)
-
-    def post(self, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
-        return post(url=url, max_retries=max_retries, **kwargs)
-
-    def put(self, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
-        return put(url=url, max_retries=max_retries, **kwargs)
-
-    def delete(self, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
-        return delete(url=url, max_retries=max_retries, **kwargs)
-
-    def patch(self, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
-        return patch(url=url, max_retries=max_retries, **kwargs)
-
-
-ssrf_proxy = SSRFProxy()

api/core/indexing_runner.py

@@ -369,78 +369,77 @@ class IndexingRunner:
         # Generate summary preview
         summary_index_setting = tmp_processing_rule.get("summary_index_setting")
         if summary_index_setting and summary_index_setting.get("enable") and preview_texts:
-            preview_texts = index_processor.generate_summary_preview(
-                tenant_id, preview_texts, summary_index_setting, doc_language
-            )
+            preview_texts = index_processor.generate_summary_preview(tenant_id, preview_texts, summary_index_setting)
 
         return IndexingEstimate(total_segments=total_segments, preview=preview_texts)
 
     def _extract(
         self, index_processor: BaseIndexProcessor, dataset_document: DatasetDocument, process_rule: dict
     ) -> list[Document]:
+        # load file
+        if dataset_document.data_source_type not in {"upload_file", "notion_import", "website_crawl"}:
+            return []
+
         data_source_info = dataset_document.data_source_info_dict
         text_docs = []
-        match dataset_document.data_source_type:
-            case "upload_file":
-                if not data_source_info or "upload_file_id" not in data_source_info:
-                    raise ValueError("no upload file found")
-                stmt = select(UploadFile).where(UploadFile.id == data_source_info["upload_file_id"])
-                file_detail = db.session.scalars(stmt).one_or_none()
+        if dataset_document.data_source_type == "upload_file":
+            if not data_source_info or "upload_file_id" not in data_source_info:
+                raise ValueError("no upload file found")
+            stmt = select(UploadFile).where(UploadFile.id == data_source_info["upload_file_id"])
+            file_detail = db.session.scalars(stmt).one_or_none()
 
-                if file_detail:
-                    extract_setting = ExtractSetting(
-                        datasource_type=DatasourceType.FILE,
-                        upload_file=file_detail,
-                        document_model=dataset_document.doc_form,
-                    )
-                    text_docs = index_processor.extract(extract_setting, process_rule_mode=process_rule["mode"])
-            case "notion_import":
-                if (
-                    not data_source_info
-                    or "notion_workspace_id" not in data_source_info
-                    or "notion_page_id" not in data_source_info
-                ):
-                    raise ValueError("no notion import info found")
+            if file_detail:
                 extract_setting = ExtractSetting(
-                    datasource_type=DatasourceType.NOTION,
-                    notion_info=NotionInfo.model_validate(
-                        {
-                            "credential_id": data_source_info.get("credential_id"),
-                            "notion_workspace_id": data_source_info["notion_workspace_id"],
-                            "notion_obj_id": data_source_info["notion_page_id"],
-                            "notion_page_type": data_source_info["type"],
-                            "document": dataset_document,
-                            "tenant_id": dataset_document.tenant_id,
-                        }
-                    ),
+                    datasource_type=DatasourceType.FILE,
+                    upload_file=file_detail,
                     document_model=dataset_document.doc_form,
                 )
                 text_docs = index_processor.extract(extract_setting, process_rule_mode=process_rule["mode"])
-            case "website_crawl":
-                if (
-                    not data_source_info
-                    or "provider" not in data_source_info
-                    or "url" not in data_source_info
-                    or "job_id" not in data_source_info
-                ):
-                    raise ValueError("no website import info found")
-                extract_setting = ExtractSetting(
-                    datasource_type=DatasourceType.WEBSITE,
-                    website_info=WebsiteInfo.model_validate(
-                        {
-                            "provider": data_source_info["provider"],
-                            "job_id": data_source_info["job_id"],
-                            "tenant_id": dataset_document.tenant_id,
-                            "url": data_source_info["url"],
-                            "mode": data_source_info["mode"],
-                            "only_main_content": data_source_info["only_main_content"],
-                        }
-                    ),
-                    document_model=dataset_document.doc_form,
-                )
-                text_docs = index_processor.extract(extract_setting, process_rule_mode=process_rule["mode"])
-            case _:
-                return []
+        elif dataset_document.data_source_type == "notion_import":
+            if (
+                not data_source_info
+                or "notion_workspace_id" not in data_source_info
+                or "notion_page_id" not in data_source_info
+            ):
+                raise ValueError("no notion import info found")
+            extract_setting = ExtractSetting(
+                datasource_type=DatasourceType.NOTION,
+                notion_info=NotionInfo.model_validate(
+                    {
+                        "credential_id": data_source_info.get("credential_id"),
+                        "notion_workspace_id": data_source_info["notion_workspace_id"],
+                        "notion_obj_id": data_source_info["notion_page_id"],
+                        "notion_page_type": data_source_info["type"],
+                        "document": dataset_document,
+                        "tenant_id": dataset_document.tenant_id,
+                    }
+                ),
+                document_model=dataset_document.doc_form,
+            )
+            text_docs = index_processor.extract(extract_setting, process_rule_mode=process_rule["mode"])
+        elif dataset_document.data_source_type == "website_crawl":
+            if (
+                not data_source_info
+                or "provider" not in data_source_info
+                or "url" not in data_source_info
+                or "job_id" not in data_source_info
+            ):
+                raise ValueError("no website import info found")
+            extract_setting = ExtractSetting(
+                datasource_type=DatasourceType.WEBSITE,
+                website_info=WebsiteInfo.model_validate(
+                    {
+                        "provider": data_source_info["provider"],
+                        "job_id": data_source_info["job_id"],
+                        "tenant_id": dataset_document.tenant_id,
+                        "url": data_source_info["url"],
+                        "mode": data_source_info["mode"],
+                        "only_main_content": data_source_info["only_main_content"],
+                    }
+                ),
+                document_model=dataset_document.doc_form,
+            )
+            text_docs = index_processor.extract(extract_setting, process_rule_mode=process_rule["mode"])
         # update document status to splitting
         self._update_document_index_status(
             document_id=dataset_document.id,

api/core/llm_generator/entities.py

@@ -1,20 +0,0 @@
-"""Shared payload models for LLM generator helpers and controllers."""
-
-from pydantic import BaseModel, Field
-
-from core.app.app_config.entities import ModelConfig
-
-
-class RuleGeneratePayload(BaseModel):
-    instruction: str = Field(..., description="Rule generation instruction")
-    model_config_data: ModelConfig = Field(..., alias="model_config", description="Model configuration")
-    no_variable: bool = Field(default=False, description="Whether to exclude variables")
-
-
-class RuleCodeGeneratePayload(RuleGeneratePayload):
-    code_language: str = Field(default="javascript", description="Programming language for code generation")
-
-
-class RuleStructuredOutputPayload(BaseModel):
-    instruction: str = Field(..., description="Structured output generation instruction")
-    model_config_data: ModelConfig = Field(..., alias="model_config", description="Model configuration")

api/core/llm_generator/llm_generator.py

@@ -6,8 +6,6 @@ from typing import Protocol, cast
 
 import json_repair
 
-from core.app.app_config.entities import ModelConfig
-from core.llm_generator.entities import RuleCodeGeneratePayload, RuleGeneratePayload, RuleStructuredOutputPayload
 from core.llm_generator.output_parser.rule_config_generator import RuleConfigGeneratorOutputParser
 from core.llm_generator.output_parser.suggested_questions_after_answer import SuggestedQuestionsAfterAnswerOutputParser
 from core.llm_generator.prompts import (
@@ -153,19 +151,19 @@ class LLMGenerator:
         return questions
 
     @classmethod
-    def generate_rule_config(cls, tenant_id: str, args: RuleGeneratePayload):
+    def generate_rule_config(cls, tenant_id: str, instruction: str, model_config: dict, no_variable: bool):
         output_parser = RuleConfigGeneratorOutputParser()
 
         error = ""
         error_step = ""
         rule_config = {"prompt": "", "variables": [], "opening_statement": "", "error": ""}
-        model_parameters = args.model_config_data.completion_params
-        if args.no_variable:
+        model_parameters = model_config.get("completion_params", {})
+        if no_variable:
             prompt_template = PromptTemplateParser(WORKFLOW_RULE_CONFIG_PROMPT_GENERATE_TEMPLATE)
 
             prompt_generate = prompt_template.format(
                 inputs={
-                    "TASK_DESCRIPTION": args.instruction,
+                    "TASK_DESCRIPTION": instruction,
                 },
                 remove_template_variables=False,
             )
@@ -177,8 +175,8 @@ class LLMGenerator:
             model_instance = model_manager.get_model_instance(
                 tenant_id=tenant_id,
                 model_type=ModelType.LLM,
-                provider=args.model_config_data.provider,
-                model=args.model_config_data.name,
+                provider=model_config.get("provider", ""),
+                model=model_config.get("name", ""),
             )
 
             try:
@@ -192,7 +190,7 @@ class LLMGenerator:
                 error = str(e)
                 error_step = "generate rule config"
             except Exception as e:
-                logger.exception("Failed to generate rule config, model: %s", args.model_config_data.name)
+                logger.exception("Failed to generate rule config, model: %s", model_config.get("name"))
                 rule_config["error"] = str(e)
 
             rule_config["error"] = f"Failed to {error_step}. Error: {error}" if error else ""
@@ -211,7 +209,7 @@ class LLMGenerator:
         # format the prompt_generate_prompt
         prompt_generate_prompt = prompt_template.format(
             inputs={
-                "TASK_DESCRIPTION": args.instruction,
+                "TASK_DESCRIPTION": instruction,
             },
             remove_template_variables=False,
         )
@@ -222,8 +220,8 @@ class LLMGenerator:
         model_instance = model_manager.get_model_instance(
             tenant_id=tenant_id,
             model_type=ModelType.LLM,
-            provider=args.model_config_data.provider,
-            model=args.model_config_data.name,
+            provider=model_config.get("provider", ""),
+            model=model_config.get("name", ""),
         )
 
         try:
@@ -252,7 +250,7 @@ class LLMGenerator:
             # the second step to generate the task_parameter and task_statement
             statement_generate_prompt = statement_template.format(
                 inputs={
-                    "TASK_DESCRIPTION": args.instruction,
+                    "TASK_DESCRIPTION": instruction,
                     "INPUT_TEXT": prompt_content.message.get_text_content(),
                 },
                 remove_template_variables=False,
@@ -278,7 +276,7 @@ class LLMGenerator:
                 error_step = "generate conversation opener"
 
         except Exception as e:
-            logger.exception("Failed to generate rule config, model: %s", args.model_config_data.name)
+            logger.exception("Failed to generate rule config, model: %s", model_config.get("name"))
             rule_config["error"] = str(e)
 
         rule_config["error"] = f"Failed to {error_step}. Error: {error}" if error else ""
@@ -286,20 +284,16 @@ class LLMGenerator:
         return rule_config
 
     @classmethod
-    def generate_code(
-        cls,
-        tenant_id: str,
-        args: RuleCodeGeneratePayload,
-    ):
-        if args.code_language == "python":
+    def generate_code(cls, tenant_id: str, instruction: str, model_config: dict, code_language: str = "javascript"):
+        if code_language == "python":
             prompt_template = PromptTemplateParser(PYTHON_CODE_GENERATOR_PROMPT_TEMPLATE)
         else:
             prompt_template = PromptTemplateParser(JAVASCRIPT_CODE_GENERATOR_PROMPT_TEMPLATE)
 
         prompt = prompt_template.format(
             inputs={
-                "INSTRUCTION": args.instruction,
-                "CODE_LANGUAGE": args.code_language,
+                "INSTRUCTION": instruction,
+                "CODE_LANGUAGE": code_language,
             },
             remove_template_variables=False,
         )
@@ -308,28 +302,28 @@ class LLMGenerator:
         model_instance = model_manager.get_model_instance(
             tenant_id=tenant_id,
             model_type=ModelType.LLM,
-            provider=args.model_config_data.provider,
-            model=args.model_config_data.name,
+            provider=model_config.get("provider", ""),
+            model=model_config.get("name", ""),
         )
 
         prompt_messages = [UserPromptMessage(content=prompt)]
-        model_parameters = args.model_config_data.completion_params
+        model_parameters = model_config.get("completion_params", {})
         try:
             response: LLMResult = model_instance.invoke_llm(
                 prompt_messages=list(prompt_messages), model_parameters=model_parameters, stream=False
             )
 
             generated_code = response.message.get_text_content()
-            return {"code": generated_code, "language": args.code_language, "error": ""}
+            return {"code": generated_code, "language": code_language, "error": ""}
 
         except InvokeError as e:
             error = str(e)
-            return {"code": "", "language": args.code_language, "error": f"Failed to generate code. Error: {error}"}
+            return {"code": "", "language": code_language, "error": f"Failed to generate code. Error: {error}"}
         except Exception as e:
             logger.exception(
-                "Failed to invoke LLM model, model: %s, language: %s", args.model_config_data.name, args.code_language
+                "Failed to invoke LLM model, model: %s, language: %s", model_config.get("name"), code_language
             )
-            return {"code": "", "language": args.code_language, "error": f"An unexpected error occurred: {str(e)}"}
+            return {"code": "", "language": code_language, "error": f"An unexpected error occurred: {str(e)}"}
 
     @classmethod
     def generate_qa_document(cls, tenant_id: str, query, document_language: str):
@@ -359,20 +353,20 @@ class LLMGenerator:
         return answer.strip()
 
     @classmethod
-    def generate_structured_output(cls, tenant_id: str, args: RuleStructuredOutputPayload):
+    def generate_structured_output(cls, tenant_id: str, instruction: str, model_config: dict):
         model_manager = ModelManager()
         model_instance = model_manager.get_model_instance(
             tenant_id=tenant_id,
             model_type=ModelType.LLM,
-            provider=args.model_config_data.provider,
-            model=args.model_config_data.name,
+            provider=model_config.get("provider", ""),
+            model=model_config.get("name", ""),
         )
 
         prompt_messages = [
             SystemPromptMessage(content=SYSTEM_STRUCTURED_OUTPUT_GENERATE),
-            UserPromptMessage(content=args.instruction),
+            UserPromptMessage(content=instruction),
         ]
-        model_parameters = args.model_config_data.completion_params
+        model_parameters = model_config.get("model_parameters", {})
 
         try:
             response: LLMResult = model_instance.invoke_llm(
@@ -396,17 +390,12 @@ class LLMGenerator:
             error = str(e)
             return {"output": "", "error": f"Failed to generate JSON Schema. Error: {error}"}
         except Exception as e:
-            logger.exception("Failed to invoke LLM model, model: %s", args.model_config_data.name)
+            logger.exception("Failed to invoke LLM model, model: %s", model_config.get("name"))
             return {"output": "", "error": f"An unexpected error occurred: {str(e)}"}
 
     @staticmethod
     def instruction_modify_legacy(
-        tenant_id: str,
-        flow_id: str,
-        current: str,
-        instruction: str,
-        model_config: ModelConfig,
-        ideal_output: str | None,
+        tenant_id: str, flow_id: str, current: str, instruction: str, model_config: dict, ideal_output: str | None
     ):
         last_run: Message | None = (
             db.session.query(Message).where(Message.app_id == flow_id).order_by(Message.created_at.desc()).first()
@@ -445,7 +434,7 @@ class LLMGenerator:
         node_id: str,
         current: str,
         instruction: str,
-        model_config: ModelConfig,
+        model_config: dict,
         ideal_output: str | None,
         workflow_service: WorkflowServiceInterface,
     ):
@@ -516,7 +505,7 @@ class LLMGenerator:
     @staticmethod
     def __instruction_modify_common(
         tenant_id: str,
-        model_config: ModelConfig,
+        model_config: dict,
         last_run: dict | None,
         current: str | None,
         error_message: str | None,
@@ -537,8 +526,8 @@ class LLMGenerator:
         model_instance = ModelManager().get_model_instance(
             tenant_id=tenant_id,
             model_type=ModelType.LLM,
-            provider=model_config.provider,
-            model=model_config.name,
+            provider=model_config.get("provider", ""),
+            model=model_config.get("name", ""),
         )
         match node_type:
             case "llm" | "agent":
@@ -581,5 +570,7 @@ class LLMGenerator:
             error = str(e)
             return {"error": f"Failed to generate code. Error: {error}"}
         except Exception as e:
-            logger.exception("Failed to invoke LLM model, model: %s", json.dumps(model_config.name), exc_info=True)
+            logger.exception(
+                "Failed to invoke LLM model, model: %s", json.dumps(model_config.get("name")), exc_info=True
+            )
             return {"error": f"An unexpected error occurred: {str(e)}"}

api/core/llm_generator/prompts.py

@@ -441,13 +441,11 @@ DEFAULT_GENERATOR_SUMMARY_PROMPT = (
 
 Requirements:
 1. Write a concise summary in plain text
-2. You must write in {language}. No language other than {language} should be used.
+2. Use the same language as the input content
 3. Focus on important facts, concepts, and details
 4. If images are included, describe their key information
 5. Do not use words like "好的", "ok", "I understand", "This text discusses", "The content mentions"
 6. Write directly without extra words
-7. If there is not enough content to generate a meaningful summary, 
-   return an empty string without any explanation or prompt
 
 Output only the summary text. Start summarizing now:
 

api/core/model_runtime/entities/defaults.py

@@ -88,7 +88,7 @@ PARAMETER_RULE_TEMPLATE: dict[DefaultParameterName, dict] = {
     DefaultParameterName.MAX_TOKENS: {
         "label": {
             "en_US": "Max Tokens",
-            "zh_Hans": "最大 Token 数",
+            "zh_Hans": "最大标记",
         },
         "type": "int",
         "help": {

api/core/model_runtime/model_providers/__base/large_language_model.py

@@ -92,10 +92,6 @@ def _build_llm_result_from_first_chunk(
     Build a single `LLMResult` from the first returned chunk.
 
     This is used for `stream=False` because the plugin side may still implement the response via a chunked stream.
-
-    Note:
-        This function always drains the `chunks` iterator after reading the first chunk to ensure any underlying
-        streaming resources are released (e.g., HTTP connections owned by the plugin runtime).
     """
     content = ""
     content_list: list[PromptMessageContentUnionTypes] = []
@@ -103,25 +99,18 @@ def _build_llm_result_from_first_chunk(
     system_fingerprint: str | None = None
     tools_calls: list[AssistantPromptMessage.ToolCall] = []
 
-    try:
-        first_chunk = next(chunks, None)
-        if first_chunk is not None:
-            if isinstance(first_chunk.delta.message.content, str):
-                content += first_chunk.delta.message.content
-            elif isinstance(first_chunk.delta.message.content, list):
-                content_list.extend(first_chunk.delta.message.content)
+    first_chunk = next(chunks, None)
+    if first_chunk is not None:
+        if isinstance(first_chunk.delta.message.content, str):
+            content += first_chunk.delta.message.content
+        elif isinstance(first_chunk.delta.message.content, list):
+            content_list.extend(first_chunk.delta.message.content)
 
-            if first_chunk.delta.message.tool_calls:
-                _increase_tool_call(first_chunk.delta.message.tool_calls, tools_calls)
+        if first_chunk.delta.message.tool_calls:
+            _increase_tool_call(first_chunk.delta.message.tool_calls, tools_calls)
 
-            usage = first_chunk.delta.usage or LLMUsage.empty_usage()
-            system_fingerprint = first_chunk.system_fingerprint
-    finally:
-        try:
-            for _ in chunks:
-                pass
-        except Exception:
-            logger.debug("Failed to drain non-stream plugin chunk iterator.", exc_info=True)
+        usage = first_chunk.delta.usage or LLMUsage.empty_usage()
+        system_fingerprint = first_chunk.system_fingerprint
 
     return LLMResult(
         model=model,

api/core/ops/ops_trace_manager.py

@@ -15,10 +15,7 @@ from sqlalchemy import select
 from sqlalchemy.orm import Session, sessionmaker
 
 from core.helper.encrypter import batch_decrypt_token, encrypt_token, obfuscated_token
-from core.ops.entities.config_entity import (
-    OPS_FILE_PATH,
-    TracingProviderEnum,
-)
+from core.ops.entities.config_entity import OPS_FILE_PATH, TracingProviderEnum
 from core.ops.entities.trace_entity import (
     DatasetRetrievalTraceInfo,
     GenerateNameTraceInfo,
@@ -31,8 +28,8 @@ from core.ops.entities.trace_entity import (
     WorkflowTraceInfo,
 )
 from core.ops.utils import get_message_data
-from extensions.ext_database import db
 from extensions.ext_storage import storage
+from models.engine import db
 from models.model import App, AppModelConfig, Conversation, Message, MessageFile, TraceAppConfig
 from models.workflow import WorkflowAppLog
 from tasks.ops_trace_task import process_trace_tasks
@@ -469,6 +466,8 @@ class TraceTask:
 
     @classmethod
     def _get_workflow_run_repo(cls):
+        from repositories.factory import DifyAPIRepositoryFactory
+
         if cls._workflow_run_repo is None:
             with cls._repo_lock:
                 if cls._workflow_run_repo is None:

api/core/ops/utils.py

@@ -5,7 +5,7 @@ from urllib.parse import urlparse
 
 from sqlalchemy import select
 
-from extensions.ext_database import db
+from models.engine import db
 from models.model import Message
 
 

api/core/plugin/backwards_invocation/app.py

@@ -1,3 +1,4 @@
+import uuid
 from collections.abc import Generator, Mapping
 from typing import Union
 
@@ -112,6 +113,7 @@ class PluginAppBackwardsInvocation(BaseBackwardsInvocation):
                     "conversation_id": conversation_id,
                 },
                 invoke_from=InvokeFrom.SERVICE_API,
+                workflow_run_id=str(uuid.uuid4()),
                 streaming=stream,
             )
         elif app.mode == AppMode.AGENT_CHAT:

api/core/rag/index_processor/index_processor_base.py

@@ -48,22 +48,12 @@ class BaseIndexProcessor(ABC):
 
     @abstractmethod
     def generate_summary_preview(
-        self,
-        tenant_id: str,
-        preview_texts: list[PreviewDetail],
-        summary_index_setting: dict,
-        doc_language: str | None = None,
+        self, tenant_id: str, preview_texts: list[PreviewDetail], summary_index_setting: dict
     ) -> list[PreviewDetail]:
         """
         For each segment in preview_texts, generate a summary using LLM and attach it to the segment.
         The summary can be stored in a new attribute, e.g., summary.
         This method should be implemented by subclasses.
-
-        Args:
-            tenant_id: Tenant ID
-            preview_texts: List of preview details to generate summaries for
-            summary_index_setting: Summary index configuration
-            doc_language: Optional document language to ensure summary is generated in the correct language
         """
         raise NotImplementedError
 

api/core/rag/index_processor/processor/paragraph_index_processor.py

@@ -275,11 +275,7 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
             raise ValueError("Chunks is not a list")
 
     def generate_summary_preview(
-        self,
-        tenant_id: str,
-        preview_texts: list[PreviewDetail],
-        summary_index_setting: dict,
-        doc_language: str | None = None,
+        self, tenant_id: str, preview_texts: list[PreviewDetail], summary_index_setting: dict
     ) -> list[PreviewDetail]:
         """
         For each segment, concurrently call generate_summary to generate a summary
@@ -302,15 +298,11 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
             if flask_app:
                 # Ensure Flask app context in worker thread
                 with flask_app.app_context():
-                    summary, _ = self.generate_summary(
-                        tenant_id, preview.content, summary_index_setting, document_language=doc_language
-                    )
+                    summary, _ = self.generate_summary(tenant_id, preview.content, summary_index_setting)
                     preview.summary = summary
             else:
                 # Fallback: try without app context (may fail)
-                summary, _ = self.generate_summary(
-                    tenant_id, preview.content, summary_index_setting, document_language=doc_language
-                )
+                summary, _ = self.generate_summary(tenant_id, preview.content, summary_index_setting)
                 preview.summary = summary
 
         # Generate summaries concurrently using ThreadPoolExecutor
@@ -364,7 +356,6 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
         text: str,
         summary_index_setting: dict | None = None,
         segment_id: str | None = None,
-        document_language: str | None = None,
     ) -> tuple[str, LLMUsage]:
         """
         Generate summary for the given text using ModelInstance.invoke_llm and the default or custom summary prompt,
@@ -375,8 +366,6 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
             text: Text content to summarize
             summary_index_setting: Summary index configuration
             segment_id: Optional segment ID to fetch attachments from SegmentAttachmentBinding table
-            document_language: Optional document language (e.g., "Chinese", "English")
-                to ensure summary is generated in the correct language
 
         Returns:
             Tuple of (summary_content, llm_usage) where llm_usage is LLMUsage object
@@ -392,22 +381,8 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
             raise ValueError("model_name and model_provider_name are required in summary_index_setting")
 
         # Import default summary prompt
-        is_default_prompt = False
         if not summary_prompt:
             summary_prompt = DEFAULT_GENERATOR_SUMMARY_PROMPT
-            is_default_prompt = True
-
-        # Format prompt with document language only for default prompt
-        # Custom prompts are used as-is to avoid interfering with user-defined templates
-        # If document_language is provided, use it; otherwise, use "the same language as the input content"
-        # This is especially important for image-only chunks where text is empty or minimal
-        if is_default_prompt:
-            language_for_prompt = document_language or "the same language as the input content"
-            try:
-                summary_prompt = summary_prompt.format(language=language_for_prompt)
-            except KeyError:
-                # If default prompt doesn't have {language} placeholder, use it as-is
-                pass
 
         provider_manager = ProviderManager()
         provider_model_bundle = provider_manager.get_provider_model_bundle(

api/core/rag/index_processor/processor/parent_child_index_processor.py

@@ -358,11 +358,7 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
         }
 
     def generate_summary_preview(
-        self,
-        tenant_id: str,
-        preview_texts: list[PreviewDetail],
-        summary_index_setting: dict,
-        doc_language: str | None = None,
+        self, tenant_id: str, preview_texts: list[PreviewDetail], summary_index_setting: dict
     ) -> list[PreviewDetail]:
         """
         For each parent chunk in preview_texts, concurrently call generate_summary to generate a summary
@@ -393,7 +389,6 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
                         tenant_id=tenant_id,
                         text=preview.content,
                         summary_index_setting=summary_index_setting,
-                        document_language=doc_language,
                     )
                     preview.summary = summary
             else:
@@ -402,7 +397,6 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
                     tenant_id=tenant_id,
                     text=preview.content,
                     summary_index_setting=summary_index_setting,
-                    document_language=doc_language,
                 )
                 preview.summary = summary
 

api/core/rag/index_processor/processor/qa_index_processor.py

@@ -241,11 +241,7 @@ class QAIndexProcessor(BaseIndexProcessor):
         }
 
     def generate_summary_preview(
-        self,
-        tenant_id: str,
-        preview_texts: list[PreviewDetail],
-        summary_index_setting: dict,
-        doc_language: str | None = None,
+        self, tenant_id: str, preview_texts: list[PreviewDetail], summary_index_setting: dict
     ) -> list[PreviewDetail]:
         """
         QA model doesn't generate summaries, so this method returns preview_texts unchanged.

api/core/repositories/__init__.py

@@ -1,19 +1,18 @@
-"""
-Repository implementations for data access.
+"""Repository implementations for data access."""
 
-This package contains concrete implementations of the repository interfaces
-defined in the core.workflow.repository package.
-"""
+from __future__ import annotations
 
-from core.repositories.celery_workflow_execution_repository import CeleryWorkflowExecutionRepository
-from core.repositories.celery_workflow_node_execution_repository import CeleryWorkflowNodeExecutionRepository
-from core.repositories.factory import DifyCoreRepositoryFactory, RepositoryImportError
-from core.repositories.sqlalchemy_workflow_node_execution_repository import SQLAlchemyWorkflowNodeExecutionRepository
+from .celery_workflow_execution_repository import CeleryWorkflowExecutionRepository
+from .celery_workflow_node_execution_repository import CeleryWorkflowNodeExecutionRepository
+from .factory import DifyCoreRepositoryFactory, RepositoryImportError
+from .sqlalchemy_workflow_execution_repository import SQLAlchemyWorkflowExecutionRepository
+from .sqlalchemy_workflow_node_execution_repository import SQLAlchemyWorkflowNodeExecutionRepository
 
 __all__ = [
     "CeleryWorkflowExecutionRepository",
     "CeleryWorkflowNodeExecutionRepository",
     "DifyCoreRepositoryFactory",
     "RepositoryImportError",
+    "SQLAlchemyWorkflowExecutionRepository",
     "SQLAlchemyWorkflowNodeExecutionRepository",
 ]

api/core/repositories/human_input_repository.py

@@ -0,0 +1,553 @@
+import dataclasses
+import json
+from collections.abc import Mapping, Sequence
+from datetime import datetime
+from typing import Any
+
+from sqlalchemy import Engine, select
+from sqlalchemy.orm import Session, selectinload, sessionmaker
+
+from core.workflow.nodes.human_input.entities import (
+    DeliveryChannelConfig,
+    EmailDeliveryMethod,
+    EmailRecipients,
+    ExternalRecipient,
+    FormDefinition,
+    HumanInputNodeData,
+    MemberRecipient,
+    WebAppDeliveryMethod,
+)
+from core.workflow.nodes.human_input.enums import (
+    DeliveryMethodType,
+    HumanInputFormKind,
+    HumanInputFormStatus,
+)
+from core.workflow.repositories.human_input_form_repository import (
+    FormCreateParams,
+    FormNotFoundError,
+    HumanInputFormEntity,
+    HumanInputFormRecipientEntity,
+)
+from libs.datetime_utils import naive_utc_now
+from libs.uuid_utils import uuidv7
+from models.account import Account, TenantAccountJoin
+from models.human_input import (
+    BackstageRecipientPayload,
+    ConsoleDeliveryPayload,
+    ConsoleRecipientPayload,
+    EmailExternalRecipientPayload,
+    EmailMemberRecipientPayload,
+    HumanInputDelivery,
+    HumanInputForm,
+    HumanInputFormRecipient,
+    RecipientType,
+    StandaloneWebAppRecipientPayload,
+)
+
+
+@dataclasses.dataclass(frozen=True)
+class _DeliveryAndRecipients:
+    delivery: HumanInputDelivery
+    recipients: Sequence[HumanInputFormRecipient]
+
+
+@dataclasses.dataclass(frozen=True)
+class _WorkspaceMemberInfo:
+    user_id: str
+    email: str
+
+
+class _HumanInputFormRecipientEntityImpl(HumanInputFormRecipientEntity):
+    def __init__(self, recipient_model: HumanInputFormRecipient):
+        self._recipient_model = recipient_model
+
+    @property
+    def id(self) -> str:
+        return self._recipient_model.id
+
+    @property
+    def token(self) -> str:
+        if self._recipient_model.access_token is None:
+            raise AssertionError(f"access_token should not be None for recipient {self._recipient_model.id}")
+        return self._recipient_model.access_token
+
+
+class _HumanInputFormEntityImpl(HumanInputFormEntity):
+    def __init__(self, form_model: HumanInputForm, recipient_models: Sequence[HumanInputFormRecipient]):
+        self._form_model = form_model
+        self._recipients = [_HumanInputFormRecipientEntityImpl(recipient) for recipient in recipient_models]
+        self._web_app_recipient = next(
+            (
+                recipient
+                for recipient in recipient_models
+                if recipient.recipient_type == RecipientType.STANDALONE_WEB_APP
+            ),
+            None,
+        )
+        self._console_recipient = next(
+            (recipient for recipient in recipient_models if recipient.recipient_type == RecipientType.CONSOLE),
+            None,
+        )
+        self._submitted_data: Mapping[str, Any] | None = (
+            json.loads(form_model.submitted_data) if form_model.submitted_data is not None else None
+        )
+
+    @property
+    def id(self) -> str:
+        return self._form_model.id
+
+    @property
+    def web_app_token(self):
+        if self._console_recipient is not None:
+            return self._console_recipient.access_token
+        if self._web_app_recipient is None:
+            return None
+        return self._web_app_recipient.access_token
+
+    @property
+    def recipients(self) -> list[HumanInputFormRecipientEntity]:
+        return list(self._recipients)
+
+    @property
+    def rendered_content(self) -> str:
+        return self._form_model.rendered_content
+
+    @property
+    def selected_action_id(self) -> str | None:
+        return self._form_model.selected_action_id
+
+    @property
+    def submitted_data(self) -> Mapping[str, Any] | None:
+        return self._submitted_data
+
+    @property
+    def submitted(self) -> bool:
+        return self._form_model.submitted_at is not None
+
+    @property
+    def status(self) -> HumanInputFormStatus:
+        return self._form_model.status
+
+    @property
+    def expiration_time(self) -> datetime:
+        return self._form_model.expiration_time
+
+
+@dataclasses.dataclass(frozen=True)
+class HumanInputFormRecord:
+    form_id: str
+    workflow_run_id: str | None
+    node_id: str
+    tenant_id: str
+    app_id: str
+    form_kind: HumanInputFormKind
+    definition: FormDefinition
+    rendered_content: str
+    created_at: datetime
+    expiration_time: datetime
+    status: HumanInputFormStatus
+    selected_action_id: str | None
+    submitted_data: Mapping[str, Any] | None
+    submitted_at: datetime | None
+    submission_user_id: str | None
+    submission_end_user_id: str | None
+    completed_by_recipient_id: str | None
+    recipient_id: str | None
+    recipient_type: RecipientType | None
+    access_token: str | None
+
+    @property
+    def submitted(self) -> bool:
+        return self.submitted_at is not None
+
+    @classmethod
+    def from_models(
+        cls, form_model: HumanInputForm, recipient_model: HumanInputFormRecipient | None
+    ) -> "HumanInputFormRecord":
+        definition_payload = json.loads(form_model.form_definition)
+        if "expiration_time" not in definition_payload:
+            definition_payload["expiration_time"] = form_model.expiration_time
+        return cls(
+            form_id=form_model.id,
+            workflow_run_id=form_model.workflow_run_id,
+            node_id=form_model.node_id,
+            tenant_id=form_model.tenant_id,
+            app_id=form_model.app_id,
+            form_kind=form_model.form_kind,
+            definition=FormDefinition.model_validate(definition_payload),
+            rendered_content=form_model.rendered_content,
+            created_at=form_model.created_at,
+            expiration_time=form_model.expiration_time,
+            status=form_model.status,
+            selected_action_id=form_model.selected_action_id,
+            submitted_data=json.loads(form_model.submitted_data) if form_model.submitted_data else None,
+            submitted_at=form_model.submitted_at,
+            submission_user_id=form_model.submission_user_id,
+            submission_end_user_id=form_model.submission_end_user_id,
+            completed_by_recipient_id=form_model.completed_by_recipient_id,
+            recipient_id=recipient_model.id if recipient_model else None,
+            recipient_type=recipient_model.recipient_type if recipient_model else None,
+            access_token=recipient_model.access_token if recipient_model else None,
+        )
+
+
+class _InvalidTimeoutStatusError(ValueError):
+    pass
+
+
+class HumanInputFormRepositoryImpl:
+    def __init__(
+        self,
+        session_factory: sessionmaker | Engine,
+        tenant_id: str,
+    ):
+        if isinstance(session_factory, Engine):
+            session_factory = sessionmaker(bind=session_factory)
+        self._session_factory = session_factory
+        self._tenant_id = tenant_id
+
+    def _delivery_method_to_model(
+        self,
+        session: Session,
+        form_id: str,
+        delivery_method: DeliveryChannelConfig,
+    ) -> _DeliveryAndRecipients:
+        delivery_id = str(uuidv7())
+        delivery_model = HumanInputDelivery(
+            id=delivery_id,
+            form_id=form_id,
+            delivery_method_type=delivery_method.type,
+            delivery_config_id=delivery_method.id,
+            channel_payload=delivery_method.model_dump_json(),
+        )
+        recipients: list[HumanInputFormRecipient] = []
+        if isinstance(delivery_method, WebAppDeliveryMethod):
+            recipient_model = HumanInputFormRecipient(
+                form_id=form_id,
+                delivery_id=delivery_id,
+                recipient_type=RecipientType.STANDALONE_WEB_APP,
+                recipient_payload=StandaloneWebAppRecipientPayload().model_dump_json(),
+            )
+            recipients.append(recipient_model)
+        elif isinstance(delivery_method, EmailDeliveryMethod):
+            email_recipients_config = delivery_method.config.recipients
+            recipients.extend(
+                self._build_email_recipients(
+                    session=session,
+                    form_id=form_id,
+                    delivery_id=delivery_id,
+                    recipients_config=email_recipients_config,
+                )
+            )
+
+        return _DeliveryAndRecipients(delivery=delivery_model, recipients=recipients)
+
+    def _build_email_recipients(
+        self,
+        session: Session,
+        form_id: str,
+        delivery_id: str,
+        recipients_config: EmailRecipients,
+    ) -> list[HumanInputFormRecipient]:
+        member_user_ids = [
+            recipient.user_id for recipient in recipients_config.items if isinstance(recipient, MemberRecipient)
+        ]
+        external_emails = [
+            recipient.email for recipient in recipients_config.items if isinstance(recipient, ExternalRecipient)
+        ]
+        if recipients_config.whole_workspace:
+            members = self._query_all_workspace_members(session=session)
+        else:
+            members = self._query_workspace_members_by_ids(session=session, restrict_to_user_ids=member_user_ids)
+
+        return self._create_email_recipients_from_resolved(
+            form_id=form_id,
+            delivery_id=delivery_id,
+            members=members,
+            external_emails=external_emails,
+        )
+
+    @staticmethod
+    def _create_email_recipients_from_resolved(
+        *,
+        form_id: str,
+        delivery_id: str,
+        members: Sequence[_WorkspaceMemberInfo],
+        external_emails: Sequence[str],
+    ) -> list[HumanInputFormRecipient]:
+        recipient_models: list[HumanInputFormRecipient] = []
+        seen_emails: set[str] = set()
+
+        for member in members:
+            if not member.email:
+                continue
+            if member.email in seen_emails:
+                continue
+            seen_emails.add(member.email)
+            payload = EmailMemberRecipientPayload(user_id=member.user_id, email=member.email)
+            recipient_models.append(
+                HumanInputFormRecipient.new(
+                    form_id=form_id,
+                    delivery_id=delivery_id,
+                    payload=payload,
+                )
+            )
+
+        for email in external_emails:
+            if not email:
+                continue
+            if email in seen_emails:
+                continue
+            seen_emails.add(email)
+            recipient_models.append(
+                HumanInputFormRecipient.new(
+                    form_id=form_id,
+                    delivery_id=delivery_id,
+                    payload=EmailExternalRecipientPayload(email=email),
+                )
+            )
+
+        return recipient_models
+
+    def _query_all_workspace_members(
+        self,
+        session: Session,
+    ) -> list[_WorkspaceMemberInfo]:
+        stmt = (
+            select(Account.id, Account.email)
+            .join(TenantAccountJoin, TenantAccountJoin.account_id == Account.id)
+            .where(TenantAccountJoin.tenant_id == self._tenant_id)
+        )
+        rows = session.execute(stmt).all()
+        return [_WorkspaceMemberInfo(user_id=account_id, email=email) for account_id, email in rows]
+
+    def _query_workspace_members_by_ids(
+        self,
+        session: Session,
+        restrict_to_user_ids: Sequence[str],
+    ) -> list[_WorkspaceMemberInfo]:
+        unique_ids = {user_id for user_id in restrict_to_user_ids if user_id}
+        if not unique_ids:
+            return []
+
+        stmt = (
+            select(Account.id, Account.email)
+            .join(TenantAccountJoin, TenantAccountJoin.account_id == Account.id)
+            .where(TenantAccountJoin.tenant_id == self._tenant_id)
+        )
+        stmt = stmt.where(Account.id.in_(unique_ids))
+
+        rows = session.execute(stmt).all()
+        return [_WorkspaceMemberInfo(user_id=account_id, email=email) for account_id, email in rows]
+
+    def create_form(self, params: FormCreateParams) -> HumanInputFormEntity:
+        form_config: HumanInputNodeData = params.form_config
+
+        with self._session_factory(expire_on_commit=False) as session, session.begin():
+            # Generate unique form ID
+            form_id = str(uuidv7())
+            start_time = naive_utc_now()
+            node_expiration = form_config.expiration_time(start_time)
+            form_definition = FormDefinition(
+                form_content=form_config.form_content,
+                inputs=form_config.inputs,
+                user_actions=form_config.user_actions,
+                rendered_content=params.rendered_content,
+                expiration_time=node_expiration,
+                default_values=dict(params.resolved_default_values),
+                display_in_ui=params.display_in_ui,
+                node_title=form_config.title,
+            )
+            form_model = HumanInputForm(
+                id=form_id,
+                tenant_id=self._tenant_id,
+                app_id=params.app_id,
+                workflow_run_id=params.workflow_execution_id,
+                form_kind=params.form_kind,
+                node_id=params.node_id,
+                form_definition=form_definition.model_dump_json(),
+                rendered_content=params.rendered_content,
+                expiration_time=node_expiration,
+                created_at=start_time,
+            )
+            session.add(form_model)
+            recipient_models: list[HumanInputFormRecipient] = []
+            for delivery in params.delivery_methods:
+                delivery_and_recipients = self._delivery_method_to_model(
+                    session=session,
+                    form_id=form_id,
+                    delivery_method=delivery,
+                )
+                session.add(delivery_and_recipients.delivery)
+                session.add_all(delivery_and_recipients.recipients)
+                recipient_models.extend(delivery_and_recipients.recipients)
+            if params.console_recipient_required and not any(
+                recipient.recipient_type == RecipientType.CONSOLE for recipient in recipient_models
+            ):
+                console_delivery_id = str(uuidv7())
+                console_delivery = HumanInputDelivery(
+                    id=console_delivery_id,
+                    form_id=form_id,
+                    delivery_method_type=DeliveryMethodType.WEBAPP,
+                    delivery_config_id=None,
+                    channel_payload=ConsoleDeliveryPayload().model_dump_json(),
+                )
+                console_recipient = HumanInputFormRecipient(
+                    form_id=form_id,
+                    delivery_id=console_delivery_id,
+                    recipient_type=RecipientType.CONSOLE,
+                    recipient_payload=ConsoleRecipientPayload(
+                        account_id=params.console_creator_account_id,
+                    ).model_dump_json(),
+                )
+                session.add(console_delivery)
+                session.add(console_recipient)
+                recipient_models.append(console_recipient)
+            if params.backstage_recipient_required and not any(
+                recipient.recipient_type == RecipientType.BACKSTAGE for recipient in recipient_models
+            ):
+                backstage_delivery_id = str(uuidv7())
+                backstage_delivery = HumanInputDelivery(
+                    id=backstage_delivery_id,
+                    form_id=form_id,
+                    delivery_method_type=DeliveryMethodType.WEBAPP,
+                    delivery_config_id=None,
+                    channel_payload=ConsoleDeliveryPayload().model_dump_json(),
+                )
+                backstage_recipient = HumanInputFormRecipient(
+                    form_id=form_id,
+                    delivery_id=backstage_delivery_id,
+                    recipient_type=RecipientType.BACKSTAGE,
+                    recipient_payload=BackstageRecipientPayload(
+                        account_id=params.console_creator_account_id,
+                    ).model_dump_json(),
+                )
+                session.add(backstage_delivery)
+                session.add(backstage_recipient)
+                recipient_models.append(backstage_recipient)
+            session.flush()
+
+        return _HumanInputFormEntityImpl(form_model=form_model, recipient_models=recipient_models)
+
+    def get_form(self, workflow_execution_id: str, node_id: str) -> HumanInputFormEntity | None:
+        form_query = select(HumanInputForm).where(
+            HumanInputForm.workflow_run_id == workflow_execution_id,
+            HumanInputForm.node_id == node_id,
+            HumanInputForm.tenant_id == self._tenant_id,
+        )
+        with self._session_factory(expire_on_commit=False) as session:
+            form_model: HumanInputForm | None = session.scalars(form_query).first()
+            if form_model is None:
+                return None
+
+            recipient_query = select(HumanInputFormRecipient).where(HumanInputFormRecipient.form_id == form_model.id)
+            recipient_models = session.scalars(recipient_query).all()
+        return _HumanInputFormEntityImpl(form_model=form_model, recipient_models=recipient_models)
+
+
+class HumanInputFormSubmissionRepository:
+    """Repository for fetching and submitting human input forms."""
+
+    def __init__(self, session_factory: sessionmaker | Engine):
+        if isinstance(session_factory, Engine):
+            session_factory = sessionmaker(bind=session_factory)
+        self._session_factory = session_factory
+
+    def get_by_token(self, form_token: str) -> HumanInputFormRecord | None:
+        query = (
+            select(HumanInputFormRecipient)
+            .options(selectinload(HumanInputFormRecipient.form))
+            .where(HumanInputFormRecipient.access_token == form_token)
+        )
+        with self._session_factory(expire_on_commit=False) as session:
+            recipient_model = session.scalars(query).first()
+            if recipient_model is None or recipient_model.form is None:
+                return None
+            return HumanInputFormRecord.from_models(recipient_model.form, recipient_model)
+
+    def get_by_form_id_and_recipient_type(
+        self,
+        form_id: str,
+        recipient_type: RecipientType,
+    ) -> HumanInputFormRecord | None:
+        query = (
+            select(HumanInputFormRecipient)
+            .options(selectinload(HumanInputFormRecipient.form))
+            .where(
+                HumanInputFormRecipient.form_id == form_id,
+                HumanInputFormRecipient.recipient_type == recipient_type,
+            )
+        )
+        with self._session_factory(expire_on_commit=False) as session:
+            recipient_model = session.scalars(query).first()
+            if recipient_model is None or recipient_model.form is None:
+                return None
+            return HumanInputFormRecord.from_models(recipient_model.form, recipient_model)
+
+    def mark_submitted(
+        self,
+        *,
+        form_id: str,
+        recipient_id: str | None,
+        selected_action_id: str,
+        form_data: Mapping[str, Any],
+        submission_user_id: str | None,
+        submission_end_user_id: str | None,
+    ) -> HumanInputFormRecord:
+        with self._session_factory(expire_on_commit=False) as session, session.begin():
+            form_model = session.get(HumanInputForm, form_id)
+            if form_model is None:
+                raise FormNotFoundError(f"form not found, id={form_id}")
+
+            recipient_model = session.get(HumanInputFormRecipient, recipient_id) if recipient_id else None
+
+            form_model.selected_action_id = selected_action_id
+            form_model.submitted_data = json.dumps(form_data)
+            form_model.submitted_at = naive_utc_now()
+            form_model.status = HumanInputFormStatus.SUBMITTED
+            form_model.submission_user_id = submission_user_id
+            form_model.submission_end_user_id = submission_end_user_id
+            form_model.completed_by_recipient_id = recipient_id
+
+            session.add(form_model)
+            session.flush()
+            session.refresh(form_model)
+            if recipient_model is not None:
+                session.refresh(recipient_model)
+
+            return HumanInputFormRecord.from_models(form_model, recipient_model)
+
+    def mark_timeout(
+        self,
+        *,
+        form_id: str,
+        timeout_status: HumanInputFormStatus,
+        reason: str | None = None,
+    ) -> HumanInputFormRecord:
+        with self._session_factory(expire_on_commit=False) as session, session.begin():
+            form_model = session.get(HumanInputForm, form_id)
+            if form_model is None:
+                raise FormNotFoundError(f"form not found, id={form_id}")
+
+            if timeout_status not in {HumanInputFormStatus.TIMEOUT, HumanInputFormStatus.EXPIRED}:
+                raise _InvalidTimeoutStatusError(f"invalid timeout status: {timeout_status}")
+
+            # already handled or submitted
+            if form_model.status in {HumanInputFormStatus.TIMEOUT, HumanInputFormStatus.EXPIRED}:
+                return HumanInputFormRecord.from_models(form_model, None)
+
+            if form_model.submitted_at is not None or form_model.status == HumanInputFormStatus.SUBMITTED:
+                raise FormNotFoundError(f"form already submitted, id={form_id}")
+
+            form_model.status = timeout_status
+            form_model.selected_action_id = None
+            form_model.submitted_data = None
+            form_model.submission_user_id = None
+            form_model.submission_end_user_id = None
+            form_model.completed_by_recipient_id = None
+            # Reason is recorded in status/error downstream; not stored on form.
+            session.add(form_model)
+            session.flush()
+            session.refresh(form_model)
+
+            return HumanInputFormRecord.from_models(form_model, None)

api/core/repositories/sqlalchemy_workflow_node_execution_repository.py

@@ -488,6 +488,7 @@ class SQLAlchemyWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository)
                 WorkflowNodeExecutionModel.workflow_run_id == workflow_run_id,
                 WorkflowNodeExecutionModel.tenant_id == self._tenant_id,
                 WorkflowNodeExecutionModel.triggered_from == triggered_from,
+                WorkflowNodeExecutionModel.status != WorkflowNodeExecutionStatus.PAUSED,
             )
 
             if self._app_id:

api/core/schemas/registry.py

@@ -35,7 +35,6 @@ class SchemaRegistry:
                     registry.load_all_versions()
 
                     cls._default_instance = registry
-            return cls._default_instance
 
         return cls._default_instance
 

api/core/tools/errors.py

@@ -1,4 +1,5 @@
 from core.tools.entities.tool_entities import ToolInvokeMeta
+from libs.exception import BaseHTTPException
 
 
 class ToolProviderNotFoundError(ValueError):
@@ -37,6 +38,12 @@ class ToolCredentialPolicyViolationError(ValueError):
     pass
 
 
+class WorkflowToolHumanInputNotSupportedError(BaseHTTPException):
+    error_code = "workflow_tool_human_input_not_supported"
+    description = "Workflow with Human Input nodes cannot be published as a workflow tool."
+    code = 400
+
+
 class ToolEngineInvokeError(Exception):
     meta: ToolInvokeMeta
 

api/core/tools/tool_manager.py

@@ -189,13 +189,16 @@ class ToolManager:
                 raise ToolProviderNotFoundError(f"builtin tool {tool_name} not found")
 
             if not provider_controller.need_credentials:
-                return builtin_tool.fork_tool_runtime(
-                    runtime=ToolRuntime(
-                        tenant_id=tenant_id,
-                        credentials={},
-                        invoke_from=invoke_from,
-                        tool_invoke_from=tool_invoke_from,
-                    )
+                return cast(
+                    BuiltinTool,
+                    builtin_tool.fork_tool_runtime(
+                        runtime=ToolRuntime(
+                            tenant_id=tenant_id,
+                            credentials={},
+                            invoke_from=invoke_from,
+                            tool_invoke_from=tool_invoke_from,
+                        )
+                    ),
                 )
             builtin_provider = None
             if isinstance(provider_controller, PluginToolProviderController):
@@ -297,15 +300,18 @@ class ToolManager:
                 decrypted_credentials = refreshed_credentials.credentials
                 cache.delete()
 
-            return builtin_tool.fork_tool_runtime(
-                runtime=ToolRuntime(
-                    tenant_id=tenant_id,
-                    credentials=dict(decrypted_credentials),
-                    credential_type=CredentialType.of(builtin_provider.credential_type),
-                    runtime_parameters={},
-                    invoke_from=invoke_from,
-                    tool_invoke_from=tool_invoke_from,
-                )
+            return cast(
+                BuiltinTool,
+                builtin_tool.fork_tool_runtime(
+                    runtime=ToolRuntime(
+                        tenant_id=tenant_id,
+                        credentials=dict(decrypted_credentials),
+                        credential_type=CredentialType.of(builtin_provider.credential_type),
+                        runtime_parameters={},
+                        invoke_from=invoke_from,
+                        tool_invoke_from=tool_invoke_from,
+                    )
+                ),
             )
 
         elif provider_type == ToolProviderType.API:

api/core/tools/utils/workflow_configuration_sync.py

@@ -3,10 +3,17 @@ from typing import Any
 
 from core.app.app_config.entities import VariableEntity
 from core.tools.entities.tool_entities import WorkflowToolParameterConfiguration
+from core.tools.errors import WorkflowToolHumanInputNotSupportedError
+from core.workflow.enums import NodeType
 from core.workflow.nodes.base.entities import OutputVariableEntity
 
 
 class WorkflowToolConfigurationUtils:
+    @classmethod
+    def check_parameter_configurations(cls, configurations: list[Mapping[str, Any]]):
+        for configuration in configurations:
+            WorkflowToolParameterConfiguration.model_validate(configuration)
+
     @classmethod
     def get_workflow_graph_variables(cls, graph: Mapping[str, Any]) -> Sequence[VariableEntity]:
         """
@@ -45,6 +52,13 @@ class WorkflowToolConfigurationUtils:
 
         return [outputs_by_variable[variable] for variable in variable_order]
 
+    @classmethod
+    def ensure_no_human_input_nodes(cls, graph: Mapping[str, Any]) -> None:
+        nodes = graph.get("nodes", [])
+        for node in nodes:
+            if node.get("data", {}).get("type") == NodeType.HUMAN_INPUT:
+                raise WorkflowToolHumanInputNotSupportedError()
+
     @classmethod
     def check_is_synced(
         cls, variables: list[VariableEntity], tool_configurations: list[WorkflowToolParameterConfiguration]

api/core/workflow/entities/__init__.py

@@ -2,10 +2,12 @@ from .agent import AgentNodeStrategyInit
 from .graph_init_params import GraphInitParams
 from .workflow_execution import WorkflowExecution
 from .workflow_node_execution import WorkflowNodeExecution
+from .workflow_start_reason import WorkflowStartReason
 
 __all__ = [
     "AgentNodeStrategyInit",
     "GraphInitParams",
     "WorkflowExecution",
     "WorkflowNodeExecution",
+    "WorkflowStartReason",
 ]

api/core/workflow/entities/graph_config.py

@@ -1,24 +0,0 @@
-from __future__ import annotations
-
-import sys
-
-from pydantic import TypeAdapter, with_config
-
-if sys.version_info >= (3, 12):
-    from typing import TypedDict
-else:
-    from typing_extensions import TypedDict
-
-
-@with_config(extra="allow")
-class NodeConfigData(TypedDict):
-    type: str
-
-
-@with_config(extra="allow")
-class NodeConfigDict(TypedDict):
-    id: str
-    data: NodeConfigData
-
-
-NodeConfigDictAdapter = TypeAdapter(NodeConfigDict)

api/core/workflow/entities/graph_init_params.py

@@ -5,6 +5,16 @@ from pydantic import BaseModel, Field
 
 
 class GraphInitParams(BaseModel):
+    """GraphInitParams encapsulates the configurations and contextual information
+    that remain constant throughout a single execution of the graph engine.
+
+    A single execution is defined as follows: as long as the execution has not reached
+    its conclusion, it is considered one execution. For instance, if a workflow is suspended
+    and later resumed, it is still regarded as a single execution, not two.
+
+    For the state diagram of workflow execution, refer to `WorkflowExecutionStatus`.
+    """
+
     # init params
     tenant_id: str = Field(..., description="tenant / workspace id")
     app_id: str = Field(..., description="app id")

api/core/workflow/entities/pause_reason.py

@@ -1,8 +1,11 @@
+from collections.abc import Mapping
 from enum import StrEnum, auto
-from typing import Annotated, Literal, TypeAlias
+from typing import Annotated, Any, Literal, TypeAlias
 
 from pydantic import BaseModel, Field
 
+from core.workflow.nodes.human_input.entities import FormInput, UserAction
+
 
 class PauseReasonType(StrEnum):
     HUMAN_INPUT_REQUIRED = auto()
@@ -11,10 +14,31 @@ class PauseReasonType(StrEnum):
 
 class HumanInputRequired(BaseModel):
     TYPE: Literal[PauseReasonType.HUMAN_INPUT_REQUIRED] = PauseReasonType.HUMAN_INPUT_REQUIRED
-
     form_id: str
-    # The identifier of the human input node causing the pause.
+    form_content: str
+    inputs: list[FormInput] = Field(default_factory=list)
+    actions: list[UserAction] = Field(default_factory=list)
+    display_in_ui: bool = False
     node_id: str
+    node_title: str
+
+    # The `resolved_default_values` stores the resolved values of variable defaults. It's a mapping from
+    # `output_variable_name` to their resolved values.
+    #
+    # For example, The form contains a input with output variable name `name` and placeholder type `VARIABLE`, its
+    # selector is ["start", "name"]. While the HumanInputNode is executed, the correspond value of variable
+    # `start.name` in variable pool is `John`. Thus, the resolved value of the output variable `name` is `John`. The
+    # `resolved_default_values` is `{"name": "John"}`.
+    #
+    # Only form inputs with default value type `VARIABLE` will be resolved and stored in `resolved_default_values`.
+    resolved_default_values: Mapping[str, Any] = Field(default_factory=dict)
+
+    # The `form_token` is the token used to submit the form via UI surfaces. It corresponds to
+    # `HumanInputFormRecipient.access_token`.
+    #
+    # This field is `None` if webapp delivery is not set and not
+    # in orchestrating mode.
+    form_token: str | None = None
 
 
 class SchedulingPause(BaseModel):

api/core/workflow/entities/workflow_start_reason.py

@@ -0,0 +1,8 @@
+from enum import StrEnum
+
+
+class WorkflowStartReason(StrEnum):
+    """Reason for workflow start events across graph/queue/SSE layers."""
+
+    INITIAL = "initial"  # First start of a workflow run.
+    RESUMPTION = "resumption"  # Start triggered after resuming a paused run.

api/core/workflow/graph/graph.py

@@ -5,20 +5,15 @@ from collections import defaultdict
 from collections.abc import Mapping, Sequence
 from typing import Protocol, cast, final
 
-from pydantic import TypeAdapter
-
-from core.workflow.entities.graph_config import NodeConfigDict
 from core.workflow.enums import ErrorStrategy, NodeExecutionType, NodeState, NodeType
 from core.workflow.nodes.base.node import Node
-from libs.typing import is_str
+from libs.typing import is_str, is_str_dict
 
 from .edge import Edge
 from .validation import get_graph_validator
 
 logger = logging.getLogger(__name__)
 
-_ListNodeConfigDict = TypeAdapter(list[NodeConfigDict])
-
 
 class NodeFactory(Protocol):
     """
@@ -28,7 +23,7 @@ class NodeFactory(Protocol):
     allowing for different node creation strategies while maintaining type safety.
     """
 
-    def create_node(self, node_config: NodeConfigDict) -> Node:
+    def create_node(self, node_config: dict[str, object]) -> Node:
         """
         Create a Node instance from node configuration data.
 
@@ -68,24 +63,28 @@ class Graph:
         self.root_node = root_node
 
     @classmethod
-    def _parse_node_configs(cls, node_configs: list[NodeConfigDict]) -> dict[str, NodeConfigDict]:
+    def _parse_node_configs(cls, node_configs: list[dict[str, object]]) -> dict[str, dict[str, object]]:
         """
         Parse node configurations and build a mapping of node IDs to configs.
 
         :param node_configs: list of node configuration dictionaries
         :return: mapping of node ID to node config
         """
-        node_configs_map: dict[str, NodeConfigDict] = {}
+        node_configs_map: dict[str, dict[str, object]] = {}
 
         for node_config in node_configs:
-            node_configs_map[node_config["id"]] = node_config
+            node_id = node_config.get("id")
+            if not node_id or not isinstance(node_id, str):
+                continue
+
+            node_configs_map[node_id] = node_config
 
         return node_configs_map
 
     @classmethod
     def _find_root_node_id(
         cls,
-        node_configs_map: Mapping[str, NodeConfigDict],
+        node_configs_map: Mapping[str, Mapping[str, object]],
         edge_configs: Sequence[Mapping[str, object]],
         root_node_id: str | None = None,
     ) -> str:
@@ -114,8 +113,10 @@ class Graph:
         # Prefer START node if available
         start_node_id = None
         for nid in root_candidates:
-            node_data = node_configs_map[nid]["data"]
-            node_type = node_data["type"]
+            node_data = node_configs_map[nid].get("data")
+            if not is_str_dict(node_data):
+                continue
+            node_type = node_data.get("type")
             if not isinstance(node_type, str):
                 continue
             if NodeType(node_type).is_start_node:
@@ -175,7 +176,7 @@ class Graph:
     @classmethod
     def _create_node_instances(
         cls,
-        node_configs_map: dict[str, NodeConfigDict],
+        node_configs_map: dict[str, dict[str, object]],
         node_factory: NodeFactory,
     ) -> dict[str, Node]:
         """
@@ -302,7 +303,7 @@ class Graph:
         node_configs = graph_config.get("nodes", [])
 
         edge_configs = cast(list[dict[str, object]], edge_configs)
-        node_configs = _ListNodeConfigDict.validate_python(node_configs)
+        node_configs = cast(list[dict[str, object]], node_configs)
 
         if not node_configs:
             raise ValueError("Graph must have at least one node")

api/core/workflow/graph_engine/_engine_utils.py

@@ -0,0 +1,15 @@
+import time
+
+
+def get_timestamp() -> float:
+    """Retrieve a timestamp as a float point numer representing the number of seconds
+    since the Unix epoch.
+
+    This function is primarily used to measure the execution time of the workflow engine.
+    Since workflow execution may be paused and resumed on a different machine,
+    `time.perf_counter` cannot be used as it is inconsistent across machines.
+
+    To address this, the function uses the wall clock as the time source.
+    However, it assumes that the clocks of all servers are properly synchronized.
+    """
+    return round(time.time())