test

.codex/skills/component-refactoring

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

.codex/skills/frontend-code-review

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

.codex/skills/frontend-testing

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

.codex/skills/orpc-contract-first

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

.github/CODEOWNERS

@@ -9,9 +9,6 @@
 # CODEOWNERS file
 /.github/CODEOWNERS @laipz8200 @crazywoola
 
-# Agents
-/.agents/skills/ @hyoban
-
 # Docs
 /docs/ @crazywoola
 
@@ -24,10 +21,6 @@
 /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)
@@ -238,9 +231,6 @@
 # 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/autofix.yml

@@ -79,6 +79,29 @@ 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/deploy-hitl.yml

@@ -4,7 +4,8 @@ on:
   workflow_run:
     workflows: ["Build and Push API & Web"]
     branches:
-      - "build/feat/hitl"
+      - "feat/hitl-frontend"
+      - "feat/hitl-backend"
     types:
       - completed
 
@@ -13,7 +14,10 @@ jobs:
     runs-on: ubuntu-latest
     if: |
       github.event.workflow_run.conclusion == 'success' &&
-      github.event.workflow_run.head_branch == 'build/feat/hitl'
+      (
+        github.event.workflow_run.head_branch == 'feat/hitl-frontend' ||
+        github.event.workflow_run.head_branch == 'feat/hitl-backend'
+      )
     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:ci
+        run: pnpm test:coverage
 
       - 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,workflow_based_app_execution",
+                "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",
                 "--loglevel",
                 "INFO"
             ],

api/.env.example

@@ -715,7 +715,6 @@ ANNOTATION_IMPORT_MAX_CONCURRENT=5
 # Sandbox expired records clean configuration
 SANDBOX_EXPIRED_RECORDS_CLEAN_GRACEFUL_PERIOD=21
 SANDBOX_EXPIRED_RECORDS_CLEAN_BATCH_SIZE=1000
-SANDBOX_EXPIRED_RECORDS_CLEAN_BATCH_MAX_INTERVAL=200
 SANDBOX_EXPIRED_RECORDS_RETENTION_DAYS=30
 SANDBOX_EXPIRED_RECORDS_CLEAN_TASK_LOCK_TTL=90000
 

api/.importlinter

@@ -52,12 +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
 
@@ -124,6 +126,11 @@ 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
@@ -133,6 +140,7 @@ 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
@@ -144,6 +152,7 @@ ignore_imports =
     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
@@ -159,6 +168,9 @@ 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
@@ -207,6 +219,7 @@ 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
@@ -222,6 +235,7 @@ ignore_imports =
     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.knowledge_retrieval.knowledge_retrieval_node -> core.rag.retrieval.retrieval_methods
     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
@@ -280,6 +294,8 @@ 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

api/.ruff.toml

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

api/.vscode/launch.json.example

@@ -54,7 +54,7 @@
                 "--loglevel",
                 "DEBUG",
                 "-Q",
