fix(api): serialize pipeline file-upload created_at (#32098)

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

.agents/skills/component-refactoring/SKILL.md

@@ -480,4 +480,4 @@ const useButtonState = () => {
 ### Related Skills
 
 - `frontend-testing` - For testing refactored components
-- `web/testing/testing.md` - Testing specification
+- `web/docs/test.md` - Testing specification

.agents/skills/frontend-testing/SKILL.md

@@ -7,7 +7,7 @@ description: Generate Vitest + React Testing Library tests for Dify frontend com
 
 This skill enables Claude to generate high-quality, comprehensive frontend tests for the Dify project following established conventions and best practices.
 
-> **⚠️ Authoritative Source**: This skill is derived from `web/testing/testing.md`. Use Vitest mock/timer APIs (`vi.*`).
+> **⚠️ Authoritative Source**: This skill is derived from `web/docs/test.md`. Use Vitest mock/timer APIs (`vi.*`).
 
 ## When to Apply This Skill
 
@@ -309,7 +309,7 @@ For more detailed information, refer to:
 
 ### Primary Specification (MUST follow)
 
-- **`web/testing/testing.md`** - The canonical testing specification. This skill is derived from this document.
+- **`web/docs/test.md`** - The canonical testing specification. This skill is derived from this document.
 
 ### Reference Examples in Codebase
 

.agents/skills/frontend-testing/references/workflow.md

@@ -4,7 +4,7 @@ This guide defines the workflow for generating tests, especially for complex com
 
 ## Scope Clarification
 
-This guide addresses **multi-file workflow** (how to process multiple test files). For coverage requirements within a single test file, see `web/testing/testing.md` § Coverage Goals.
+This guide addresses **multi-file workflow** (how to process multiple test files). For coverage requirements within a single test file, see `web/docs/test.md` § Coverage Goals.
 
 | Scope | Rule |
 |-------|------|

.codex/skills/component-refactoring

@@ -1 +0,0 @@
-../../.agents/skills/component-refactoring

.codex/skills/frontend-code-review

@@ -1 +0,0 @@
-../../.agents/skills/frontend-code-review

.codex/skills/frontend-testing

@@ -1 +0,0 @@
-../../.agents/skills/frontend-testing

.codex/skills/orpc-contract-first

@@ -1 +0,0 @@
-../../.agents/skills/orpc-contract-first

.github/CODEOWNERS

@@ -9,6 +9,9 @@
 # CODEOWNERS file
 /.github/CODEOWNERS @laipz8200 @crazywoola
 
+# Agents
+/.agents/skills/ @hyoban
+
 # Docs
 /docs/ @crazywoola
 
@@ -21,6 +24,10 @@
 /api/services/tools/mcp_tools_manage_service.py @Nov1c444
 /api/controllers/mcp/ @Nov1c444
 /api/controllers/console/app/mcp_server.py @Nov1c444
+
+# Backend - Tests
+/api/tests/ @laipz8200 @QuantumGhost
+
 /api/tests/**/*mcp* @Nov1c444
 
 # Backend - Workflow - Engine (Core graph execution engine)
@@ -231,6 +238,9 @@
 # Frontend - Base Components
 /web/app/components/base/ @iamjoel @zxhlyh
 
+# Frontend - Base Components Tests
+/web/app/components/base/**/*.spec.tsx @hyoban @CodingOnStar
+
 # Frontend - Utils and Hooks
 /web/utils/classnames.ts @iamjoel @zxhlyh
 /web/utils/time.ts @iamjoel @zxhlyh

.github/workflows/api-tests.yml

@@ -72,6 +72,7 @@ jobs:
           OPENDAL_FS_ROOT: /tmp/dify-storage
         run: |
           uv run --project api pytest \
+            -n auto \
             --timeout "${PYTEST_TIMEOUT:-180}" \
             api/tests/integration_tests/workflow \
             api/tests/integration_tests/tools \

.github/workflows/autofix.yml

@@ -79,29 +79,6 @@ jobs:
           find . -name "*.py" -type f -exec sed -i.bak -E 's/"([^"]+)" \| None/Optional["\1"]/g; s/'"'"'([^'"'"']+)'"'"' \| None/Optional['"'"'\1'"'"']/g' {} \;
           find . -name "*.py.bak" -type f -delete
 
-      - name: Install pnpm
-        uses: pnpm/action-setup@v4
-        with:
-          package_json_file: web/package.json
-          run_install: false
-
-      - name: Setup Node.js
-        uses: actions/setup-node@v6
-        with:
-          node-version: 24
-          cache: pnpm
-          cache-dependency-path: ./web/pnpm-lock.yaml
-
-      - name: Install web dependencies
-        run: |
-          cd web
-          pnpm install --frozen-lockfile
-
-      - name: ESLint autofix
-        run: |
-          cd web
-          pnpm lint:fix || true
-
       # mdformat breaks YAML front matter in markdown files. Add --exclude for directories containing YAML front matter.
       - name: mdformat
         run: |

.github/workflows/build-push.yml

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

.github/workflows/deploy-hitl.yml

@@ -4,8 +4,7 @@ on:
   workflow_run:
     workflows: ["Build and Push API & Web"]
     branches:
-      - "feat/hitl-frontend"
-      - "feat/hitl-backend"
+      - "build/feat/hitl"
     types:
       - completed
 
@@ -14,10 +13,7 @@ jobs:
     runs-on: ubuntu-latest
     if: |
       github.event.workflow_run.conclusion == 'success' &&
-      (
-        github.event.workflow_run.head_branch == 'feat/hitl-frontend' ||
-        github.event.workflow_run.head_branch == 'feat/hitl-backend'
-      )
+      github.event.workflow_run.head_branch == 'build/feat/hitl'
     steps:
       - name: Deploy to server
         uses: appleboy/ssh-action@v1

.github/workflows/web-tests.yml

@@ -39,7 +39,7 @@ jobs:
         run: pnpm install --frozen-lockfile
 
       - name: Run tests
-        run: pnpm test:coverage
+        run: pnpm test:ci
 
       - name: Coverage Summary
         if: always()

.vscode/launch.json.template

@@ -37,7 +37,7 @@
                 "-c",
                 "1",
                 "-Q",
-                "dataset,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor,retention",
+                "dataset,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor,retention,workflow_based_app_execution",
                 "--loglevel",
                 "INFO"
             ],

AGENTS.md

@@ -7,7 +7,7 @@ Dify is an open-source platform for developing LLM applications with an intuitiv
 The codebase is split into:
 
 - **Backend API** (`/api`): Python Flask application organized with Domain-Driven Design
-- **Frontend Web** (`/web`): Next.js 15 application using TypeScript and React 19
+- **Frontend Web** (`/web`): Next.js application using TypeScript and React
 - **Docker deployment** (`/docker`): Containerized deployment configurations
 
 ## Backend Workflow
@@ -18,36 +18,7 @@ The codebase is split into:
 
 ## Frontend Workflow
 
-```bash
-cd web
-pnpm lint:fix
-pnpm type-check:tsgo
-pnpm test
-```
-
-### Frontend Linting
-
-ESLint is used for frontend code quality. Available commands:
-
-```bash
-# Lint all files (report only)
-pnpm lint
-
-# Lint and auto-fix issues
-pnpm lint:fix
-
-# Lint specific files or directories
-pnpm lint:fix app/components/base/button/
-pnpm lint:fix app/components/base/button/index.tsx
-
-# Lint quietly (errors only, no warnings)
-pnpm lint:quiet
-
-# Check code complexity
-pnpm lint:complexity
-```
-
-**Important**: Always run `pnpm lint:fix` before committing. The pre-commit hook runs `lint-staged` which only lints staged files.
+- Read `web/AGENTS.md` for details
 
 ## Testing & Quality Practices
 

CONTRIBUTING.md

@@ -77,7 +77,7 @@ How we prioritize:
 
 For setting up the frontend service, please refer to our comprehensive [guide](https://github.com/langgenius/dify/blob/main/web/README.md) in the `web/README.md` file. This document provides detailed instructions to help you set up the frontend environment properly.
 
-**Testing**: All React components must have comprehensive test coverage. See [web/testing/testing.md](https://github.com/langgenius/dify/blob/main/web/testing/testing.md) for the canonical frontend testing guidelines and follow every requirement described there.
+**Testing**: All React components must have comprehensive test coverage. See [web/docs/test.md](https://github.com/langgenius/dify/blob/main/web/docs/test.md) for the canonical frontend testing guidelines and follow every requirement described there.
 
 #### Backend
 

Makefile

@@ -3,7 +3,6 @@ DOCKER_REGISTRY=langgenius
 WEB_IMAGE=$(DOCKER_REGISTRY)/dify-web
 API_IMAGE=$(DOCKER_REGISTRY)/dify-api
 VERSION=latest
-PATH_TO_CHECK ?= api/
 
 # Default target - show help
 .DEFAULT_GOAL := help
@@ -81,7 +80,7 @@ test:
 		echo "Target: $(TARGET_TESTS)"; \
 		uv run --project api --dev pytest $(TARGET_TESTS); \
 	else \
-		uv run --project api --dev dev/pytest/pytest_unit_tests.sh; \
+		PYTEST_XDIST_ARGS="-n auto" uv run --project api --dev dev/pytest/pytest_unit_tests.sh; \
 	fi
 	@echo "✅ Tests complete"
 

api/.env.example

@@ -617,6 +617,7 @@ PLUGIN_DAEMON_URL=http://127.0.0.1:5002
 PLUGIN_REMOTE_INSTALL_PORT=5003
 PLUGIN_REMOTE_INSTALL_HOST=localhost
 PLUGIN_MAX_PACKAGE_SIZE=15728640
+PLUGIN_MODEL_SCHEMA_CACHE_TTL=3600
 INNER_API_KEY_FOR_PLUGIN=QaHbTe77CtuXmsfyhR7+vRjI/+XbV1AaFy691iy+kGDv2Jvy0/eAh8Y1
 
 # Marketplace configuration
@@ -717,3 +718,27 @@ 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
@@ -50,14 +52,14 @@ ignore_imports =
     core.workflow.nodes.agent.agent_node -> extensions.ext_database
     core.workflow.nodes.datasource.datasource_node -> extensions.ext_database
     core.workflow.nodes.knowledge_index.knowledge_index_node -> extensions.ext_database
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> extensions.ext_database
     core.workflow.nodes.llm.file_saver -> extensions.ext_database
     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.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
@@ -122,11 +124,6 @@ ignore_imports =
     core.workflow.nodes.http_request.node -> core.tools.tool_file_manager
     core.workflow.nodes.iteration.iteration_node -> core.app.workflow.node_factory
     core.workflow.nodes.knowledge_index.knowledge_index_node -> core.rag.index_processor.index_processor_factory
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.rag.datasource.retrieval_service
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.rag.retrieval.dataset_retrieval
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> models.dataset
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> services.feature_service
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.model_runtime.model_providers.__base.large_language_model
     core.workflow.nodes.llm.llm_utils -> configs
     core.workflow.nodes.llm.llm_utils -> core.app.entities.app_invoke_entities
     core.workflow.nodes.llm.llm_utils -> core.file.models
@@ -136,7 +133,6 @@ ignore_imports =
     core.workflow.nodes.llm.llm_utils -> models.provider
     core.workflow.nodes.llm.llm_utils -> services.credit_pool_service
     core.workflow.nodes.llm.node -> core.tools.signature
-    core.workflow.nodes.template_transform.template_transform_node -> configs
     core.workflow.nodes.tool.tool_node -> core.callback_handler.workflow_tool_callback_handler
     core.workflow.nodes.tool.tool_node -> core.tools.tool_engine
     core.workflow.nodes.tool.tool_node -> core.tools.tool_manager
@@ -145,9 +141,9 @@ 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
     core.workflow.nodes.llm.node -> core.app.entities.app_invoke_entities
     core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.app.entities.app_invoke_entities
     core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.prompt.advanced_prompt_transform
@@ -163,9 +159,6 @@ ignore_imports =
     core.workflow.workflow_entry -> core.app.workflow.node_factory
     core.workflow.nodes.datasource.datasource_node -> core.datasource.datasource_manager
     core.workflow.nodes.datasource.datasource_node -> core.datasource.utils.message_transformer
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.entities.agent_entities
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.entities.model_entities
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.model_manager
     core.workflow.nodes.llm.llm_utils -> core.entities.provider_entities
     core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.model_manager
     core.workflow.nodes.question_classifier.question_classifier_node -> core.model_manager
@@ -214,7 +207,6 @@ ignore_imports =
     core.workflow.nodes.llm.node -> core.llm_generator.output_parser.structured_output
     core.workflow.nodes.llm.node -> core.model_manager
     core.workflow.nodes.agent.entities -> core.prompt.entities.advanced_prompt_entities
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.prompt.simple_prompt_transform
     core.workflow.nodes.llm.entities -> core.prompt.entities.advanced_prompt_entities
     core.workflow.nodes.llm.llm_utils -> core.prompt.entities.advanced_prompt_entities
     core.workflow.nodes.llm.node -> core.prompt.entities.advanced_prompt_entities
@@ -227,7 +219,9 @@ ignore_imports =
     core.workflow.nodes.knowledge_index.entities -> core.rag.retrieval.retrieval_methods
     core.workflow.nodes.knowledge_index.knowledge_index_node -> core.rag.retrieval.retrieval_methods
     core.workflow.nodes.knowledge_index.knowledge_index_node -> models.dataset
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.rag.retrieval.retrieval_methods
+    core.workflow.nodes.knowledge_index.knowledge_index_node -> services.summary_index_service
+    core.workflow.nodes.knowledge_index.knowledge_index_node -> tasks.generate_summary_index_task
+    core.workflow.nodes.knowledge_index.knowledge_index_node -> core.rag.index_processor.processor.paragraph_index_processor
     core.workflow.nodes.llm.node -> models.dataset
     core.workflow.nodes.agent.agent_node -> core.tools.utils.message_transformer
     core.workflow.nodes.llm.file_saver -> core.tools.signature
@@ -245,6 +239,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
@@ -285,12 +280,12 @@ ignore_imports =
     core.workflow.nodes.agent.agent_node -> extensions.ext_database
     core.workflow.nodes.datasource.datasource_node -> extensions.ext_database
     core.workflow.nodes.knowledge_index.knowledge_index_node -> extensions.ext_database
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> extensions.ext_database
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> extensions.ext_redis
     core.workflow.nodes.llm.file_saver -> extensions.ext_database
     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
@@ -300,6 +295,58 @@ ignore_imports =
     core.workflow.nodes.agent.agent_node -> services
     core.workflow.nodes.tool.tool_node -> services
 
+[importlinter:contract:model-runtime-no-internal-imports]
+name = Model Runtime Internal Imports
+type = forbidden
+source_modules =
+    core.model_runtime
+forbidden_modules =
+    configs
+    controllers
+    extensions
+    models
+    services
+    tasks
+    core.agent
+    core.app
+    core.base
+    core.callback_handler
+    core.datasource
+    core.db
+    core.entities
+    core.errors
+    core.extension
+    core.external_data_tool
+    core.file
+    core.helper
+    core.hosting_configuration
+    core.indexing_runner
+    core.llm_generator
+    core.logging
+    core.mcp
+    core.memory
+    core.model_manager
+    core.moderation
+    core.ops
+    core.plugin
+    core.prompt
+    core.provider_manager
+    core.rag
+    core.repositories
+    core.schemas
+    core.tools
+    core.trigger
+    core.variables
+    core.workflow
+ignore_imports =
+    core.model_runtime.model_providers.__base.ai_model -> configs
+    core.model_runtime.model_providers.__base.ai_model -> extensions.ext_redis
+    core.model_runtime.model_providers.__base.large_language_model -> configs
+    core.model_runtime.model_providers.__base.text_embedding_model -> core.entities.embedding_type
+    core.model_runtime.model_providers.model_provider_factory -> configs
+    core.model_runtime.model_providers.model_provider_factory -> extensions.ext_redis
+    core.model_runtime.model_providers.model_provider_factory -> models.provider_ids
+
 [importlinter:contract:rsc]
 name = RSC
 type = layers

api/.ruff.toml

@@ -53,6 +53,7 @@ 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
 
 ]
 
@@ -88,6 +89,7 @@ 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]
@@ -109,10 +111,20 @@ 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/README.md

@@ -122,7 +122,7 @@ These commands assume you start from the repository root.
 
    ```bash
    cd api
-   uv run celery -A app.celery worker -P threads -c 2 --loglevel INFO -Q dataset,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor,retention
+   uv run celery -A app.celery worker -P threads -c 2 --loglevel INFO -Q api_token,dataset,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor,retention
    ```
 
 1. Optional: start Celery Beat (scheduled tasks, in a new terminal).

api/app.py

@@ -1,4 +1,12 @@
+from __future__ import annotations
+
 import sys
+from typing import TYPE_CHECKING, cast
+
+if TYPE_CHECKING:
+    from celery import Celery
+
+    celery: Celery
 
 
 def is_db_command() -> bool:
@@ -23,7 +31,7 @@ else:
     from app_factory import create_app
 
     app = create_app()
-    celery = app.extensions["celery"]
+    celery = cast("Celery", app.extensions["celery"])
 
 if __name__ == "__main__":
     app.run(host="0.0.0.0", port=5001)

api/app_factory.py

@@ -149,7 +149,7 @@ def initialize_extensions(app: DifyApp):
             logger.info("Loaded %s (%s ms)", short_name, round((end_time - start_time) * 1000, 2))
 
 
-def create_migrations_app():
+def create_migrations_app() -> DifyApp:
     app = create_flask_app_with_configs()
     from extensions import ext_database, ext_migrate
 

api/commands.py

@@ -739,8 +739,10 @@ def upgrade_db():
 
             click.echo(click.style("Database migration successful!", fg="green"))
 
-        except Exception:
+        except Exception as e:
             logger.exception("Failed to execute database migration")
+            click.echo(click.style(f"Database migration failed: {e}", fg="red"))
+            raise SystemExit(1)
         finally:
             lock.release()
     else:
@@ -1450,54 +1452,58 @@ def clear_orphaned_file_records(force: bool):
         all_ids_in_tables = []
         for ids_table in ids_tables:
             query = ""
-            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"
+            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",
+                        )
                     )
-                )
-                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",
+                    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 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",
+                    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']}::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})
+                    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
         click.echo(click.style(f"Found {len(all_ids_in_tables)} file ids in tables.", fg="white"))
 
     except Exception as e:
@@ -1737,59 +1743,18 @@ def file_usage(
                 if src_filter != src:
                     continue
 
-        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:
+        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 ref_file_id not in file_key_map:
                             continue
                         storage_key = file_key_map[ref_file_id]
@@ -1812,6 +1777,50 @@ 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,
     )
 
+    HUMAN_INPUT_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=7).total_seconds()),
+        ge=1,
+    )
+
 
 class CodeExecutionSandboxConfig(BaseSettings):
     """
@@ -243,6 +260,11 @@ class PluginConfig(BaseSettings):
         default=15728640 * 12,
     )
 
+    PLUGIN_MODEL_SCHEMA_CACHE_TTL: PositiveInt = Field(
+        description="TTL in seconds for caching plugin model schemas in Redis",
+        default=60 * 60,
+    )
+
 
 class MarketplaceConfig(BaseSettings):
     """
@@ -1129,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,
@@ -1150,6 +1180,16 @@ class CeleryScheduleTasksConfig(BaseSettings):
         default=0,
     )
 
+    # API token last_used_at batch update
+    ENABLE_API_TOKEN_LAST_USED_UPDATE_TASK: bool = Field(
+        description="Enable periodic batch update of API token last_used_at timestamps",
+        default=True,
+    )
+    API_TOKEN_LAST_USED_UPDATE_INTERVAL: int = Field(
+        description="Interval in minutes for batch updating API token last_used_at (default 30)",
+        default=30,
+    )
+
     # Trigger provider refresh (simple version)
     ENABLE_TRIGGER_PROVIDER_REFRESH_TASK: bool = Field(
         description="Enable trigger provider refresh poller",

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/contexts/__init__.py

@@ -6,7 +6,6 @@ from contexts.wrapper import RecyclableContextVar
 
 if TYPE_CHECKING:
     from core.datasource.__base.datasource_provider import DatasourcePluginProviderController
-    from core.model_runtime.entities.model_entities import AIModelEntity
     from core.plugin.entities.plugin_daemon import PluginModelProviderEntity
     from core.tools.plugin_tool.provider import PluginToolProviderController
     from core.trigger.provider import PluginTriggerProviderController
@@ -29,12 +28,6 @@ plugin_model_providers_lock: RecyclableContextVar[Lock] = RecyclableContextVar(
     ContextVar("plugin_model_providers_lock")
 )
 
-plugin_model_schema_lock: RecyclableContextVar[Lock] = RecyclableContextVar(ContextVar("plugin_model_schema_lock"))
-
-plugin_model_schemas: RecyclableContextVar[dict[str, "AIModelEntity"]] = RecyclableContextVar(
-    ContextVar("plugin_model_schemas")
-)
-
 datasource_plugin_providers: RecyclableContextVar[dict[str, "DatasourcePluginProviderController"]] = (
     RecyclableContextVar(ContextVar("datasource_plugin_providers"))
 )

api/controllers/common/schema.py

@@ -5,8 +5,6 @@ from enum import StrEnum
 from flask_restx import Namespace
 from pydantic import BaseModel, TypeAdapter
 
-from controllers.console import console_ns
-
 DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
 
@@ -24,6 +22,9 @@ def register_schema_models(namespace: Namespace, *models: type[BaseModel]) -> No
 
 
 def get_or_create_model(model_name: str, field_def):
+    # Import lazily to avoid circular imports between console controllers and schema helpers.
+    from controllers.console import console_ns
+
     existing = console_ns.models.get(model_name)
     if existing is None:
         existing = console_ns.model(model_name, field_def)

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/admin.py

@@ -243,15 +243,13 @@ class InsertExploreBannerApi(Resource):
     def post(self):
         payload = InsertExploreBannerPayload.model_validate(console_ns.payload)
 
-        content = {
-            "category": payload.category,
-            "title": payload.title,
-            "description": payload.description,
-            "img-src": payload.img_src,
-        }
-
         banner = ExporleBanner(
-            content=content,
+            content={
+                "category": payload.category,
+                "title": payload.title,
+                "description": payload.description,
+                "img-src": payload.img_src,
+            },
             link=payload.link,
             sort=payload.sort,
             language=payload.language,

api/controllers/console/apikey.py

@@ -10,6 +10,7 @@ from libs.helper import TimestampField
 from libs.login import current_account_with_tenant, login_required
 from models.dataset import Dataset
 from models.model import ApiToken, App
+from services.api_token_service import ApiTokenCache
 
 from . import console_ns
 from .wraps import account_initialization_required, edit_permission_required, setup_required
@@ -131,6 +132,11 @@ class BaseApiKeyResource(Resource):
         if key is None:
             flask_restx.abort(HTTPStatus.NOT_FOUND, message="API key not found")
 
+        # Invalidate cache before deleting from database
+        # Type assertion: key is guaranteed to be non-None here because abort() raises
+        assert key is not None  # nosec - for type checker only
+        ApiTokenCache.delete(key.token, key.type)
+
         db.session.query(ApiToken).where(ApiToken.id == api_key_id).delete()
         db.session.commit()
 

api/controllers/console/app/annotation.py

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

api/controllers/console/app/app.py

@@ -1,3 +1,4 @@
+import logging
 import uuid
 from datetime import datetime
 from typing import Any, Literal, TypeAlias
@@ -54,6 +55,8 @@ ALLOW_CREATE_APP_MODES = ["chat", "agent-chat", "advanced-chat", "workflow", "co
 
 register_enum_models(console_ns, IconType)
 
+_logger = logging.getLogger(__name__)
+
 
 class AppListQuery(BaseModel):
     page: int = Field(default=1, ge=1, le=99999, description="Page number (1-99999)")
@@ -499,6 +502,7 @@ class AppListApi(Resource):
                     select(Workflow).where(
                         Workflow.version == Workflow.VERSION_DRAFT,
                         Workflow.app_id.in_(workflow_capable_app_ids),
+                        Workflow.tenant_id == current_tenant_id,
                     )
                 )
                 .scalars()
@@ -510,12 +514,14 @@ class AppListApi(Resource):
                 NodeType.TRIGGER_PLUGIN,
             }
             for workflow in draft_workflows:
+                node_id = None
                 try:
-                    for _, node_data in workflow.walk_nodes():
+                    for node_id, node_data in workflow.walk_nodes():
                         if node_data.get("type") in trigger_node_types:
                             draft_trigger_app_ids.add(str(workflow.app_id))
                             break
                 except Exception:
+                    _logger.exception("error while walking nodes, workflow_id=%s, node_id=%s", workflow.id, node_id)
                     continue
 
         for app in app_pagination.items:

api/controllers/console/app/audio.py

@@ -6,6 +6,7 @@ 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,
@@ -33,7 +34,6 @@ from services.errors.audio import (
 )
 
 logger = logging.getLogger(__name__)
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
 
 class TextToSpeechPayload(BaseModel):
@@ -47,13 +47,11 @@ class TextToSpeechVoiceQuery(BaseModel):
     language: str = Field(..., description="Language code")
 
 
-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),
-)
+class AudioTranscriptResponse(BaseModel):
+    text: str = Field(description="Transcribed text from audio")
+
+
+register_schema_models(console_ns, AudioTranscriptResponse, TextToSpeechPayload, TextToSpeechVoiceQuery)
 
 
 @console_ns.route("/apps/<uuid:app_id>/audio-to-text")