-                "dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,workflow_based_app_execution,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor"
+                "dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor"
             ]
         }
     ]

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 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
+   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
    ```
 
 1. Optional: start Celery Beat (scheduled tasks, in a new terminal).

api/commands.py

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

api/configs/feature/__init__.py

@@ -93,9 +93,9 @@ class AppExecutionConfig(BaseSettings):
         default=0,
     )
 
-    HUMAN_INPUT_GLOBAL_TIMEOUT_SECONDS: PositiveInt = Field(
+    HITL_GLOBAL_TIMEOUT_SECONDS: PositiveInt = Field(
         description="Maximum seconds a workflow run can stay paused waiting for human input before global timeout.",
-        default=int(timedelta(days=7).total_seconds()),
+        default=int(timedelta(days=3).total_seconds()),
         ge=1,
     )
 
@@ -1180,16 +1180,6 @@ 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",
@@ -1344,10 +1334,6 @@ class SandboxExpiredRecordsCleanConfig(BaseSettings):
         description="Maximum number of records to process in each batch",
         default=1000,
     )
-    SANDBOX_EXPIRED_RECORDS_CLEAN_BATCH_MAX_INTERVAL: PositiveInt = Field(
-        description="Maximum interval in milliseconds between batches",
-        default=200,
-    )
     SANDBOX_EXPIRED_RECORDS_RETENTION_DAYS: PositiveInt = Field(
         description="Retention days for sandbox expired workflow_run records and message records",
         default=30,

api/configs/middleware/__init__.py

@@ -259,20 +259,11 @@ class CeleryConfig(DatabaseConfig):
         description="Password of the Redis Sentinel master.",
         default=None,
     )
-
     CELERY_SENTINEL_SOCKET_TIMEOUT: PositiveFloat | None = Field(
         description="Timeout for Redis Sentinel socket operations in seconds.",
         default=0.1,
     )
 
-    CELERY_TASK_ANNOTATIONS: dict[str, Any] | None = Field(
-        description=(
-            "Annotations for Celery tasks as a JSON mapping of task name -> options "
-            "(for example, rate limits or other task-specific settings)."
-        ),
-        default=None,
-    )
-
     @computed_field
     def CELERY_RESULT_BACKEND(self) -> str | None:
         if self.CELERY_BACKEND in ("database", "rabbitmq"):

api/constants/languages.py

@@ -21,7 +21,6 @@ language_timezone_mapping = {
     "th-TH": "Asia/Bangkok",
     "id-ID": "Asia/Jakarta",
     "ar-TN": "Africa/Tunis",
-    "nl-NL": "Europe/Amsterdam",
 }
 
 languages = list(language_timezone_mapping.keys())

api/controllers/common/schema.py

@@ -5,6 +5,8 @@ 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}"
 
 
@@ -22,9 +24,6 @@ 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/apikey.py

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

api/controllers/console/app/app.py

@@ -1,4 +1,3 @@
-import logging
 import uuid
 from datetime import datetime
 from typing import Any, Literal, TypeAlias
@@ -55,8 +54,6 @@ 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)")
@@ -502,7 +499,6 @@ 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()
@@ -514,14 +510,12 @@ class AppListApi(Resource):
                 NodeType.TRIGGER_PLUGIN,
             }
             for workflow in draft_workflows:
-                node_id = None
                 try:
-                    for node_id, node_data in workflow.walk_nodes():
+                    for _, 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,7 +6,6 @@ from pydantic import BaseModel, Field
 from werkzeug.exceptions import InternalServerError
 
 import services
-from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.error import (
     AppUnavailableError,
@@ -34,6 +33,7 @@ from services.errors.audio import (
 )
 
 logger = logging.getLogger(__name__)
+DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
 
 class TextToSpeechPayload(BaseModel):
@@ -47,11 +47,13 @@ class TextToSpeechVoiceQuery(BaseModel):
     language: str = Field(..., description="Language code")
 
 
-class AudioTranscriptResponse(BaseModel):
-    text: str = Field(description="Transcribed text from audio")
-
-
-register_schema_models(console_ns, AudioTranscriptResponse, TextToSpeechPayload, TextToSpeechVoiceQuery)
+console_ns.schema_model(
+    TextToSpeechPayload.__name__, TextToSpeechPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
+console_ns.schema_model(
+    TextToSpeechVoiceQuery.__name__,
+    TextToSpeechVoiceQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
 
 
 @console_ns.route("/apps/<uuid:app_id>/audio-to-text")
@@ -62,7 +64,7 @@ class ChatMessageAudioApi(Resource):
     @console_ns.response(
         200,
         "Audio transcription successful",
-        console_ns.models[AudioTranscriptResponse.__name__],
+        console_ns.model("AudioTranscriptResponse", {"text": fields.String(description="Transcribed text from audio")}),
     )
     @console_ns.response(400, "Bad request - No audio uploaded or unsupported type")
     @console_ns.response(413, "Audio file too large")

api/controllers/console/app/conversation.py

@@ -509,19 +509,16 @@ class ChatConversationApi(Resource):
                 case "created_at" | "-created_at" | _:
                     query = query.where(Conversation.created_at <= end_datetime_utc)
 
-        match args.annotation_status:
-            case "annotated":
-                query = query.options(joinedload(Conversation.message_annotations)).join(  # type: ignore
-                    MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id
-                )
-            case "not_annotated":
-                query = (
-                    query.outerjoin(MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id)
-                    .group_by(Conversation.id)
-                    .having(func.count(MessageAnnotation.id) == 0)
-                )
-            case "all":
-                pass
+        if args.annotation_status == "annotated":
+            query = query.options(joinedload(Conversation.message_annotations)).join(  # type: ignore
+                MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id
+            )
+        elif args.annotation_status == "not_annotated":
+            query = (
+                query.outerjoin(MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id)
+                .group_by(Conversation.id)
+                .having(func.count(MessageAnnotation.id) == 0)
+            )
 
         if app_model.mode == AppMode.ADVANCED_CHAT:
             query = query.where(Conversation.invoke_from != InvokeFrom.DEBUGGER)
@@ -599,12 +596,7 @@ def _get_conversation(app_model, conversation_id):
     db.session.execute(
         sa.update(Conversation)
         .where(Conversation.id == conversation_id, Conversation.read_at.is_(None))
-        # Keep updated_at unchanged when only marking a conversation as read.
-        .values(
-            read_at=naive_utc_now(),
-            read_account_id=current_user.id,
-            updated_at=Conversation.updated_at,
-        )
+        .values(read_at=naive_utc_now(), read_account_id=current_user.id)
     )
     db.session.commit()
     db.session.refresh(conversation)

api/controllers/console/app/generator.py

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

api/controllers/console/app/message.py

@@ -7,7 +7,6 @@ from pydantic import BaseModel, Field, field_validator
 from sqlalchemy import exists, select
 from werkzeug.exceptions import InternalServerError, NotFound
 
-from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.error import (
     CompletionRequestError,
@@ -36,6 +35,7 @@ from services.errors.message import MessageNotExistsError, SuggestedQuestionsAft
 from services.message_service import MessageService, attach_message_extra_contents
 
 logger = logging.getLogger(__name__)
+DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
 
 class ChatMessagesQuery(BaseModel):
@@ -90,22 +90,13 @@ class FeedbackExportQuery(BaseModel):
         raise ValueError("has_comment must be a boolean value")
 
 
-class AnnotationCountResponse(BaseModel):
-    count: int = Field(description="Number of annotations")
+def reg(cls: type[BaseModel]):
+    console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
 
 
-class SuggestedQuestionsResponse(BaseModel):
-    data: list[str] = Field(description="Suggested question")
-
-
-register_schema_models(
-    console_ns,
-    ChatMessagesQuery,
-    MessageFeedbackPayload,
-    FeedbackExportQuery,
-    AnnotationCountResponse,
-    SuggestedQuestionsResponse,
-)
+reg(ChatMessagesQuery)
+reg(MessageFeedbackPayload)
+reg(FeedbackExportQuery)
 
 # Register models for flask_restx to avoid dict type issues in Swagger
 # Register in dependency order: base models first, then dependent models
@@ -241,7 +232,7 @@ class ChatMessageListApi(Resource):
     @marshal_with(message_infinite_scroll_pagination_model)
     @edit_permission_required
     def get(self, app_model):
-        args = ChatMessagesQuery.model_validate(request.args.to_dict())
+        args = ChatMessagesQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
 
         conversation = (
             db.session.query(Conversation)
@@ -367,7 +358,7 @@ class MessageAnnotationCountApi(Resource):
     @console_ns.response(
         200,
         "Annotation count retrieved successfully",
-        console_ns.models[AnnotationCountResponse.__name__],
+        console_ns.model("AnnotationCountResponse", {"count": fields.Integer(description="Number of annotations")}),
     )
     @get_app_model
     @setup_required
@@ -387,7 +378,9 @@ class MessageSuggestedQuestionApi(Resource):
     @console_ns.response(
         200,
         "Suggested questions retrieved successfully",
-        console_ns.models[SuggestedQuestionsResponse.__name__],
+        console_ns.model(
+            "SuggestedQuestionsResponse", {"data": fields.List(fields.String(description="Suggested question"))}
+        ),
     )
     @console_ns.response(404, "Message or conversation not found")
     @setup_required
@@ -437,7 +430,7 @@ class MessageFeedbackExportApi(Resource):
     @login_required
     @account_initialization_required
     def get(self, app_model):
-        args = FeedbackExportQuery.model_validate(request.args.to_dict())
+        args = FeedbackExportQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
 
         # Import the service function
         from services.feedback_service import FeedbackService

api/controllers/console/app/workflow_run.py

@@ -463,9 +463,8 @@ class WorkflowRunNodeExecutionListApi(Resource):
 class ConsoleWorkflowPauseDetailsApi(Resource):
     """Console API for getting workflow pause details."""
 
-    @setup_required
-    @login_required
     @account_initialization_required
+    @login_required
     def get(self, workflow_run_id: str):
         """
         Get workflow pause details.
@@ -478,14 +477,10 @@ class ConsoleWorkflowPauseDetailsApi(Resource):
         # 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:

api/controllers/console/auth/data_source_oauth.py

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

api/controllers/console/auth/forgot_password.py

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

api/controllers/console/auth/oauth_server.py

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

api/controllers/console/datasets/data_source.py

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

api/controllers/console/datasets/datasets.py

@@ -55,7 +55,6 @@ 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
@@ -821,11 +820,6 @@ 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

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

api/controllers/console/datasets/metadata.py

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

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

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

api/controllers/console/explore/trial.py

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

api/controllers/console/init_validate.py

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

api/controllers/console/remote_files.py

@@ -11,59 +11,68 @@ from controllers.common.errors import (
     RemoteFileUploadError,
     UnsupportedFileTypeError,
 )
-from controllers.console import console_ns
+from controllers.common.schema import register_schema_models
 from core.file import helpers as file_helpers
 from core.helper import ssrf_proxy
 from extensions.ext_database import db
 from fields.file_fields import FileWithSignedUrl, RemoteFileInfo
-from libs.login import current_account_with_tenant, login_required
+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.route("/remote-files/<path:url>")
-class GetRemoteFileInfo(Resource):
-    @login_required
-    def get(self, url: str):
-        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)),
-        ).model_dump(mode="json")
+console_ns.schema_model(
+    RemoteFileUploadPayload.__name__,
+    RemoteFileUploadPayload.model_json_schema(ref_template="#/definitions/{model}"),
+)
 
 
 @console_ns.route("/remote-files/upload")
-class RemoteFileUpload(Resource):
-    @login_required
+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):
-        payload = RemoteFileUploadPayload.model_validate(console_ns.payload)
-        url = payload.url
+        args = RemoteFileUploadPayload.model_validate(console_ns.payload)
+        url = args.url
 
-        # Try to fetch remote file metadata/content first
         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:
-                # Normalize into a user-friendly error message expected by tests
                 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)
 
-        # Enforce file size limit with 400 (Bad Request) per tests' expectation
         if not FileService.is_file_size_within_limit(extension=file_info.extension, file_size=file_info.size):
-            raise FileTooLargeError()
+            raise FileTooLargeError
 
-        # Load content if needed
         content = resp.content if resp.request.method == "GET" else ssrf_proxy.get(url).content
 
         try:
@@ -80,17 +89,14 @@ class RemoteFileUpload(Resource):
         except services.errors.file.UnsupportedFileTypeError:
             raise UnsupportedFileTypeError()
 
-        # Success: return created resource with 201 status
-        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()),
-            ).model_dump(mode="json"),
-            201,
+        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()),
         )
+        return payload.model_dump(mode="json"), 201

api/controllers/console/setup.py

@@ -42,15 +42,7 @@ class SetupResponse(BaseModel):
     tags=["console"],
 )
 def get_setup_status_api() -> SetupStatusResponse:
-    """Get system setup status.
-
-    NOTE: This endpoint is unauthenticated by design.
-
-    During first-time bootstrap there is no admin account yet, so frontend initialization must be
-    able to query setup progress before any login flow exists.
-
-    Only bootstrap-safe status information should be returned by this endpoint.
-    """
+    """Get system setup status."""
     if dify_config.EDITION == "SELF_HOSTED":
         setup_status = get_setup_status()
         if setup_status and not isinstance(setup_status, bool):
@@ -69,12 +61,7 @@ def get_setup_status_api() -> SetupStatusResponse:
 )
 @only_edition_self_hosted
 def setup_system(payload: SetupRequestPayload) -> SetupResponse:
-    """Initialize system setup with admin account.
-
-    NOTE: This endpoint is unauthenticated by design for first-time bootstrap.
-    Access is restricted by deployment mode (`SELF_HOSTED`), one-time setup guards,
-    and init-password validation rather than user session authentication.
-    """
+    """Initialize system setup with admin account."""
     if get_setup_status():
         raise AlreadySetupError()
 

api/controllers/console/tag/tags.py

@@ -1,27 +1,17 @@
 from typing import Literal
 
 from flask import request
-from flask_restx import Namespace, Resource, fields, marshal_with
+from flask_restx import Resource, marshal_with
 from pydantic import BaseModel, Field
 from werkzeug.exceptions import Forbidden
 
 from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
+from fields.tag_fields import dataset_tag_fields
 from libs.login import current_account_with_tenant, login_required
 from services.tag_service import TagService
 
-dataset_tag_fields = {
-    "id": fields.String,
-    "name": fields.String,
-    "type": fields.String,
-    "binding_count": fields.String,
-}
-
-
-def build_dataset_tag_fields(api_or_ns: Namespace):
-    return api_or_ns.model("DataSetTag", dataset_tag_fields)
-
 
 class TagBasePayload(BaseModel):
     name: str = Field(description="Tag name", min_length=1, max_length=50)
@@ -120,7 +110,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,7 +12,6 @@ from sqlalchemy.orm import Session
 
 from configs import dify_config
 from constants.languages import supported_language
-from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.auth.error import (
     EmailAlreadyInUseError,
@@ -38,7 +37,7 @@ from controllers.console.wraps import (
     setup_required,
 )
 from extensions.ext_database import db
-from fields.member_fields import Account as AccountResponse
+from fields.member_fields import account_fields
 from libs.datetime_utils import naive_utc_now
 from libs.helper import EmailStr, TimestampField, extract_remote_ip, timezone
 from libs.login import current_account_with_tenant, login_required
@@ -171,12 +170,6 @@ reg(ChangeEmailSendPayload)
 reg(ChangeEmailValidityPayload)
 reg(ChangeEmailResetPayload)
 reg(CheckEmailUniquePayload)
-register_schema_models(console_ns, AccountResponse)
-
-
-def _serialize_account(account) -> dict:
-    return AccountResponse.model_validate(account, from_attributes=True).model_dump(mode="json")
-
 
 integrate_fields = {
     "provider": fields.String,
@@ -243,11 +236,11 @@ class AccountProfileApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
+    @marshal_with(account_fields)
     @enterprise_license_required
     def get(self):
         current_user, _ = current_account_with_tenant()
-        return _serialize_account(current_user)
+        return current_user
 
 
 @console_ns.route("/account/name")
@@ -256,14 +249,14 @@ class AccountNameApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
+    @marshal_with(account_fields)
     def post(self):
         current_user, _ = current_account_with_tenant()
         payload = console_ns.payload or {}
         args = AccountNamePayload.model_validate(payload)
         updated_account = AccountService.update_account(current_user, name=args.name)
 
-        return _serialize_account(updated_account)
+        return updated_account
 
 
 @console_ns.route("/account/avatar")
@@ -272,7 +265,7 @@ class AccountAvatarApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
+    @marshal_with(account_fields)
     def post(self):
         current_user, _ = current_account_with_tenant()
         payload = console_ns.payload or {}
@@ -280,7 +273,7 @@ class AccountAvatarApi(Resource):
 
         updated_account = AccountService.update_account(current_user, avatar=args.avatar)
 
-        return _serialize_account(updated_account)
+        return updated_account
 
 
 @console_ns.route("/account/interface-language")
@@ -289,7 +282,7 @@ class AccountInterfaceLanguageApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
+    @marshal_with(account_fields)
     def post(self):
         current_user, _ = current_account_with_tenant()
         payload = console_ns.payload or {}
@@ -297,7 +290,7 @@ class AccountInterfaceLanguageApi(Resource):
 
         updated_account = AccountService.update_account(current_user, interface_language=args.interface_language)
 
-        return _serialize_account(updated_account)
+        return updated_account
 
 
 @console_ns.route("/account/interface-theme")
@@ -306,7 +299,7 @@ class AccountInterfaceThemeApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
+    @marshal_with(account_fields)
     def post(self):
         current_user, _ = current_account_with_tenant()
         payload = console_ns.payload or {}
@@ -314,7 +307,7 @@ class AccountInterfaceThemeApi(Resource):
 
         updated_account = AccountService.update_account(current_user, interface_theme=args.interface_theme)
 
-        return _serialize_account(updated_account)
+        return updated_account
 
 
 @console_ns.route("/account/timezone")
@@ -323,7 +316,7 @@ class AccountTimezoneApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
+    @marshal_with(account_fields)
     def post(self):
         current_user, _ = current_account_with_tenant()
         payload = console_ns.payload or {}
@@ -331,7 +324,7 @@ class AccountTimezoneApi(Resource):
 
         updated_account = AccountService.update_account(current_user, timezone=args.timezone)
 
-        return _serialize_account(updated_account)
+        return updated_account
 
 
 @console_ns.route("/account/password")
@@ -340,7 +333,7 @@ class AccountPasswordApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
+    @marshal_with(account_fields)
     def post(self):
         current_user, _ = current_account_with_tenant()
         payload = console_ns.payload or {}
@@ -351,7 +344,7 @@ class AccountPasswordApi(Resource):
         except ServiceCurrentPasswordIncorrectError:
             raise CurrentPasswordIncorrectError()
 
-        return _serialize_account(current_user)
+        return {"result": "success"}
 
 
 @console_ns.route("/account/integrates")
@@ -627,7 +620,7 @@ class ChangeEmailResetApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
+    @marshal_with(account_fields)
     def post(self):
         payload = console_ns.payload or {}
         args = ChangeEmailResetPayload.model_validate(payload)
@@ -656,7 +649,7 @@ class ChangeEmailResetApi(Resource):
             email=normalized_new_email,
         )
 
-        return _serialize_account(updated_account)
+        return updated_account
 
 
 @console_ns.route("/account/change-email/check-email-unique")

api/controllers/console/workspace/endpoint.py

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

api/controllers/console/workspace/members.py

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

api/controllers/service_api/__init__.py

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

api/controllers/service_api/app/completion.py

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

api/controllers/service_api/app/conversation.py

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

api/controllers/service_api/app/message.py

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

api/controllers/service_api/dataset/dataset.py

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

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

api/controllers/service_api/dataset/metadata.py

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

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

@@ -1,3 +1,5 @@
+import string
+import uuid
 from collections.abc import Generator
 from typing import Any
 
@@ -10,7 +12,6 @@ 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
@@ -40,7 +41,7 @@ register_schema_model(service_api_ns, DatasourceNodeRunPayload)
 register_schema_model(service_api_ns, PipelineRunApiEntity)
 
 
-@service_api_ns.route("/datasets/<uuid:dataset_id>/pipeline/datasource-plugins")
+@service_api_ns.route(f"/datasets/{uuid:dataset_id}/pipeline/datasource-plugins")
 class DatasourcePluginsApi(DatasetApiResource):
     """Resource for datasource plugins."""
 
@@ -75,7 +76,7 @@ class DatasourcePluginsApi(DatasetApiResource):
         return datasource_plugins, 200
 
 
-@service_api_ns.route("/datasets/<uuid:dataset_id>/pipeline/datasource/nodes/<string:node_id>/run")
+@service_api_ns.route(f"/datasets/{uuid:dataset_id}/pipeline/datasource/nodes/{string:node_id}/run")
 class DatasourceNodeRunApi(DatasetApiResource):
     """Resource for datasource node run."""
 
@@ -130,7 +131,7 @@ class DatasourceNodeRunApi(DatasetApiResource):
         )
 
 
-@service_api_ns.route("/datasets/<uuid:dataset_id>/pipeline/run")
+@service_api_ns.route(f"/datasets/{uuid:dataset_id}/pipeline/run")
 class PipelineRunApi(DatasetApiResource):
     """Resource for datasource node run."""
 
@@ -231,4 +232,12 @@ class KnowledgebasePipelineFileUploadApi(DatasetApiResource):
         except services.errors.file.UnsupportedFileTypeError:
             raise UnsupportedFileTypeError()
 
-        return serialize_upload_file(upload_file), 201
+        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

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

@@ -1,22 +0,0 @@
-"""
-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

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

api/controllers/service_api/end_user/end_user.py

@@ -1,41 +0,0 @@
-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

@@ -1,7 +0,0 @@
-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,24 +1,27 @@
 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, cast
+from typing import Concatenate, ParamSpec, TypeVar
 
 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
 
@@ -70,14 +73,14 @@ def validate_app_token(view: Callable[P, R] | None = None, *, fetch_user_arg: Fe
 
             # If caller needs end-user context, attach EndUser to current_user
             if fetch_user_arg:
-                user_id = None
-                match fetch_user_arg.fetch_from:
-                    case WhereisUserArg.QUERY:
-                        user_id = request.args.get("user")
-                    case WhereisUserArg.JSON:
-                        user_id = request.get_json().get("user")
-                    case WhereisUserArg.FORM:
-                        user_id = request.form.get("user")
+                if fetch_user_arg.fetch_from == WhereisUserArg.QUERY:
+                    user_id = request.args.get("user")
+                elif fetch_user_arg.fetch_from == WhereisUserArg.JSON:
+                    user_id = request.get_json().get("user")
+                elif fetch_user_arg.fetch_from == WhereisUserArg.FORM:
+                    user_id = request.form.get("user")
+                else:
+                    user_id = None
 
                 if not user_id and fetch_user_arg.required:
                     raise ValueError("Arg user must be provided.")
@@ -217,8 +220,6 @@ 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
@@ -255,18 +256,12 @@ 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,
-                        Dataset.tenant_id == api_token.tenant_id,
-                    )
-                    .first()
-                )
+                dataset = db.session.query(Dataset).where(Dataset.id == dataset_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)
@@ -301,14 +296,7 @@ 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 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.
+    Validate and get API token.
     """
     auth_header = request.headers.get("Authorization")
     if auth_header is None or " " not in auth_header:
@@ -320,18 +308,29 @@ def validate_and_get_api_token(scope: str | None = None):
     if auth_scheme != "bearer":
         raise Unauthorized("Authorization scheme must be 'Bearer'")
 
-    # 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)
+    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)
 
-    # 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)
+        if hasattr(result, "rowcount") and result.rowcount > 0:
+            session.commit()
+
+        if not api_token:
+            raise Unauthorized("Access token is invalid")
+
+    return api_token
 
 
 class DatasetApiResource(Resource):

api/controllers/web/human_input_form.py

@@ -65,12 +65,15 @@ def _jsonify_form_definition(form: Form, site_payload: dict | None = None) -> Re
     return Response(json.dumps(payload, ensure_ascii=False), mimetype="application/json")
 
 
+# TODO(QuantumGhost): disable authorization for web app
+# form api temporarily
+
+
 @web_ns.route("/form/human_input/<string:form_token>")
+# class HumanInputFormApi(WebApiResource):
 class HumanInputFormApi(Resource):
     """API for getting and submitting human input forms via the web app."""
 
-    # 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):
         """

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

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

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

@@ -268,7 +268,7 @@ class WorkflowResponseConverter:
             data=WorkflowFinishStreamResponse.Data(
                 id=run_id,
                 workflow_id=workflow_id,
-                status=status,
+                status=status.value,
                 outputs=encoded_outputs,
                 error=error,
                 elapsed_time=elapsed_time,
@@ -346,7 +346,7 @@ class WorkflowResponseConverter:
                     paused_nodes=list(event.paused_nodes),
                     outputs=encoded_outputs,
                     reasons=pause_reasons,
-                    status=WorkflowExecutionStatus.PAUSED,
+                    status=WorkflowExecutionStatus.PAUSED.value,
                     created_at=int(started_at.timestamp()),
                     elapsed_time=elapsed_time,
                     total_tokens=graph_runtime_state.total_tokens,
@@ -422,7 +422,7 @@ class WorkflowResponseConverter:
             data=WorkflowFinishStreamResponse.Data(
                 id=run_id,
                 workflow_id=workflow_run.workflow_id,
-                status=workflow_run.status,
+                status=workflow_run.status.value,
                 outputs=encoded_outputs,
                 error=workflow_run.error,
                 elapsed_time=elapsed_time,
@@ -512,13 +512,13 @@ class WorkflowResponseConverter:
         metadata = self._merge_metadata(event.execution_metadata, snapshot)
 
         if isinstance(event, QueueNodeSucceededEvent):
-            status = WorkflowNodeExecutionStatus.SUCCEEDED
+            status = WorkflowNodeExecutionStatus.SUCCEEDED.value
             error_message = event.error
         elif isinstance(event, QueueNodeFailedEvent):
-            status = WorkflowNodeExecutionStatus.FAILED
+            status = WorkflowNodeExecutionStatus.FAILED.value
             error_message = event.error
         else:
-            status = WorkflowNodeExecutionStatus.EXCEPTION
+            status = WorkflowNodeExecutionStatus.EXCEPTION.value
             error_message = event.error
 
         return NodeFinishStreamResponse(
@@ -585,7 +585,7 @@ class WorkflowResponseConverter:
                 process_data_truncated=process_data_truncated,
                 outputs=outputs,
                 outputs_truncated=outputs_truncated,
-                status=WorkflowNodeExecutionStatus.RETRY,
+                status=WorkflowNodeExecutionStatus.RETRY.value,
                 error=event.error,
                 elapsed_time=elapsed_time,
                 execution_metadata=metadata,

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

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

api/core/app/apps/streaming_utils.py

@@ -34,7 +34,7 @@ def stream_topic_events(
             on_subscribe()
         while True:
             try:
-                msg = sub.receive(timeout=1)
+                msg = sub.receive(timeout=0.1)
             except SubscriptionClosedError:
                 return
             if msg is None:

api/core/app/entities/task_entities.py

@@ -8,7 +8,7 @@ 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.entities.workflow_start_reason import WorkflowStartReason
-from core.workflow.enums import WorkflowExecutionStatus, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
+from core.workflow.enums import WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
 from core.workflow.nodes.human_input.entities import FormInput, UserAction
 
 
@@ -231,7 +231,7 @@ class WorkflowFinishStreamResponse(StreamResponse):
 
         id: str
         workflow_id: str
-        status: WorkflowExecutionStatus
+        status: str
         outputs: Mapping[str, Any] | None = None
         error: str | None = None
         elapsed_time: float
@@ -262,7 +262,7 @@ class WorkflowPauseStreamResponse(StreamResponse):
         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
+        status: str
         created_at: int
         elapsed_time: float
         total_tokens: int
@@ -398,7 +398,7 @@ class NodeFinishStreamResponse(StreamResponse):
         process_data_truncated: bool = False
         outputs: Mapping[str, Any] | None = None
         outputs_truncated: bool = True
-        status: WorkflowNodeExecutionStatus
+        status: str
         error: str | None = None
         elapsed_time: float
         execution_metadata: Mapping[WorkflowNodeExecutionMetadataKey, Any] | None = None
@@ -462,7 +462,7 @@ class NodeRetryStreamResponse(StreamResponse):
         process_data_truncated: bool = False
         outputs: Mapping[str, Any] | None = None
         outputs_truncated: bool = False
-        status: WorkflowNodeExecutionStatus
+        status: str
         error: str | None = None
         elapsed_time: float
         execution_metadata: Mapping[WorkflowNodeExecutionMetadataKey, Any] | None = None
@@ -806,7 +806,7 @@ class WorkflowAppBlockingResponse(AppBlockingResponse):
 
         id: str
         workflow_id: str
-        status: WorkflowExecutionStatus
+        status: str
         outputs: Mapping[str, Any] | None = None
         error: str | None = None
         elapsed_time: float

api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py

@@ -45,8 +45,6 @@ from core.app.entities.task_entities import (
 from core.app.task_pipeline.based_generate_task_pipeline import BasedGenerateTaskPipeline
 from core.app.task_pipeline.message_cycle_manager import MessageCycleManager
 from core.base.tts import AppGeneratorTTSPublisher, AudioTrunk
-from core.file import helpers as file_helpers
-from core.file.enums import FileTransferMethod
 from core.model_manager import ModelInstance
 from core.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk, LLMResultChunkDelta, LLMUsage
 from core.model_runtime.entities.message_entities import (
@@ -58,11 +56,10 @@ from core.ops.entities.trace_entity import TraceTaskName
 from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
 from core.prompt.utils.prompt_message_util import PromptMessageUtil
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
-from core.tools.signature import sign_tool_file
 from events.message_event import message_was_created
 from extensions.ext_database import db
 from libs.datetime_utils import naive_utc_now
-from models.model import AppMode, Conversation, Message, MessageAgentThought, MessageFile, UploadFile
+from models.model import AppMode, Conversation, Message, MessageAgentThought
 
 logger = logging.getLogger(__name__)
 
@@ -157,7 +154,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
                             id=self._message_id,
                             mode=self._conversation_mode,
                             message_id=self._message_id,
-                            answer=self._task_state.llm_result.message.get_text_content(),
+                            answer=cast(str, self._task_state.llm_result.message.content),
                             created_at=self._message_created_at,
                             **extras,
                         ),
@@ -170,7 +167,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
                             mode=self._conversation_mode,
                             conversation_id=self._conversation_id,
                             message_id=self._message_id,
-                            answer=self._task_state.llm_result.message.get_text_content(),
+                            answer=cast(str, self._task_state.llm_result.message.content),
                             created_at=self._message_created_at,
                             **extras,
                         ),
@@ -283,7 +280,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
 
                 # handle output moderation
                 output_moderation_answer = self.handle_output_moderation_when_task_finished(
-                    self._task_state.llm_result.message.get_text_content()
+                    cast(str, self._task_state.llm_result.message.content)
                 )
                 if output_moderation_answer:
                     self._task_state.llm_result.message.content = output_moderation_answer
@@ -397,7 +394,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
         message.message_unit_price = usage.prompt_unit_price
         message.message_price_unit = usage.prompt_price_unit
         message.answer = (
-            PromptTemplateParser.remove_template_variables(llm_result.message.get_text_content().strip())
+            PromptTemplateParser.remove_template_variables(cast(str, llm_result.message.content).strip())
             if llm_result.message.content
             else ""
         )
@@ -466,85 +463,6 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
             metadata=metadata_dict,
         )
 
-    def _record_files(self):
-        with Session(db.engine, expire_on_commit=False) as session:
-            message_files = session.scalars(select(MessageFile).where(MessageFile.message_id == self._message_id)).all()
-            if not message_files:
-                return None
-
-            files_list = []
-            upload_file_ids = [
-                mf.upload_file_id
-                for mf in message_files
-                if mf.transfer_method == FileTransferMethod.LOCAL_FILE and mf.upload_file_id
-            ]
-            upload_files_map = {}
-            if upload_file_ids:
-                upload_files = session.scalars(select(UploadFile).where(UploadFile.id.in_(upload_file_ids))).all()
-                upload_files_map = {uf.id: uf for uf in upload_files}
-
-            for message_file in message_files:
-                upload_file = None
-                if message_file.transfer_method == FileTransferMethod.LOCAL_FILE and message_file.upload_file_id:
-                    upload_file = upload_files_map.get(message_file.upload_file_id)
-
-                url = None
-                filename = "file"
-                mime_type = "application/octet-stream"
-                size = 0
-                extension = ""
-
-                if message_file.transfer_method == FileTransferMethod.REMOTE_URL:
-                    url = message_file.url
-                    if message_file.url:
-                        filename = message_file.url.split("/")[-1].split("?")[0]  # Remove query params
-                elif message_file.transfer_method == FileTransferMethod.LOCAL_FILE:
-                    if upload_file:
-                        url = file_helpers.get_signed_file_url(upload_file_id=str(upload_file.id))
-                        filename = upload_file.name
-                        mime_type = upload_file.mime_type or "application/octet-stream"
-                        size = upload_file.size or 0
-                        extension = f".{upload_file.extension}" if upload_file.extension else ""
-                    elif message_file.upload_file_id:
-                        # Fallback: generate URL even if upload_file not found
-                        url = file_helpers.get_signed_file_url(upload_file_id=str(message_file.upload_file_id))
-                elif message_file.transfer_method == FileTransferMethod.TOOL_FILE and message_file.url:
-                    # For tool files, use URL directly if it's HTTP, otherwise sign it
-                    if message_file.url.startswith("http"):
-                        url = message_file.url
-                        filename = message_file.url.split("/")[-1].split("?")[0]
-                    else:
-                        # Extract tool file id and extension from URL
-                        url_parts = message_file.url.split("/")
-                        if url_parts:
-                            file_part = url_parts[-1].split("?")[0]  # Remove query params first
-                            # Use rsplit to correctly handle filenames with multiple dots
-                            if "." in file_part:
-                                tool_file_id, ext = file_part.rsplit(".", 1)
-                                extension = f".{ext}"
-                            else:
-                                tool_file_id = file_part
-                                extension = ".bin"
-                            url = sign_tool_file(tool_file_id=tool_file_id, extension=extension)
-                            filename = file_part
-
-                transfer_method_value = message_file.transfer_method
-                remote_url = message_file.url if message_file.transfer_method == FileTransferMethod.REMOTE_URL else ""
-                file_dict = {
-                    "related_id": message_file.id,
-                    "extension": extension,
-                    "filename": filename,
-                    "size": size,
-                    "mime_type": mime_type,
-                    "transfer_method": transfer_method_value,
-                    "type": message_file.type,
-                    "url": url or "",
-                    "upload_file_id": message_file.upload_file_id or message_file.id,
-                    "remote_url": remote_url,
-                }
-                files_list.append(file_dict)
-            return files_list or None
-
     def _agent_message_to_stream_response(self, answer: str, message_id: str) -> AgentMessageStreamResponse:
         """
         Agent message to stream response.

api/core/app/task_pipeline/message_cycle_manager.py

@@ -64,13 +64,7 @@ class MessageCycleManager:
 
         # Use SQLAlchemy 2.x style session.scalar(select(...))
         with session_factory.create_session() as session:
-            message_file = session.scalar(
-                select(MessageFile)
-                .where(
-                    MessageFile.message_id == message_id,
-                )
-                .where(MessageFile.belongs_to == "assistant")
-            )
+            message_file = session.scalar(select(MessageFile).where(MessageFile.message_id == message_id))
 
         if message_file:
             self._message_has_file.add(message_id)

api/core/app/workflow/node_factory.py

@@ -4,20 +4,17 @@ from typing import TYPE_CHECKING, final
 from typing_extensions import override
 
 from configs import dify_config
-from core.file.file_manager import file_manager
+from core.file import file_manager
+from core.helper import ssrf_proxy
 from core.helper.code_executor.code_executor import CodeExecutor
 from core.helper.code_executor.code_node_provider import CodeNodeProvider
-from core.helper.ssrf_proxy import ssrf_proxy
-from core.rag.retrieval.dataset_retrieval import DatasetRetrieval
 from core.tools.tool_file_manager import ToolFileManager
-from core.workflow.entities.graph_config import NodeConfigDict
 from core.workflow.enums import NodeType
-from core.workflow.graph.graph import NodeFactory
+from core.workflow.graph import NodeFactory
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.code.code_node import CodeNode
 from core.workflow.nodes.code.limits import CodeNodeLimits
 from core.workflow.nodes.http_request.node import HttpRequestNode
-from core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node import KnowledgeRetrievalNode
 from core.workflow.nodes.node_mapping import LATEST_VERSION, NODE_TYPE_CLASSES_MAPPING
 from core.workflow.nodes.protocols import FileManagerProtocol, HttpClientProtocol
 from core.workflow.nodes.template_transform.template_renderer import (
@@ -25,6 +22,7 @@ from core.workflow.nodes.template_transform.template_renderer import (
     Jinja2TemplateRenderer,
 )
 from core.workflow.nodes.template_transform.template_transform_node import TemplateTransformNode
+from libs.typing import is_str, is_str_dict
 
 if TYPE_CHECKING:
     from core.workflow.entities import GraphInitParams
@@ -49,10 +47,9 @@ class DifyNodeFactory(NodeFactory):
         code_providers: Sequence[type[CodeNodeProvider]] | None = None,
         code_limits: CodeNodeLimits | None = None,
         template_renderer: Jinja2TemplateRenderer | None = None,
-        template_transform_max_output_length: int | None = None,
-        http_request_http_client: HttpClientProtocol | None = None,
+        http_request_http_client: HttpClientProtocol = ssrf_proxy,
         http_request_tool_file_manager_factory: Callable[[], ToolFileManager] = ToolFileManager,
-        http_request_file_manager: FileManagerProtocol | None = None,
+        http_request_file_manager: FileManagerProtocol = file_manager,
     ) -> None:
         self.graph_init_params = graph_init_params
         self.graph_runtime_state = graph_runtime_state
@@ -71,16 +68,12 @@ class DifyNodeFactory(NodeFactory):
             max_object_array_length=dify_config.CODE_MAX_OBJECT_ARRAY_LENGTH,
         )
         self._template_renderer = template_renderer or CodeExecutorJinja2TemplateRenderer()
-        self._template_transform_max_output_length = (
-            template_transform_max_output_length or dify_config.TEMPLATE_TRANSFORM_MAX_LENGTH
-        )
-        self._http_request_http_client = http_request_http_client or ssrf_proxy
+        self._http_request_http_client = http_request_http_client
         self._http_request_tool_file_manager_factory = http_request_tool_file_manager_factory
-        self._http_request_file_manager = http_request_file_manager or file_manager
-        self._rag_retrieval = DatasetRetrieval()
+        self._http_request_file_manager = http_request_file_manager
 
     @override
-    def create_node(self, node_config: NodeConfigDict) -> Node:
+    def create_node(self, node_config: dict[str, object]) -> Node:
         """
         Create a Node instance from node configuration data using the traditional mapping.
 
@@ -89,14 +82,23 @@ class DifyNodeFactory(NodeFactory):
         :raises ValueError: if node type is unknown or configuration is invalid
         """
         # Get node_id from config
-        node_id = node_config["id"]
+        node_id = node_config.get("id")
+        if not is_str(node_id):
+            raise ValueError("Node config missing id")
 
         # Get node type from config
-        node_data = node_config["data"]
+        node_data = node_config.get("data", {})
+        if not is_str_dict(node_data):
+            raise ValueError(f"Node {node_id} missing data information")
+
+        node_type_str = node_data.get("type")
+        if not is_str(node_type_str):
+            raise ValueError(f"Node {node_id} missing or invalid type information")
+
         try:
-            node_type = NodeType(node_data["type"])
+            node_type = NodeType(node_type_str)
         except ValueError:
-            raise ValueError(f"Unknown node type: {node_data['type']}")
+            raise ValueError(f"Unknown node type: {node_type_str}")
 
         # Get node class
         node_mapping = NODE_TYPE_CLASSES_MAPPING.get(node_type)
@@ -129,7 +131,6 @@ class DifyNodeFactory(NodeFactory):
                 graph_init_params=self.graph_init_params,
                 graph_runtime_state=self.graph_runtime_state,
                 template_renderer=self._template_renderer,
-                max_output_length=self._template_transform_max_output_length,
             )
 
         if node_type == NodeType.HTTP_REQUEST:
@@ -143,15 +144,6 @@ class DifyNodeFactory(NodeFactory):
                 file_manager=self._http_request_file_manager,
             )
 
-        if node_type == NodeType.KNOWLEDGE_RETRIEVAL:
-            return KnowledgeRetrievalNode(
-                id=node_id,
-                config=node_config,
-                graph_init_params=self.graph_init_params,
-                graph_runtime_state=self.graph_runtime_state,
-                rag_retrieval=self._rag_retrieval,
-            )
-
         return node_class(
             id=node_id,
             config=node_config,

api/core/file/file_manager.py

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

api/core/helper/code_executor/code_node_provider.py

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

api/core/helper/marketplace.py

@@ -6,8 +6,7 @@ from yarl import URL
 
 from configs import dify_config
 from core.helper.download import download_with_size_limit
-from core.plugin.entities.marketplace import MarketplacePluginDeclaration, MarketplacePluginSnapshot
-from extensions.ext_redis import redis_client
+from core.plugin.entities.marketplace import MarketplacePluginDeclaration
 
 marketplace_api_url = URL(str(dify_config.MARKETPLACE_API_URL))
 logger = logging.getLogger(__name__)
@@ -44,37 +43,28 @@ def batch_fetch_plugin_by_ids(plugin_ids: list[str]) -> list[dict]:
     return data.get("data", {}).get("plugins", [])
 
 
+def batch_fetch_plugin_manifests_ignore_deserialization_error(
+    plugin_ids: list[str],
+) -> Sequence[MarketplacePluginDeclaration]:
+    if len(plugin_ids) == 0:
+        return []
+
+    url = str(marketplace_api_url / "api/v1/plugins/batch")
+    response = httpx.post(url, json={"plugin_ids": plugin_ids}, headers={"X-Dify-Version": dify_config.project.version})
+    response.raise_for_status()
+    result: list[MarketplacePluginDeclaration] = []
+    for plugin in response.json()["data"]["plugins"]:
+        try:
+            result.append(MarketplacePluginDeclaration.model_validate(plugin))
+        except Exception:
+            logger.exception(
+                "Failed to deserialize marketplace plugin manifest for %s", plugin.get("plugin_id", "unknown")
+            )
+
+    return result
+
+
 def record_install_plugin_event(plugin_unique_identifier: str):
     url = str(marketplace_api_url / "api/v1/stats/plugins/install_count")
     response = httpx.post(url, json={"unique_identifier": plugin_unique_identifier})
     response.raise_for_status()
-
-
-def fetch_global_plugin_manifest(cache_key_prefix: str, cache_ttl: int) -> None:
-    """
-    Fetch all plugin manifests from marketplace and cache them in Redis.
-    This should be called once per check cycle to populate the instance-level cache.
-
-    Args:
-        cache_key_prefix: Redis key prefix for caching plugin manifests
-        cache_ttl: Cache TTL in seconds
-
-    Raises:
-        httpx.HTTPError: If the HTTP request fails
-        Exception: If any other error occurs during fetching or caching
-    """
-    url = str(marketplace_api_url / "api/v1/dist/plugins/manifest.json")
-    response = httpx.get(url, headers={"X-Dify-Version": dify_config.project.version}, timeout=30)
-    response.raise_for_status()
-
-    raw_json = response.json()
-    plugins_data = raw_json.get("plugins", [])
-
-    # Parse and cache all plugin snapshots
-    for plugin_data in plugins_data:
-        plugin_snapshot = MarketplacePluginSnapshot.model_validate(plugin_data)
-        redis_client.setex(
-            name=f"{cache_key_prefix}{plugin_snapshot.plugin_id}",
-            time=cache_ttl,
-            value=plugin_snapshot.model_dump_json(),
-        )

api/core/helper/ssrf_proxy.py

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

api/core/indexing_runner.py

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

api/core/llm_generator/entities.py

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

api/core/llm_generator/llm_generator.py

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

api/core/llm_generator/prompts.py

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

api/core/model_runtime/entities/defaults.py

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

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

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

api/core/plugin/backwards_invocation/app.py

@@ -12,7 +12,6 @@ from core.app.apps.chat.app_generator import ChatAppGenerator
 from core.app.apps.completion.app_generator import CompletionAppGenerator
 from core.app.apps.workflow.app_generator import WorkflowAppGenerator
 from core.app.entities.app_invoke_entities import InvokeFrom
-from core.app.layers.pause_state_persist_layer import PauseStateLayerConfig
 from core.plugin.backwards_invocation.base import BaseBackwardsInvocation
 from extensions.ext_database import db
 from models import Account
@@ -103,11 +102,6 @@ class PluginAppBackwardsInvocation(BaseBackwardsInvocation):
             if not workflow:
                 raise ValueError("unexpected app type")
 
-            pause_config = PauseStateLayerConfig(
-                session_factory=db.engine,
-                state_owner_user_id=workflow.created_by,
-            )
-
             return AdvancedChatAppGenerator().generate(
                 app_model=app,
                 workflow=workflow,
@@ -121,7 +115,6 @@ class PluginAppBackwardsInvocation(BaseBackwardsInvocation):
                 invoke_from=InvokeFrom.SERVICE_API,
                 workflow_run_id=str(uuid.uuid4()),
                 streaming=stream,
-                pause_state_config=pause_config,
             )
         elif app.mode == AppMode.AGENT_CHAT:
             return AgentChatAppGenerator().generate(
@@ -168,11 +161,6 @@ class PluginAppBackwardsInvocation(BaseBackwardsInvocation):
         if not workflow:
             raise ValueError("unexpected app type")
 
-        pause_config = PauseStateLayerConfig(
-            session_factory=db.engine,
-            state_owner_user_id=workflow.created_by,
-        )
-
         return WorkflowAppGenerator().generate(
             app_model=app,
             workflow=workflow,
@@ -181,7 +169,6 @@ class PluginAppBackwardsInvocation(BaseBackwardsInvocation):
             invoke_from=InvokeFrom.SERVICE_API,
             streaming=stream,
             call_depth=1,
-            pause_state_config=pause_config,
         )
 
     @classmethod

api/core/plugin/entities/marketplace.py

@@ -1,4 +1,4 @@
-from pydantic import BaseModel, Field, computed_field, model_validator
+from pydantic import BaseModel, Field, model_validator
 
 from core.model_runtime.entities.provider_entities import ProviderEntity
 from core.plugin.entities.endpoint import EndpointProviderDeclaration
@@ -48,15 +48,3 @@ class MarketplacePluginDeclaration(BaseModel):
         if "tool" in data and not data["tool"]:
             del data["tool"]
         return data
-
-
-class MarketplacePluginSnapshot(BaseModel):
-    org: str
-    name: str
-    latest_version: str
-    latest_package_identifier: str
-    latest_package_url: str
-
-    @computed_field
-    def plugin_id(self) -> str:
-        return f"{self.org}/{self.name}"

api/core/rag/index_processor/index_processor_base.py

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

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

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

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

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

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

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

api/core/rag/retrieval/dataset_retrieval.py

@@ -1,15 +1,13 @@
 import json
-import logging
 import math
 import re
 import threading
-import time
 from collections import Counter, defaultdict
 from collections.abc import Generator, Mapping
 from typing import Any, Union, cast
 
 from flask import Flask, current_app
-from sqlalchemy import and_, func, literal, or_, select
+from sqlalchemy import and_, literal, or_, select
 from sqlalchemy.orm import Session
 
 from core.app.app_config.entities import (
@@ -20,7 +18,6 @@ from core.app.app_config.entities import (
 )
 from core.app.entities.app_invoke_entities import InvokeFrom, ModelConfigWithCredentialsEntity
 from core.callback_handler.index_tool_callback_handler import DatasetIndexToolCallbackHandler
-from core.db.session_factory import session_factory
 from core.entities.agent_entities import PlanningStrategy
 from core.entities.model_entities import ModelStatus
 from core.file import File, FileTransferMethod, FileType
@@ -61,30 +58,12 @@ from core.rag.retrieval.template_prompts import (
 )
 from core.tools.signature import sign_upload_file
 from core.tools.utils.dataset_retriever.dataset_retriever_base_tool import DatasetRetrieverBaseTool
-from core.workflow.nodes.knowledge_retrieval import exc
-from core.workflow.repositories.rag_retrieval_protocol import (
-    KnowledgeRetrievalRequest,
-    Source,
-    SourceChildChunk,
-    SourceMetadata,
-)
 from extensions.ext_database import db
-from extensions.ext_redis import redis_client
 from libs.json_in_md_parser import parse_and_check_json_markdown
 from models import UploadFile
-from models.dataset import (
-    ChildChunk,
-    Dataset,
-    DatasetMetadata,
-    DatasetQuery,
-    DocumentSegment,
-    RateLimitLog,
-    SegmentAttachmentBinding,
-)
+from models.dataset import ChildChunk, Dataset, DatasetMetadata, DatasetQuery, DocumentSegment, SegmentAttachmentBinding
 from models.dataset import Document as DatasetDocument
-from models.dataset import Document as DocumentModel
 from services.external_knowledge_service import ExternalDatasetService
-from services.feature_service import FeatureService
 
 default_retrieval_model: dict[str, Any] = {
     "search_method": RetrievalMethod.SEMANTIC_SEARCH,
@@ -94,8 +73,6 @@ default_retrieval_model: dict[str, Any] = {
     "score_threshold_enabled": False,
 }
 
-logger = logging.getLogger(__name__)
-
 
 class DatasetRetrieval:
     def __init__(self, application_generate_entity=None):
@@ -114,233 +91,6 @@ class DatasetRetrieval:
         else:
             self._llm_usage = self._llm_usage.plus(usage)
 
-    def knowledge_retrieval(self, request: KnowledgeRetrievalRequest) -> list[Source]:
-        self._check_knowledge_rate_limit(request.tenant_id)
-        available_datasets = self._get_available_datasets(request.tenant_id, request.dataset_ids)
-        available_datasets_ids = [i.id for i in available_datasets]
-        if not available_datasets_ids:
-            return []
-
-        if not request.query:
-            return []
-
-        metadata_filter_document_ids, metadata_condition = None, None
-
-        if request.metadata_filtering_mode != "disabled":
-            # Convert workflow layer types to app_config layer types
-            if not request.metadata_model_config:
-                raise ValueError("metadata_model_config is required for this method")
-
-            app_metadata_model_config = ModelConfig.model_validate(request.metadata_model_config.model_dump())
-
-            app_metadata_filtering_conditions = None
-            if request.metadata_filtering_conditions is not None:
-                app_metadata_filtering_conditions = MetadataFilteringCondition.model_validate(
-                    request.metadata_filtering_conditions.model_dump()
-                )
-
-            query = request.query if request.query is not None else ""
-
-            metadata_filter_document_ids, metadata_condition = self.get_metadata_filter_condition(
-                dataset_ids=available_datasets_ids,
-                query=query,
-                tenant_id=request.tenant_id,
-                user_id=request.user_id,
-                metadata_filtering_mode=request.metadata_filtering_mode,
-                metadata_model_config=app_metadata_model_config,
-                metadata_filtering_conditions=app_metadata_filtering_conditions,
-                inputs={},
-            )
-
-        if request.retrieval_mode == DatasetRetrieveConfigEntity.RetrieveStrategy.SINGLE:
-            planning_strategy = PlanningStrategy.REACT_ROUTER
-            # Ensure required fields are not None for single retrieval mode
-            if request.model_provider is None or request.model_name is None or request.query is None:
-                raise ValueError("model_provider, model_name, and query are required for single retrieval mode")
-
-            model_manager = ModelManager()
-            model_instance = model_manager.get_model_instance(
-                tenant_id=request.tenant_id,
-                model_type=ModelType.LLM,
-                provider=request.model_provider,
-                model=request.model_name,
-            )
-
-            provider_model_bundle = model_instance.provider_model_bundle
-            model_type_instance = model_instance.model_type_instance
-            model_type_instance = cast(LargeLanguageModel, model_type_instance)
-
-            model_credentials = model_instance.credentials
-
-            # check model
-            provider_model = provider_model_bundle.configuration.get_provider_model(
-                model=request.model_name, model_type=ModelType.LLM
-            )
-
-            if provider_model is None:
-                raise exc.ModelNotExistError(f"Model {request.model_name} not exist.")
-
-            if provider_model.status == ModelStatus.NO_CONFIGURE:
-                raise exc.ModelCredentialsNotInitializedError(
-                    f"Model {request.model_name} credentials is not initialized."
-                )
-            elif provider_model.status == ModelStatus.NO_PERMISSION:
-                raise exc.ModelNotSupportedError(f"Dify Hosted OpenAI {request.model_name} currently not support.")
-            elif provider_model.status == ModelStatus.QUOTA_EXCEEDED:
-                raise exc.ModelQuotaExceededError(f"Model provider {request.model_provider} quota exceeded.")
-
-            stop = []
-            completion_params = (request.completion_params or {}).copy()
-            if "stop" in completion_params:
-                stop = completion_params["stop"]
-                del completion_params["stop"]
-
-            model_schema = model_type_instance.get_model_schema(request.model_name, model_credentials)
-
-            if not model_schema:
-                raise exc.ModelNotExistError(f"Model {request.model_name} not exist.")
-
-            model_config = ModelConfigWithCredentialsEntity(
-                provider=request.model_provider,
-                model=request.model_name,
-                model_schema=model_schema,
-                mode=request.model_mode or "chat",
-                provider_model_bundle=provider_model_bundle,
-                credentials=model_credentials,
-                parameters=completion_params,
-                stop=stop,
-            )
-            all_documents = self.single_retrieve(
-                request.app_id,
-                request.tenant_id,
-                request.user_id,
-                request.user_from,
-                request.query,
-                available_datasets,
-                model_instance,
-                model_config,
-                planning_strategy,
-                None,  # message_id
-                metadata_filter_document_ids,
-                metadata_condition,
-            )
-        else:
-            all_documents = self.multiple_retrieve(
-                app_id=request.app_id,
-                tenant_id=request.tenant_id,
-                user_id=request.user_id,
-                user_from=request.user_from,
-                available_datasets=available_datasets,
-                query=request.query,
-                top_k=request.top_k,
-                score_threshold=request.score_threshold,
-                reranking_mode=request.reranking_mode,
-                reranking_model=request.reranking_model,
-                weights=request.weights,
-                reranking_enable=request.reranking_enable,
-                metadata_filter_document_ids=metadata_filter_document_ids,
-                metadata_condition=metadata_condition,
-                attachment_ids=request.attachment_ids,
-            )
-
-        dify_documents = [item for item in all_documents if item.provider == "dify"]
-        external_documents = [item for item in all_documents if item.provider == "external"]
-        retrieval_resource_list = []
-        # deal with external documents
-        for item in external_documents:
-            source = Source(
-                metadata=SourceMetadata(
-                    source="knowledge",
-                    dataset_id=item.metadata.get("dataset_id"),
-                    dataset_name=item.metadata.get("dataset_name"),
-                    document_id=item.metadata.get("document_id"),
-                    document_name=item.metadata.get("title"),
-                    data_source_type="external",
-                    retriever_from="workflow",
-                    score=item.metadata.get("score"),
-                    doc_metadata=item.metadata,
-                ),
-                title=item.metadata.get("title"),
-                content=item.page_content,
-            )
-            retrieval_resource_list.append(source)
-        # deal with dify documents
-        if dify_documents:
-            records = RetrievalService.format_retrieval_documents(dify_documents)
-            dataset_ids = [i.segment.dataset_id for i in records]
-            document_ids = [i.segment.document_id for i in records]
-
-            with session_factory.create_session() as session:
-                datasets = session.query(Dataset).where(Dataset.id.in_(dataset_ids)).all()
-                documents = session.query(DatasetDocument).where(DatasetDocument.id.in_(document_ids)).all()
-
-            dataset_map = {i.id: i for i in datasets}
-            document_map = {i.id: i for i in documents}
-
-            if records:
-                for record in records:
-                    segment = record.segment
-                    dataset = dataset_map.get(segment.dataset_id)
-                    document = document_map.get(segment.document_id)
-
-                    if dataset and document:
-                        source = Source(
-                            metadata=SourceMetadata(
-                                source="knowledge",
-                                dataset_id=dataset.id,
-                                dataset_name=dataset.name,
-                                document_id=document.id,
-                                document_name=document.name,
-                                data_source_type=document.data_source_type,
-                                segment_id=segment.id,
-                                retriever_from="workflow",
-                                score=record.score or 0.0,
-                                segment_hit_count=segment.hit_count,
-                                segment_word_count=segment.word_count,
-                                segment_position=segment.position,
-                                segment_index_node_hash=segment.index_node_hash,
-                                doc_metadata=document.doc_metadata,
-                                child_chunks=[
-                                    SourceChildChunk(
-                                        id=str(getattr(chunk, "id", "")),
-                                        content=str(getattr(chunk, "content", "")),
-                                        position=int(getattr(chunk, "position", 0)),
-                                        score=float(getattr(chunk, "score", 0.0)),
-                                    )
-                                    for chunk in (record.child_chunks or [])
-                                ],
-                                position=None,
-                            ),
-                            title=document.name,
-                            files=list(record.files) if record.files else None,
-                            content=segment.get_sign_content(),
-                        )
-                        if segment.answer:
-                            source.content = f"question:{segment.get_sign_content()} \nanswer:{segment.answer}"
-
-                        if record.summary:
-                            source.summary = record.summary
-
-                        retrieval_resource_list.append(source)
-
-        if retrieval_resource_list:
-
-            def _score(item: Source) -> float:
-                meta = item.metadata
-                score = meta.score
-                if isinstance(score, (int, float)):
-                    return float(score)
-                return 0.0
-
-            retrieval_resource_list = sorted(
-                retrieval_resource_list,
-                key=_score,  # type: ignore[arg-type, return-value]
-                reverse=True,
-            )
-            for position, item in enumerate(retrieval_resource_list, start=1):
-                item.metadata.position = position  # type: ignore[index]
-        return retrieval_resource_list
-
     def retrieve(
         self,
         app_id: str,
@@ -400,7 +150,14 @@ class DatasetRetrieval:
         if features:
             if ModelFeature.TOOL_CALL in features or ModelFeature.MULTI_TOOL_CALL in features:
                 planning_strategy = PlanningStrategy.ROUTER
-        available_datasets = self._get_available_datasets(tenant_id, dataset_ids)
+        available_datasets = []
+
+        dataset_stmt = select(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id.in_(dataset_ids))
+        datasets: list[Dataset] = db.session.execute(dataset_stmt).scalars().all()  # type: ignore
+        for dataset in datasets:
+            if dataset.available_document_count == 0 and dataset.provider != "external":
+                continue
+            available_datasets.append(dataset)
 
         if inputs:
             inputs = {key: str(value) for key, value in inputs.items()}
@@ -1404,6 +1161,7 @@ class DatasetRetrieval:
             query=query or "",
         )
 
+        result_text = ""
         try:
             # handle invoke result
             invoke_result = cast(
@@ -1434,8 +1192,7 @@ class DatasetRetrieval:
                                 "condition": item.get("comparison_operator"),
                             }
                         )
-        except Exception as e:
-            logger.warning(e, exc_info=True)
+        except Exception:
             return None
         return automatic_metadata_filters
 
@@ -1649,12 +1406,7 @@ class DatasetRetrieval:
         usage = None
         for result in invoke_result:
             text = result.delta.message.content
-            if isinstance(text, str):
-                full_text += text
-            elif isinstance(text, list):
-                for i in text:
-                    if i.data:
-                        full_text += i.data
+            full_text += text
 
             if not model:
                 model = result.model
@@ -1772,53 +1524,3 @@ class DatasetRetrieval:
                 cancel_event.set()
             if thread_exceptions is not None:
                 thread_exceptions.append(e)
-
-    def _get_available_datasets(self, tenant_id: str, dataset_ids: list[str]) -> list[Dataset]:
-        with session_factory.create_session() as session:
-            subquery = (
-                session.query(DocumentModel.dataset_id, func.count(DocumentModel.id).label("available_document_count"))
-                .where(
-                    DocumentModel.indexing_status == "completed",
-                    DocumentModel.enabled == True,
-                    DocumentModel.archived == False,
-                    DocumentModel.dataset_id.in_(dataset_ids),
-                )
-                .group_by(DocumentModel.dataset_id)
-                .having(func.count(DocumentModel.id) > 0)
-                .subquery()
-            )
-
-            results = (
-                session.query(Dataset)
-                .outerjoin(subquery, Dataset.id == subquery.c.dataset_id)
-                .where(Dataset.tenant_id == tenant_id, Dataset.id.in_(dataset_ids))
-                .where((subquery.c.available_document_count > 0) | (Dataset.provider == "external"))
-                .all()
-            )
-
-        available_datasets = []
-        for dataset in results:
-            if not dataset:
-                continue
-            available_datasets.append(dataset)
-        return available_datasets
-
-    def _check_knowledge_rate_limit(self, tenant_id: str):
-        knowledge_rate_limit = FeatureService.get_knowledge_rate_limit(tenant_id)
-        if knowledge_rate_limit.enabled:
-            current_time = int(time.time() * 1000)
-            key = f"rate_limit_{tenant_id}"
-            redis_client.zadd(key, {current_time: current_time})
-            redis_client.zremrangebyscore(key, 0, current_time - 60000)
-            request_count = redis_client.zcard(key)
-            if request_count > knowledge_rate_limit.limit:
-                with session_factory.create_session() as session:
-                    rate_limit_log = RateLimitLog(
-                        tenant_id=tenant_id,
-                        subscription_plan=knowledge_rate_limit.subscription_plan,
-                        operation="knowledge",
-                    )
-                    session.add(rate_limit_log)
-                raise exc.RateLimitExceededError(
-                    "you have reached the knowledge base request rate limit of your subscription."
-                )

api/core/schemas/registry.py

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

api/core/tools/__base/tool.py

@@ -5,7 +5,7 @@ from collections.abc import Generator
 from copy import deepcopy
 from typing import TYPE_CHECKING, Any
 
-if TYPE_CHECKING:  # pragma: no cover
+if TYPE_CHECKING:
     from models.model import File
 
 from core.tools.__base.tool_runtime import ToolRuntime
@@ -171,7 +171,7 @@ class Tool(ABC):
     def create_file_message(self, file: File) -> ToolInvokeMessage:
         return ToolInvokeMessage(
             type=ToolInvokeMessage.MessageType.FILE,
-            message=ToolInvokeMessage.FileMessage(file_marker="file_marker"),
+            message=ToolInvokeMessage.FileMessage(),
             meta={"file": file},
         )
 

api/core/tools/mcp_tool/tool.py

@@ -3,8 +3,8 @@ from __future__ import annotations
 import base64
 import json
 import logging
-from collections.abc import Generator, Mapping
-from typing import Any, cast
+from collections.abc import Generator
+from typing import Any
 
 from core.mcp.auth_client import MCPClientWithAuthRetry
 from core.mcp.error import MCPConnectionError
@@ -17,7 +17,6 @@ from core.mcp.types import (
     TextContent,
     TextResourceContents,
 )
-from core.model_runtime.entities.llm_entities import LLMUsage, LLMUsageMetadata
 from core.tools.__base.tool import Tool
 from core.tools.__base.tool_runtime import ToolRuntime
 from core.tools.entities.tool_entities import ToolEntity, ToolInvokeMessage, ToolProviderType
@@ -47,7 +46,6 @@ class MCPTool(Tool):
         self.headers = headers or {}
         self.timeout = timeout
         self.sse_read_timeout = sse_read_timeout
-        self._latest_usage = LLMUsage.empty_usage()
 
     def tool_provider_type(self) -> ToolProviderType:
         return ToolProviderType.MCP
@@ -61,10 +59,6 @@ class MCPTool(Tool):
         message_id: str | None = None,
     ) -> Generator[ToolInvokeMessage, None, None]:
         result = self.invoke_remote_mcp_tool(tool_parameters)
-
-        # Extract usage metadata from MCP protocol's _meta field
-        self._latest_usage = self._derive_usage_from_result(result)
-
         # handle dify tool output
         for content in result.content:
             if isinstance(content, TextContent):
@@ -126,99 +120,6 @@ class MCPTool(Tool):
         for item in json_list:
             yield self.create_json_message(item)
 
-    @property
-    def latest_usage(self) -> LLMUsage:
-        return self._latest_usage
-
-    @classmethod
-    def _derive_usage_from_result(cls, result: CallToolResult) -> LLMUsage:
-        """
-        Extract usage metadata from MCP tool result's _meta field.
-
-        The MCP protocol's _meta field (aliased as 'meta' in Python) can contain
-        usage information such as token counts, costs, and other metadata.
-
-        Args:
-            result: The CallToolResult from MCP tool invocation
-
-        Returns:
-            LLMUsage instance with values from meta or empty_usage if not found
-        """
-        # Extract usage from the meta field if present
-        if result.meta:
-            usage_dict = cls._extract_usage_dict(result.meta)
-            if usage_dict is not None:
-                return LLMUsage.from_metadata(cast(LLMUsageMetadata, cast(object, dict(usage_dict))))
-
-        return LLMUsage.empty_usage()
-
-    @classmethod
-    def _extract_usage_dict(cls, payload: Mapping[str, Any]) -> Mapping[str, Any] | None:
-        """
-        Recursively search for usage dictionary in the payload.
-
-        The MCP protocol's _meta field can contain usage data in various formats:
-        - Direct usage field: {"usage": {...}}
-        - Nested in metadata: {"metadata": {"usage": {...}}}
-        - Or nested within other fields
-
-        Args:
-            payload: The payload to search for usage data
-
-        Returns:
-            The usage dictionary if found, None otherwise
-        """
-        # Check for direct usage field
-        usage_candidate = payload.get("usage")
-        if isinstance(usage_candidate, Mapping):
-            return usage_candidate
-
-        # Check for metadata nested usage
-        metadata_candidate = payload.get("metadata")
-        if isinstance(metadata_candidate, Mapping):
-            usage_candidate = metadata_candidate.get("usage")
-            if isinstance(usage_candidate, Mapping):
-                return usage_candidate
-
-        # Check for common token counting fields directly in payload
-        # Some MCP servers may include token counts directly
-        if "total_tokens" in payload or "prompt_tokens" in payload or "completion_tokens" in payload:
-            usage_dict: dict[str, Any] = {}
-            for key in (
-                "prompt_tokens",
-                "completion_tokens",
-                "total_tokens",
-                "prompt_unit_price",
-                "completion_unit_price",
-                "total_price",
-                "currency",
-                "prompt_price_unit",
-                "completion_price_unit",
-                "prompt_price",
-                "completion_price",
-                "latency",
-                "time_to_first_token",
-                "time_to_generate",
-            ):
-                if key in payload:
-                    usage_dict[key] = payload[key]
-            if usage_dict:
-                return usage_dict
-
-        # Recursively search through nested structures
-        for value in payload.values():
-            if isinstance(value, Mapping):
-                found = cls._extract_usage_dict(value)
-                if found is not None:
-                    return found
-            elif isinstance(value, list) and not isinstance(value, (str, bytes, bytearray)):
-                for item in value:
-                    if isinstance(item, Mapping):
-                        found = cls._extract_usage_dict(item)
-                        if found is not None:
-                            return found
-        return None
-
     def fork_tool_runtime(self, runtime: ToolRuntime) -> MCPTool:
         return MCPTool(
             entity=self.entity,

api/core/tools/tool_manager.py

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

api/core/tools/utils/workflow_configuration_sync.py

@@ -9,6 +9,11 @@ from core.workflow.nodes.base.entities import OutputVariableEntity
 
 
 class WorkflowToolConfigurationUtils:
+    @classmethod
+    def check_parameter_configurations(cls, configurations: list[Mapping[str, Any]]):
+        for configuration in configurations:
+            WorkflowToolParameterConfiguration.model_validate(configuration)
+
     @classmethod
     def get_workflow_graph_variables(cls, graph: Mapping[str, Any]) -> Sequence[VariableEntity]:
         """

api/core/tools/workflow_as_tool/tool.py

@@ -98,10 +98,6 @@ class WorkflowTool(Tool):
             invoke_from=self.runtime.invoke_from,
             streaming=False,
             call_depth=self.workflow_call_depth + 1,
-            # NOTE(QuantumGhost): We explicitly set `pause_state_config` to `None`
-            # because workflow pausing mechanisms (such as HumanInput) are not
-            # supported within WorkflowTool execution context.
-            pause_state_config=None,
         )
         assert isinstance(result, dict)
         data = result.get("data", {})

api/core/variables/variables.py

@@ -112,7 +112,7 @@ class ArrayBooleanVariable(ArrayBooleanSegment, ArrayVariable):
 
 class RAGPipelineVariable(BaseModel):
     belong_to_node_id: str = Field(description="belong to which node id, shared means public")
-    type: str = Field(description="variable type, text-input, paragraph, select, number, file, file-list")
+    type: str = Field(description="variable type, text-input, paragraph, select, number,  file, file-list")
     label: str = Field(description="label")
     description: str | None = Field(description="description", default="")
     variable: str = Field(description="variable key", default="")

api/core/workflow/entities/graph_config.py

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

api/core/workflow/graph/graph.py

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

api/core/workflow/graph_engine/domain/graph_execution.py

@@ -10,7 +10,6 @@ from pydantic import BaseModel, Field
 
 from core.workflow.entities.pause_reason import PauseReason
 from core.workflow.enums import NodeState
-from core.workflow.runtime.graph_runtime_state import GraphExecutionProtocol
 
 from .node_execution import NodeExecution
 
@@ -237,6 +236,3 @@ class GraphExecution:
     def record_node_failure(self) -> None:
         """Increment the count of node failures encountered during execution."""
         self.exceptions_count += 1
-
-
-_: GraphExecutionProtocol = GraphExecution(workflow_id="")

api/core/workflow/graph_engine/graph_engine.py

@@ -47,6 +47,7 @@ from .graph_traversal import EdgeProcessor, SkipPropagator
 from .layers.base import GraphEngineLayer
 from .orchestration import Dispatcher, ExecutionCoordinator
 from .protocols.command_channel import CommandChannel
+from .ready_queue import ReadyQueue
 from .worker_management import WorkerPool
 
 if TYPE_CHECKING:
@@ -93,7 +94,7 @@ class GraphEngine:
         self._graph_execution.workflow_id = workflow_id
 
         # === Execution Queues ===
-        self._ready_queue = self._graph_runtime_state.ready_queue
+        self._ready_queue = cast(ReadyQueue, self._graph_runtime_state.ready_queue)
 
         # Queue for events generated during execution
         self._event_queue: queue.Queue[GraphNodeEventBase] = queue.Queue()

api/core/workflow/graph_engine/response_coordinator/coordinator.py

@@ -15,10 +15,10 @@ from uuid import uuid4
 from pydantic import BaseModel, Field
 
 from core.workflow.enums import NodeExecutionType, NodeState
+from core.workflow.graph import Graph
 from core.workflow.graph_events import NodeRunStreamChunkEvent, NodeRunSucceededEvent
 from core.workflow.nodes.base.template import TextSegment, VariableSegment
 from core.workflow.runtime import VariablePool
-from core.workflow.runtime.graph_runtime_state import GraphProtocol
 
 from .path import Path
 from .session import ResponseSession
@@ -75,7 +75,7 @@ class ResponseStreamCoordinator:
     Ensures ordered streaming of responses based on upstream node outputs and constants.
     """
 
-    def __init__(self, variable_pool: "VariablePool", graph: GraphProtocol) -> None:
+    def __init__(self, variable_pool: "VariablePool", graph: "Graph") -> None:
         """
         Initialize coordinator with variable pool.
 

api/core/workflow/graph_engine/response_coordinator/session.py

@@ -10,10 +10,10 @@ from __future__ import annotations
 from dataclasses import dataclass
 
 from core.workflow.nodes.answer.answer_node import AnswerNode
+from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.base.template import Template
 from core.workflow.nodes.end.end_node import EndNode
 from core.workflow.nodes.knowledge_index import KnowledgeIndexNode
-from core.workflow.runtime.graph_runtime_state import NodeProtocol
 
 
 @dataclass
@@ -29,26 +29,21 @@ class ResponseSession:
     index: int = 0  # Current position in the template segments
 
     @classmethod
-    def from_node(cls, node: NodeProtocol) -> ResponseSession:
+    def from_node(cls, node: Node) -> ResponseSession:
         """
-        Create a ResponseSession from a response-capable node.
-
-        The parameter is typed as `NodeProtocol` because the graph is exposed behind a protocol at the runtime layer,
-        but at runtime this must be an `AnswerNode`, `EndNode`, or `KnowledgeIndexNode` that provides:
-        - `id: str`
-        - `get_streaming_template() -> Template`
+        Create a ResponseSession from an AnswerNode or EndNode.
 
         Args:
-            node: Node from the materialized workflow graph.
+            node: Must be either an AnswerNode or EndNode instance
 
         Returns:
             ResponseSession configured with the node's streaming template
 
         Raises:
-            TypeError: If node is not a supported response node type.
+            TypeError: If node is not an AnswerNode or EndNode
         """
         if not isinstance(node, AnswerNode | EndNode | KnowledgeIndexNode):
-            raise TypeError("ResponseSession.from_node only supports AnswerNode, EndNode, or KnowledgeIndexNode")
+            raise TypeError
         return cls(
             node_id=node.id,
             template=node.get_streaming_template(),

api/core/workflow/nodes/agent/agent_node.py

@@ -192,33 +192,32 @@ class AgentNode(Node[AgentNodeData]):
                 result[parameter_name] = None
                 continue
             agent_input = node_data.agent_parameters[parameter_name]
-            match agent_input.type:
-                case "variable":
-                    variable = variable_pool.get(agent_input.value)  # type: ignore
-                    if variable is None:
-                        raise AgentVariableNotFoundError(str(agent_input.value))
-                    parameter_value = variable.value
-                case "mixed" | "constant":
-                    # variable_pool.convert_template expects a string template,
-                    # but if passing a dict, convert to JSON string first before rendering
-                    try:
-                        if not isinstance(agent_input.value, str):
-                            parameter_value = json.dumps(agent_input.value, ensure_ascii=False)
-                        else:
-                            parameter_value = str(agent_input.value)
-                    except TypeError:
+            if agent_input.type == "variable":
+                variable = variable_pool.get(agent_input.value)  # type: ignore
+                if variable is None:
+                    raise AgentVariableNotFoundError(str(agent_input.value))
+                parameter_value = variable.value
+            elif agent_input.type in {"mixed", "constant"}:
+                # variable_pool.convert_template expects a string template,
+                # but if passing a dict, convert to JSON string first before rendering
+                try:
+                    if not isinstance(agent_input.value, str):
+                        parameter_value = json.dumps(agent_input.value, ensure_ascii=False)
+                    else:
                         parameter_value = str(agent_input.value)
-                    segment_group = variable_pool.convert_template(parameter_value)
-                    parameter_value = segment_group.log if for_log else segment_group.text
-                    # variable_pool.convert_template returns a string,
-                    # so we need to convert it back to a dictionary
-                    try:
-                        if not isinstance(agent_input.value, str):
-                            parameter_value = json.loads(parameter_value)
-                    except json.JSONDecodeError:
-                        parameter_value = parameter_value
-                case _:
-                    raise AgentInputTypeError(agent_input.type)
+                except TypeError:
+                    parameter_value = str(agent_input.value)
+                segment_group = variable_pool.convert_template(parameter_value)
+                parameter_value = segment_group.log if for_log else segment_group.text
+                # variable_pool.convert_template returns a string,
+                # so we need to convert it back to a dictionary
+                try:
+                    if not isinstance(agent_input.value, str):
+                        parameter_value = json.loads(parameter_value)
+                except json.JSONDecodeError:
+                    parameter_value = parameter_value
+            else:
+                raise AgentInputTypeError(agent_input.type)
             value = parameter_value
             if parameter.type == "array[tools]":
                 value = cast(list[dict[str, Any]], value)
@@ -375,13 +374,12 @@ class AgentNode(Node[AgentNodeData]):
         result: dict[str, Any] = {}
         for parameter_name in typed_node_data.agent_parameters:
             input = typed_node_data.agent_parameters[parameter_name]
-            match input.type:
-                case "mixed" | "constant":
-                    selectors = VariableTemplateParser(str(input.value)).extract_variable_selectors()
-                    for selector in selectors:
-                        result[selector.variable] = selector.value_selector
-                case "variable":
-                    result[parameter_name] = input.value
+            if input.type in ["mixed", "constant"]:
+                selectors = VariableTemplateParser(str(input.value)).extract_variable_selectors()
+                for selector in selectors:
+                    result[selector.variable] = selector.value_selector
+            elif input.type == "variable":
+                result[parameter_name] = input.value
 
         result = {node_id + "." + key: value for key, value in result.items()}