@@ -64,7 +62,7 @@ class ChatMessageAudioApi(Resource):
     @console_ns.response(
         200,
         "Audio transcription successful",
-        console_ns.model("AudioTranscriptResponse", {"text": fields.String(description="Transcribed text from audio")}),
+        console_ns.models[AudioTranscriptResponse.__name__],
     )
     @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,16 +509,19 @@ class ChatConversationApi(Resource):
                 case "created_at" | "-created_at" | _:
                     query = query.where(Conversation.created_at <= end_datetime_utc)
 
-        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)
-            )
+        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 app_model.mode == AppMode.ADVANCED_CHAT:
             query = query.where(Conversation.invoke_from != InvokeFrom.DEBUGGER)

api/controllers/console/app/generator.py

@@ -1,5 +1,4 @@
 from collections.abc import Sequence
-from typing import Any
 
 from flask_restx import Resource
 from pydantic import BaseModel, Field
@@ -12,10 +11,12 @@ 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
@@ -26,28 +27,13 @@ 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: dict[str, Any] = Field(..., alias="model_config", description="Model configuration")
+    model_config_data: ModelConfig = Field(..., alias="model_config", description="Model configuration")
     ideal_output: str = Field(default="", description="Expected ideal output")
 
 
@@ -64,6 +50,7 @@ reg(RuleCodeGeneratePayload)
 reg(RuleStructuredOutputPayload)
 reg(InstructionGeneratePayload)
 reg(InstructionTemplatePayload)
+reg(ModelConfig)
 
 
 @console_ns.route("/rule-generate")
@@ -82,12 +69,7 @@ class RuleGenerateApi(Resource):
         _, current_tenant_id = current_account_with_tenant()
 
         try:
-            rules = LLMGenerator.generate_rule_config(
-                tenant_id=current_tenant_id,
-                instruction=args.instruction,
-                model_config=args.model_config_data,
-                no_variable=args.no_variable,
-            )
+            rules = LLMGenerator.generate_rule_config(tenant_id=current_tenant_id, args=args)
         except ProviderTokenNotInitError as ex:
             raise ProviderNotInitializeError(ex.description)
         except QuotaExceededError:
@@ -118,9 +100,7 @@ class RuleCodeGenerateApi(Resource):
         try:
             code_result = LLMGenerator.generate_code(
                 tenant_id=current_tenant_id,
-                instruction=args.instruction,
-                model_config=args.model_config_data,
-                code_language=args.code_language,
+                args=args,
             )
         except ProviderTokenNotInitError as ex:
             raise ProviderNotInitializeError(ex.description)
@@ -152,8 +132,7 @@ class RuleStructuredOutputGenerateApi(Resource):
         try:
             structured_output = LLMGenerator.generate_structured_output(
                 tenant_id=current_tenant_id,
-                instruction=args.instruction,
-                model_config=args.model_config_data,
+                args=args,
             )
         except ProviderTokenNotInitError as ex:
             raise ProviderNotInitializeError(ex.description)
@@ -204,23 +183,29 @@ class InstructionGenerateApi(Resource):
                     case "llm":
                         return LLMGenerator.generate_rule_config(
                             current_tenant_id,
-                            instruction=args.instruction,
-                            model_config=args.model_config_data,
-                            no_variable=True,
+                            args=RuleGeneratePayload(
+                                instruction=args.instruction,
+                                model_config=args.model_config_data,
+                                no_variable=True,
+                            ),
                         )
                     case "agent":
                         return LLMGenerator.generate_rule_config(
                             current_tenant_id,
-                            instruction=args.instruction,
-                            model_config=args.model_config_data,
-                            no_variable=True,
+                            args=RuleGeneratePayload(
+                                instruction=args.instruction,
+                                model_config=args.model_config_data,
+                                no_variable=True,
+                            ),
                         )
                     case "code":
                         return LLMGenerator.generate_code(
                             tenant_id=current_tenant_id,
-                            instruction=args.instruction,
-                            model_config=args.model_config_data,
-                            code_language=args.language,
+                            args=RuleCodeGeneratePayload(
+                                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,6 +7,7 @@ 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,
@@ -32,10 +33,9 @@ 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,13 +90,22 @@ class FeedbackExportQuery(BaseModel):
         raise ValueError("has_comment must be a boolean value")
 
 
-def reg(cls: type[BaseModel]):
-    console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
+class AnnotationCountResponse(BaseModel):
+    count: int = Field(description="Number of annotations")
 
 
-reg(ChatMessagesQuery)
-reg(MessageFeedbackPayload)
-reg(FeedbackExportQuery)
+class SuggestedQuestionsResponse(BaseModel):
+    data: list[str] = Field(description="Suggested question")
+
+
+register_schema_models(
+    console_ns,
+    ChatMessagesQuery,
+    MessageFeedbackPayload,
+    FeedbackExportQuery,
+    AnnotationCountResponse,
+    SuggestedQuestionsResponse,
+)
 
 # Register models for flask_restx to avoid dict type issues in Swagger
 # Register in dependency order: base models first, then dependent models
@@ -198,6 +207,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,
@@ -231,7 +241,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(flat=True))  # type: ignore
+        args = ChatMessagesQuery.model_validate(request.args.to_dict())
 
         conversation = (
             db.session.query(Conversation)
@@ -290,6 +300,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)
 
@@ -356,7 +367,7 @@ class MessageAnnotationCountApi(Resource):
     @console_ns.response(
         200,
         "Annotation count retrieved successfully",
-        console_ns.model("AnnotationCountResponse", {"count": fields.Integer(description="Number of annotations")}),
+        console_ns.models[AnnotationCountResponse.__name__],
     )
     @get_app_model
     @setup_required
@@ -376,9 +387,7 @@ class MessageSuggestedQuestionApi(Resource):
     @console_ns.response(
         200,
         "Suggested questions retrieved successfully",
-        console_ns.model(
-            "SuggestedQuestionsResponse", {"data": fields.List(fields.String(description="Suggested question"))}
-        ),
+        console_ns.models[SuggestedQuestionsResponse.__name__],
     )
     @console_ns.response(404, "Message or conversation not found")
     @setup_required
@@ -428,7 +437,7 @@ class MessageFeedbackExportApi(Resource):
     @login_required
     @account_initialization_required
     def get(self, app_model):
-        args = FeedbackExportQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = FeedbackExportQuery.model_validate(request.args.to_dict())
 
         # Import the service function
         from services.feedback_service import FeedbackService
@@ -474,4 +483,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,68 @@ 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."""
+
+    @setup_required
+    @login_required
+    @account_initialization_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")
+
+        if workflow_run.tenant_id != current_user.current_tenant_id:
+            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,9 +2,11 @@ import logging
 
 import httpx
 from flask import current_app, redirect, request
-from flask_restx import Resource, fields
+from flask_restx import Resource
+from pydantic import BaseModel, Field
 
 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
 
@@ -14,6 +16,26 @@ 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(
@@ -34,10 +56,7 @@ class OAuthDataSource(Resource):
     @console_ns.response(
         200,
         "Authorization URL or internal setup success",
-        console_ns.model(
-            "OAuthDataSourceResponse",
-            {"data": fields.Raw(description="Authorization URL or 'internal' for internal setup")},
-        ),
+        console_ns.models[OAuthDataSourceResponse.__name__],
     )
     @console_ns.response(400, "Invalid provider")
     @console_ns.response(403, "Admin privileges required")
@@ -101,7 +120,7 @@ class OAuthDataSourceBinding(Resource):
     @console_ns.response(
         200,
         "Data source binding success",
-        console_ns.model("OAuthDataSourceBindingResponse", {"result": fields.String(description="Operation result")}),
+        console_ns.models[OAuthDataSourceBindingResponse.__name__],
     )
     @console_ns.response(400, "Invalid provider or code")
     def get(self, provider: str):
@@ -133,7 +152,7 @@ class OAuthDataSourceSync(Resource):
     @console_ns.response(
         200,
         "Data source sync success",
-        console_ns.model("OAuthDataSourceSyncResponse", {"result": fields.String(description="Operation result")}),
+        console_ns.models[OAuthDataSourceSyncResponse.__name__],
     )
     @console_ns.response(400, "Invalid provider or sync failed")
     @setup_required

api/controllers/console/auth/forgot_password.py

@@ -2,10 +2,11 @@ import base64
 import secrets
 
 from flask import request
-from flask_restx import Resource, fields
+from flask_restx import Resource
 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,
@@ -48,8 +49,31 @@ class ForgotPasswordResetPayload(BaseModel):
         return valid_password(value)
 
 
-for model in (ForgotPasswordSendPayload, ForgotPasswordCheckPayload, ForgotPasswordResetPayload):
-    console_ns.schema_model(model.__name__, model.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
+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,
+)
 
 
 @console_ns.route("/forgot-password")
@@ -60,14 +84,7 @@ class ForgotPasswordSendEmailApi(Resource):
     @console_ns.response(
         200,
         "Email sent successfully",
-        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.models[ForgotPasswordEmailResponse.__name__],
     )
     @console_ns.response(400, "Invalid email or rate limit exceeded")
     @setup_required
@@ -106,14 +123,7 @@ class ForgotPasswordCheckApi(Resource):
     @console_ns.response(
         200,
         "Code verified successfully",
-        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.models[ForgotPasswordCheckResponse.__name__],
     )
     @console_ns.response(400, "Invalid code or token")
     @setup_required
@@ -163,7 +173,7 @@ class ForgotPasswordResetApi(Resource):
     @console_ns.response(
         200,
         "Password reset successfully",
-        console_ns.model("ForgotPasswordResetResponse", {"result": fields.String(description="Operation result")}),
+        console_ns.models[ForgotPasswordResetResponse.__name__],
     )
     @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 grant_type == 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 payload.client_secret != oauth_provider_app.client_secret:
-                raise BadRequest("client_secret is invalid")
+                if payload.redirect_uri not in oauth_provider_app.redirect_uris:
+                    raise BadRequest("redirect_uri is invalid")
 
-            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, 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")
 
-            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,
-                }
-            )
+                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, cast
+from typing import Any, Literal, cast
 
 from flask import request
 from flask_restx import Resource, fields, marshal_with
@@ -157,9 +157,8 @@ class DataSourceApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    def patch(self, binding_id, action):
+    def patch(self, binding_id, action: Literal["enable", "disable"]):
         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)
@@ -167,23 +166,24 @@ class DataSourceApi(Resource):
         if data_source_binding is None:
             raise NotFound("Data source binding not found.")
         # enable binding
-        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.")
+        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.")
         return {"result": "success"}, 200
 
 

api/controllers/console/datasets/datasets.py

@@ -55,6 +55,7 @@ from libs.login import current_account_with_tenant, login_required
 from models import ApiToken, Dataset, Document, DocumentSegment, UploadFile
 from models.dataset import DatasetPermissionEnum
 from models.provider_ids import ModelProviderID
+from services.api_token_service import ApiTokenCache
 from services.dataset_service import DatasetPermissionService, DatasetService, DocumentService
 
 # Register models for flask_restx to avoid dict type issues in Swagger
@@ -148,6 +149,7 @@ class DatasetUpdatePayload(BaseModel):
     embedding_model: str | None = None
     embedding_model_provider: str | None = None
     retrieval_model: dict[str, Any] | None = None
+    summary_index_setting: dict[str, Any] | None = None
     partial_member_list: list[dict[str, str]] | None = None
     external_retrieval_model: dict[str, Any] | None = None
     external_knowledge_id: str | None = None
@@ -288,7 +290,14 @@ class DatasetListApi(Resource):
     @enterprise_license_required
     def get(self):
         current_user, current_tenant_id = current_account_with_tenant()
-        query = ConsoleDatasetListQuery.model_validate(request.args.to_dict())
+        # Convert query parameters to dict, handling list parameters correctly
+        query_params: dict[str, str | list[str]] = dict(request.args.to_dict())
+        # Handle ids and tag_ids as lists (Flask request.args.getlist returns list even for single value)
+        if "ids" in request.args:
+            query_params["ids"] = request.args.getlist("ids")
+        if "tag_ids" in request.args:
+            query_params["tag_ids"] = request.args.getlist("tag_ids")
+        query = ConsoleDatasetListQuery.model_validate(query_params)
         # provider = request.args.get("provider", default="vendor")
         if query.ids:
             datasets, total = DatasetService.get_datasets_by_ids(query.ids, current_tenant_id)
@@ -812,6 +821,11 @@ class DatasetApiDeleteApi(Resource):
         if key is None:
             console_ns.abort(404, message="API key not found")
 
+        # Invalidate cache before deleting from database
+        # Type assertion: key is guaranteed to be non-None here because abort() raises
+        assert key is not None  # nosec - for type checker only
+        ApiTokenCache.delete(key.token, key.type)
+
         db.session.query(ApiToken).where(ApiToken.id == api_key_id).delete()
         db.session.commit()
 

api/controllers/console/datasets/datasets_document.py

@@ -45,6 +45,7 @@ from models.dataset import DocumentPipelineExecutionLog
 from services.dataset_service import DatasetService, DocumentService
 from services.entities.knowledge_entities.knowledge_entities import KnowledgeConfig, ProcessRule, RetrievalModel
 from services.file_service import FileService
+from tasks.generate_summary_index_task import generate_summary_index_task
 
 from ..app.error import (
     ProviderModelCurrentlyNotSupportError,
@@ -103,6 +104,10 @@ class DocumentRenamePayload(BaseModel):
     name: str
 
 
+class GenerateSummaryPayload(BaseModel):
+    document_list: list[str]
+
+
 class DocumentBatchDownloadZipPayload(BaseModel):
     """Request payload for bulk downloading documents as a zip archive."""
 
@@ -125,6 +130,7 @@ register_schema_models(
     RetrievalModel,
     DocumentRetryPayload,
     DocumentRenamePayload,
+    GenerateSummaryPayload,
     DocumentBatchDownloadZipPayload,
 )
 
@@ -312,6 +318,13 @@ class DatasetDocumentListApi(Resource):
 
         paginated_documents = db.paginate(select=query, page=page, per_page=limit, max_per_page=100, error_out=False)
         documents = paginated_documents.items
+
+        DocumentService.enrich_documents_with_summary_index_status(
+            documents=documents,
+            dataset=dataset,
+            tenant_id=current_tenant_id,
+        )
+
         if fetch:
             for document in documents:
                 completed_segments = (
@@ -563,63 +576,62 @@ 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 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()
-                )
+                    if file_detail is None:
+                        raise NotFound("File not found.")
 
-                if file_detail is None:
-                    raise NotFound("File not found.")
+                    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)
 
-                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")
+                case _:
+                    raise ValueError("Data source type not support")
             indexing_runner = IndexingRunner()
             try:
                 response = indexing_runner.indexing_estimate(
@@ -797,6 +809,7 @@ class DocumentApi(DocumentResource):
                 "display_status": document.display_status,
                 "doc_form": document.doc_form,
                 "doc_language": document.doc_language,
+                "need_summary": document.need_summary if document.need_summary is not None else False,
             }
         else:
             dataset_process_rules = DatasetService.get_process_rules(dataset_id)
@@ -832,6 +845,7 @@ class DocumentApi(DocumentResource):
                 "display_status": document.display_status,
                 "doc_form": document.doc_form,
                 "doc_language": document.doc_language,
+                "need_summary": document.need_summary if document.need_summary is not None else False,
             }
 
         return response, 200
@@ -939,23 +953,24 @@ class DocumentProcessingApi(DocumentResource):
         if not current_user.is_dataset_editor:
             raise Forbidden()
 
-        if action == "pause":
-            if document.indexing_status != "indexing":
-                raise InvalidActionError("Document not in indexing state.")
+        match action:
+            case "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()
 
-        elif action == "resume":
-            if document.indexing_status not in {"paused", "error"}:
-                raise InvalidActionError("Document not in paused or error state.")
+            case "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
 
@@ -1255,3 +1270,149 @@ class DocumentPipelineExecutionLogApi(DocumentResource):
             "input_data": log.input_data,
             "datasource_node_id": log.datasource_node_id,
         }, 200
+
+
+@console_ns.route("/datasets/<uuid:dataset_id>/documents/generate-summary")
+class DocumentGenerateSummaryApi(Resource):
+    @console_ns.doc("generate_summary_for_documents")
+    @console_ns.doc(description="Generate summary index for documents")
+    @console_ns.doc(params={"dataset_id": "Dataset ID"})
+    @console_ns.expect(console_ns.models[GenerateSummaryPayload.__name__])
+    @console_ns.response(200, "Summary generation started successfully")
+    @console_ns.response(400, "Invalid request or dataset configuration")
+    @console_ns.response(403, "Permission denied")
+    @console_ns.response(404, "Dataset not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @cloud_edition_billing_rate_limit_check("knowledge")
+    def post(self, dataset_id):
+        """
+        Generate summary index for specified documents.
+
+        This endpoint checks if the dataset configuration supports summary generation
+        (indexing_technique must be 'high_quality' and summary_index_setting.enable must be true),
+        then asynchronously generates summary indexes for the provided documents.
+        """
+        current_user, _ = current_account_with_tenant()
+        dataset_id = str(dataset_id)
+
+        # Get dataset
+        dataset = DatasetService.get_dataset(dataset_id)
+        if not dataset:
+            raise NotFound("Dataset not found.")
+
+        # Check permissions
+        if not current_user.is_dataset_editor:
+            raise Forbidden()
+
+        try:
+            DatasetService.check_dataset_permission(dataset, current_user)
+        except services.errors.account.NoPermissionError as e:
+            raise Forbidden(str(e))
+
+        # Validate request payload
+        payload = GenerateSummaryPayload.model_validate(console_ns.payload or {})
+        document_list = payload.document_list
+
+        if not document_list:
+            from werkzeug.exceptions import BadRequest
+
+            raise BadRequest("document_list cannot be empty.")
+
+        # Check if dataset configuration supports summary generation
+        if dataset.indexing_technique != "high_quality":
+            raise ValueError(
+                f"Summary generation is only available for 'high_quality' indexing technique. "
+                f"Current indexing technique: {dataset.indexing_technique}"
+            )
+
+        summary_index_setting = dataset.summary_index_setting
+        if not summary_index_setting or not summary_index_setting.get("enable"):
+            raise ValueError("Summary index is not enabled for this dataset. Please enable it in the dataset settings.")
+
+        # Verify all documents exist and belong to the dataset
+        documents = DocumentService.get_documents_by_ids(dataset_id, document_list)
+
+        if len(documents) != len(document_list):
+            found_ids = {doc.id for doc in documents}
+            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
+            if document.doc_form == "qa_model":
+                logger.info("Skipping summary generation for qa_model document %s", document.id)
+                continue
+
+            # Dispatch async task
+            generate_summary_index_task.delay(dataset_id, document.id)
+            logger.info(
+                "Dispatched summary generation task for document %s in dataset %s",
+                document.id,
+                dataset_id,
+            )
+
+        return {"result": "success"}, 200
+
+
+@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/summary-status")
+class DocumentSummaryStatusApi(DocumentResource):
+    @console_ns.doc("get_document_summary_status")
+    @console_ns.doc(description="Get summary index generation status for a document")
+    @console_ns.doc(params={"dataset_id": "Dataset ID", "document_id": "Document ID"})
+    @console_ns.response(200, "Summary status retrieved successfully")
+    @console_ns.response(404, "Document not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, dataset_id, document_id):
+        """
+        Get summary index generation status for a document.
+
+        Returns:
+        - total_segments: Total number of segments in the document
+        - summary_status: Dictionary with status counts
+          - completed: Number of summaries completed
+          - generating: Number of summaries being generated
+          - error: Number of summaries with errors
+          - not_started: Number of segments without summary records
+        - summaries: List of summary records with status and content preview
+        """
+        current_user, _ = current_account_with_tenant()
+        dataset_id = str(dataset_id)
+        document_id = str(document_id)
+
+        # Get dataset
+        dataset = DatasetService.get_dataset(dataset_id)
+        if not dataset:
+            raise NotFound("Dataset not found.")
+
+        # Check permissions
+        try:
+            DatasetService.check_dataset_permission(dataset, current_user)
+        except services.errors.account.NoPermissionError as e:
+            raise Forbidden(str(e))
+
+        # Get summary status detail from service
+        from services.summary_index_service import SummaryIndexService
+
+        result = SummaryIndexService.get_document_summary_status_detail(
+            document_id=document_id,
+            dataset_id=dataset_id,
+        )
+
+        return result, 200

api/controllers/console/datasets/datasets_segments.py

@@ -41,6 +41,17 @@ from services.errors.chunk import ChildChunkIndexingError as ChildChunkIndexingS
 from tasks.batch_create_segment_to_index_task import batch_create_segment_to_index_task
 
 
+def _get_segment_with_summary(segment, dataset_id):
+    """Helper function to marshal segment and add summary information."""
+    from services.summary_index_service import SummaryIndexService
+
+    segment_dict = dict(marshal(segment, segment_fields))
+    # Query summary for this segment (only enabled summaries)
+    summary = SummaryIndexService.get_segment_summary(segment_id=segment.id, dataset_id=dataset_id)
+    segment_dict["summary"] = summary.summary_content if summary else None
+    return segment_dict
+
+
 class SegmentListQuery(BaseModel):
     limit: int = Field(default=20, ge=1, le=100)
     status: list[str] = Field(default_factory=list)
@@ -63,6 +74,7 @@ class SegmentUpdatePayload(BaseModel):
     keywords: list[str] | None = None
     regenerate_child_chunks: bool = False
     attachment_ids: list[str] | None = None
+    summary: str | None = None  # Summary content for summary index
 
 
 class BatchImportPayload(BaseModel):
@@ -181,8 +193,25 @@ class DatasetDocumentSegmentListApi(Resource):
 
         segments = db.paginate(select=query, page=page, per_page=limit, max_per_page=100, error_out=False)
 
+        # Query summaries for all segments in this page (batch query for efficiency)
+        segment_ids = [segment.id for segment in segments.items]
+        summaries = {}
+        if segment_ids:
+            from services.summary_index_service import SummaryIndexService
+
+            summary_records = SummaryIndexService.get_segments_summaries(segment_ids=segment_ids, dataset_id=dataset_id)
+            # Only include enabled summaries (already filtered by service)
+            summaries = {chunk_id: summary.summary_content for chunk_id, summary in summary_records.items()}
+
+        # Add summary to each segment
+        segments_with_summary = []
+        for segment in segments.items:
+            segment_dict = dict(marshal(segment, segment_fields))
+            segment_dict["summary"] = summaries.get(segment.id)
+            segments_with_summary.append(segment_dict)
+
         response = {
-            "data": marshal(segments.items, segment_fields),
+            "data": segments_with_summary,
             "limit": limit,
             "total": segments.total,
             "total_pages": segments.pages,
@@ -328,7 +357,7 @@ class DatasetDocumentSegmentAddApi(Resource):
         payload_dict = payload.model_dump(exclude_none=True)
         SegmentService.segment_create_args_validate(payload_dict, document)
         segment = SegmentService.create_segment(payload_dict, document, dataset)
-        return {"data": marshal(segment, segment_fields), "doc_form": document.doc_form}, 200
+        return {"data": _get_segment_with_summary(segment, dataset_id), "doc_form": document.doc_form}, 200
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/segments/<uuid:segment_id>")
@@ -390,10 +419,12 @@ class DatasetDocumentSegmentUpdateApi(Resource):
         payload = SegmentUpdatePayload.model_validate(console_ns.payload or {})
         payload_dict = payload.model_dump(exclude_none=True)
         SegmentService.segment_create_args_validate(payload_dict, document)
+
+        # Update segment (summary update with change detection is handled in SegmentService.update_segment)
         segment = SegmentService.update_segment(
             SegmentUpdateArgs.model_validate(payload.model_dump(exclude_none=True)), segment, document, dataset
         )
-        return {"data": marshal(segment, segment_fields), "doc_form": document.doc_form}, 200
+        return {"data": _get_segment_with_summary(segment, dataset_id), "doc_form": document.doc_form}, 200
 
     @setup_required
     @login_required

api/controllers/console/datasets/hit_testing.py

@@ -1,6 +1,13 @@
-from flask_restx import Resource
+from flask_restx import Resource, fields
 
 from controllers.common.schema import register_schema_model
+from fields.hit_testing_fields import (
+    child_chunk_fields,
+    document_fields,
+    files_fields,
+    hit_testing_record_fields,
+    segment_fields,
+)
 from libs.login import login_required
 
 from .. import console_ns
@@ -14,13 +21,45 @@ from ..wraps import (
 register_schema_model(console_ns, HitTestingPayload)
 
 
+def _get_or_create_model(model_name: str, field_def):
+    """Get or create a flask_restx model to avoid dict type issues in Swagger."""
+    existing = console_ns.models.get(model_name)
+    if existing is None:
+        existing = console_ns.model(model_name, field_def)
+    return existing
+
+
+# Register models for flask_restx to avoid dict type issues in Swagger
+document_model = _get_or_create_model("HitTestingDocument", document_fields)
+
+segment_fields_copy = segment_fields.copy()
+segment_fields_copy["document"] = fields.Nested(document_model)
+segment_model = _get_or_create_model("HitTestingSegment", segment_fields_copy)
+
+child_chunk_model = _get_or_create_model("HitTestingChildChunk", child_chunk_fields)
+files_model = _get_or_create_model("HitTestingFile", files_fields)
+
+hit_testing_record_fields_copy = hit_testing_record_fields.copy()
+hit_testing_record_fields_copy["segment"] = fields.Nested(segment_model)
+hit_testing_record_fields_copy["child_chunks"] = fields.List(fields.Nested(child_chunk_model))
+hit_testing_record_fields_copy["files"] = fields.List(fields.Nested(files_model))
+hit_testing_record_model = _get_or_create_model("HitTestingRecord", hit_testing_record_fields_copy)
+
+# Response model for hit testing API
+hit_testing_response_fields = {
+    "query": fields.String,
+    "records": fields.List(fields.Nested(hit_testing_record_model)),
+}
+hit_testing_response_model = _get_or_create_model("HitTestingResponse", hit_testing_response_fields)
+
+
 @console_ns.route("/datasets/<uuid:dataset_id>/hit-testing")
 class HitTestingApi(Resource, DatasetsHitTestingBase):
     @console_ns.doc("test_dataset_retrieval")
     @console_ns.doc(description="Test dataset knowledge retrieval")
     @console_ns.doc(params={"dataset_id": "Dataset ID"})
     @console_ns.expect(console_ns.models[HitTestingPayload.__name__])
-    @console_ns.response(200, "Hit testing completed successfully")
+    @console_ns.response(200, "Hit testing completed successfully", model=hit_testing_response_model)
     @console_ns.response(404, "Dataset not found")
     @console_ns.response(400, "Invalid parameters")
     @setup_required

api/controllers/console/datasets/metadata.py

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

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

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

api/controllers/console/explore/trial.py

@@ -1,8 +1,9 @@
 import logging
-from typing import Any, cast
+from typing import Any, Literal, cast
 
 from flask import request
-from flask_restx import Resource, fields, marshal, marshal_with, reqparse
+from flask_restx import Resource, fields, marshal, marshal_with
+from pydantic import BaseModel
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
 
 import services
@@ -51,7 +52,7 @@ from fields.app_fields import (
     tag_fields,
 )
 from fields.dataset_fields import dataset_fields
-from fields.member_fields import build_simple_account_model
+from fields.member_fields import simple_account_fields
 from fields.workflow_fields import (
     conversation_variable_fields,
     pipeline_variable_fields,
@@ -103,7 +104,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 = build_simple_account_model(console_ns)
+simple_account_model = get_or_create_model("SimpleAccount", simple_account_fields)
 conversation_variable_model = get_or_create_model("TrialConversationVariable", conversation_variable_fields)
 pipeline_variable_model = get_or_create_model("TrialPipelineVariable", pipeline_variable_fields)
 
@@ -117,7 +118,56 @@ 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
@@ -129,10 +179,8 @@ class TrialAppWorkflowRunApi(TrialAppResource):
         if app_mode != AppMode.WORKFLOW:
             raise NotWorkflowAppError()
 
-        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()
+        request_data = WorkflowRunRequest.model_validate(console_ns.payload)
+        args = request_data.model_dump()
         assert current_user is not None
         try:
             app_id = app_model.id
@@ -183,6 +231,7 @@ 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
@@ -190,14 +239,14 @@ class TrialChatApi(TrialAppResource):
         if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
-        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()
+        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"])
 
         args["auto_generate_name"] = False
 
@@ -320,20 +369,16 @@ 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:
-            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()
+            request_data = TextToSpeechRequest.model_validate(console_ns.payload)
 
-            message_id = args.get("message_id", None)
-            text = args.get("text", None)
-            voice = args.get("voice", None)
+            message_id = request_data.message_id
+            text = request_data.text
+            voice = request_data.voice
             if not isinstance(current_user, Account):
                 raise ValueError("current_user must be an Account instance")
 
@@ -371,19 +416,15 @@ 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()
 
-        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()
+        request_data = CompletionRequest.model_validate(console_ns.payload)
+        args = request_data.model_dump()
 
         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,87 +1,74 @@
 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)
+    password: str = Field(..., max_length=30, description="Initialization password")
 
 
-console_ns.schema_model(
-    InitValidatePayload.__name__,
-    InitValidatePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+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"],
 )
+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_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"}
+@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.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()
+    if payload.password != os.environ.get("INIT_PASSWORD"):
+        session["is_init_validated"] = False
+        raise InitValidateFailedError()
 
-        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
+    session["is_init_validated"] = True
+    return InitValidateResponse(result="success")
 
 
-def get_init_validate_status():
+def get_init_validate_status() -> bool:
     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()
+                return db_session.execute(select(DifySetup)).scalar_one_or_none() is not None
 
     return True

api/controllers/console/remote_files.py

@@ -1,7 +1,6 @@
 import urllib.parse
 
 import httpx
-from flask_restx import Resource
 from pydantic import BaseModel, Field
 
 import services
@@ -11,7 +10,7 @@ from controllers.common.errors import (
     RemoteFileUploadError,
     UnsupportedFileTypeError,
 )
-from controllers.common.schema import register_schema_models
+from controllers.fastopenapi import console_router
 from core.file import helpers as file_helpers
 from core.helper import ssrf_proxy
 from extensions.ext_database import db
@@ -19,84 +18,74 @@ 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_ns.schema_model(
-    RemoteFileUploadPayload.__name__,
-    RemoteFileUploadPayload.model_json_schema(ref_template="#/definitions/{model}"),
+@console_router.get(
+    "/remote-files/<path:url>",
+    response_model=RemoteFileInfo,
+    tags=["console"],
 )
+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_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
+@console_router.post(
+    "/remote-files/upload",
+    response_model=FileWithSignedUrl,
+    tags=["console"],
+    status_code=201,
+)
+def upload_remote_file(payload: RemoteFileUploadPayload) -> FileWithSignedUrl:
+    url = payload.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,
-            )
-        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()),
+    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,
         )
-        return payload.model_dump(mode="json"), 201
+    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()),
+    )

api/controllers/console/tag/tags.py

@@ -1,17 +1,27 @@
 from typing import Literal
 
 from flask import request
-from flask_restx import Resource, marshal_with
+from flask_restx import Namespace, Resource, fields, 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)
@@ -110,7 +120,7 @@ class TagUpdateDeleteApi(Resource):
 
         TagService.delete_tag(tag_id)
 
-        return 204
+        return "", 204
 
 
 @console_ns.route("/tag-bindings/create")

api/controllers/console/workspace/account.py

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

api/controllers/console/workspace/endpoint.py

@@ -1,9 +1,10 @@
 from typing import Any
 
 from flask import request
-from flask_restx import Resource, fields
+from flask_restx import Resource
 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
@@ -38,15 +39,53 @@ 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))
 
 
-reg(EndpointCreatePayload)
-reg(EndpointIdPayload)
-reg(EndpointUpdatePayload)
-reg(EndpointListQuery)
-reg(EndpointListForPluginQuery)
+register_schema_models(
+    console_ns,
+    EndpointCreatePayload,
+    EndpointIdPayload,
+    EndpointUpdatePayload,
+    EndpointListQuery,
+    EndpointListForPluginQuery,
+    EndpointCreateResponse,
+    EndpointListResponse,
+    PluginEndpointListResponse,
+    EndpointDeleteResponse,
+    EndpointUpdateResponse,
+    EndpointEnableResponse,
+    EndpointDisableResponse,
+)
 
 
 @console_ns.route("/workspaces/current/endpoints/create")
@@ -57,7 +96,7 @@ class EndpointCreateApi(Resource):
     @console_ns.response(
         200,
         "Endpoint created successfully",
-        console_ns.model("EndpointCreateResponse", {"success": fields.Boolean(description="Operation success")}),
+        console_ns.models[EndpointCreateResponse.__name__],
     )
     @console_ns.response(403, "Admin privileges required")
     @setup_required
@@ -91,9 +130,7 @@ class EndpointListApi(Resource):
     @console_ns.response(
         200,
         "Success",
-        console_ns.model(
-            "EndpointListResponse", {"endpoints": fields.List(fields.Raw(description="Endpoint information"))}
-        ),
+        console_ns.models[EndpointListResponse.__name__],
     )
     @setup_required
     @login_required
@@ -126,9 +163,7 @@ class EndpointListForSinglePluginApi(Resource):
     @console_ns.response(
         200,
         "Success",
-        console_ns.model(
-            "PluginEndpointListResponse", {"endpoints": fields.List(fields.Raw(description="Endpoint information"))}
-        ),
+        console_ns.models[PluginEndpointListResponse.__name__],
     )
     @setup_required
     @login_required
@@ -163,7 +198,7 @@ class EndpointDeleteApi(Resource):
     @console_ns.response(
         200,
         "Endpoint deleted successfully",
-        console_ns.model("EndpointDeleteResponse", {"success": fields.Boolean(description="Operation success")}),
+        console_ns.models[EndpointDeleteResponse.__name__],
     )
     @console_ns.response(403, "Admin privileges required")
     @setup_required
@@ -190,7 +225,7 @@ class EndpointUpdateApi(Resource):
     @console_ns.response(
         200,
         "Endpoint updated successfully",
-        console_ns.model("EndpointUpdateResponse", {"success": fields.Boolean(description="Operation success")}),
+        console_ns.models[EndpointUpdateResponse.__name__],
     )
     @console_ns.response(403, "Admin privileges required")
     @setup_required
@@ -221,7 +256,7 @@ class EndpointEnableApi(Resource):
     @console_ns.response(
         200,
         "Endpoint enabled successfully",
-        console_ns.model("EndpointEnableResponse", {"success": fields.Boolean(description="Operation success")}),
+        console_ns.models[EndpointEnableResponse.__name__],
     )
     @console_ns.response(403, "Admin privileges required")
     @setup_required
@@ -248,7 +283,7 @@ class EndpointDisableApi(Resource):
     @console_ns.response(
         200,
         "Endpoint disabled successfully",
-        console_ns.model("EndpointDisableResponse", {"success": fields.Boolean(description="Operation success")}),
+        console_ns.models[EndpointDisableResponse.__name__],
     )
     @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, fields, marshal_with
-from pydantic import BaseModel, Field
+from flask_restx import Resource
+from pydantic import BaseModel, Field, TypeAdapter
 
 import services
 from configs import dify_config
-from controllers.common.schema import get_or_create_model, register_enum_models
+from controllers.common.schema import register_enum_models, register_schema_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 account_with_role_fields, account_with_role_list_fields
+from fields.member_fields import AccountWithRole, AccountWithRoleList
 from libs.helper import extract_remote_ip
 from libs.login import current_account_with_tenant, login_required
 from models.account import Account, TenantAccountRole
@@ -69,12 +69,7 @@ reg(OwnerTransferEmailPayload)
 reg(OwnerTransferCheckPayload)
 reg(OwnerTransferPayload)
 register_enum_models(console_ns, TenantAccountRole)
-
-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)
+register_schema_models(console_ns, AccountWithRole, AccountWithRoleList)
 
 
 @console_ns.route("/workspaces/current/members")
@@ -84,13 +79,15 @@ class MemberListApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @marshal_with(account_with_role_list_model)
+    @console_ns.response(200, "Success", console_ns.models[AccountWithRoleList.__name__])
     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)
-        return {"result": "success", "accounts": members}, 200
+        member_models = TypeAdapter(list[AccountWithRole]).validate_python(members, from_attributes=True)
+        response = AccountWithRoleList(accounts=member_models)
+        return response.model_dump(mode="json"), 200
 
 
 @console_ns.route("/workspaces/current/members/invite-email")
@@ -235,13 +232,15 @@ class DatasetOperatorMemberListApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @marshal_with(account_with_role_list_model)
+    @console_ns.response(200, "Success", console_ns.models[AccountWithRoleList.__name__])
     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)
-        return {"result": "success", "accounts": members}, 200
+        member_models = TypeAdapter(list[AccountWithRole]).validate_python(members, from_attributes=True)
+        response = AccountWithRoleList(accounts=member_models)
+        return response.model_dump(mode="json"), 200
 
 
 @console_ns.route("/workspaces/current/members/send-owner-transfer-confirm-email")

api/controllers/service_api/__init__.py

@@ -34,6 +34,8 @@ from .dataset import (
     metadata,
     segment,
 )
+from .dataset.rag_pipeline import rag_pipeline_workflow
+from .end_user import end_user
 from .workspace import models
 
 __all__ = [
@@ -44,6 +46,7 @@ __all__ = [
     "conversation",
     "dataset",
     "document",
+    "end_user",
     "file",
     "file_preview",
     "hit_testing",
@@ -51,6 +54,7 @@ __all__ = [
     "message",
     "metadata",
     "models",
+    "rag_pipeline_workflow",
     "segment",
     "site",
     "workflow",

api/controllers/service_api/app/annotation.py

@@ -1,16 +1,16 @@
 from typing import Literal
 
 from flask import request
-from flask_restx import Namespace, Resource, fields
+from flask_restx import Resource
 from flask_restx.api import HTTPStatus
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, TypeAdapter
 
 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_fields, build_annotation_model
+from fields.annotation_fields import Annotation, AnnotationList
 from models.model import App
 from services.annotation_service import AppAnnotationService
 
@@ -26,7 +26,9 @@ class AnnotationReplyActionPayload(BaseModel):
     embedding_model_name: str = Field(description="Embedding model name")
 
 
-register_schema_models(service_api_ns, AnnotationCreatePayload, AnnotationReplyActionPayload)
+register_schema_models(
+    service_api_ns, AnnotationCreatePayload, AnnotationReplyActionPayload, Annotation, AnnotationList
+)
 
 
 @service_api_ns.route("/apps/annotation-reply/<string:action>")
@@ -45,10 +47,11 @@ 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()
-        if action == "enable":
-            result = AppAnnotationService.enable_app_annotation(args, app_model.id)
-        elif action == "disable":
-            result = AppAnnotationService.disable_app_annotation(app_model.id)
+        match action:
+            case "enable":
+                result = AppAnnotationService.enable_app_annotation(args, app_model.id)
+            case "disable":
+                result = AppAnnotationService.disable_app_annotation(app_model.id)
         return result, 200
 
 
@@ -82,23 +85,6 @@ 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")
@@ -109,8 +95,12 @@ 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)
@@ -118,13 +108,15 @@ 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)
-        return {
-            "data": annotation_list,
-            "has_more": len(annotation_list) == limit,
-            "limit": limit,
-            "total": total,
-            "page": page,
-        }
+        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")
 
     @service_api_ns.expect(service_api_ns.models[AnnotationCreatePayload.__name__])
     @service_api_ns.doc("create_annotation")
@@ -135,13 +127,18 @@ 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)
-        return annotation, 201
+        response = Annotation.model_validate(annotation, from_attributes=True)
+        return response.model_dump(mode="json"), HTTPStatus.CREATED
 
 
 @service_api_ns.route("/apps/annotations/<uuid:annotation_id>")
@@ -158,14 +155,19 @@ 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)
-        return annotation
+        response = Annotation.model_validate(annotation, from_attributes=True)
+        return response.model_dump(mode="json")
 
     @service_api_ns.doc("delete_annotation")
     @service_api_ns.doc(description="Delete an annotation")

api/controllers/service_api/app/completion.py

@@ -30,6 +30,7 @@ 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
@@ -52,7 +53,7 @@ class ChatRequestPayload(BaseModel):
     query: str
     files: list[dict[str, Any]] | None = None
     response_mode: Literal["blocking", "streaming"] | None = None
-    conversation_id: str | None = Field(default=None, description="Conversation UUID")
+    conversation_id: UUIDStrOrEmpty | 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,5 +1,4 @@
 from typing import Any, Literal
-from uuid import UUID
 
 from flask import request
 from flask_restx import Resource
@@ -23,12 +22,13 @@ 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: UUID | None = Field(default=None, description="Last conversation ID for pagination")
+    last_id: UUIDStrOrEmpty | 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: UUID | None = Field(default=None, description="Last variable ID for pagination")
+    last_id: UUIDStrOrEmpty | 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,6 +1,5 @@
 import logging
 from typing import Literal
-from uuid import UUID
 
 from flask import request
 from flask_restx import Resource
@@ -15,6 +14,7 @@ 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: UUID
-    first_id: UUID | None = None
+    conversation_id: UUIDStrOrEmpty
+    first_id: UUIDStrOrEmpty | 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 build_dataset_tag_fields
+from fields.tag_fields import DataSetTag
 from libs.login import current_user
 from models.account import Account
 from models.dataset import DatasetPermissionEnum
@@ -46,6 +46,7 @@ class DatasetCreatePayload(BaseModel):
     retrieval_model: RetrievalModel | None = None
     embedding_model: str | None = None
     embedding_model_provider: str | None = None
+    summary_index_setting: dict | None = None
 
 
 class DatasetUpdatePayload(BaseModel):
@@ -113,6 +114,7 @@ register_schema_models(
     TagBindingPayload,
     TagUnbindingPayload,
     DatasetListQuery,
+    DataSetTag,
 )
 
 
@@ -217,6 +219,7 @@ class DatasetListApi(DatasetApiResource):
                 embedding_model_provider=payload.embedding_model_provider,
                 embedding_model_name=payload.embedding_model,
                 retrieval_model=payload.retrieval_model,
+                summary_index_setting=payload.summary_index_setting,
             )
         except services.errors.dataset.DatasetNameDuplicateError:
             raise DatasetNameDuplicateError()
@@ -393,7 +396,7 @@ class DatasetApi(DatasetApiResource):
         try:
             if DatasetService.delete_dataset(dataset_id_str, current_user):
                 DatasetPermissionService.clear_partial_member_list(dataset_id_str)
-                return 204
+                return "", 204
             else:
                 raise NotFound("Dataset not found.")
         except services.errors.dataset.DatasetInUseError:
@@ -478,15 +481,14 @@ 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)
-
-        return tags, 200
+        tag_models = TypeAdapter(list[DataSetTag]).validate_python(tags, from_attributes=True)
+        return [tag.model_dump(mode="json") for tag in tag_models], 200
 
     @service_api_ns.expect(service_api_ns.models[TagCreatePayload.__name__])
     @service_api_ns.doc("create_dataset_tag")
@@ -498,7 +500,6 @@ 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)
@@ -508,7 +509,9 @@ class DatasetTagsApi(DatasetApiResource):
         payload = TagCreatePayload.model_validate(service_api_ns.payload or {})
         tag = TagService.save_tags({"name": payload.name, "type": "knowledge"})
 
-        response = {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": 0}
+        response = DataSetTag.model_validate(
+            {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": 0}
+        ).model_dump(mode="json")
         return response, 200
 
     @service_api_ns.expect(service_api_ns.models[TagUpdatePayload.__name__])
@@ -521,7 +524,6 @@ 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):
@@ -534,8 +536,9 @@ class DatasetTagsApi(DatasetApiResource):
 
         binding_count = TagService.get_tag_binding_count(tag_id)
 
-        response = {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": binding_count}
-
+        response = DataSetTag.model_validate(
+            {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": binding_count}
+        ).model_dump(mode="json")
         return response, 200
 
     @service_api_ns.expect(service_api_ns.models[TagDeletePayload.__name__])
@@ -554,7 +557,7 @@ class DatasetTagsApi(DatasetApiResource):
         payload = TagDeletePayload.model_validate(service_api_ns.payload or {})
         TagService.delete_tag(payload.tag_id)
 
-        return 204
+        return "", 204
 
 
 @service_api_ns.route("/datasets/tags/binding")
@@ -578,7 +581,7 @@ class DatasetTagBindingApi(DatasetApiResource):
         payload = TagBindingPayload.model_validate(service_api_ns.payload or {})
         TagService.save_tag_binding({"tag_ids": payload.tag_ids, "target_id": payload.target_id, "type": "knowledge"})
 
-        return 204
+        return "", 204
 
 
 @service_api_ns.route("/datasets/tags/unbinding")
@@ -602,7 +605,7 @@ class DatasetTagUnbindingApi(DatasetApiResource):
         payload = TagUnbindingPayload.model_validate(service_api_ns.payload or {})
         TagService.delete_tag_binding({"tag_id": payload.tag_id, "target_id": payload.target_id, "type": "knowledge"})
 
-        return 204
+        return "", 204
 
 
 @service_api_ns.route("/datasets/<uuid:dataset_id>/tags")

api/controllers/service_api/dataset/document.py

@@ -45,6 +45,7 @@ from services.entities.knowledge_entities.knowledge_entities import (
     Segmentation,
 )
 from services.file_service import FileService
+from services.summary_index_service import SummaryIndexService
 
 
 class DocumentTextCreatePayload(BaseModel):
@@ -508,6 +509,12 @@ class DocumentListApi(DatasetApiResource):
         )
         documents = paginated_documents.items
 
+        DocumentService.enrich_documents_with_summary_index_status(
+            documents=documents,
+            dataset=dataset,
+            tenant_id=tenant_id,
+        )
+
         response = {
             "data": marshal(documents, document_fields),
             "has_more": len(documents) == query_params.limit,
@@ -612,6 +619,16 @@ class DocumentApi(DatasetApiResource):
         if metadata not in self.METADATA_CHOICES:
             raise InvalidMetadataError(f"Invalid metadata value: {metadata}")
 
+        # Calculate summary_index_status if needed
+        summary_index_status = None
+        has_summary_index = dataset.summary_index_setting and dataset.summary_index_setting.get("enable") is True
+        if has_summary_index and document.need_summary is True:
+            summary_index_status = SummaryIndexService.get_document_summary_index_status(
+                document_id=document_id,
+                dataset_id=dataset_id,
+                tenant_id=tenant_id,
+            )
+
         if metadata == "only":
             response = {"id": document.id, "doc_type": document.doc_type, "doc_metadata": document.doc_metadata_details}
         elif metadata == "without":
@@ -646,6 +663,8 @@ class DocumentApi(DatasetApiResource):
                 "display_status": document.display_status,
                 "doc_form": document.doc_form,
                 "doc_language": document.doc_language,
+                "summary_index_status": summary_index_status,
+                "need_summary": document.need_summary if document.need_summary is not None else False,
             }
         else:
             dataset_process_rules = DatasetService.get_process_rules(dataset_id)
@@ -681,6 +700,8 @@ class DocumentApi(DatasetApiResource):
                 "display_status": document.display_status,
                 "doc_form": document.doc_form,
                 "doc_language": document.doc_language,
+                "summary_index_status": summary_index_status,
+                "need_summary": document.need_summary if document.need_summary is not None else False,
             }
 
         return response
@@ -725,4 +746,4 @@ class DocumentApi(DatasetApiResource):
         except services.errors.document.DocumentIndexingError:
             raise DocumentIndexingError("Cannot delete document during indexing.")
 
-        return 204
+        return "", 204

api/controllers/service_api/dataset/hit_testing.py

@@ -1,7 +1,10 @@
-from controllers.console.datasets.hit_testing_base import DatasetsHitTestingBase
+from controllers.common.schema import register_schema_model
+from controllers.console.datasets.hit_testing_base import DatasetsHitTestingBase, HitTestingPayload
 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):
@@ -15,6 +18,7 @@ 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

@@ -128,7 +128,7 @@ class DatasetMetadataServiceApi(DatasetApiResource):
         DatasetService.check_dataset_permission(dataset, current_user)
 
         MetadataService.delete_metadata(dataset_id_str, metadata_id_str)
-        return 204
+        return "", 204
 
 
 @service_api_ns.route("/datasets/<uuid:dataset_id>/metadata/built-in")
@@ -168,10 +168,11 @@ class DatasetMetadataBuiltInFieldActionServiceApi(DatasetApiResource):
             raise NotFound("Dataset not found.")
         DatasetService.check_dataset_permission(dataset, current_user)
 
-        if action == "enable":
-            MetadataService.enable_built_in_field(dataset)
-        elif action == "disable":
-            MetadataService.disable_built_in_field(dataset)
+        match action:
+            case "enable":
+                MetadataService.enable_built_in_field(dataset)
+            case "disable":
+                MetadataService.disable_built_in_field(dataset)
         return {"result": "success"}, 200
 
 

api/controllers/service_api/dataset/rag_pipeline/rag_pipeline_workflow.py

@@ -1,5 +1,3 @@
-import string
-import uuid
 from collections.abc import Generator
 from typing import Any
 
@@ -12,6 +10,7 @@ from controllers.common.errors import FilenameNotExistsError, NoFileUploadedErro
 from controllers.common.schema import register_schema_model
 from controllers.service_api import service_api_ns
 from controllers.service_api.dataset.error import PipelineRunError
+from controllers.service_api.dataset.rag_pipeline.serializers import serialize_upload_file
 from controllers.service_api.wraps import DatasetApiResource
 from core.app.apps.pipeline.pipeline_generator import PipelineGenerator
 from core.app.entities.app_invoke_entities import InvokeFrom
@@ -41,7 +40,7 @@ register_schema_model(service_api_ns, DatasourceNodeRunPayload)
 register_schema_model(service_api_ns, PipelineRunApiEntity)
 
 
-@service_api_ns.route(f"/datasets/{uuid:dataset_id}/pipeline/datasource-plugins")
+@service_api_ns.route("/datasets/<uuid:dataset_id>/pipeline/datasource-plugins")
 class DatasourcePluginsApi(DatasetApiResource):
     """Resource for datasource plugins."""
 
@@ -76,7 +75,7 @@ class DatasourcePluginsApi(DatasetApiResource):
         return datasource_plugins, 200
 
 
-@service_api_ns.route(f"/datasets/{uuid:dataset_id}/pipeline/datasource/nodes/{string:node_id}/run")
+@service_api_ns.route("/datasets/<uuid:dataset_id>/pipeline/datasource/nodes/<string:node_id>/run")
 class DatasourceNodeRunApi(DatasetApiResource):
     """Resource for datasource node run."""
 
@@ -131,7 +130,7 @@ class DatasourceNodeRunApi(DatasetApiResource):
         )
 
 
-@service_api_ns.route(f"/datasets/{uuid:dataset_id}/pipeline/run")
+@service_api_ns.route("/datasets/<uuid:dataset_id>/pipeline/run")
 class PipelineRunApi(DatasetApiResource):
     """Resource for datasource node run."""
 
@@ -232,12 +231,4 @@ class KnowledgebasePipelineFileUploadApi(DatasetApiResource):
         except services.errors.file.UnsupportedFileTypeError:
             raise UnsupportedFileTypeError()
 
-        return {
-            "id": upload_file.id,
-            "name": upload_file.name,
-            "size": upload_file.size,
-            "extension": upload_file.extension,
-            "mime_type": upload_file.mime_type,
-            "created_by": upload_file.created_by,
-            "created_at": upload_file.created_at,
-        }, 201
+        return serialize_upload_file(upload_file), 201

api/controllers/service_api/dataset/rag_pipeline/serializers.py

@@ -0,0 +1,22 @@
+"""
+Serialization helpers for Service API knowledge pipeline endpoints.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from models.model import UploadFile
+
+
+def serialize_upload_file(upload_file: UploadFile) -> dict[str, Any]:
+    return {
+        "id": upload_file.id,
+        "name": upload_file.name,
+        "size": upload_file.size,
+        "extension": upload_file.extension,
+        "mime_type": upload_file.mime_type,
+        "created_by": upload_file.created_by,
+        "created_at": upload_file.created_at.isoformat() if upload_file.created_at else None,
+    }

api/controllers/service_api/dataset/segment.py

@@ -233,7 +233,7 @@ class DatasetSegmentApi(DatasetApiResource):
         if not segment:
             raise NotFound("Segment not found.")
         SegmentService.delete_segment(segment, document, dataset)
-        return 204
+        return "", 204
 
     @service_api_ns.expect(service_api_ns.models[SegmentUpdatePayload.__name__])
     @service_api_ns.doc("update_segment")
@@ -499,7 +499,7 @@ class DatasetChildChunkApi(DatasetApiResource):
         except ChildChunkDeleteIndexServiceError as e:
             raise ChildChunkDeleteIndexError(str(e))
 
-        return 204
+        return "", 204
 
     @service_api_ns.expect(service_api_ns.models[ChildChunkUpdatePayload.__name__])
     @service_api_ns.doc("update_child_chunk")

api/controllers/service_api/end_user/__init__.py

@@ -0,0 +1,3 @@
+from . import end_user
+
+__all__ = ["end_user"]

api/controllers/service_api/end_user/end_user.py

@@ -0,0 +1,41 @@
+from uuid import UUID
+
+from flask_restx import Resource
+
+from controllers.service_api import service_api_ns
+from controllers.service_api.end_user.error import EndUserNotFoundError
+from controllers.service_api.wraps import validate_app_token
+from fields.end_user_fields import EndUserDetail
+from models.model import App
+from services.end_user_service import EndUserService
+
+
+@service_api_ns.route("/end-users/<uuid:end_user_id>")
+class EndUserApi(Resource):
+    """Resource for retrieving end user details by ID."""
+
+    @service_api_ns.doc("get_end_user")
+    @service_api_ns.doc(description="Get an end user by ID")
+    @service_api_ns.doc(
+        params={"end_user_id": "End user ID"},
+        responses={
+            200: "End user retrieved successfully",
+            401: "Unauthorized - invalid API token",
+            404: "End user not found",
+        },
+    )
+    @validate_app_token
+    def get(self, app_model: App, end_user_id: UUID):
+        """Get end user detail.
+
+        This endpoint is scoped to the current app token's tenant/app to prevent
+        cross-tenant/app access when an end-user ID is known.
+        """
+
+        end_user = EndUserService.get_end_user_by_id(
+            tenant_id=app_model.tenant_id, app_id=app_model.id, end_user_id=str(end_user_id)
+        )
+        if end_user is None:
+            raise EndUserNotFoundError()
+
+        return EndUserDetail.model_validate(end_user).model_dump(mode="json")

api/controllers/service_api/end_user/error.py

@@ -0,0 +1,7 @@
+from libs.exception import BaseHTTPException
+
+
+class EndUserNotFoundError(BaseHTTPException):
+    error_code = "end_user_not_found"
+    description = "End user not found."
+    code = 404

api/controllers/service_api/wraps.py

@@ -1,27 +1,24 @@
 import logging
 import time
 from collections.abc import Callable
-from datetime import timedelta
 from enum import StrEnum, auto
 from functools import wraps
-from typing import Concatenate, ParamSpec, TypeVar
+from typing import Concatenate, ParamSpec, TypeVar, cast
 
 from flask import current_app, request
 from flask_login import user_logged_in
 from flask_restx import Resource
 from pydantic import BaseModel
-from sqlalchemy import select, update
-from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden, NotFound, Unauthorized
 
 from enums.cloud_plan import CloudPlan
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
-from libs.datetime_utils import naive_utc_now
 from libs.login import current_user
 from models import Account, Tenant, TenantAccountJoin, TenantStatus
 from models.dataset import Dataset, RateLimitLog
 from models.model import ApiToken, App
+from services.api_token_service import ApiTokenCache, fetch_token_with_single_flight, record_token_usage
 from services.end_user_service import EndUserService
 from services.feature_service import FeatureService
 
@@ -73,14 +70,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:
-                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
+                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 not user_id and fetch_user_arg.required:
                     raise ValueError("Arg user must be provided.")
@@ -220,6 +217,8 @@ def validate_dataset_token(view: Callable[Concatenate[T, P], R] | None = None):
     def decorator(view: Callable[Concatenate[T, P], R]):
         @wraps(view)
         def decorated(*args: P.args, **kwargs: P.kwargs):
+            api_token = validate_and_get_api_token("dataset")
+
             # get url path dataset_id from positional args or kwargs
             # Flask passes URL path parameters as positional arguments
             dataset_id = None
@@ -256,12 +255,18 @@ def validate_dataset_token(view: Callable[Concatenate[T, P], R] | None = None):
             # Validate dataset if dataset_id is provided
             if dataset_id:
                 dataset_id = str(dataset_id)
-                dataset = db.session.query(Dataset).where(Dataset.id == dataset_id).first()
+                dataset = (
+                    db.session.query(Dataset)
+                    .where(
+                        Dataset.id == dataset_id,
+                        Dataset.tenant_id == api_token.tenant_id,
+                    )
+                    .first()
+                )
                 if not dataset:
                     raise NotFound("Dataset not found.")
                 if not dataset.enable_api:
                     raise Forbidden("Dataset api access is not enabled.")
-            api_token = validate_and_get_api_token("dataset")
             tenant_account_join = (
                 db.session.query(Tenant, TenantAccountJoin)
                 .where(Tenant.id == api_token.tenant_id)
@@ -296,7 +301,14 @@ def validate_dataset_token(view: Callable[Concatenate[T, P], R] | None = None):
 
 def validate_and_get_api_token(scope: str | None = None):
     """
-    Validate and get API token.
+    Validate and get API token with Redis caching.
+
+    This function uses a two-tier approach:
+    1. First checks Redis cache for the token
+    2. If not cached, queries database and caches the result
+
+    The last_used_at field is updated asynchronously via Celery task
+    to avoid blocking the request.
     """
     auth_header = request.headers.get("Authorization")
     if auth_header is None or " " not in auth_header:
@@ -308,29 +320,18 @@ def validate_and_get_api_token(scope: str | None = None):
     if auth_scheme != "bearer":
         raise Unauthorized("Authorization scheme must be 'Bearer'")
 
-    current_time = naive_utc_now()
-    cutoff_time = current_time - timedelta(minutes=1)
-    with Session(db.engine, expire_on_commit=False) as session:
-        update_stmt = (
-            update(ApiToken)
-            .where(
-                ApiToken.token == auth_token,
-                (ApiToken.last_used_at.is_(None) | (ApiToken.last_used_at < cutoff_time)),
-                ApiToken.type == scope,
-            )
-            .values(last_used_at=current_time)
-        )
-        stmt = select(ApiToken).where(ApiToken.token == auth_token, ApiToken.type == scope)
-        result = session.execute(update_stmt)
-        api_token = session.scalar(stmt)
+    # Try to get token from cache first
+    # Returns a CachedApiToken (plain Python object), not a SQLAlchemy model
+    cached_token = ApiTokenCache.get(auth_token, scope)
+    if cached_token is not None:
+        logger.debug("Token validation served from cache for scope: %s", scope)
+        # Record usage in Redis for later batch update (no Celery task per request)
+        record_token_usage(auth_token, scope)
+        return cast(ApiToken, cached_token)
 
-        if hasattr(result, "rowcount") and result.rowcount > 0:
-            session.commit()
-
-        if not api_token:
-            raise Unauthorized("Access token is invalid")
-
-    return api_token
+    # Cache miss - use Redis lock for single-flight mode
+    # This ensures only one request queries DB for the same token concurrently
+    return fetch_token_with_single_flight(auth_token, scope)
 
 
 class DatasetApiResource(Resource):

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,161 @@
+"""
+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")
+
+
+@web_ns.route("/form/human_input/<string:form_token>")
+class HumanInputFormApi(Resource):
+    """API for getting and submitting human input forms via the web app."""
+
+    # NOTE(QuantumGhost): this endpoint is unauthenticated on purpose for now.
+
+    # 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,16 +14,17 @@ class AgentConfigManager:
             agent_dict = config.get("agent_mode", {})
             agent_strategy = agent_dict.get("strategy", "cot")
 
-            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":
+            match agent_strategy:
+                case "function_call":
                     strategy = AgentEntity.Strategy.FUNCTION_CALLING
-                else:
+                case "cot" | "react":
                     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/base_app_generate_response_converter.py

@@ -79,6 +79,7 @@ class AppGenerateResponseConverter(ABC):
                         "document_name": resource["document_name"],
                         "score": resource["score"],
                         "content": resource["content"],
+                        "summary": resource.get("summary"),
                     }
                 )
             metadata["retriever_resources"] = updated_resources

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.value,
+                status=status,
                 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,
+                    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,
+                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.value
+            status = WorkflowNodeExecutionStatus.SUCCEEDED
             error_message = event.error
         elif isinstance(event, QueueNodeFailedEvent):
-            status = WorkflowNodeExecutionStatus.FAILED.value
+            status = WorkflowNodeExecutionStatus.FAILED
             error_message = event.error
         else:
-            status = WorkflowNodeExecutionStatus.EXCEPTION.value
+            status = WorkflowNodeExecutionStatus.EXCEPTION
             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.value,
+                status=WorkflowNodeExecutionStatus.RETRY,
                 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: str = args["datasource_type"]
+        datasource_type = DatasourceProviderType(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: str,
+        datasource_type: DatasourceProviderType,
         datasource_info: Mapping[str, Any],
         created_from: str,
         position: int,
@@ -668,17 +668,17 @@ class PipelineGenerator(BaseAppGenerator):
         batch: str,
         document_form: str,
     ):
-        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}")
-
+        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}")
         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: str,
+        datasource_type: DatasourceProviderType,
         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 == "online_drive":
+        if datasource_type == DatasourceProviderType.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 WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
+from core.workflow.entities.workflow_start_reason import WorkflowStartReason
+from core.workflow.enums import WorkflowExecutionStatus, 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: str
+        status: WorkflowExecutionStatus
         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: WorkflowExecutionStatus
+        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: str
+        status: WorkflowNodeExecutionStatus
         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: str
+        status: WorkflowNodeExecutionStatus
         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: str
+        status: WorkflowExecutionStatus
         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