fix: provide json as text

Signed-off-by: Stream <Stream_2@qq.com>

.github/CODEOWNERS

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

.github/workflows/build-push.yml

@@ -75,7 +75,9 @@ jobs:
         with:
           context: "{{defaultContext}}:${{ matrix.context }}"
           platforms: ${{ matrix.platform }}
-          build-args: COMMIT_SHA=${{ fromJSON(steps.meta.outputs.json).labels['org.opencontainers.image.revision'] }}
+          build-args: |
+            COMMIT_SHA=${{ fromJSON(steps.meta.outputs.json).labels['org.opencontainers.image.revision'] }}
+            ENABLE_PROD_SOURCEMAP=${{ matrix.context == 'web' && github.ref_name == 'deploy/dev' }}
           labels: ${{ steps.meta.outputs.labels }}
           outputs: type=image,name=${{ env[matrix.image_name_env] }},push-by-digest=true,name-canonical=true,push=true
           cache-from: type=gha,scope=${{ matrix.service_name }}

.gitignore

@@ -209,6 +209,7 @@ api/.vscode
 .history
 
 .idea/
+web/migration/
 
 # pnpm
 /.pnpm-store

api/.env.example

@@ -33,6 +33,9 @@ TRIGGER_URL=http://localhost:5001
 # The time in seconds after the signature is rejected
 FILES_ACCESS_TIMEOUT=300
 
+# Collaboration mode toggle
+ENABLE_COLLABORATION_MODE=false
+
 # Access token expiration time in minutes
 ACCESS_TOKEN_EXPIRE_MINUTES=60
 
@@ -717,3 +720,14 @@ SANDBOX_EXPIRED_RECORDS_CLEAN_GRACEFUL_PERIOD=21
 SANDBOX_EXPIRED_RECORDS_CLEAN_BATCH_SIZE=1000
 SANDBOX_EXPIRED_RECORDS_RETENTION_DAYS=30
 SANDBOX_EXPIRED_RECORDS_CLEAN_TASK_LOCK_TTL=90000
+
+# Sandbox Dify CLI configuration
+# Directory containing dify CLI binaries (dify-cli-<os>-<arch>). Defaults to api/bin when unset.
+SANDBOX_DIFY_CLI_ROOT=
+
+# CLI API URL for sandbox (dify-sandbox or e2b) to call back to Dify API.
+# This URL must be accessible from the sandbox environment.
+# For local development: use http://localhost:5001 or http://127.0.0.1:5001
+# For Docker deployment: use http://api:5001 (internal Docker network)
+# For external sandbox (e.g., e2b): use a publicly accessible URL
+CLI_API_URL=http://localhost:5001

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/agent-notes/configs/feature/__init__.py.md

@@ -0,0 +1,9 @@
+Summary:
+Summary:
+- Application configuration definitions, including file access settings.
+
+Invariants:
+- File access settings drive signed URL expiration and base URLs.
+
+Tests:
+- Config parsing tests under tests/unit_tests/configs.

api/agent-notes/controllers/files/__init__.py.md

@@ -0,0 +1,9 @@
+Summary:
+- Registers file-related API namespaces and routes for files service.
+- Includes app-assets and sandbox archive proxy controllers.
+
+Invariants:
+- files_ns must include all file controller modules to register routes.
+
+Tests:
+- Coverage via controller unit tests and route registration smoke checks.

api/agent-notes/controllers/files/app_assets_download.py.md

@@ -0,0 +1,14 @@
+Summary:
+- App assets download proxy endpoint (signed URL verification, stream from storage).
+
+Invariants:
+- Validates AssetPath fields (UUIDs, asset_type allowlist).
+- Verifies tenant-scoped signature and expiration before reading storage.
+- URL uses expires_at/nonce/sign query params.
+
+Edge Cases:
+- Missing files return NotFound.
+- Invalid signature or expired link returns Forbidden.
+
+Tests:
+- Verify signature validation and invalid/expired cases.

api/agent-notes/controllers/files/app_assets_upload.py.md

@@ -0,0 +1,13 @@
+Summary:
+- App assets upload proxy endpoint (signed URL verification, upload to storage).
+
+Invariants:
+- Validates AssetPath fields (UUIDs, asset_type allowlist).
+- Verifies tenant-scoped signature and expiration before writing storage.
+- URL uses expires_at/nonce/sign query params.
+
+Edge Cases:
+- Invalid signature or expired link returns Forbidden.
+
+Tests:
+- Verify signature validation and invalid/expired cases.

api/agent-notes/controllers/files/sandbox_archive.py.md

@@ -0,0 +1,14 @@
+Summary:
+- Sandbox archive upload/download proxy endpoints (signed URL verification, stream to storage).
+
+Invariants:
+- Validates tenant_id and sandbox_id UUIDs.
+- Verifies tenant-scoped signature and expiration before storage access.
+- URL uses expires_at/nonce/sign query params.
+
+Edge Cases:
+- Missing archive returns NotFound.
+- Invalid signature or expired link returns Forbidden.
+
+Tests:
+- Add unit tests for signature validation if needed.

api/agent-notes/core/app_assets/builder/file_builder.py.md

@@ -0,0 +1,9 @@
+Summary:
+Summary:
+- Collects file assets and emits FileAsset entries with storage keys.
+
+Invariants:
+- Storage keys are derived via AppAssetStorage for draft files.
+
+Tests:
+- Covered by asset build pipeline tests.

api/agent-notes/core/app_assets/builder/skill_builder.py.md

@@ -0,0 +1,14 @@
+Summary:
+Summary:
+- Builds skill artifacts from markdown assets and uploads resolved outputs.
+
+Invariants:
+- Reads draft asset content via AppAssetStorage refs.
+- Writes resolved artifacts via AppAssetStorage refs.
+- FileAsset storage keys are derived via AppAssetStorage.
+
+Edge Cases:
+- Missing or invalid JSON content yields empty skill content/metadata.
+
+Tests:
+- Build pipeline unit tests covering compile/upload paths.

api/agent-notes/core/app_assets/converters.py.md

@@ -0,0 +1,9 @@
+Summary:
+Summary:
+- Converts AppAssetFileTree to FileAsset items for packaging.
+
+Invariants:
+- Storage keys for assets are derived via AppAssetStorage.
+
+Tests:
+- Used in packaging/service tests for asset bundles.

api/agent-notes/core/app_assets/packager/zip_packager.py.md

@@ -0,0 +1,14 @@
+# Zip Packager Notes
+
+## Purpose
+- Builds a ZIP archive of asset contents stored via the configured storage backend.
+
+## Key Decisions
+- Packaging writes assets into an in-memory zip buffer returned as bytes.
+- Asset fetch + zip writing are executed via a thread pool with a lock guarding `ZipFile` writes.
+
+## Edge Cases
+- ZIP writes are serialized by the lock; storage reads still run in parallel.
+
+## Tests/Verification
+- None yet.

api/agent-notes/core/app_assets/parser/asset_parser.py.md

@@ -0,0 +1,9 @@
+Summary:
+Summary:
+- Builds AssetItem entries for asset trees using AssetPath-derived storage keys.
+
+Invariants:
+- Uses AssetPath to compute draft storage keys.
+
+Tests:
+- Covered by asset parsing and packaging tests.

api/agent-notes/core/app_assets/storage.py.md

@@ -0,0 +1,20 @@
+Summary:
+- Defines AssetPath facade + typed asset path classes for app-asset storage access.
+- Maps asset paths to storage keys and generates presigned or signed-proxy URLs.
+- Signs proxy URLs using tenant private keys and enforces expiration.
+- Exposes app_asset_storage singleton for reuse.
+
+Invariants:
+- AssetPathBase fields (tenant_id/app_id/resource_id/node_id) must be UUIDs.
+- AssetPath.from_components enforces valid types and resolved node_id presence.
+- Storage keys are derived internally via AssetPathBase.get_storage_key; callers never supply raw paths.
+- AppAssetStorage.storage returns the cached presign wrapper (not the raw storage).
+
+Edge Cases:
+- Storage backends without presign support must fall back to signed proxy URLs.
+- Signed proxy verification enforces expiration and tenant-scoped signing keys.
+- Upload URLs also fall back to signed proxy endpoints when presign is unsupported.
+- load_or_none treats SilentStorage "File Not Found" bytes as missing.
+
+Tests:
+- Unit tests for ref validation, storage key mapping, and signed URL verification.

api/agent-notes/core/app_bundle/source_zip_extractor.py.md

@@ -0,0 +1,10 @@
+Summary:
+Summary:
+- Extracts asset files from a zip and persists them into app asset storage.
+
+Invariants:
+- Rejects path traversal/absolute/backslash paths.
+- Saves extracted files via AppAssetStorage draft refs.
+
+Tests:
+- Zip security edge cases and tree construction tests.

api/agent-notes/core/sandbox/initializer/app_assets_initializer.py.md

@@ -0,0 +1,9 @@
+Summary:
+Summary:
+- Downloads published app asset zip into sandbox and extracts it.
+
+Invariants:
+- Uses AppAssetStorage to generate download URLs for build zips (internal URL).
+
+Tests:
+- Sandbox initialization integration tests.

api/agent-notes/core/sandbox/initializer/draft_app_assets_initializer.py.md

@@ -0,0 +1,12 @@
+Summary:
+Summary:
+- Downloads draft/resolved assets into sandbox for draft execution.
+
+Invariants:
+- Uses AppAssetStorage to generate download URLs for draft/resolved refs (internal URL).
+
+Edge Cases:
+- No nodes -> returns early.
+
+Tests:
+- Sandbox draft initialization tests.

api/agent-notes/core/sandbox/sandbox.py.md

@@ -0,0 +1,9 @@
+Summary:
+- Sandbox lifecycle wrapper (ready/cancel/fail signals, mount/unmount, release).
+
+Invariants:
+- wait_ready raises with the original initialization error as the cause.
+- release always attempts unmount and environment release, logging failures.
+
+Tests:
+- Covered by sandbox lifecycle/unit tests and workflow execution error handling.

api/agent-notes/core/sandbox/security/__init__.py.md

@@ -0,0 +1,2 @@
+Summary:
+- Sandbox security helper modules.

api/agent-notes/core/sandbox/security/archive_signer.py.md

@@ -0,0 +1,13 @@
+Summary:
+- Generates and verifies signed URLs for sandbox archive upload/download.
+
+Invariants:
+- tenant_id and sandbox_id must be UUIDs.
+- Signatures are tenant-scoped and include operation, expiry, and nonce.
+
+Edge Cases:
+- Missing tenant private key raises ValueError.
+- Expired or tampered signatures are rejected.
+
+Tests:
+- Add unit tests if sandbox archive signature behavior expands.

api/agent-notes/core/sandbox/storage/archive_storage.py.md

@@ -0,0 +1,12 @@
+Summary:
+- Manages sandbox archive uploads/downloads for workspace persistence.
+
+Invariants:
+- Archive storage key is sandbox/<tenant_id>/<sandbox_id>.tar.gz.
+- Signed URLs are tenant-scoped and use external files URL.
+
+Edge Cases:
+- Missing archive skips mount.
+
+Tests:
+- Covered indirectly via sandbox integration tests.

api/agent-notes/core/skill/skill_manager.py.md

@@ -0,0 +1,9 @@
+Summary:
+Summary:
+- Loads/saves skill bundles to app asset storage.
+
+Invariants:
+- Skill bundles use AppAssetStorage refs and JSON serialization.
+
+Tests:
+- Covered by skill bundle build/load unit tests.

api/agent-notes/core/virtual_environment/providers/e2b_sandbox.py.md

@@ -0,0 +1,16 @@
+# E2B Sandbox Provider Notes
+
+## Purpose
+- Implements the E2B-backed `VirtualEnvironment` provider and bootstraps sandbox metadata, file I/O, and command execution.
+
+## Key Decisions
+- Sandbox metadata is gathered during `_construct_environment` using the E2B SDK before returning `Metadata`.
+- Architecture/OS detection uses a single `uname -m -s` call split by whitespace to reduce round-trips.
+- Command execution streams stdout/stderr through `QueueTransportReadCloser`; stdin is unsupported.
+
+## Edge Cases
+- `release_environment` raises when sandbox termination fails.
+- `execute_command` runs in a background thread; consumers must read stdout/stderr until EOF.
+
+## Tests/Verification
+- None yet. Add targeted service tests when behavior changes.

api/agent-notes/services/app_asset_service.py.md

@@ -0,0 +1,14 @@
+Summary:
+- App asset CRUD, publish/build pipeline, and presigned URL generation.
+
+Invariants:
+- Asset storage access goes through AppAssetStorage + AssetPath, using app_asset_storage singleton.
+- Tree operations require tenant/app scoping and lock for mutation.
+- Asset zips are packaged via raw storage with storage keys from AppAssetStorage.
+
+Edge Cases:
+- File nodes larger than preview limit are rejected.
+- Deletion runs asynchronously; storage failures are logged.
+
+Tests:
+- Unit tests for storage URL generation and publish/build flows.

api/agent-notes/services/app_bundle_service.py.md

@@ -0,0 +1,10 @@
+Summary:
+Summary:
+- Imports app bundles, including asset extraction into app asset storage.
+
+Invariants:
+- Asset imports respect zip security checks and tenant/app scoping.
+- Draft asset packaging uses AppAssetStorage for key mapping.
+
+Tests:
+- Bundle import unit tests and zip validation coverage.

api/agent-notes/tests/unit_tests/core/app_assets/test_storage.py.md

@@ -0,0 +1,6 @@
+Summary:
+Summary:
+- Unit tests for AppAssetStorage ref validation, key mapping, and signing.
+
+Tests:
+- Covers valid/invalid refs, signature verify, expiration handling, and proxy URL generation.

api/app.py

@@ -1,12 +1,5 @@
-from __future__ import annotations
-
+import os
 import sys
-from typing import TYPE_CHECKING, cast
-
-if TYPE_CHECKING:
-    from celery import Celery
-
-    celery: Celery
 
 
 def is_db_command() -> bool:
@@ -16,10 +9,15 @@ def is_db_command() -> bool:
 
 
 # create app
+flask_app = None
+socketio_app = None
+
 if is_db_command():
     from app_factory import create_migrations_app
 
     app = create_migrations_app()
+    socketio_app = app
+    flask_app = app
 else:
     # Gunicorn and Celery handle monkey patching automatically in production by
     # specifying the `gevent` worker class. Manual monkey patching is not required here.
@@ -30,8 +28,15 @@ else:
 
     from app_factory import create_app
 
-    app = create_app()
-    celery = cast("Celery", app.extensions["celery"])
+    socketio_app, flask_app = create_app()
+    app = flask_app
+    celery = flask_app.extensions["celery"]
 
 if __name__ == "__main__":
-    app.run(host="0.0.0.0", port=5001)
+    from gevent import pywsgi
+    from geventwebsocket.handler import WebSocketHandler  # type: ignore[reportMissingTypeStubs]
+
+    host = os.environ.get("HOST", "0.0.0.0")
+    port = int(os.environ.get("PORT", 5001))
+    server = pywsgi.WSGIServer((host, port), socketio_app, handler_class=WebSocketHandler)
+    server.serve_forever()

api/app_factory.py

@@ -1,6 +1,7 @@
 import logging
 import time
 
+import socketio  # type: ignore[reportMissingTypeStubs]
 from opentelemetry.trace import get_current_span
 from opentelemetry.trace.span import INVALID_SPAN_ID, INVALID_TRACE_ID
 
@@ -8,6 +9,7 @@ from configs import dify_config
 from contexts.wrapper import RecyclableContextVar
 from core.logging.context import init_request_context
 from dify_app import DifyApp
+from extensions.ext_socketio import sio
 
 logger = logging.getLogger(__name__)
 
@@ -60,14 +62,18 @@ def create_flask_app_with_configs() -> DifyApp:
     return dify_app
 
 
-def create_app() -> DifyApp:
+def create_app() -> tuple[socketio.WSGIApp, DifyApp]:
     start_time = time.perf_counter()
     app = create_flask_app_with_configs()
     initialize_extensions(app)
+
+    sio.app = app
+    socketio_app = socketio.WSGIApp(sio, app)
+
     end_time = time.perf_counter()
     if dify_config.DEBUG:
         logger.info("Finished create_app (%s ms)", round((end_time - start_time) * 1000, 2))
-    return app
+    return socketio_app, app
 
 
 def initialize_extensions(app: DifyApp):
@@ -149,7 +155,7 @@ def initialize_extensions(app: DifyApp):
             logger.info("Loaded %s (%s ms)", short_name, round((end_time - start_time) * 1000, 2))
 
 
-def create_migrations_app() -> DifyApp:
+def create_migrations_app():
     app = create_flask_app_with_configs()
     from extensions import ext_database, ext_migrate
 

api/commands.py

@@ -23,7 +23,8 @@ from core.rag.datasource.vdb.vector_factory import Vector
 from core.rag.datasource.vdb.vector_type import VectorType
 from core.rag.index_processor.constant.built_in_field import BuiltInField
 from core.rag.models.document import ChildDocument, Document
-from core.tools.utils.system_oauth_encryption import encrypt_system_oauth_params
+from core.sandbox import SandboxBuilder, SandboxType
+from core.tools.utils.system_encryption import encrypt_system_params
 from events.app_event import app_was_created
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
@@ -1450,58 +1451,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:
@@ -1611,7 +1608,7 @@ def remove_orphaned_files_on_storage(force: bool):
             click.echo(click.style(f"- Scanning files on storage path {storage_path}", fg="white"))
             files = storage.scan(path=storage_path, files=True, directories=False)
             all_files_on_storage.extend(files)
-        except FileNotFoundError as e:
+        except FileNotFoundError:
             click.echo(click.style(f"  -> Skipping path {storage_path} as it does not exist.", fg="yellow"))
             continue
         except Exception as e:
@@ -1741,18 +1738,59 @@ def file_usage(
                 if src_filter != src:
                     continue
 
-        match ids_table["type"]:
-            case "uuid":
-                # Direct UUID match
-                query = (
-                    f"SELECT {ids_table['pk_column']}, {ids_table['column']} "
-                    f"FROM {ids_table['table']} WHERE {ids_table['column']} IS NOT NULL"
-                )
-                with db.engine.begin() as conn:
-                    rs = conn.execute(sa.text(query))
-                    for row in rs:
-                        record_id = str(row[0])
-                        ref_file_id = str(row[1])
+        if ids_table["type"] == "uuid":
+            # Direct UUID match
+            query = (
+                f"SELECT {ids_table['pk_column']}, {ids_table['column']} "
+                f"FROM {ids_table['table']} WHERE {ids_table['column']} IS NOT NULL"
+            )
+            with db.engine.begin() as conn:
+                rs = conn.execute(sa.text(query))
+                for row in rs:
+                    record_id = str(row[0])
+                    ref_file_id = str(row[1])
+                    if ref_file_id not in file_key_map:
+                        continue
+                    storage_key = file_key_map[ref_file_id]
+
+                    # Apply filters
+                    if file_id and ref_file_id != file_id:
+                        continue
+                    if key and not storage_key.endswith(key):
+                        continue
+
+                    # Only collect items within the requested page range
+                    if offset <= total_count < offset + limit:
+                        paginated_usages.append(
+                            {
+                                "src": f"{ids_table['table']}.{ids_table['column']}",
+                                "record_id": record_id,
+                                "file_id": ref_file_id,
+                                "key": storage_key,
+                            }
+                        )
+                    total_count += 1
+
+        elif ids_table["type"] in ("text", "json"):
+            # Extract UUIDs from text/json content
+            column_cast = f"{ids_table['column']}::text" if ids_table["type"] == "json" else ids_table["column"]
+            query = (
+                f"SELECT {ids_table['pk_column']}, {column_cast} "
+                f"FROM {ids_table['table']} WHERE {ids_table['column']} IS NOT NULL"
+            )
+            with db.engine.begin() as conn:
+                rs = conn.execute(sa.text(query))
+                for row in rs:
+                    record_id = str(row[0])
+                    content = str(row[1])
+
+                    # Find all UUIDs in the content
+                    import re
+
+                    uuid_pattern = re.compile(guid_regexp, re.IGNORECASE)
+                    matches = uuid_pattern.findall(content)
+
+                    for ref_file_id in matches:
                         if ref_file_id not in file_key_map:
                             continue
                         storage_key = file_key_map[ref_file_id]
@@ -1775,50 +1813,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 = {
@@ -1862,6 +1856,57 @@ def file_usage(
             click.echo(click.style(f"Use --offset {offset + limit} to see next page", fg="white"))
 
 
+@click.command("setup-sandbox-system-config", help="Setup system-level sandbox provider configuration.")
+@click.option(
+    "--provider-type", prompt=True, type=click.Choice(["e2b", "docker", "local"]), help="Sandbox provider type"
+)
+@click.option("--config", prompt=True, help='Configuration JSON (e.g., {"api_key": "xxx"} for e2b)')
+def setup_sandbox_system_config(provider_type: str, config: str):
+    """
+    Setup system-level sandbox provider configuration.
+
+    Examples:
+        flask setup-sandbox-system-config --provider-type e2b --config '{"api_key": "e2b_xxx"}'
+        flask setup-sandbox-system-config --provider-type docker --config '{"docker_sock": "unix:///var/run/docker.sock"}'
+        flask setup-sandbox-system-config --provider-type local --config '{}'
+    """
+    from models.sandbox import SandboxProviderSystemConfig
+
+    try:
+        click.echo(click.style(f"Validating config: {config}", fg="yellow"))
+        config_dict = TypeAdapter(dict[str, Any]).validate_json(config)
+        click.echo(click.style("Config validated successfully.", fg="green"))
+
+        click.echo(click.style(f"Validating config schema for provider type: {provider_type}", fg="yellow"))
+        SandboxBuilder.validate(SandboxType(provider_type), config_dict)
+        click.echo(click.style("Config schema validated successfully.", fg="green"))
+
+        click.echo(click.style("Encrypting config...", fg="yellow"))
+        click.echo(click.style(f"Using SECRET_KEY: `{dify_config.SECRET_KEY}`", fg="yellow"))
+        encrypted_config = encrypt_system_params(config_dict)
+        click.echo(click.style("Config encrypted successfully.", fg="green"))
+    except Exception as e:
+        click.echo(click.style(f"Error validating/encrypting config: {str(e)}", fg="red"))
+        return
+
+    deleted_count = db.session.query(SandboxProviderSystemConfig).filter_by(provider_type=provider_type).delete()
+    if deleted_count > 0:
+        click.echo(
+            click.style(
+                f"Deleted {deleted_count} existing system config for provider type: {provider_type}", fg="yellow"
+            )
+        )
+
+    system_config = SandboxProviderSystemConfig(
+        provider_type=provider_type,
+        encrypted_config=encrypted_config,
+    )
+    db.session.add(system_config)
+    db.session.commit()
+    click.echo(click.style(f"Sandbox system config setup successfully. id: {system_config.id}", fg="green"))
+    click.echo(click.style(f"Provider type: {provider_type}", fg="green"))
+
+
 @click.command("setup-system-tool-oauth-client", help="Setup system tool oauth client.")
 @click.option("--provider", prompt=True, help="Provider name")
 @click.option("--client-params", prompt=True, help="Client Params")
@@ -1881,7 +1926,7 @@ def setup_system_tool_oauth_client(provider, client_params):
 
         click.echo(click.style(f"Encrypting client params: {client_params}", fg="yellow"))
         click.echo(click.style(f"Using SECRET_KEY: `{dify_config.SECRET_KEY}`", fg="yellow"))
-        oauth_client_params = encrypt_system_oauth_params(client_params_dict)
+        oauth_client_params = encrypt_system_params(client_params_dict)
         click.echo(click.style("Client params encrypted successfully.", fg="green"))
     except Exception as e:
         click.echo(click.style(f"Error parsing client params: {str(e)}", fg="red"))
@@ -1930,7 +1975,7 @@ def setup_system_trigger_oauth_client(provider, client_params):
 
         click.echo(click.style(f"Encrypting client params: {client_params}", fg="yellow"))
         click.echo(click.style(f"Using SECRET_KEY: `{dify_config.SECRET_KEY}`", fg="yellow"))
-        oauth_client_params = encrypt_system_oauth_params(client_params_dict)
+        oauth_client_params = encrypt_system_params(client_params_dict)
         click.echo(click.style("Client params encrypted successfully.", fg="green"))
     except Exception as e:
         click.echo(click.style(f"Error parsing client params: {str(e)}", fg="red"))

api/configs/app_config.py

@@ -2,6 +2,7 @@ import logging
 from pathlib import Path
 from typing import Any
 
+from pydantic import Field
 from pydantic.fields import FieldInfo
 from pydantic_settings import BaseSettings, PydanticBaseSettingsSource, SettingsConfigDict, TomlConfigSettingsSource
 
@@ -82,6 +83,17 @@ class DifyConfig(
         extra="ignore",
     )
 
+    SANDBOX_DIFY_CLI_ROOT: str | None = Field(
+        default=None,
+        description=(
+            "Filesystem directory containing dify CLI binaries named dify-cli-<os>-<arch>. "
+            "Defaults to api/bin when unset."
+        ),
+    )
+    DIFY_PORT: int = Field(
+        default=5001,
+        description="Port used by Dify to communicate with the host machine.",
+    )
     # Before adding any config,
     # please consider to arrange it in the proper config group of existed or added
     # for better readability and maintainability.

api/configs/feature/__init__.py

@@ -249,6 +249,17 @@ class PluginConfig(BaseSettings):
     )
 
 
+class CliApiConfig(BaseSettings):
+    """
+    Configuration for CLI API (for dify-cli to call back from external sandbox environments)
+    """
+
+    CLI_API_URL: str = Field(
+        description="CLI API URL for external sandbox (e.g., e2b) to call back.",
+        default="http://localhost:5001",
+    )
+
+
 class MarketplaceConfig(BaseSettings):
     """
     Configuration for marketplace
@@ -1234,6 +1245,13 @@ class PositionConfig(BaseSettings):
         return {item.strip() for item in self.POSITION_TOOL_EXCLUDES.split(",") if item.strip() != ""}
 
 
+class CollaborationConfig(BaseSettings):
+    ENABLE_COLLABORATION_MODE: bool = Field(
+        description="Whether to enable collaboration mode features across the workspace",
+        default=False,
+    )
+
+
 class LoginConfig(BaseSettings):
     ENABLE_EMAIL_CODE_LOGIN: bool = Field(
         description="whether to enable email code login",
@@ -1328,6 +1346,7 @@ class FeatureConfig(
     TriggerConfig,
     AsyncWorkflowConfig,
     PluginConfig,
+    CliApiConfig,
     MarketplaceConfig,
     DataSetConfig,
     EndpointConfig,
@@ -1352,6 +1371,7 @@ class FeatureConfig(
     WorkflowConfig,
     WorkflowNodeExecutionConfig,
     WorkspaceConfig,
+    CollaborationConfig,
     LoginConfig,
     AccountConfig,
     SwaggerUIConfig,

api/controllers/cli_api/__init__.py

@@ -0,0 +1,27 @@
+from flask import Blueprint
+from flask_restx import Namespace
+
+from libs.external_api import ExternalApi
+
+bp = Blueprint("cli_api", __name__, url_prefix="/cli/api")
+
+api = ExternalApi(
+    bp,
+    version="1.0",
+    title="CLI API",
+    description="APIs for Dify CLI to call back from external sandbox environments (e.g., e2b)",
+)
+
+# Create namespace
+cli_api_ns = Namespace("cli_api", description="CLI API operations", path="/")
+
+from .dify_cli import cli_api as _plugin
+
+api.add_namespace(cli_api_ns)
+
+__all__ = [
+    "_plugin",
+    "api",
+    "bp",
+    "cli_api_ns",
+]

api/controllers/cli_api/dify_cli/cli_api.py

@@ -0,0 +1,192 @@
+from flask import abort
+from flask_restx import Resource
+from pydantic import BaseModel
+
+from controllers.cli_api import cli_api_ns
+from controllers.cli_api.dify_cli.wraps import get_cli_user_tenant, plugin_data
+from controllers.cli_api.wraps import cli_api_only
+from controllers.console.wraps import setup_required
+from core.app.entities.app_invoke_entities import InvokeFrom
+from core.file.helpers import get_signed_file_url_for_plugin
+from core.plugin.backwards_invocation.app import PluginAppBackwardsInvocation
+from core.plugin.backwards_invocation.base import BaseBackwardsInvocationResponse
+from core.plugin.backwards_invocation.model import PluginModelBackwardsInvocation
+from core.plugin.backwards_invocation.tool import PluginToolBackwardsInvocation
+from core.plugin.entities.request import (
+    RequestInvokeApp,
+    RequestInvokeLLM,
+    RequestInvokeTool,
+    RequestRequestUploadFile,
+)
+from core.sandbox.bash.dify_cli import DifyCliToolConfig
+from core.session.cli_api import CliContext
+from core.skill.entities import ToolInvocationRequest
+from core.tools.entities.tool_entities import ToolProviderType
+from core.tools.tool_manager import ToolManager
+from libs.helper import length_prefixed_response
+from models.account import Account
+from models.model import EndUser, Tenant
+
+
+class FetchToolItem(BaseModel):
+    tool_type: str
+    tool_provider: str
+    tool_name: str
+    credential_id: str | None = None
+
+
+class FetchToolBatchRequest(BaseModel):
+    tools: list[FetchToolItem]
+
+
+@cli_api_ns.route("/invoke/llm")
+class CliInvokeLLMApi(Resource):
+    @cli_api_only
+    @get_cli_user_tenant
+    @setup_required
+    @plugin_data(payload_type=RequestInvokeLLM)
+    def post(
+        self,
+        user_model: Account | EndUser,
+        tenant_model: Tenant,
+        payload: RequestInvokeLLM,
+        cli_context: CliContext,
+    ):
+        def generator():
+            response = PluginModelBackwardsInvocation.invoke_llm(user_model.id, tenant_model, payload)
+            return PluginModelBackwardsInvocation.convert_to_event_stream(response)
+
+        return length_prefixed_response(0xF, generator())
+
+
+@cli_api_ns.route("/invoke/tool")
+class CliInvokeToolApi(Resource):
+    @cli_api_only
+    @get_cli_user_tenant
+    @setup_required
+    @plugin_data(payload_type=RequestInvokeTool)
+    def post(
+        self,
+        user_model: Account | EndUser,
+        tenant_model: Tenant,
+        payload: RequestInvokeTool,
+        cli_context: CliContext,
+    ):
+        tool_type = ToolProviderType.value_of(payload.tool_type)
+
+        request = ToolInvocationRequest(
+            tool_type=tool_type,
+            provider=payload.provider,
+            tool_name=payload.tool,
+            credential_id=payload.credential_id,
+        )
+        if cli_context.tool_access and not cli_context.tool_access.is_allowed(request):
+            abort(403, description=f"Access denied for tool: {payload.provider}/{payload.tool}")
+
+        def generator():
+            return PluginToolBackwardsInvocation.convert_to_event_stream(
+                PluginToolBackwardsInvocation.invoke_tool(
+                    tenant_id=tenant_model.id,
+                    user_id=user_model.id,
+                    tool_type=tool_type,
+                    provider=payload.provider,
+                    tool_name=payload.tool,
+                    tool_parameters=payload.tool_parameters,
+                    credential_id=payload.credential_id,
+                ),
+            )
+
+        return length_prefixed_response(0xF, generator())
+
+
+@cli_api_ns.route("/invoke/app")
+class CliInvokeAppApi(Resource):
+    @cli_api_only
+    @get_cli_user_tenant
+    @setup_required
+    @plugin_data(payload_type=RequestInvokeApp)
+    def post(
+        self,
+        user_model: Account | EndUser,
+        tenant_model: Tenant,
+        payload: RequestInvokeApp,
+        cli_context: CliContext,
+    ):
+        response = PluginAppBackwardsInvocation.invoke_app(
+            app_id=payload.app_id,
+            user_id=user_model.id,
+            tenant_id=tenant_model.id,
+            conversation_id=payload.conversation_id,
+            query=payload.query,
+            stream=payload.response_mode == "streaming",
+            inputs=payload.inputs,
+            files=payload.files,
+        )
+
+        return length_prefixed_response(0xF, PluginAppBackwardsInvocation.convert_to_event_stream(response))
+
+
+@cli_api_ns.route("/upload/file/request")
+class CliUploadFileRequestApi(Resource):
+    @cli_api_only
+    @get_cli_user_tenant
+    @setup_required
+    @plugin_data(payload_type=RequestRequestUploadFile)
+    def post(
+        self,
+        user_model: Account | EndUser,
+        tenant_model: Tenant,
+        payload: RequestRequestUploadFile,
+        cli_context: CliContext,
+    ):
+        url = get_signed_file_url_for_plugin(
+            filename=payload.filename,
+            mimetype=payload.mimetype,
+            tenant_id=tenant_model.id,
+            user_id=user_model.id,
+        )
+        return BaseBackwardsInvocationResponse(data={"url": url}).model_dump()
+
+
+@cli_api_ns.route("/fetch/tools/batch")
+class CliFetchToolsBatchApi(Resource):
+    @cli_api_only
+    @get_cli_user_tenant
+    @setup_required
+    @plugin_data(payload_type=FetchToolBatchRequest)
+    def post(
+        self,
+        user_model: Account | EndUser,
+        tenant_model: Tenant,
+        payload: FetchToolBatchRequest,
+        cli_context: CliContext,
+    ):
+        tools: list[dict] = []
+
+        for item in payload.tools:
+            provider_type = ToolProviderType.value_of(item.tool_type)
+
+            request = ToolInvocationRequest(
+                tool_type=provider_type,
+                provider=item.tool_provider,
+                tool_name=item.tool_name,
+                credential_id=item.credential_id,
+            )
+            if cli_context.tool_access and not cli_context.tool_access.is_allowed(request):
+                abort(403, description=f"Access denied for tool: {item.tool_provider}/{item.tool_name}")
+
+            try:
+                tool_runtime = ToolManager.get_tool_runtime(
+                    tenant_id=tenant_model.id,
+                    provider_type=provider_type,
+                    provider_id=item.tool_provider,
+                    tool_name=item.tool_name,
+                    invoke_from=InvokeFrom.AGENT,
+                    credential_id=item.credential_id,
+                )
+                tool_config = DifyCliToolConfig.create_from_tool(tool_runtime)
+                tools.append(tool_config.model_dump())
+            except Exception:
+                continue
+
+        return BaseBackwardsInvocationResponse(data={"tools": tools}).model_dump()

api/controllers/cli_api/dify_cli/wraps.py

@@ -0,0 +1,137 @@
+from collections.abc import Callable
+from functools import wraps
+from typing import ParamSpec, TypeVar
+
+from flask import current_app, g, request
+from flask_login import user_logged_in
+from pydantic import BaseModel
+from sqlalchemy.orm import Session
+
+from core.session.cli_api import CliApiSession, CliContext
+from extensions.ext_database import db
+from libs.login import current_user
+from models.account import Tenant
+from models.model import DefaultEndUserSessionID, EndUser
+
+P = ParamSpec("P")
+R = TypeVar("R")
+
+
+class TenantUserPayload(BaseModel):
+    tenant_id: str
+    user_id: str
+
+
+def get_user(tenant_id: str, user_id: str | None) -> EndUser:
+    """
+    Get current user
+
+    NOTE: user_id is not trusted, it could be maliciously set to any value.
+    As a result, it could only be considered as an end user id.
+    """
+    if not user_id:
+        user_id = DefaultEndUserSessionID.DEFAULT_SESSION_ID
+    is_anonymous = user_id == DefaultEndUserSessionID.DEFAULT_SESSION_ID
+    try:
+        with Session(db.engine) as session:
+            user_model = None
+
+            if is_anonymous:
+                user_model = (
+                    session.query(EndUser)
+                    .where(
+                        EndUser.session_id == user_id,
+                        EndUser.tenant_id == tenant_id,
+                    )
+                    .first()
+                )
+            else:
+                user_model = (
+                    session.query(EndUser)
+                    .where(
+                        EndUser.id == user_id,
+                        EndUser.tenant_id == tenant_id,
+                    )
+                    .first()
+                )
+
+            if not user_model:
+                user_model = EndUser(
+                    tenant_id=tenant_id,
+                    type="service_api",
+                    is_anonymous=is_anonymous,
+                    session_id=user_id,
+                )
+                session.add(user_model)
+                session.commit()
+                session.refresh(user_model)
+
+    except Exception:
+        raise ValueError("user not found")
+
+    return user_model
+
+
+def get_cli_user_tenant(view_func: Callable[P, R]):
+    @wraps(view_func)
+    def decorated_view(*args: P.args, **kwargs: P.kwargs):
+        session: CliApiSession | None = getattr(g, "cli_api_session", None)
+        if session is None:
+            raise ValueError("session not found")
+
+        user_id = session.user_id
+        tenant_id = session.tenant_id
+        cli_context = CliContext.model_validate(session.context)
+
+        if not user_id:
+            user_id = DefaultEndUserSessionID.DEFAULT_SESSION_ID
+
+        try:
+            tenant_model = (
+                db.session.query(Tenant)
+                .where(
+                    Tenant.id == tenant_id,
+                )
+                .first()
+            )
+        except Exception:
+            raise ValueError("tenant not found")
+
+        if not tenant_model:
+            raise ValueError("tenant not found")
+
+        kwargs["tenant_model"] = tenant_model
+        kwargs["user_model"] = get_user(tenant_id, user_id)
+        kwargs["cli_context"] = cli_context
+
+        current_app.login_manager._update_request_context_with_user(kwargs["user_model"])  # type: ignore
+        user_logged_in.send(current_app._get_current_object(), user=current_user)  # type: ignore
+
+        return view_func(*args, **kwargs)
+
+    return decorated_view
+
+
+def plugin_data(view: Callable[P, R] | None = None, *, payload_type: type[BaseModel]):
+    def decorator(view_func: Callable[P, R]):
+        @wraps(view_func)
+        def decorated_view(*args: P.args, **kwargs: P.kwargs):
+            try:
+                data = request.get_json()
+            except Exception:
+                raise ValueError("invalid json")
+
+            try:
+                payload = payload_type.model_validate(data)
+            except Exception as e:
+                raise ValueError(f"invalid payload: {str(e)}")
+
+            kwargs["payload"] = payload
+            return view_func(*args, **kwargs)
+
+        return decorated_view
+
+    if view is None:
+        return decorator
+    else:
+        return decorator(view)

api/controllers/cli_api/wraps.py

@@ -0,0 +1,56 @@
+import hashlib
+import hmac
+import time
+from collections.abc import Callable
+from functools import wraps
+from typing import ParamSpec, TypeVar
+
+from flask import abort, g, request
+
+from core.session.cli_api import CliApiSessionManager
+
+P = ParamSpec("P")
+R = TypeVar("R")
+
+SIGNATURE_TTL_SECONDS = 300
+
+
+def _verify_signature(session_secret: str, timestamp: str, body: bytes, signature: str) -> bool:
+    expected = hmac.new(
+        session_secret.encode(),
+        f"{timestamp}.".encode() + body,
+        hashlib.sha256,
+    ).hexdigest()
+    return hmac.compare_digest(f"sha256={expected}", signature)
+
+
+def cli_api_only(view: Callable[P, R]):
+    @wraps(view)
+    def decorated(*args: P.args, **kwargs: P.kwargs):
+        session_id = request.headers.get("X-Cli-Api-Session-Id")
+        timestamp = request.headers.get("X-Cli-Api-Timestamp")
+        signature = request.headers.get("X-Cli-Api-Signature")
+
+        if not session_id or not timestamp or not signature:
+            abort(401)
+
+        try:
+            ts = int(timestamp)
+            if abs(time.time() - ts) > SIGNATURE_TTL_SECONDS:
+                abort(401)
+        except ValueError:
+            abort(401)
+
+        session = CliApiSessionManager().get(session_id)
+        if not session:
+            abort(401)
+
+        body = request.get_data()
+        if not _verify_signature(session.secret, timestamp, body, signature):
+            abort(401)
+
+        g.cli_api_session = session
+
+        return view(*args, **kwargs)
+
+    return decorated

api/controllers/console/__init__.py

@@ -32,6 +32,7 @@ for module_name in RESOURCE_MODULES:
 
 # Ensure resource modules are imported so route decorators are evaluated.
 # Import other controllers
+# Sandbox file browser
 from . import (
     admin,
     apikey,
@@ -39,6 +40,7 @@ from . import (
     feature,
     init_validate,
     ping,
+    sandbox_files,
     setup,
     spec,
     version,
@@ -50,6 +52,7 @@ from .app import (
     agent,
     annotation,
     app,
+    app_asset,
     audio,
     completion,
     conversation,
@@ -60,9 +63,11 @@ from .app import (
     model_config,
     ops_trace,
     site,
+    skills,
     statistic,
     workflow,
     workflow_app_log,
+    workflow_comment,
     workflow_draft_variable,
     workflow_run,
     workflow_statistic,
@@ -114,6 +119,7 @@ from .explore import (
     saved_message,
     trial,
 )
+from .socketio import workflow as socketio_workflow  # pyright: ignore[reportUnusedImport]
 
 # Import tag controllers
 from .tag import tags
@@ -128,6 +134,7 @@ from .workspace import (
     model_providers,
     models,
     plugin,
+    sandbox_providers,
     tool_providers,
     trigger_providers,
     workspace,
@@ -146,6 +153,7 @@ __all__ = [
     "api",
     "apikey",
     "app",
+    "app_asset",
     "audio",
     "banner",
     "billing",
@@ -194,9 +202,12 @@ __all__ = [
     "rag_pipeline_import",
     "rag_pipeline_workflow",
     "recommended_app",
+    "sandbox_files",
+    "sandbox_providers",
     "saved_message",
     "setup",
     "site",
+    "skills",
     "spec",
     "statistic",
     "tags",
@@ -207,6 +218,7 @@ __all__ = [
     "website",
     "workflow",
     "workflow_app_log",
+    "workflow_comment",
     "workflow_draft_variable",
     "workflow_run",
     "workflow_statistic",

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,5 +1,6 @@
 import uuid
 from datetime import datetime
+from enum import StrEnum
 from typing import Any, Literal, TypeAlias
 
 from flask import request
@@ -30,6 +31,7 @@ from extensions.ext_database import db
 from libs.login import current_account_with_tenant, login_required
 from models import App, DatasetPermissionEnum, Workflow
 from models.model import IconType
+from models.workflow_features import WorkflowFeatures
 from services.app_dsl_service import AppDslService, ImportMode
 from services.app_service import AppService
 from services.enterprise.enterprise_service import EnterpriseService
@@ -55,6 +57,11 @@ ALLOW_CREATE_APP_MODES = ["chat", "agent-chat", "advanced-chat", "workflow", "co
 register_enum_models(console_ns, IconType)
 
 
+class RuntimeType(StrEnum):
+    CLASSIC = "classic"
+    SANDBOXED = "sandboxed"
+
+
 class AppListQuery(BaseModel):
     page: int = Field(default=1, ge=1, le=99999, description="Page number (1-99999)")
     limit: int = Field(default=20, ge=1, le=100, description="Page size (1-100)")
@@ -119,6 +126,11 @@ class AppExportQuery(BaseModel):
     workflow_id: str | None = Field(default=None, description="Specific workflow ID to export")
 
 
+class AppExportBundleQuery(BaseModel):
+    include_secret: bool = Field(default=False, description="Include secrets in export")
+    workflow_id: str | None = Field(default=None, description="Specific workflow ID to export")
+
+
 class AppNamePayload(BaseModel):
     name: str = Field(..., min_length=1, description="Name to check")
 
@@ -344,6 +356,7 @@ class AppPartial(ResponseModel):
     create_user_name: str | None = None
     author_name: str | None = None
     has_draft_trigger: bool | None = None
+    runtime_type: RuntimeType = RuntimeType.CLASSIC
 
     @computed_field(return_type=str | None)  # type: ignore
     @property
@@ -493,6 +506,7 @@ class AppListApi(Resource):
             str(app.id) for app in app_pagination.items if app.mode in {"workflow", "advanced-chat"}
         ]
         draft_trigger_app_ids: set[str] = set()
+        sandbox_app_ids: set[str] = set()
         if workflow_capable_app_ids:
             draft_workflows = (
                 db.session.execute(
@@ -510,6 +524,10 @@ class AppListApi(Resource):
                 NodeType.TRIGGER_PLUGIN,
             }
             for workflow in draft_workflows:
+                # Check sandbox feature
+                if workflow.get_feature(WorkflowFeatures.SANDBOX).enabled:
+                    sandbox_app_ids.add(str(workflow.app_id))
+
                 try:
                     for _, node_data in workflow.walk_nodes():
                         if node_data.get("type") in trigger_node_types:
@@ -520,6 +538,7 @@ class AppListApi(Resource):
 
         for app in app_pagination.items:
             app.has_draft_trigger = str(app.id) in draft_trigger_app_ids
+            app.runtime_type = RuntimeType.SANDBOXED if str(app.id) in sandbox_app_ids else RuntimeType.CLASSIC
 
         pagination_model = AppPagination.model_validate(app_pagination, from_attributes=True)
         return pagination_model.model_dump(mode="json"), 200
@@ -688,6 +707,29 @@ class AppExportApi(Resource):
         return payload.model_dump(mode="json")
 
 
+@console_ns.route("/apps/<uuid:app_id>/export-bundle")
+class AppExportBundleApi(Resource):
+    @get_app_model
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def get(self, app_model):
+        from services.app_bundle_service import AppBundleService
+
+        args = AppExportBundleQuery.model_validate(request.args.to_dict(flat=True))
+        current_user, _ = current_account_with_tenant()
+
+        result = AppBundleService.export_bundle(
+            app_model=app_model,
+            account_id=str(current_user.id),
+            include_secret=args.include_secret,
+            workflow_id=args.workflow_id,
+        )
+
+        return result.model_dump(mode="json")
+
+
 @console_ns.route("/apps/<uuid:app_id>/name")
 class AppNameApi(Resource):
     @console_ns.doc("check_app_name")

api/controllers/console/app/app_asset.py

@@ -0,0 +1,321 @@
+from flask import request
+from flask_restx import Resource
+from pydantic import BaseModel, Field, field_validator
+
+from controllers.console import console_ns
+from controllers.console.app.error import (
+    AppAssetNodeNotFoundError,
+    AppAssetPathConflictError,
+)
+from controllers.console.app.wraps import get_app_model
+from controllers.console.wraps import account_initialization_required, setup_required
+from core.app.entities.app_asset_entities import BatchUploadNode
+from libs.login import current_account_with_tenant, login_required
+from models import App
+from models.model import AppMode
+from services.app_asset_service import AppAssetService
+from services.errors.app_asset import (
+    AppAssetNodeNotFoundError as ServiceNodeNotFoundError,
+)
+from services.errors.app_asset import (
+    AppAssetParentNotFoundError,
+)
+from services.errors.app_asset import (
+    AppAssetPathConflictError as ServicePathConflictError,
+)
+
+DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
+
+
+class CreateFolderPayload(BaseModel):
+    name: str = Field(..., min_length=1, max_length=255)
+    parent_id: str | None = None
+
+
+class CreateFilePayload(BaseModel):
+    name: str = Field(..., min_length=1, max_length=255)
+    parent_id: str | None = None
+
+    @field_validator("name", mode="before")
+    @classmethod
+    def strip_name(cls, v: str) -> str:
+        return v.strip() if isinstance(v, str) else v
+
+    @field_validator("parent_id", mode="before")
+    @classmethod
+    def empty_to_none(cls, v: str | None) -> str | None:
+        return v or None
+
+
+class GetUploadUrlPayload(BaseModel):
+    name: str = Field(..., min_length=1, max_length=255)
+    size: int = Field(..., ge=0)
+    parent_id: str | None = None
+
+    @field_validator("name", mode="before")
+    @classmethod
+    def strip_name(cls, v: str) -> str:
+        return v.strip() if isinstance(v, str) else v
+
+    @field_validator("parent_id", mode="before")
+    @classmethod
+    def empty_to_none(cls, v: str | None) -> str | None:
+        return v or None
+
+
+class BatchUploadPayload(BaseModel):
+    children: list[BatchUploadNode] = Field(..., min_length=1)
+
+
+class UpdateFileContentPayload(BaseModel):
+    content: str
+
+
+class RenameNodePayload(BaseModel):
+    name: str = Field(..., min_length=1, max_length=255)
+
+
+class MoveNodePayload(BaseModel):
+    parent_id: str | None = None
+
+
+class ReorderNodePayload(BaseModel):
+    after_node_id: str | None = Field(default=None, description="Place after this node, None for first position")
+
+
+def reg(cls: type[BaseModel]) -> None:
+    console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
+
+
+reg(CreateFolderPayload)
+reg(CreateFilePayload)
+reg(GetUploadUrlPayload)
+reg(BatchUploadNode)
+reg(BatchUploadPayload)
+reg(UpdateFileContentPayload)
+reg(RenameNodePayload)
+reg(MoveNodePayload)
+reg(ReorderNodePayload)
+
+
+@console_ns.route("/apps/<string:app_id>/assets/tree")
+class AppAssetTreeResource(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    def get(self, app_model: App):
+        current_user, _ = current_account_with_tenant()
+        tree = AppAssetService.get_asset_tree(app_model, current_user.id)
+        return {"children": [view.model_dump() for view in tree.transform()]}
+
+
+@console_ns.route("/apps/<string:app_id>/assets/folders")
+class AppAssetFolderResource(Resource):
+    @console_ns.expect(console_ns.models[CreateFolderPayload.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    def post(self, app_model: App):
+        current_user, _ = current_account_with_tenant()
+        payload = CreateFolderPayload.model_validate(console_ns.payload or {})
+
+        try:
+            node = AppAssetService.create_folder(app_model, current_user.id, payload.name, payload.parent_id)
+            return node.model_dump(), 201
+        except AppAssetParentNotFoundError:
+            raise AppAssetNodeNotFoundError()
+        except ServicePathConflictError:
+            raise AppAssetPathConflictError()
+
+
+@console_ns.route("/apps/<string:app_id>/assets/files/<string:node_id>")
+class AppAssetFileDetailResource(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    def get(self, app_model: App, node_id: str):
+        current_user, _ = current_account_with_tenant()
+        try:
+            content = AppAssetService.get_file_content(app_model, current_user.id, node_id)
+            return {"content": content.decode("utf-8", errors="replace")}
+        except ServiceNodeNotFoundError:
+            raise AppAssetNodeNotFoundError()
+
+    @console_ns.expect(console_ns.models[UpdateFileContentPayload.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    def put(self, app_model: App, node_id: str):
+        current_user, _ = current_account_with_tenant()
+
+        file = request.files.get("file")
+        if file:
+            content = file.read()
+        else:
+            payload = UpdateFileContentPayload.model_validate(console_ns.payload or {})
+            content = payload.content.encode("utf-8")
+
+        try:
+            node = AppAssetService.update_file_content(app_model, current_user.id, node_id, content)
+            return node.model_dump()
+        except ServiceNodeNotFoundError:
+            raise AppAssetNodeNotFoundError()
+
+
+@console_ns.route("/apps/<string:app_id>/assets/nodes/<string:node_id>")
+class AppAssetNodeResource(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    def delete(self, app_model: App, node_id: str):
+        current_user, _ = current_account_with_tenant()
+        try:
+            AppAssetService.delete_node(app_model, current_user.id, node_id)
+            return {"result": "success"}, 200
+        except ServiceNodeNotFoundError:
+            raise AppAssetNodeNotFoundError()
+
+
+@console_ns.route("/apps/<string:app_id>/assets/nodes/<string:node_id>/rename")
+class AppAssetNodeRenameResource(Resource):
+    @console_ns.expect(console_ns.models[RenameNodePayload.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    def post(self, app_model: App, node_id: str):
+        current_user, _ = current_account_with_tenant()
+        payload = RenameNodePayload.model_validate(console_ns.payload or {})
+
+        try:
+            node = AppAssetService.rename_node(app_model, current_user.id, node_id, payload.name)
+            return node.model_dump()
+        except ServiceNodeNotFoundError:
+            raise AppAssetNodeNotFoundError()
+        except ServicePathConflictError:
+            raise AppAssetPathConflictError()
+
+
+@console_ns.route("/apps/<string:app_id>/assets/nodes/<string:node_id>/move")
+class AppAssetNodeMoveResource(Resource):
+    @console_ns.expect(console_ns.models[MoveNodePayload.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    def post(self, app_model: App, node_id: str):
+        current_user, _ = current_account_with_tenant()
+        payload = MoveNodePayload.model_validate(console_ns.payload or {})
+
+        try:
+            node = AppAssetService.move_node(app_model, current_user.id, node_id, payload.parent_id)
+            return node.model_dump()
+        except ServiceNodeNotFoundError:
+            raise AppAssetNodeNotFoundError()
+        except AppAssetParentNotFoundError:
+            raise AppAssetNodeNotFoundError()
+        except ServicePathConflictError:
+            raise AppAssetPathConflictError()
+
+
+@console_ns.route("/apps/<string:app_id>/assets/nodes/<string:node_id>/reorder")
+class AppAssetNodeReorderResource(Resource):
+    @console_ns.expect(console_ns.models[ReorderNodePayload.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    def post(self, app_model: App, node_id: str):
+        current_user, _ = current_account_with_tenant()
+        payload = ReorderNodePayload.model_validate(console_ns.payload or {})
+
+        try:
+            node = AppAssetService.reorder_node(app_model, current_user.id, node_id, payload.after_node_id)
+            return node.model_dump()
+        except ServiceNodeNotFoundError:
+            raise AppAssetNodeNotFoundError()
+
+
+@console_ns.route("/apps/<string:app_id>/assets/files/<string:node_id>/download-url")
+class AppAssetFileDownloadUrlResource(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    def get(self, app_model: App, node_id: str):
+        current_user, _ = current_account_with_tenant()
+        try:
+            download_url = AppAssetService.get_file_download_url(app_model, current_user.id, node_id)
+            return {"download_url": download_url}
+        except ServiceNodeNotFoundError:
+            raise AppAssetNodeNotFoundError()
+
+
+@console_ns.route("/apps/<string:app_id>/assets/files/upload")
+class AppAssetFileUploadUrlResource(Resource):
+    @console_ns.expect(console_ns.models[GetUploadUrlPayload.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    def post(self, app_model: App):
+        current_user, _ = current_account_with_tenant()
+        payload = GetUploadUrlPayload.model_validate(console_ns.payload or {})
+
+        try:
+            node, upload_url = AppAssetService.get_file_upload_url(
+                app_model, current_user.id, payload.name, payload.size, payload.parent_id
+            )
+            return {"node": node.model_dump(), "upload_url": upload_url}, 201
+        except AppAssetParentNotFoundError:
+            raise AppAssetNodeNotFoundError()
+        except ServicePathConflictError:
+            raise AppAssetPathConflictError()
+
+
+@console_ns.route("/apps/<string:app_id>/assets/batch-upload")
+class AppAssetBatchUploadResource(Resource):
+    @console_ns.expect(console_ns.models[BatchUploadPayload.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    def post(self, app_model: App):
+        """
+        Create nodes from tree structure and return upload URLs.
+
+        Input:
+        {
+            "children": [
+                {"name": "folder1", "node_type": "folder", "children": [
+                    {"name": "file1.txt", "node_type": "file", "size": 1024}
+                ]},
+                {"name": "root.txt", "node_type": "file", "size": 512}
+            ]
+        }
+
+        Output:
+        {
+            "children": [
+                {"id": "xxx", "name": "folder1", "node_type": "folder", "children": [
+                    {"id": "yyy", "name": "file1.txt", "node_type": "file", "size": 1024, "upload_url": "..."}
+                ]},
+                {"id": "zzz", "name": "root.txt", "node_type": "file", "size": 512, "upload_url": "..."}
+            ]
+        }
+        """
+        current_user, _ = current_account_with_tenant()
+        payload = BatchUploadPayload.model_validate(console_ns.payload or {})
+
+        try:
+            result_children = AppAssetService.batch_create_from_tree(app_model, current_user.id, payload.children)
+            return {"children": [child.model_dump() for child in result_children]}, 201
+        except AppAssetParentNotFoundError:
+            raise AppAssetNodeNotFoundError()
+        except ServicePathConflictError:
+            raise AppAssetPathConflictError()

api/controllers/console/app/app_import.py

@@ -51,6 +51,14 @@ class AppImportPayload(BaseModel):
     app_id: str | None = Field(None)
 
 
+class AppImportBundleConfirmPayload(BaseModel):
+    name: str | None = None
+    description: str | None = None
+    icon_type: str | None = None
+    icon: str | None = None
+    icon_background: str | None = None
+
+
 console_ns.schema_model(
     AppImportPayload.__name__, AppImportPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
 )
@@ -139,3 +147,68 @@ class AppImportCheckDependenciesApi(Resource):
             result = import_service.check_dependencies(app_model=app_model)
 
         return result.model_dump(mode="json"), 200
+
+
+@console_ns.route("/apps/imports-bundle/prepare")
+class AppImportBundlePrepareApi(Resource):
+    """Step 1: Get upload URL for bundle import."""
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def post(self):
+        from services.app_bundle_service import AppBundleService
+
+        current_user, current_tenant_id = current_account_with_tenant()
+
+        result = AppBundleService.prepare_import(
+            tenant_id=current_tenant_id,
+            account_id=current_user.id,
+        )
+
+        return {"import_id": result.import_id, "upload_url": result.upload_url}, 200
+
+
+@console_ns.route("/apps/imports-bundle/<string:import_id>/confirm")
+class AppImportBundleConfirmApi(Resource):
+    """Step 2: Confirm bundle import after upload."""
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @marshal_with(app_import_model)
+    @cloud_edition_billing_resource_check("apps")
+    @edit_permission_required
+    def post(self, import_id: str):
+        from flask import request
+
+        from core.app.entities.app_bundle_entities import BundleFormatError
+        from services.app_bundle_service import AppBundleService
+
+        current_user, _ = current_account_with_tenant()
+
+        args = AppImportBundleConfirmPayload.model_validate(request.get_json() or {})
+
+        try:
+            result = AppBundleService.confirm_import(
+                import_id=import_id,
+                account=current_user,
+                name=args.name,
+                description=args.description,
+                icon_type=args.icon_type,
+                icon=args.icon,
+                icon_background=args.icon_background,
+            )
+        except BundleFormatError as e:
+            return {"error": str(e)}, 400
+
+        if result.app_id and FeatureService.get_system_features().webapp_auth.enabled:
+            EnterpriseService.WebAppAuth.update_app_access_mode(result.app_id, "private")
+
+        status = result.status
+        if status == ImportStatus.FAILED:
+            return result.model_dump(mode="json"), 400
+        elif status == ImportStatus.PENDING:
+            return result.model_dump(mode="json"), 202
+        return result.model_dump(mode="json"), 200

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

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

api/controllers/console/app/error.py

@@ -110,8 +110,6 @@ class TracingConfigCheckError(BaseHTTPException):
 
 
 class InvokeRateLimitError(BaseHTTPException):
-    """Raised when the Invoke returns rate limit error."""
-
     error_code = "rate_limit_error"
     description = "Rate Limit Error"
     code = 429
@@ -121,3 +119,21 @@ class NeedAddIdsError(BaseHTTPException):
     error_code = "need_add_ids"
     description = "Need to add ids."
     code = 400
+
+
+class AppAssetNodeNotFoundError(BaseHTTPException):
+    error_code = "app_asset_node_not_found"
+    description = "App asset node not found."
+    code = 404
+
+
+class AppAssetFileRequiredError(BaseHTTPException):
+    error_code = "app_asset_file_required"
+    description = "File is required."
+    code = 400
+
+
+class AppAssetPathConflictError(BaseHTTPException):
+    error_code = "app_asset_path_conflict"
+    description = "Path already exists."
+    code = 409

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,15 @@ 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.context_models import (
+    AvailableVarPayload,
+    CodeContextPayload,
+    ParameterInfoPayload,
+)
 from core.llm_generator.llm_generator import LLMGenerator
 from core.model_runtime.errors.invoke import InvokeError
 from extensions.ext_database import db
@@ -27,13 +31,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")
 
 
@@ -41,6 +60,34 @@ class InstructionTemplatePayload(BaseModel):
     type: str = Field(..., description="Instruction template type")
 
 
+class ContextGeneratePayload(BaseModel):
+    """Payload for generating extractor code node."""
+
+    language: str = Field(default="python3", description="Code language (python3/javascript)")
+    prompt_messages: list[dict[str, Any]] = Field(
+        ..., description="Multi-turn conversation history, last message is the current instruction"
+    )
+    model_config_data: dict[str, Any] = Field(..., alias="model_config", description="Model configuration")
+    available_vars: list[AvailableVarPayload] = Field(..., description="Available variables from upstream nodes")
+    parameter_info: ParameterInfoPayload = Field(..., description="Target parameter metadata from the frontend")
+    code_context: CodeContextPayload = Field(description="Existing code node context for incremental generation")
+
+
+class SuggestedQuestionsPayload(BaseModel):
+    """Payload for generating suggested questions."""
+
+    language: str = Field(
+        default="English", description="Language for generated questions (e.g. English, Chinese, Japanese)"
+    )
+    model_config_data: dict[str, Any] = Field(
+        default_factory=dict,
+        alias="model_config",
+        description="Model configuration (optional, uses system default if not provided)",
+    )
+    available_vars: list[AvailableVarPayload] = Field(..., description="Available variables from upstream nodes")
+    parameter_info: ParameterInfoPayload = Field(..., description="Target parameter metadata from the frontend")
+
+
 def reg(cls: type[BaseModel]):
     console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
 
@@ -50,7 +97,8 @@ reg(RuleCodeGeneratePayload)
 reg(RuleStructuredOutputPayload)
 reg(InstructionGeneratePayload)
 reg(InstructionTemplatePayload)
-reg(ModelConfig)
+reg(ContextGeneratePayload)
+reg(SuggestedQuestionsPayload)
 
 
 @console_ns.route("/rule-generate")
@@ -69,7 +117,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 +153,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 +187,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 +239,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}"}
@@ -263,3 +313,70 @@ class InstructionGenerationTemplateApi(Resource):
                 return {"data": INSTRUCTION_GENERATE_TEMPLATE_CODE}
             case _:
                 raise ValueError(f"Invalid type: {args.type}")
+
+
+@console_ns.route("/context-generate")
+class ContextGenerateApi(Resource):
+    @console_ns.doc("generate_with_context")
+    @console_ns.doc(description="Generate with multi-turn conversation context")
+    @console_ns.expect(console_ns.models[ContextGeneratePayload.__name__])
+    @console_ns.response(200, "Content generated successfully")
+    @console_ns.response(400, "Invalid request parameters or workflow not found")
+    @console_ns.response(402, "Provider quota exceeded")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self):
+        from core.llm_generator.utils import deserialize_prompt_messages
+
+        args = ContextGeneratePayload.model_validate(console_ns.payload)
+        _, current_tenant_id = current_account_with_tenant()
+
+        try:
+            return LLMGenerator.generate_with_context(
+                tenant_id=current_tenant_id,
+                language=args.language,
+                prompt_messages=deserialize_prompt_messages(args.prompt_messages),
+                model_config=args.model_config_data,
+                available_vars=args.available_vars,
+                parameter_info=args.parameter_info,
+                code_context=args.code_context,
+            )
+        except ProviderTokenNotInitError as ex:
+            raise ProviderNotInitializeError(ex.description)
+        except QuotaExceededError:
+            raise ProviderQuotaExceededError()
+        except ModelCurrentlyNotSupportError:
+            raise ProviderModelCurrentlyNotSupportError()
+        except InvokeError as e:
+            raise CompletionRequestError(e.description)
+
+
+@console_ns.route("/context-generate/suggested-questions")
+class SuggestedQuestionsApi(Resource):
+    @console_ns.doc("generate_suggested_questions")
+    @console_ns.doc(description="Generate suggested questions for context generation")
+    @console_ns.expect(console_ns.models[SuggestedQuestionsPayload.__name__])
+    @console_ns.response(200, "Questions generated successfully")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self):
+        args = SuggestedQuestionsPayload.model_validate(console_ns.payload)
+        _, current_tenant_id = current_account_with_tenant()
+        try:
+            return LLMGenerator.generate_suggested_questions(
+                tenant_id=current_tenant_id,
+                language=args.language,
+                available_vars=args.available_vars,
+                parameter_info=args.parameter_info,
+                model_config=args.model_config_data,
+            )
+        except ProviderTokenNotInitError as ex:
+            raise ProviderNotInitializeError(ex.description)
+        except QuotaExceededError:
+            raise ProviderQuotaExceededError()
+        except ModelCurrentlyNotSupportError:
+            raise ProviderModelCurrentlyNotSupportError()
+        except InvokeError as e:
+            raise CompletionRequestError(e.description)

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
 
 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
@@ -211,6 +202,7 @@ message_detail_model = console_ns.model(
         "status": fields.String,
         "error": fields.String,
         "parent_message_id": fields.String,
+        "generation_detail": fields.Raw,
     },
 )
 
@@ -240,7 +232,7 @@ class ChatMessageListApi(Resource):
     @marshal_with(message_infinite_scroll_pagination_model)
     @edit_permission_required
     def get(self, app_model):
-        args = ChatMessagesQuery.model_validate(request.args.to_dict())
+        args = ChatMessagesQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
 
         conversation = (
             db.session.query(Conversation)
@@ -365,7 +357,7 @@ class MessageAnnotationCountApi(Resource):
     @console_ns.response(
         200,
         "Annotation count retrieved successfully",
-        console_ns.models[AnnotationCountResponse.__name__],
+        console_ns.model("AnnotationCountResponse", {"count": fields.Integer(description="Number of annotations")}),
     )
     @get_app_model
     @setup_required
@@ -385,7 +377,9 @@ class MessageSuggestedQuestionApi(Resource):
     @console_ns.response(
         200,
         "Suggested questions retrieved successfully",
-        console_ns.models[SuggestedQuestionsResponse.__name__],
+        console_ns.model(
+            "SuggestedQuestionsResponse", {"data": fields.List(fields.String(description="Suggested question"))}
+        ),
     )
     @console_ns.response(404, "Message or conversation not found")
     @setup_required
@@ -435,7 +429,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/skills.py

@@ -0,0 +1,83 @@
+from flask_restx import Resource
+
+from controllers.console import console_ns
+from controllers.console.app.error import DraftWorkflowNotExist
+from controllers.console.app.wraps import get_app_model
+from controllers.console.wraps import account_initialization_required, current_account_with_tenant, setup_required
+from libs.login import login_required
+from models import App
+from models.model import AppMode
+from services.skill_service import SkillService
+from services.workflow_service import WorkflowService
+
+
+@console_ns.route("/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/skills")
+class NodeSkillsApi(Resource):
+    """API for retrieving skill references for a specific workflow node."""
+
+    @console_ns.doc("get_node_skills")
+    @console_ns.doc(description="Get skill references for a specific node in the draft workflow")
+    @console_ns.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
+    @console_ns.response(200, "Node skills retrieved successfully")
+    @console_ns.response(404, "Workflow or node not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    def get(self, app_model: App, node_id: str):
+        """
+        Get skill information for a specific node in the draft workflow.
+
+        Returns information about skill references in the node, including:
+        - skill_references: List of prompt messages marked as skills
+        - tool_references: Aggregated tool references from all skill prompts
+        - file_references: Aggregated file references from all skill prompts
+        """
+        current_user, _ = current_account_with_tenant()
+        workflow_service = WorkflowService()
+        workflow = workflow_service.get_draft_workflow(app_model=app_model)
+
+        if not workflow:
+            raise DraftWorkflowNotExist()
+
+        skill_info = SkillService.get_node_skill_info(
+            app=app_model,
+            workflow=workflow,
+            node_id=node_id,
+            user_id=current_user.id,
+        )
+        return skill_info.model_dump()
+
+
+@console_ns.route("/apps/<uuid:app_id>/workflows/draft/skills")
+class WorkflowSkillsApi(Resource):
+    """API for retrieving all skill references in a workflow."""
+
+    @console_ns.doc("get_workflow_skills")
+    @console_ns.doc(description="Get all skill references in the draft workflow")
+    @console_ns.doc(params={"app_id": "Application ID"})
+    @console_ns.response(200, "Workflow skills retrieved successfully")
+    @console_ns.response(404, "Workflow not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    def get(self, app_model: App):
+        """
+        Get skill information for all nodes in the draft workflow that have skill references.
+
+        Returns a list of nodes with their skill information.
+        """
+        current_user, _ = current_account_with_tenant()
+        workflow_service = WorkflowService()
+        workflow = workflow_service.get_draft_workflow(app_model=app_model)
+
+        if not workflow:
+            raise DraftWorkflowNotExist()
+
+        skills_info = SkillService.get_workflow_skills(
+            app=app_model,
+            workflow=workflow,
+            user_id=current_user.id,
+        )
+        return {"nodes": [info.model_dump() for info in skills_info]}

api/controllers/console/app/workflow.py

@@ -33,8 +33,10 @@ from core.trigger.debug.event_selectors import (
 from core.workflow.enums import NodeType
 from core.workflow.graph_engine.manager import GraphEngineManager
 from extensions.ext_database import db
+from extensions.ext_redis import redis_client
 from factories import file_factory, variable_factory
 from fields.member_fields import simple_account_fields
+from fields.online_user_fields import online_user_list_fields
 from fields.workflow_fields import workflow_fields, workflow_pagination_fields
 from libs import helper
 from libs.datetime_utils import naive_utc_now
@@ -43,9 +45,12 @@ from libs.login import current_account_with_tenant, login_required
 from models import App
 from models.model import AppMode
 from models.workflow import Workflow
+from repositories.workflow_collaboration_repository import WORKFLOW_ONLINE_USERS_PREFIX
 from services.app_generate_service import AppGenerateService
 from services.errors.app import WorkflowHashNotEqualError
 from services.errors.llm import InvokeRateLimitError
+from services.workflow.entities import NestedNodeGraphRequest, NestedNodeParameterSchema
+from services.workflow.nested_node_graph_service import NestedNodeGraphService
 from services.workflow_service import DraftWorkflowDeletionError, WorkflowInUseError, WorkflowService
 
 logger = logging.getLogger(__name__)
@@ -160,6 +165,14 @@ class WorkflowUpdatePayload(BaseModel):
     marked_comment: str | None = Field(default=None, max_length=100)
 
 
+class WorkflowFeaturesPayload(BaseModel):
+    features: dict[str, Any] = Field(..., description="Workflow feature configuration")
+
+
+class WorkflowOnlineUsersQuery(BaseModel):
+    workflow_ids: str = Field(..., description="Comma-separated workflow IDs")
+
+
 class DraftWorkflowTriggerRunPayload(BaseModel):
     node_id: str
 
@@ -168,6 +181,15 @@ class DraftWorkflowTriggerRunAllPayload(BaseModel):
     node_ids: list[str]
 
 
+class NestedNodeGraphPayload(BaseModel):
+    """Request payload for generating nested node graph."""
+
+    parent_node_id: str = Field(description="ID of the parent node that uses the extracted value")
+    parameter_key: str = Field(description="Key of the parameter being extracted")
+    context_source: list[str] = Field(description="Variable selector for the context source")
+    parameter_schema: dict[str, Any] = Field(description="Schema of the parameter to extract")
+
+
 def reg(cls: type[BaseModel]):
     console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
 
@@ -183,8 +205,11 @@ reg(DefaultBlockConfigQuery)
 reg(ConvertToWorkflowPayload)
 reg(WorkflowListQuery)
 reg(WorkflowUpdatePayload)
+reg(WorkflowFeaturesPayload)
+reg(WorkflowOnlineUsersQuery)
 reg(DraftWorkflowTriggerRunPayload)
 reg(DraftWorkflowTriggerRunAllPayload)
+reg(NestedNodeGraphPayload)
 
 
 # TODO(QuantumGhost): Refactor existing node run API to handle file parameter parsing
@@ -654,13 +679,14 @@ class PublishedWorkflowApi(Resource):
         """
         Publish workflow
         """
+        from services.app_bundle_service import AppBundleService
+
         current_user, _ = current_account_with_tenant()
 
         args = PublishWorkflowPayload.model_validate(console_ns.payload or {})
 
-        workflow_service = WorkflowService()
         with Session(db.engine) as session:
-            workflow = workflow_service.publish_workflow(
+            workflow = AppBundleService.publish(
                 session=session,
                 app_model=app_model,
                 account=current_user,
@@ -771,6 +797,31 @@ class ConvertToWorkflowApi(Resource):
         }
 
 
+@console_ns.route("/apps/<uuid:app_id>/workflows/draft/features")
+class WorkflowFeaturesApi(Resource):
+    """Update draft workflow features."""
+
+    @console_ns.expect(console_ns.models[WorkflowFeaturesPayload.__name__])
+    @console_ns.doc("update_workflow_features")
+    @console_ns.doc(description="Update draft workflow features")
+    @console_ns.doc(params={"app_id": "Application ID"})
+    @console_ns.response(200, "Workflow features updated successfully")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    def post(self, app_model: App):
+        current_user, _ = current_account_with_tenant()
+
+        args = WorkflowFeaturesPayload.model_validate(console_ns.payload or {})
+        features = args.features
+
+        workflow_service = WorkflowService()
+        workflow_service.update_draft_workflow_features(app_model=app_model, features=features, account=current_user)
+
+        return {"result": "success"}
+
+
 @console_ns.route("/apps/<uuid:app_id>/workflows")
 class PublishedAllWorkflowApi(Resource):
     @console_ns.expect(console_ns.models[WorkflowListQuery.__name__])
@@ -1148,3 +1199,83 @@ class DraftWorkflowTriggerRunAllApi(Resource):
                     "status": "error",
                 }
             ), 400
+
+
+@console_ns.route("/apps/<uuid:app_id>/workflows/draft/nested-node-graph")
+class NestedNodeGraphApi(Resource):
+    """
+    API for generating Nested Node LLM graph structures.
+
+    This endpoint creates a complete graph structure containing an LLM node
+    configured to extract values from list[PromptMessage] variables.
+    """
+
+    @console_ns.doc("generate_nested_node_graph")
+    @console_ns.doc(description="Generate a Nested Node LLM graph structure")
+    @console_ns.doc(params={"app_id": "Application ID"})
+    @console_ns.expect(console_ns.models[NestedNodeGraphPayload.__name__])
+    @console_ns.response(200, "Nested node graph generated successfully")
+    @console_ns.response(400, "Invalid request parameters")
+    @console_ns.response(403, "Permission denied")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    @edit_permission_required
+    def post(self, app_model: App):
+        """
+        Generate a Nested Node LLM graph structure.
+
+        Returns a complete graph structure containing a single LLM node
+        configured for extracting values from list[PromptMessage] context.
+        """
+
+        payload = NestedNodeGraphPayload.model_validate(console_ns.payload or {})
+
+        parameter_schema = NestedNodeParameterSchema(
+            name=payload.parameter_schema.get("name", payload.parameter_key),
+            type=payload.parameter_schema.get("type", "string"),
+            description=payload.parameter_schema.get("description", ""),
+        )
+
+        request = NestedNodeGraphRequest(
+            parent_node_id=payload.parent_node_id,
+            parameter_key=payload.parameter_key,
+            context_source=payload.context_source,
+            parameter_schema=parameter_schema,
+        )
+
+        with Session(db.engine) as session:
+            service = NestedNodeGraphService(session)
+            response = service.generate_nested_node_graph(tenant_id=app_model.tenant_id, request=request)
+
+        return response.model_dump()
+
+
+@console_ns.route("/apps/workflows/online-users")
+class WorkflowOnlineUsersApi(Resource):
+    @console_ns.expect(console_ns.models[WorkflowOnlineUsersQuery.__name__])
+    @console_ns.doc("get_workflow_online_users")
+    @console_ns.doc(description="Get workflow online users")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @marshal_with(online_user_list_fields)
+    def get(self):
+        args = WorkflowOnlineUsersQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+
+        workflow_ids = [workflow_id.strip() for workflow_id in args.workflow_ids.split(",") if workflow_id.strip()]
+
+        results = []
+        for workflow_id in workflow_ids:
+            users_json = redis_client.hgetall(f"{WORKFLOW_ONLINE_USERS_PREFIX}{workflow_id}")
+
+            users = []
+            for _, user_info_json in users_json.items():
+                try:
+                    users.append(json.loads(user_info_json))
+                except Exception:
+                    continue
+            results.append({"workflow_id": workflow_id, "users": users})
+
+        return {"data": results}

api/controllers/console/app/workflow_comment.py

@@ -0,0 +1,317 @@
+import logging
+
+from flask_restx import Resource, fields, marshal_with
+from pydantic import BaseModel, Field
+
+from controllers.console import console_ns
+from controllers.console.app.wraps import get_app_model
+from controllers.console.wraps import account_initialization_required, setup_required
+from fields.member_fields import account_with_role_fields
+from fields.workflow_comment_fields import (
+    workflow_comment_basic_fields,
+    workflow_comment_create_fields,
+    workflow_comment_detail_fields,
+    workflow_comment_reply_create_fields,
+    workflow_comment_reply_update_fields,
+    workflow_comment_resolve_fields,
+    workflow_comment_update_fields,
+)
+from libs.login import current_user, login_required
+from models import App
+from services.account_service import TenantService
+from services.workflow_comment_service import WorkflowCommentService
+
+logger = logging.getLogger(__name__)
+DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
+
+
+class WorkflowCommentCreatePayload(BaseModel):
+    position_x: float = Field(..., description="Comment X position")
+    position_y: float = Field(..., description="Comment Y position")
+    content: str = Field(..., description="Comment content")
+    mentioned_user_ids: list[str] = Field(default_factory=list, description="Mentioned user IDs")
+
+
+class WorkflowCommentUpdatePayload(BaseModel):
+    content: str = Field(..., description="Comment content")
+    position_x: float | None = Field(default=None, description="Comment X position")
+    position_y: float | None = Field(default=None, description="Comment Y position")
+    mentioned_user_ids: list[str] = Field(default_factory=list, description="Mentioned user IDs")
+
+
+class WorkflowCommentReplyCreatePayload(BaseModel):
+    content: str = Field(..., description="Reply content")
+    mentioned_user_ids: list[str] = Field(default_factory=list, description="Mentioned user IDs")
+
+
+class WorkflowCommentReplyUpdatePayload(BaseModel):
+    content: str = Field(..., description="Reply content")
+    mentioned_user_ids: list[str] = Field(default_factory=list, description="Mentioned user IDs")
+
+
+for model in (
+    WorkflowCommentCreatePayload,
+    WorkflowCommentUpdatePayload,
+    WorkflowCommentReplyCreatePayload,
+    WorkflowCommentReplyUpdatePayload,
+):
+    console_ns.schema_model(model.__name__, model.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
+
+workflow_comment_basic_model = console_ns.model("WorkflowCommentBasic", workflow_comment_basic_fields)
+workflow_comment_detail_model = console_ns.model("WorkflowCommentDetail", workflow_comment_detail_fields)
+workflow_comment_create_model = console_ns.model("WorkflowCommentCreate", workflow_comment_create_fields)
+workflow_comment_update_model = console_ns.model("WorkflowCommentUpdate", workflow_comment_update_fields)
+workflow_comment_resolve_model = console_ns.model("WorkflowCommentResolve", workflow_comment_resolve_fields)
+workflow_comment_reply_create_model = console_ns.model(
+    "WorkflowCommentReplyCreate", workflow_comment_reply_create_fields
+)
+workflow_comment_reply_update_model = console_ns.model(
+    "WorkflowCommentReplyUpdate", workflow_comment_reply_update_fields
+)
+workflow_comment_mention_users_model = console_ns.model(
+    "WorkflowCommentMentionUsers",
+    {"users": fields.List(fields.Nested(account_with_role_fields))},
+)
+
+
+@console_ns.route("/apps/<uuid:app_id>/workflow/comments")
+class WorkflowCommentListApi(Resource):
+    """API for listing and creating workflow comments."""
+
+    @console_ns.doc("list_workflow_comments")
+    @console_ns.doc(description="Get all comments for a workflow")
+    @console_ns.doc(params={"app_id": "Application ID"})
+    @console_ns.response(200, "Comments retrieved successfully", workflow_comment_basic_model)
+    @login_required
+    @setup_required
+    @account_initialization_required
+    @get_app_model()
+    @marshal_with(workflow_comment_basic_model, envelope="data")
+    def get(self, app_model: App):
+        """Get all comments for a workflow."""
+        comments = WorkflowCommentService.get_comments(tenant_id=current_user.current_tenant_id, app_id=app_model.id)
+
+        return comments
+
+    @console_ns.doc("create_workflow_comment")
+    @console_ns.doc(description="Create a new workflow comment")
+    @console_ns.doc(params={"app_id": "Application ID"})
+    @console_ns.expect(console_ns.models[WorkflowCommentCreatePayload.__name__])
+    @console_ns.response(201, "Comment created successfully", workflow_comment_create_model)
+    @login_required
+    @setup_required
+    @account_initialization_required
+    @get_app_model()
+    @marshal_with(workflow_comment_create_model)
+    def post(self, app_model: App):
+        """Create a new workflow comment."""
+        payload = WorkflowCommentCreatePayload.model_validate(console_ns.payload or {})
+
+        result = WorkflowCommentService.create_comment(
+            tenant_id=current_user.current_tenant_id,
+            app_id=app_model.id,
+            created_by=current_user.id,
+            content=payload.content,
+            position_x=payload.position_x,
+            position_y=payload.position_y,
+            mentioned_user_ids=payload.mentioned_user_ids,
+        )
+
+        return result, 201
+
+
+@console_ns.route("/apps/<uuid:app_id>/workflow/comments/<string:comment_id>")
+class WorkflowCommentDetailApi(Resource):
+    """API for managing individual workflow comments."""
+
+    @console_ns.doc("get_workflow_comment")
+    @console_ns.doc(description="Get a specific workflow comment")
+    @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID"})
+    @console_ns.response(200, "Comment retrieved successfully", workflow_comment_detail_model)
+    @login_required
+    @setup_required
+    @account_initialization_required
+    @get_app_model()
+    @marshal_with(workflow_comment_detail_model)
+    def get(self, app_model: App, comment_id: str):
+        """Get a specific workflow comment."""
+        comment = WorkflowCommentService.get_comment(
+            tenant_id=current_user.current_tenant_id, app_id=app_model.id, comment_id=comment_id
+        )
+
+        return comment
+
+    @console_ns.doc("update_workflow_comment")
+    @console_ns.doc(description="Update a workflow comment")
+    @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID"})
+    @console_ns.expect(console_ns.models[WorkflowCommentUpdatePayload.__name__])
+    @console_ns.response(200, "Comment updated successfully", workflow_comment_update_model)
+    @login_required
+    @setup_required
+    @account_initialization_required
+    @get_app_model()
+    @marshal_with(workflow_comment_update_model)
+    def put(self, app_model: App, comment_id: str):
+        """Update a workflow comment."""
+        payload = WorkflowCommentUpdatePayload.model_validate(console_ns.payload or {})
+
+        result = WorkflowCommentService.update_comment(
+            tenant_id=current_user.current_tenant_id,
+            app_id=app_model.id,
+            comment_id=comment_id,
+            user_id=current_user.id,
+            content=payload.content,
+            position_x=payload.position_x,
+            position_y=payload.position_y,
+            mentioned_user_ids=payload.mentioned_user_ids,
+        )
+
+        return result
+
+    @console_ns.doc("delete_workflow_comment")
+    @console_ns.doc(description="Delete a workflow comment")
+    @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID"})
+    @console_ns.response(204, "Comment deleted successfully")
+    @login_required
+    @setup_required
+    @account_initialization_required
+    @get_app_model()
+    def delete(self, app_model: App, comment_id: str):
+        """Delete a workflow comment."""
+        WorkflowCommentService.delete_comment(
+            tenant_id=current_user.current_tenant_id,
+            app_id=app_model.id,
+            comment_id=comment_id,
+            user_id=current_user.id,
+        )
+
+        return {"result": "success"}, 204
+
+
+@console_ns.route("/apps/<uuid:app_id>/workflow/comments/<string:comment_id>/resolve")
+class WorkflowCommentResolveApi(Resource):
+    """API for resolving and reopening workflow comments."""
+
+    @console_ns.doc("resolve_workflow_comment")
+    @console_ns.doc(description="Resolve a workflow comment")
+    @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID"})
+    @console_ns.response(200, "Comment resolved successfully", workflow_comment_resolve_model)
+    @login_required
+    @setup_required
+    @account_initialization_required
+    @get_app_model()
+    @marshal_with(workflow_comment_resolve_model)
+    def post(self, app_model: App, comment_id: str):
+        """Resolve a workflow comment."""
+        comment = WorkflowCommentService.resolve_comment(
+            tenant_id=current_user.current_tenant_id,
+            app_id=app_model.id,
+            comment_id=comment_id,
+            user_id=current_user.id,
+        )
+
+        return comment
+
+
+@console_ns.route("/apps/<uuid:app_id>/workflow/comments/<string:comment_id>/replies")
+class WorkflowCommentReplyApi(Resource):
+    """API for managing comment replies."""
+
+    @console_ns.doc("create_workflow_comment_reply")
+    @console_ns.doc(description="Add a reply to a workflow comment")
+    @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID"})
+    @console_ns.expect(console_ns.models[WorkflowCommentReplyCreatePayload.__name__])
+    @console_ns.response(201, "Reply created successfully", workflow_comment_reply_create_model)
+    @login_required
+    @setup_required
+    @account_initialization_required
+    @get_app_model()
+    @marshal_with(workflow_comment_reply_create_model)
+    def post(self, app_model: App, comment_id: str):
+        """Add a reply to a workflow comment."""
+        # Validate comment access first
+        WorkflowCommentService.validate_comment_access(
+            comment_id=comment_id, tenant_id=current_user.current_tenant_id, app_id=app_model.id
+        )
+
+        payload = WorkflowCommentReplyCreatePayload.model_validate(console_ns.payload or {})
+
+        result = WorkflowCommentService.create_reply(
+            comment_id=comment_id,
+            content=payload.content,
+            created_by=current_user.id,
+            mentioned_user_ids=payload.mentioned_user_ids,
+        )
+
+        return result, 201
+
+
+@console_ns.route("/apps/<uuid:app_id>/workflow/comments/<string:comment_id>/replies/<string:reply_id>")
+class WorkflowCommentReplyDetailApi(Resource):
+    """API for managing individual comment replies."""
+
+    @console_ns.doc("update_workflow_comment_reply")
+    @console_ns.doc(description="Update a comment reply")
+    @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID", "reply_id": "Reply ID"})
+    @console_ns.expect(console_ns.models[WorkflowCommentReplyUpdatePayload.__name__])
+    @console_ns.response(200, "Reply updated successfully", workflow_comment_reply_update_model)
+    @login_required
+    @setup_required
+    @account_initialization_required
+    @get_app_model()
+    @marshal_with(workflow_comment_reply_update_model)
+    def put(self, app_model: App, comment_id: str, reply_id: str):
+        """Update a comment reply."""
+        # Validate comment access first
+        WorkflowCommentService.validate_comment_access(
+            comment_id=comment_id, tenant_id=current_user.current_tenant_id, app_id=app_model.id
+        )
+
+        payload = WorkflowCommentReplyUpdatePayload.model_validate(console_ns.payload or {})
+
+        reply = WorkflowCommentService.update_reply(
+            reply_id=reply_id,
+            user_id=current_user.id,
+            content=payload.content,
+            mentioned_user_ids=payload.mentioned_user_ids,
+        )
+
+        return reply
+
+    @console_ns.doc("delete_workflow_comment_reply")
+    @console_ns.doc(description="Delete a comment reply")
+    @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID", "reply_id": "Reply ID"})
+    @console_ns.response(204, "Reply deleted successfully")
+    @login_required
+    @setup_required
+    @account_initialization_required
+    @get_app_model()
+    def delete(self, app_model: App, comment_id: str, reply_id: str):
+        """Delete a comment reply."""
+        # Validate comment access first
+        WorkflowCommentService.validate_comment_access(
+            comment_id=comment_id, tenant_id=current_user.current_tenant_id, app_id=app_model.id
+        )
+
+        WorkflowCommentService.delete_reply(reply_id=reply_id, user_id=current_user.id)
+
+        return {"result": "success"}, 204
+
+
+@console_ns.route("/apps/<uuid:app_id>/workflow/comments/mention-users")
+class WorkflowCommentMentionUsersApi(Resource):
+    """API for getting mentionable users for workflow comments."""
+
+    @console_ns.doc("workflow_comment_mention_users")
+    @console_ns.doc(description="Get all users in current tenant for mentions")
+    @console_ns.doc(params={"app_id": "Application ID"})
+    @console_ns.response(200, "Mentionable users retrieved successfully", workflow_comment_mention_users_model)
+    @login_required
+    @setup_required
+    @account_initialization_required
+    @get_app_model()
+    @marshal_with(workflow_comment_mention_users_model)
+    def get(self, app_model: App):
+        """Get all users in current tenant for mentions."""
+        members = TenantService.get_tenant_members(current_user.current_tenant)
+        return {"users": members}

api/controllers/console/app/workflow_draft_variable.py

@@ -17,15 +17,16 @@ from controllers.console.wraps import account_initialization_required, edit_perm
 from controllers.web.error import InvalidArgumentError, NotFoundError
 from core.file import helpers as file_helpers
 from core.variables.segment_group import SegmentGroup
-from core.variables.segments import ArrayFileSegment, FileSegment, Segment
+from core.variables.segments import ArrayFileSegment, ArrayPromptMessageSegment, FileSegment, Segment
 from core.variables.types import SegmentType
 from core.workflow.constants import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
 from extensions.ext_database import db
+from factories import variable_factory
 from factories.file_factory import build_from_mapping, build_from_mappings
-from factories.variable_factory import build_segment_with_type
-from libs.login import login_required
+from libs.login import current_account_with_tenant, login_required
 from models import App, AppMode
 from models.workflow import WorkflowDraftVariable
+from services.sandbox.sandbox_service import SandboxService
 from services.workflow_draft_variable_service import WorkflowDraftVariableList, WorkflowDraftVariableService
 from services.workflow_service import WorkflowService
 
@@ -43,6 +44,16 @@ class WorkflowDraftVariableUpdatePayload(BaseModel):
     value: Any | None = Field(default=None, description="Variable value")
 
 
+class ConversationVariableUpdatePayload(BaseModel):
+    conversation_variables: list[dict[str, Any]] = Field(
+        ..., description="Conversation variables for the draft workflow"
+    )
+
+
+class EnvironmentVariableUpdatePayload(BaseModel):
+    environment_variables: list[dict[str, Any]] = Field(..., description="Environment variables for the draft workflow")
+
+
 console_ns.schema_model(
     WorkflowDraftVariableListQuery.__name__,
     WorkflowDraftVariableListQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
@@ -51,6 +62,14 @@ console_ns.schema_model(
     WorkflowDraftVariableUpdatePayload.__name__,
     WorkflowDraftVariableUpdatePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
 )
+console_ns.schema_model(
+    ConversationVariableUpdatePayload.__name__,
+    ConversationVariableUpdatePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+console_ns.schema_model(
+    EnvironmentVariableUpdatePayload.__name__,
+    EnvironmentVariableUpdatePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
 
 
 def _convert_values_to_json_serializable_object(value: Segment):
@@ -58,6 +77,8 @@ def _convert_values_to_json_serializable_object(value: Segment):
         return value.value.model_dump()
     elif isinstance(value, ArrayFileSegment):
         return [i.model_dump() for i in value.value]
+    elif isinstance(value, ArrayPromptMessageSegment):
+        return value.to_object()
     elif isinstance(value, SegmentGroup):
         return [_convert_values_to_json_serializable_object(i) for i in value.value]
     else:
@@ -247,6 +268,8 @@ class WorkflowVariableCollectionApi(Resource):
     @console_ns.response(204, "Workflow variables deleted successfully")
     @_api_prerequisite
     def delete(self, app_model: App):
+        current_user, _ = current_account_with_tenant()
+        SandboxService.delete_draft_storage(app_model.tenant_id, current_user.id)
         draft_var_srv = WorkflowDraftVariableService(
             session=db.session(),
         )
@@ -383,7 +406,7 @@ class VariableApi(Resource):
                 if len(raw_value) > 0 and not isinstance(raw_value[0], dict):
                     raise InvalidArgumentError(description=f"expected dict for files[0], got {type(raw_value)}")
                 raw_value = build_from_mappings(mappings=raw_value, tenant_id=app_model.tenant_id)
-            new_value = build_segment_with_type(variable.value_type, raw_value)
+            new_value = variable_factory.build_segment_with_type(variable.value_type, raw_value)
         draft_var_srv.update_variable(variable, name=new_name, value=new_value)
         db.session.commit()
         return variable
@@ -476,6 +499,34 @@ class ConversationVariableCollectionApi(Resource):
         db.session.commit()
         return _get_variable_list(app_model, CONVERSATION_VARIABLE_NODE_ID)
 
+    @console_ns.expect(console_ns.models[ConversationVariableUpdatePayload.__name__])
+    @console_ns.doc("update_conversation_variables")
+    @console_ns.doc(description="Update conversation variables for workflow draft")
+    @console_ns.doc(params={"app_id": "Application ID"})
+    @console_ns.response(200, "Conversation variables updated successfully")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    @get_app_model(mode=AppMode.ADVANCED_CHAT)
+    def post(self, app_model: App):
+        payload = ConversationVariableUpdatePayload.model_validate(console_ns.payload or {})
+
+        workflow_service = WorkflowService()
+
+        conversation_variables_list = payload.conversation_variables
+        conversation_variables = [
+            variable_factory.build_conversation_variable_from_mapping(obj) for obj in conversation_variables_list
+        ]
+
+        workflow_service.update_draft_workflow_conversation_variables(
+            app_model=app_model,
+            account=current_user,
+            conversation_variables=conversation_variables,
+        )
+
+        return {"result": "success"}
+
 
 @console_ns.route("/apps/<uuid:app_id>/workflows/draft/system-variables")
 class SystemVariableCollectionApi(Resource):
@@ -527,3 +578,31 @@ class EnvironmentVariableCollectionApi(Resource):
             )
 
         return {"items": env_vars_list}
+
+    @console_ns.expect(console_ns.models[EnvironmentVariableUpdatePayload.__name__])
+    @console_ns.doc("update_environment_variables")
+    @console_ns.doc(description="Update environment variables for workflow draft")
+    @console_ns.doc(params={"app_id": "Application ID"})
+    @console_ns.response(200, "Environment variables updated successfully")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    def post(self, app_model: App):
+        payload = EnvironmentVariableUpdatePayload.model_validate(console_ns.payload or {})
+
+        workflow_service = WorkflowService()
+
+        environment_variables_list = payload.environment_variables
+        environment_variables = [
+            variable_factory.build_environment_variable_from_mapping(obj) for obj in environment_variables_list
+        ]
+
+        workflow_service.update_draft_workflow_environment_variables(
+            app_model=app_model,
+            account=current_user,
+            environment_variables=environment_variables,
+        )
+
+        return {"result": "success"}

api/controllers/console/auth/data_source_oauth.py

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

api/controllers/console/auth/forgot_password.py

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

api/controllers/console/auth/oauth_server.py

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

api/controllers/console/datasets/data_source.py

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

api/controllers/console/datasets/datasets_document.py

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

api/controllers/console/datasets/metadata.py

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

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

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

api/controllers/console/explore/trial.py

@@ -9,7 +9,7 @@ import services
 from controllers.common.fields import Parameters as ParametersResponse
 from controllers.common.fields import Site as SiteResponse
 from controllers.common.schema import get_or_create_model
-from controllers.console import api
+from controllers.console import api, console_ns
 from controllers.console.app.error import (
     AppUnavailableError,
     AudioTooLargeError,
@@ -51,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,
@@ -103,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)
 

api/controllers/console/init_validate.py

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

api/controllers/console/remote_files.py

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

api/controllers/console/sandbox_files.py

@@ -0,0 +1,86 @@
+from __future__ import annotations
+
+from flask import request
+from flask_restx import Resource, fields
+from pydantic import BaseModel, Field
+
+from controllers.console import console_ns
+from controllers.console.wraps import account_initialization_required, setup_required
+from libs.login import current_account_with_tenant, login_required
+from services.sandbox.sandbox_file_service import SandboxFileService
+
+DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
+
+
+class SandboxFileListQuery(BaseModel):
+    path: str | None = Field(default=None, description="Workspace relative path")
+    recursive: bool = Field(default=False, description="List recursively")
+
+
+class SandboxFileDownloadRequest(BaseModel):
+    path: str = Field(..., description="Workspace relative file path")
+
+
+console_ns.schema_model(
+    SandboxFileListQuery.__name__,
+    SandboxFileListQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+console_ns.schema_model(
+    SandboxFileDownloadRequest.__name__,
+    SandboxFileDownloadRequest.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+
+
+SANDBOX_FILE_NODE_FIELDS = {
+    "path": fields.String,
+    "is_dir": fields.Boolean,
+    "size": fields.Raw,
+    "mtime": fields.Raw,
+    "extension": fields.String,
+}
+
+
+SANDBOX_FILE_DOWNLOAD_TICKET_FIELDS = {
+    "download_url": fields.String,
+    "expires_in": fields.Integer,
+    "export_id": fields.String,
+}
+
+
+sandbox_file_node_model = console_ns.model("SandboxFileNode", SANDBOX_FILE_NODE_FIELDS)
+sandbox_file_download_ticket_model = console_ns.model("SandboxFileDownloadTicket", SANDBOX_FILE_DOWNLOAD_TICKET_FIELDS)
+
+
+@console_ns.route("/sandboxes/<string:sandbox_id>/files")
+class SandboxFilesApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @console_ns.expect(console_ns.models[SandboxFileListQuery.__name__])
+    @console_ns.marshal_list_with(sandbox_file_node_model)
+    def get(self, sandbox_id: str):
+        args = SandboxFileListQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore[arg-type]
+        _, tenant_id = current_account_with_tenant()
+        return [
+            e.__dict__
+            for e in SandboxFileService.list_files(
+                tenant_id=tenant_id,
+                sandbox_id=sandbox_id,
+                path=args.path,
+                recursive=args.recursive,
+            )
+        ]
+
+
+@console_ns.route("/sandboxes/<string:sandbox_id>/files/download")
+class SandboxFileDownloadApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @console_ns.expect(console_ns.models[SandboxFileDownloadRequest.__name__])
+    @console_ns.marshal_with(sandbox_file_download_ticket_model)
+    def post(self, sandbox_id: str):
+        payload = SandboxFileDownloadRequest.model_validate(console_ns.payload or {})
+        _, tenant_id = current_account_with_tenant()
+        res = SandboxFileService.download_file(tenant_id=tenant_id, sandbox_id=sandbox_id, path=payload.path)
+        return res.__dict__

api/controllers/console/socketio/__init__.py

@@ -0,0 +1 @@
+

api/controllers/console/socketio/workflow.py

@@ -0,0 +1,119 @@
+import logging
+from collections.abc import Callable
+from typing import cast
+
+from flask import Request as FlaskRequest
+
+from extensions.ext_socketio import sio
+from libs.passport import PassportService
+from libs.token import extract_access_token
+from repositories.workflow_collaboration_repository import WorkflowCollaborationRepository
+from services.account_service import AccountService
+from services.workflow_collaboration_service import WorkflowCollaborationService
+
+repository = WorkflowCollaborationRepository()
+collaboration_service = WorkflowCollaborationService(repository, sio)
+
+
+def _sio_on(event: str) -> Callable[[Callable[..., object]], Callable[..., object]]:
+    return cast(Callable[[Callable[..., object]], Callable[..., object]], sio.on(event))
+
+
+@_sio_on("connect")
+def socket_connect(sid, environ, auth):
+    """
+    WebSocket connect event, do authentication here.
+    """
+    try:
+        request_environ = FlaskRequest(environ)
+        token = extract_access_token(request_environ)
+    except Exception:
+        logging.exception("Failed to extract token")
+        token = None
+
+    if not token:
+        logging.warning("Socket connect rejected: missing token (sid=%s)", sid)
+        return False
+
+    try:
+        decoded = PassportService().verify(token)
+        user_id = decoded.get("user_id")
+        if not user_id:
+            logging.warning("Socket connect rejected: missing user_id (sid=%s)", sid)
+            return False
+
+        with sio.app.app_context():
+            user = AccountService.load_logged_in_account(account_id=user_id)
+            if not user:
+                logging.warning("Socket connect rejected: user not found (user_id=%s, sid=%s)", user_id, sid)
+                return False
+            if not user.has_edit_permission:
+                logging.warning("Socket connect rejected: no edit permission (user_id=%s, sid=%s)", user_id, sid)
+                return False
+
+            collaboration_service.save_session(sid, user)
+            return True
+
+    except Exception:
+        logging.exception("Socket authentication failed")
+        return False
+
+
+@_sio_on("user_connect")
+def handle_user_connect(sid, data):
+    """
+    Handle user connect event. Each session (tab) is treated as an independent collaborator.
+    """
+    workflow_id = data.get("workflow_id")
+    if not workflow_id:
+        return {"msg": "workflow_id is required"}, 400
+
+    result = collaboration_service.register_session(workflow_id, sid)
+    if not result:
+        return {"msg": "unauthorized"}, 401
+
+    user_id, is_leader = result
+    return {"msg": "connected", "user_id": user_id, "sid": sid, "isLeader": is_leader}
+
+
+@_sio_on("disconnect")
+def handle_disconnect(sid):
+    """
+    Handle session disconnect event. Remove the specific session from online users.
+    """
+    collaboration_service.disconnect_session(sid)
+
+
+@_sio_on("collaboration_event")
+def handle_collaboration_event(sid, data):
+    """
+    Handle general collaboration events, include:
+    1. mouse_move
+    2. vars_and_features_update
+    3. sync_request (ask leader to update graph)
+    4. app_state_update
+    5. mcp_server_update
+    6. workflow_update
+    7. comments_update
+    8. node_panel_presence
+    9. skill_file_active
+    10. skill_sync_request
+    11. skill_resync_request
+    """
+    return collaboration_service.relay_collaboration_event(sid, data)
+
+
+@_sio_on("graph_event")
+def handle_graph_event(sid, data):
+    """
+    Handle graph events - simple broadcast relay.
+    """
+    return collaboration_service.relay_graph_event(sid, data)
+
+
+@_sio_on("skill_event")
+def handle_skill_event(sid, data):
+    """
+    Handle skill events - simple broadcast relay.
+    """
+    return collaboration_service.relay_skill_event(sid, data)

api/controllers/console/tag/tags.py

@@ -1,11 +1,14 @@
 from typing import Literal
-from uuid import UUID
 
+from flask import request
+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 controllers.fastopenapi import console_router
+from fields.tag_fields import dataset_tag_fields
 from libs.login import current_account_with_tenant, login_required
 from services.tag_service import TagService
 
@@ -32,129 +35,115 @@ class TagListQueryParam(BaseModel):
     keyword: str | None = Field(None, description="Search keyword")
 
 
-class TagResponse(BaseModel):
-    id: str = Field(description="Tag ID")
-    name: str = Field(description="Tag name")
-    type: str = Field(description="Tag type")
-    binding_count: int = Field(description="Number of bindings")
-
-
-class TagBindingResult(BaseModel):
-    result: Literal["success"] = Field(description="Operation result", examples=["success"])
-
-
-@console_router.get(
-    "/tags",
-    response_model=list[TagResponse],
-    tags=["console"],
+register_schema_models(
+    console_ns,
+    TagBasePayload,
+    TagBindingPayload,
+    TagBindingRemovePayload,
+    TagListQueryParam,
 )
-@setup_required
-@login_required
-@account_initialization_required
-def list_tags(query: TagListQueryParam) -> list[TagResponse]:
-    _, current_tenant_id = current_account_with_tenant()
-    tags = TagService.get_tags(query.type, current_tenant_id, query.keyword)
-
-    return [
-        TagResponse(
-            id=tag.id,
-            name=tag.name,
-            type=tag.type,
-            binding_count=int(tag.binding_count),
-        )
-        for tag in tags
-    ]
 
 
-@console_router.post(
-    "/tags",
-    response_model=TagResponse,
-    tags=["console"],
-)
-@setup_required
-@login_required
-@account_initialization_required
-def create_tag(payload: TagBasePayload) -> TagResponse:
-    current_user, _ = current_account_with_tenant()
-    # The role of the current user in the tag table must be admin, owner, or editor
-    if not (current_user.has_edit_permission or current_user.is_dataset_editor):
-        raise Forbidden()
+@console_ns.route("/tags")
+class TagListApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @console_ns.doc(
+        params={"type": 'Tag type filter. Can be "knowledge" or "app".', "keyword": "Search keyword for tag name."}
+    )
+    @marshal_with(dataset_tag_fields)
+    def get(self):
+        _, current_tenant_id = current_account_with_tenant()
+        raw_args = request.args.to_dict()
+        param = TagListQueryParam.model_validate(raw_args)
+        tags = TagService.get_tags(param.type, current_tenant_id, param.keyword)
 
-    tag = TagService.save_tags(payload.model_dump())
+        return tags, 200
 
-    return TagResponse(id=tag.id, name=tag.name, type=tag.type, binding_count=0)
+    @console_ns.expect(console_ns.models[TagBasePayload.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self):
+        current_user, _ = current_account_with_tenant()
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not (current_user.has_edit_permission or current_user.is_dataset_editor):
+            raise Forbidden()
+
+        payload = TagBasePayload.model_validate(console_ns.payload or {})
+        tag = TagService.save_tags(payload.model_dump())
+
+        response = {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": 0}
+
+        return response, 200
 
 
-@console_router.patch(
-    "/tags/<uuid:tag_id>",
-    response_model=TagResponse,
-    tags=["console"],
-)
-@setup_required
-@login_required
-@account_initialization_required
-def update_tag(tag_id: UUID, payload: TagBasePayload) -> TagResponse:
-    current_user, _ = current_account_with_tenant()
-    tag_id_str = str(tag_id)
-    # The role of the current user in the ta table must be admin, owner, or editor
-    if not (current_user.has_edit_permission or current_user.is_dataset_editor):
-        raise Forbidden()
+@console_ns.route("/tags/<uuid:tag_id>")
+class TagUpdateDeleteApi(Resource):
+    @console_ns.expect(console_ns.models[TagBasePayload.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def patch(self, tag_id):
+        current_user, _ = current_account_with_tenant()
+        tag_id = str(tag_id)
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not (current_user.has_edit_permission or current_user.is_dataset_editor):
+            raise Forbidden()
 
-    tag = TagService.update_tags(payload.model_dump(), tag_id_str)
+        payload = TagBasePayload.model_validate(console_ns.payload or {})
+        tag = TagService.update_tags(payload.model_dump(), tag_id)
 
-    binding_count = TagService.get_tag_binding_count(tag_id_str)
+        binding_count = TagService.get_tag_binding_count(tag_id)
 
-    return TagResponse(id=tag.id, name=tag.name, type=tag.type, binding_count=binding_count)
+        response = {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": binding_count}
+
+        return response, 200
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def delete(self, tag_id):
+        tag_id = str(tag_id)
+
+        TagService.delete_tag(tag_id)
+
+        return 204
 
 
-@console_router.delete(
-    "/tags/<uuid:tag_id>",
-    tags=["console"],
-    status_code=204,
-)
-@setup_required
-@login_required
-@account_initialization_required
-@edit_permission_required
-def delete_tag(tag_id: UUID) -> None:
-    tag_id_str = str(tag_id)
+@console_ns.route("/tag-bindings/create")
+class TagBindingCreateApi(Resource):
+    @console_ns.expect(console_ns.models[TagBindingPayload.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self):
+        current_user, _ = current_account_with_tenant()
+        # The role of the current user in the ta table must be admin, owner, editor, or dataset_operator
+        if not (current_user.has_edit_permission or current_user.is_dataset_editor):
+            raise Forbidden()
 
-    TagService.delete_tag(tag_id_str)
+        payload = TagBindingPayload.model_validate(console_ns.payload or {})
+        TagService.save_tag_binding(payload.model_dump())
+
+        return {"result": "success"}, 200
 
 
-@console_router.post(
-    "/tag-bindings/create",
-    response_model=TagBindingResult,
-    tags=["console"],
-)
-@setup_required
-@login_required
-@account_initialization_required
-def create_tag_binding(payload: TagBindingPayload) -> TagBindingResult:
-    current_user, _ = current_account_with_tenant()
-    # The role of the current user in the tag table must be admin, owner, editor, or dataset_operator
-    if not (current_user.has_edit_permission or current_user.is_dataset_editor):
-        raise Forbidden()
+@console_ns.route("/tag-bindings/remove")
+class TagBindingDeleteApi(Resource):
+    @console_ns.expect(console_ns.models[TagBindingRemovePayload.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self):
+        current_user, _ = current_account_with_tenant()
+        # The role of the current user in the ta table must be admin, owner, editor, or dataset_operator
+        if not (current_user.has_edit_permission or current_user.is_dataset_editor):
+            raise Forbidden()
 
-    TagService.save_tag_binding(payload.model_dump())
+        payload = TagBindingRemovePayload.model_validate(console_ns.payload or {})
+        TagService.delete_tag_binding(payload.model_dump())
 
-    return TagBindingResult(result="success")
-
-
-@console_router.post(
-    "/tag-bindings/remove",
-    response_model=TagBindingResult,
-    tags=["console"],
-)
-@setup_required
-@login_required
-@account_initialization_required
-def delete_tag_binding(payload: TagBindingRemovePayload) -> TagBindingResult:
-    current_user, _ = current_account_with_tenant()
-    # The role of the current user in the tag table must be admin, owner, editor, or dataset_operator
-    if not (current_user.has_edit_permission or current_user.is_dataset_editor):
-        raise Forbidden()
-
-    TagService.delete_tag_binding(payload.model_dump())
-
-    return TagBindingResult(result="success")
+        return {"result": "success"}, 200

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,
@@ -37,8 +36,9 @@ from controllers.console.wraps import (
     only_edition_cloud,
     setup_required,
 )
+from core.file import helpers as file_helpers
 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
@@ -74,6 +74,10 @@ class AccountAvatarPayload(BaseModel):
     avatar: str
 
 
+class AccountAvatarQuery(BaseModel):
+    avatar: str = Field(..., description="Avatar file ID")
+
+
 class AccountInterfaceLanguagePayload(BaseModel):
     interface_language: str
 
@@ -159,6 +163,7 @@ def reg(cls: type[BaseModel]):
 reg(AccountInitPayload)
 reg(AccountNamePayload)
 reg(AccountAvatarPayload)
+reg(AccountAvatarQuery)
 reg(AccountInterfaceLanguagePayload)
 reg(AccountInterfaceThemePayload)
 reg(AccountTimezonePayload)
@@ -171,12 +176,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 +242,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,23 +255,35 @@ 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")
 class AccountAvatarApi(Resource):
+    @console_ns.expect(console_ns.models[AccountAvatarQuery.__name__])
+    @console_ns.doc("get_account_avatar")
+    @console_ns.doc(description="Get account avatar url")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self):
+        args = AccountAvatarQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+
+        avatar_url = file_helpers.get_signed_file_url(args.avatar)
+        return {"avatar_url": avatar_url}
+
     @console_ns.expect(console_ns.models[AccountAvatarPayload.__name__])
     @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 +291,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 +300,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 +308,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 +317,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 +325,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 +334,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 +342,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 +351,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 +362,7 @@ class AccountPasswordApi(Resource):
         except ServiceCurrentPasswordIncorrectError:
             raise CurrentPasswordIncorrectError()
 
-        return _serialize_account(current_user)
+        return {"result": "success"}
 
 
 @console_ns.route("/account/integrates")
@@ -627,7 +638,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 +667,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/dsl.py

@@ -0,0 +1,65 @@
+import json
+
+import httpx
+import yaml
+from flask_restx import Resource, reqparse
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import Forbidden
+
+from controllers.console import console_ns
+from controllers.console.wraps import account_initialization_required, setup_required
+from core.plugin.impl.exc import PluginPermissionDeniedError
+from extensions.ext_database import db
+from libs.login import current_account_with_tenant, login_required
+from models.model import App
+from models.workflow import Workflow
+from services.app_dsl_service import AppDslService
+
+
+@console_ns.route("/workspaces/current/dsl/predict")
+class DSLPredictApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self):
+        user, _ = current_account_with_tenant()
+        if not user.is_admin_or_owner:
+            raise Forbidden()
+
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("app_id", type=str, required=True, location="json")
+            .add_argument("current_node_id", type=str, required=True, location="json")
+        )
+        args = parser.parse_args()
+
+        app_id: str = args["app_id"]
+        current_node_id: str = args["current_node_id"]
+
+        with Session(db.engine) as session:
+            app = session.query(App).filter_by(id=app_id).first()
+            workflow = session.query(Workflow).filter_by(app_id=app_id, version=Workflow.VERSION_DRAFT).first()
+
+        if not app:
+            raise ValueError("App not found")
+        if not workflow:
+            raise ValueError("Workflow not found")
+
+        try:
+            i = 0
+            for node_id, _ in workflow.walk_nodes():
+                if node_id == current_node_id:
+                    break
+                i += 1
+
+            dsl = yaml.safe_load(AppDslService.export_dsl(app_model=app))
+
+            response = httpx.post(
+                "http://spark-832c:8000/predict",
+                json={"graph_data": dsl, "source_node_index": i},
+            )
+            return {
+                "nodes": json.loads(response.json()),
+            }
+        except PluginPermissionDeniedError as e:
+            raise ValueError(e.description) from e

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/console/workspace/sandbox_providers.py

@@ -0,0 +1,103 @@
+import logging
+
+from flask_restx import Resource, fields, reqparse
+
+from controllers.console import console_ns
+from controllers.console.wraps import account_initialization_required, setup_required
+from core.model_runtime.utils.encoders import jsonable_encoder
+from libs.login import current_account_with_tenant, login_required
+from services.sandbox.sandbox_provider_service import SandboxProviderService
+
+logger = logging.getLogger(__name__)
+
+
+@console_ns.route("/workspaces/current/sandbox-providers")
+class SandboxProviderListApi(Resource):
+    @console_ns.doc("list_sandbox_providers")
+    @console_ns.doc(description="Get list of available sandbox providers with configuration status")
+    @console_ns.response(200, "Success", fields.List(fields.Raw(description="Sandbox provider information")))
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self):
+        _, current_tenant_id = current_account_with_tenant()
+        providers = SandboxProviderService.list_providers(current_tenant_id)
+        return jsonable_encoder([p.model_dump() for p in providers])
+
+
+config_parser = reqparse.RequestParser()
+config_parser.add_argument("config", type=dict, required=True, location="json")
+config_parser.add_argument("activate", type=bool, required=False, default=False, location="json")
+
+
+@console_ns.route("/workspaces/current/sandbox-provider/<string:provider_type>/config")
+class SandboxProviderConfigApi(Resource):
+    @console_ns.doc("save_sandbox_provider_config")
+    @console_ns.doc(description="Save or update configuration for a sandbox provider")
+    @console_ns.expect(config_parser)
+    @console_ns.response(200, "Success")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, provider_type: str):
+        _, current_tenant_id = current_account_with_tenant()
+        args = config_parser.parse_args()
+
+        try:
+            result = SandboxProviderService.save_config(
+                tenant_id=current_tenant_id,
+                provider_type=provider_type,
+                config=args["config"],
+                activate=args["activate"],
+            )
+            return result
+        except ValueError as e:
+            return {"message": str(e)}, 400
+
+    @console_ns.doc("delete_sandbox_provider_config")
+    @console_ns.doc(description="Delete configuration for a sandbox provider")
+    @console_ns.response(200, "Success")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def delete(self, provider_type: str):
+        _, current_tenant_id = current_account_with_tenant()
+
+        try:
+            result = SandboxProviderService.delete_config(
+                tenant_id=current_tenant_id,
+                provider_type=provider_type,
+            )
+            return result
+        except ValueError as e:
+            return {"message": str(e)}, 400
+
+
+activate_parser = reqparse.RequestParser()
+activate_parser.add_argument("type", type=str, required=True, location="json")
+
+
+@console_ns.route("/workspaces/current/sandbox-provider/<string:provider_type>/activate")
+class SandboxProviderActivateApi(Resource):
+    """Activate a sandbox provider."""
+
+    @console_ns.doc("activate_sandbox_provider")
+    @console_ns.doc(description="Activate a sandbox provider for the current workspace")
+    @console_ns.response(200, "Success")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, provider_type: str):
+        """Activate a sandbox provider."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        try:
+            args = activate_parser.parse_args()
+            result = SandboxProviderService.activate_provider(
+                tenant_id=current_tenant_id,
+                provider_type=provider_type,
+                type=args["type"],
+            )
+            return result
+        except ValueError as e:
+            return {"message": str(e)}, 400

api/controllers/files/__init__.py

@@ -14,7 +14,12 @@ api = ExternalApi(
 
 files_ns = Namespace("files", description="File operations", path="/")
 
-from . import image_preview, tool_files, upload
+from . import (
+    image_preview,
+    storage_files,
+    tool_files,
+    upload,
+)
 
 api.add_namespace(files_ns)
 
@@ -23,6 +28,7 @@ __all__ = [
     "bp",
     "files_ns",
     "image_preview",
+    "storage_files",
     "tool_files",
     "upload",
 ]

api/controllers/files/storage_files.py

@@ -0,0 +1,80 @@
+"""Token-based file proxy controller for storage operations.
+
+This controller handles file download and upload operations using opaque UUID tokens.
+The token maps to the real storage key in Redis, so the actual storage path is never
+exposed in the URL.
+
+Routes:
+    GET  /files/storage-files/{token} - Download a file
+    PUT  /files/storage-files/{token} - Upload a file
+
+The operation type (download/upload) is determined by the ticket stored in Redis,
+not by the HTTP method. This ensures a download ticket cannot be used for upload
+and vice versa.
+"""
+
+from urllib.parse import quote
+
+from flask import Response, request
+from flask_restx import Resource
+from werkzeug.exceptions import Forbidden, NotFound, RequestEntityTooLarge
+
+from controllers.files import files_ns
+from extensions.ext_storage import storage
+from services.storage_ticket_service import StorageTicketService
+
+
+@files_ns.route("/storage-files/<string:token>")
+class StorageFilesApi(Resource):
+    """Handle file operations through token-based URLs."""
+
+    def get(self, token: str):
+        """Download a file using a token.
+
+        The ticket must have op="download", otherwise returns 403.
+        """
+        ticket = StorageTicketService.get_ticket(token)
+        if ticket is None:
+            raise Forbidden("Invalid or expired token")
+
+        if ticket.op != "download":
+            raise Forbidden("This token is not valid for download")
+
+        try:
+            generator = storage.load_stream(ticket.storage_key)
+        except FileNotFoundError:
+            raise NotFound("File not found")
+
+        filename = ticket.filename or ticket.storage_key.rsplit("/", 1)[-1]
+        encoded_filename = quote(filename)
+
+        return Response(
+            generator,
+            mimetype="application/octet-stream",
+            direct_passthrough=True,
+            headers={
+                "Content-Disposition": f"attachment; filename*=UTF-8''{encoded_filename}",
+            },
+        )
+
+    def put(self, token: str):
+        """Upload a file using a token.
+
+        The ticket must have op="upload", otherwise returns 403.
+        If the request body exceeds max_bytes, returns 413.
+        """
+        ticket = StorageTicketService.get_ticket(token)
+        if ticket is None:
+            raise Forbidden("Invalid or expired token")
+
+        if ticket.op != "upload":
+            raise Forbidden("This token is not valid for upload")
+
+        content = request.get_data()
+
+        if ticket.max_bytes is not None and len(content) > ticket.max_bytes:
+            raise RequestEntityTooLarge(f"Upload exceeds maximum size of {ticket.max_bytes} bytes")
+
+        storage.save(ticket.storage_key, content)
+
+        return Response(status=204)

api/controllers/inner_api/plugin/plugin.py

@@ -448,3 +448,53 @@ class PluginFetchAppInfoApi(Resource):
         return BaseBackwardsInvocationResponse(
             data=PluginAppBackwardsInvocation.fetch_app_info(payload.app_id, tenant_model.id)
         ).model_dump()
+
+
+@inner_api_ns.route("/fetch/tools/list")
+class PluginFetchToolsListApi(Resource):
+    @get_user_tenant
+    @setup_required
+    @plugin_inner_api_only
+    @inner_api_ns.doc("plugin_fetch_tools_list")
+    @inner_api_ns.doc(description="Fetch all available tools through plugin interface")
+    @inner_api_ns.doc(
+        responses={
+            200: "Tools list retrieved successfully",
+            401: "Unauthorized - invalid API key",
+            404: "Service not available",
+        }
+    )
+    def post(self, user_model: Account | EndUser, tenant_model: Tenant):
+        from sqlalchemy.orm import Session
+
+        from extensions.ext_database import db
+        from services.tools.api_tools_manage_service import ApiToolManageService
+        from services.tools.builtin_tools_manage_service import BuiltinToolManageService
+        from services.tools.mcp_tools_manage_service import MCPToolManageService
+        from services.tools.workflow_tools_manage_service import WorkflowToolManageService
+
+        providers = []
+
+        # Get builtin tools
+        builtin_providers = BuiltinToolManageService.list_builtin_tools(user_model.id, tenant_model.id)
+        for provider in builtin_providers:
+            providers.append(provider.to_dict())
+
+        # Get API tools
+        api_providers = ApiToolManageService.list_api_tools(tenant_model.id)
+        for provider in api_providers:
+            providers.append(provider.to_dict())
+
+        # Get workflow tools
+        workflow_providers = WorkflowToolManageService.list_tenant_workflow_tools(user_model.id, tenant_model.id)
+        for provider in workflow_providers:
+            providers.append(provider.to_dict())
+
+        # Get MCP tools
+        with Session(db.engine) as session:
+            mcp_service = MCPToolManageService(session)
+            mcp_providers = mcp_service.list_providers(tenant_id=tenant_model.id, for_list=True)
+            for provider in mcp_providers:
+                providers.append(provider.to_dict())
+
+        return BaseBackwardsInvocationResponse(data={"providers": providers}).model_dump()

api/controllers/inner_api/plugin/wraps.py

@@ -75,7 +75,6 @@ def get_user_tenant(view_func: Callable[P, R]):
     @wraps(view_func)
     def decorated_view(*args: P.args, **kwargs: P.kwargs):
         payload = TenantUserPayload.model_validate(request.get_json(silent=True) or {})
-
         user_id = payload.user_id
         tenant_id = payload.tenant_id
 

api/controllers/inner_api/wraps.py

@@ -5,14 +5,15 @@ from hashlib import sha1
 from hmac import new as hmac_new
 from typing import ParamSpec, TypeVar
 
-P = ParamSpec("P")
-R = TypeVar("R")
 from flask import abort, request
 
 from configs import dify_config
 from extensions.ext_database import db
 from models.model import EndUser
 
+P = ParamSpec("P")
+R = TypeVar("R")
+
 
 def billing_inner_api_only(view: Callable[P, R]):
     @wraps(view)
@@ -88,11 +89,11 @@ def plugin_inner_api_only(view: Callable[P, R]):
         if not dify_config.PLUGIN_DAEMON_KEY:
             abort(404)
 
-        # get header 'X-Inner-Api-Key'
+        # validate using inner api key
         inner_api_key = request.headers.get("X-Inner-Api-Key")
-        if not inner_api_key or inner_api_key != dify_config.INNER_API_KEY_FOR_PLUGIN:
-            abort(404)
+        if inner_api_key and inner_api_key == dify_config.INNER_API_KEY_FOR_PLUGIN:
+            return view(*args, **kwargs)
 
-        return view(*args, **kwargs)
+        abort(401)
 
     return decorated

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,
 )
 
 
@@ -481,14 +480,15 @@ class DatasetTagsApi(DatasetApiResource):
             401: "Unauthorized - invalid API token",
         }
     )
+    @service_api_ns.marshal_with(build_dataset_tag_fields(service_api_ns))
     def get(self, _):
         """Get all knowledge type tags."""
         assert isinstance(current_user, Account)
         cid = current_user.current_tenant_id
         assert cid is not None
         tags = TagService.get_tags("knowledge", cid)
-        tag_models = TypeAdapter(list[DataSetTag]).validate_python(tags, from_attributes=True)
-        return [tag.model_dump(mode="json") for tag in tag_models], 200
+
+        return tags, 200
 
     @service_api_ns.expect(service_api_ns.models[TagCreatePayload.__name__])
     @service_api_ns.doc("create_dataset_tag")
@@ -500,6 +500,7 @@ class DatasetTagsApi(DatasetApiResource):
             403: "Forbidden - insufficient permissions",
         }
     )
+    @service_api_ns.marshal_with(build_dataset_tag_fields(service_api_ns))
     def post(self, _):
         """Add a knowledge type tag."""
         assert isinstance(current_user, Account)
@@ -509,9 +510,7 @@ class DatasetTagsApi(DatasetApiResource):
         payload = TagCreatePayload.model_validate(service_api_ns.payload or {})
         tag = TagService.save_tags({"name": payload.name, "type": "knowledge"})
 
-        response = DataSetTag.model_validate(
-            {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": 0}
-        ).model_dump(mode="json")
+        response = {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": 0}
         return response, 200
 
     @service_api_ns.expect(service_api_ns.models[TagUpdatePayload.__name__])
@@ -524,6 +523,7 @@ class DatasetTagsApi(DatasetApiResource):
             403: "Forbidden - insufficient permissions",
         }
     )
+    @service_api_ns.marshal_with(build_dataset_tag_fields(service_api_ns))
     def patch(self, _):
         assert isinstance(current_user, Account)
         if not (current_user.has_edit_permission or current_user.is_dataset_editor):
@@ -536,9 +536,8 @@ class DatasetTagsApi(DatasetApiResource):
 
         binding_count = TagService.get_tag_binding_count(tag_id)
 
-        response = DataSetTag.model_validate(
-            {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": binding_count}
-        ).model_dump(mode="json")
+        response = {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": binding_count}
+
         return response, 200
 
     @service_api_ns.expect(service_api_ns.models[TagDeletePayload.__name__])

api/controllers/service_api/dataset/hit_testing.py

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

api/controllers/service_api/dataset/metadata.py

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

api/controllers/service_api/wraps.py

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

api/core/agent/agent_app_runner.py

@@ -0,0 +1,398 @@
+import logging
+from collections.abc import Generator
+from copy import deepcopy
+from typing import Any
+
+from core.agent.base_agent_runner import BaseAgentRunner
+from core.agent.entities import AgentEntity, AgentLog, AgentOutputKind, AgentResult
+from core.agent.patterns.strategy_factory import StrategyFactory
+from core.app.apps.base_app_queue_manager import PublishFrom
+from core.app.entities.queue_entities import QueueAgentThoughtEvent, QueueMessageEndEvent, QueueMessageFileEvent
+from core.file import file_manager
+from core.model_runtime.entities import (
+    AssistantPromptMessage,
+    LLMResult,
+    LLMResultChunk,
+    LLMResultChunkDelta,
+    LLMUsage,
+    PromptMessage,
+    PromptMessageContentType,
+    SystemPromptMessage,
+    TextPromptMessageContent,
+    UserPromptMessage,
+)
+from core.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
+from core.prompt.agent_history_prompt_transform import AgentHistoryPromptTransform
+from core.tools.__base.tool import Tool
+from core.tools.entities.tool_entities import ToolInvokeMeta
+from core.tools.tool_engine import ToolEngine
+from models.model import Message
+
+logger = logging.getLogger(__name__)
+
+
+class AgentAppRunner(BaseAgentRunner):
+    def _create_tool_invoke_hook(self, message: Message):
+        """
+        Create a tool invoke hook that uses ToolEngine.agent_invoke.
+        This hook handles file creation and returns proper meta information.
+        """
+        # Get trace manager from app generate entity
+        trace_manager = self.application_generate_entity.trace_manager
+
+        def tool_invoke_hook(
+            tool: Tool, tool_args: dict[str, Any], tool_name: str
+        ) -> tuple[str, list[str], ToolInvokeMeta]:
+            """Hook that uses agent_invoke for proper file and meta handling."""
+            tool_invoke_response, message_files, tool_invoke_meta = ToolEngine.agent_invoke(
+                tool=tool,
+                tool_parameters=tool_args,
+                user_id=self.user_id,
+                tenant_id=self.tenant_id,
+                message=message,
+                invoke_from=self.application_generate_entity.invoke_from,
+                agent_tool_callback=self.agent_callback,
+                trace_manager=trace_manager,
+                app_id=self.application_generate_entity.app_config.app_id,
+                message_id=message.id,
+                conversation_id=self.conversation.id,
+            )
+
+            # Publish files and track IDs
+            for message_file_id in message_files:
+                self.queue_manager.publish(
+                    QueueMessageFileEvent(message_file_id=message_file_id),
+                    PublishFrom.APPLICATION_MANAGER,
+                )
+                self._current_message_file_ids.append(message_file_id)
+
+            return tool_invoke_response, message_files, tool_invoke_meta
+
+        return tool_invoke_hook
+
+    def run(self, message: Message, query: str, **kwargs: Any) -> Generator[LLMResultChunk, None, None]:
+        """
+        Run Agent application
+        """
+        self.query = query
+        app_generate_entity = self.application_generate_entity
+
+        app_config = self.app_config
+        assert app_config is not None, "app_config is required"
+        assert app_config.agent is not None, "app_config.agent is required"
+
+        # convert tools into ModelRuntime Tool format
+        tool_instances, _ = self._init_prompt_tools()
+
+        assert app_config.agent
+
+        # Create tool invoke hook for agent_invoke
+        tool_invoke_hook = self._create_tool_invoke_hook(message)
+
+        # Get instruction for ReAct strategy
+        instruction = self.app_config.prompt_template.simple_prompt_template or ""
+
+        # Use factory to create appropriate strategy
+        strategy = StrategyFactory.create_strategy(
+            model_features=self.model_features,
+            model_instance=self.model_instance,
+            tools=list(tool_instances.values()),
+            files=list(self.files),
+            max_iterations=app_config.agent.max_iteration,
+            context=self.build_execution_context(),
+            agent_strategy=self.config.strategy,
+            tool_invoke_hook=tool_invoke_hook,
+            instruction=instruction,
+        )
+
+        # Initialize state variables
+        current_agent_thought_id = None
+        current_tool_name: str | None = None
+        self._current_message_file_ids: list[str] = []
+
+        # organize prompt messages
+        prompt_messages = self._organize_prompt_messages()
+
+        # Run strategy
+        generator = strategy.run(
+            prompt_messages=prompt_messages,
+            model_parameters=app_generate_entity.model_conf.parameters,
+            stop=app_generate_entity.model_conf.stop,
+            stream=False,
+        )
+
+        # Consume generator and collect result
+        result: AgentResult | None = None
+        try:
+            while True:
+                try:
+                    output = next(generator)
+                except StopIteration as e:
+                    # Generator finished, get the return value
+                    result = e.value
+                    break
+
+                if isinstance(output, LLMResultChunk):
+                    # No more expect streaming data
+                    continue
+
+                else:
+                    # Handle Agent Log using log_type for type-safe dispatch
+                    if output.status == AgentLog.LogStatus.START:
+                        if output.log_type == AgentLog.LogType.ROUND:
+                            # Start of a new round
+                            message_file_ids: list[str] = []
+                            current_agent_thought_id = self.create_agent_thought(
+                                message_id=message.id,
+                                message="",
+                                tool_name="",
+                                tool_input="",
+                                messages_ids=message_file_ids,
+                            )
+
+                        elif output.log_type == AgentLog.LogType.TOOL_CALL:
+                            if current_agent_thought_id is None:
+                                continue
+
+                            # Tool call start - extract data from structured fields
+                            current_tool_name = output.data.get("tool_name", "")
+                            tool_input = output.data.get("tool_args", {})
+
+                            self.save_agent_thought(
+                                agent_thought_id=current_agent_thought_id,
+                                tool_name=current_tool_name,
+                                tool_input=tool_input,
+                                thought=None,
+                                observation=None,
+                                tool_invoke_meta=None,
+                                answer=None,
+                                messages_ids=[],
+                            )
+                            self.queue_manager.publish(
+                                QueueAgentThoughtEvent(agent_thought_id=current_agent_thought_id),
+                                PublishFrom.APPLICATION_MANAGER,
+                            )
+
+                    elif output.status == AgentLog.LogStatus.SUCCESS:
+                        if output.log_type == AgentLog.LogType.THOUGHT:
+                            if current_agent_thought_id is None:
+                                continue
+
+                            thought_text = output.data.get("thought")
+                            self.save_agent_thought(
+                                agent_thought_id=current_agent_thought_id,
+                                tool_name=None,
+                                tool_input=None,
+                                thought=thought_text,
+                                observation=None,
+                                tool_invoke_meta=None,
+                                answer=None,
+                                messages_ids=[],
+                            )
+                            self.queue_manager.publish(
+                                QueueAgentThoughtEvent(agent_thought_id=current_agent_thought_id),
+                                PublishFrom.APPLICATION_MANAGER,
+                            )
+
+                        elif output.log_type == AgentLog.LogType.TOOL_CALL:
+                            if current_agent_thought_id is None:
+                                continue
+
+                            # Tool call finished
+                            tool_output = output.data.get("output")
+                            # Get meta from strategy output (now properly populated)
+                            tool_meta = output.data.get("meta")
+
+                            # Wrap tool_meta with tool_name as key (required by agent_service)
+                            if tool_meta and current_tool_name:
+                                tool_meta = {current_tool_name: tool_meta}
+
+                            self.save_agent_thought(
+                                agent_thought_id=current_agent_thought_id,
+                                tool_name=None,
+                                tool_input=None,
+                                thought=None,
+                                observation=tool_output,
+                                tool_invoke_meta=tool_meta,
+                                answer=None,
+                                messages_ids=self._current_message_file_ids,
+                            )
+                            # Clear message file ids after saving
+                            self._current_message_file_ids = []
+                            current_tool_name = None
+
+                            self.queue_manager.publish(
+                                QueueAgentThoughtEvent(agent_thought_id=current_agent_thought_id),
+                                PublishFrom.APPLICATION_MANAGER,
+                            )
+
+                        elif output.log_type == AgentLog.LogType.ROUND:
+                            if current_agent_thought_id is None:
+                                continue
+
+                            # Round finished - save LLM usage and answer
+                            llm_usage = output.metadata.get(AgentLog.LogMetadata.LLM_USAGE)
+                            llm_result = output.data.get("llm_result")
+                            final_answer = output.data.get("final_answer")
+
+                            self.save_agent_thought(
+                                agent_thought_id=current_agent_thought_id,
+                                tool_name=None,
+                                tool_input=None,
+                                thought=llm_result,
+                                observation=None,
+                                tool_invoke_meta=None,
+                                answer=final_answer,
+                                messages_ids=[],
+                                llm_usage=llm_usage,
+                            )
+                            self.queue_manager.publish(
+                                QueueAgentThoughtEvent(agent_thought_id=current_agent_thought_id),
+                                PublishFrom.APPLICATION_MANAGER,
+                            )
+
+        except Exception:
+            # Re-raise any other exceptions
+            raise
+
+        # Process final result
+        if isinstance(result, AgentResult):
+            output_payload = result.output
+            if isinstance(output_payload, AgentResult.StructuredOutput):
+                if output_payload.output_kind == AgentOutputKind.ILLEGAL_OUTPUT:
+                    raise ValueError("Agent returned illegal output")
+                if output_payload.output_kind not in {
+                    AgentOutputKind.FINAL_OUTPUT_ANSWER,
+                    AgentOutputKind.OUTPUT_TEXT,
+                }:
+                    raise ValueError("Agent did not return text output")
+                if not output_payload.output_text:
+                    raise ValueError("Agent returned empty text output")
+                final_answer = output_payload.output_text
+            else:
+                if not output_payload:
+                    raise ValueError("Agent returned empty output")
+                final_answer = str(output_payload)
+            usage = result.usage or LLMUsage.empty_usage()
+
+            # Publish end event
+            self.queue_manager.publish(
+                QueueMessageEndEvent(
+                    llm_result=LLMResult(
+                        model=self.model_instance.model,
+                        prompt_messages=prompt_messages,
+                        message=AssistantPromptMessage(content=final_answer),
+                        usage=usage,
+                        system_fingerprint="",
+                    )
+                ),
+                PublishFrom.APPLICATION_MANAGER,
+            )
+
+        if False:
+            yield LLMResultChunk(
+                model="",
+                prompt_messages=[],
+                delta=LLMResultChunkDelta(
+                    index=0,
+                    message=AssistantPromptMessage(content=""),
+                    usage=None,
+                ),
+            )
+
+    def _init_system_message(self, prompt_template: str, prompt_messages: list[PromptMessage]) -> list[PromptMessage]:
+        """
+        Initialize system message
+        """
+        if not prompt_template:
+            return prompt_messages or []
+
+        prompt_messages = prompt_messages or []
+
+        if prompt_messages and isinstance(prompt_messages[0], SystemPromptMessage):
+            prompt_messages[0] = SystemPromptMessage(content=prompt_template)
+            return prompt_messages
+
+        if not prompt_messages:
+            return [SystemPromptMessage(content=prompt_template)]
+
+        prompt_messages.insert(0, SystemPromptMessage(content=prompt_template))
+        return prompt_messages
+
+    def _organize_user_query(self, query: str, prompt_messages: list[PromptMessage]) -> list[PromptMessage]:
+        """
+        Organize user query
+        """
+        if self.files:
+            # get image detail config
+            image_detail_config = (
+                self.application_generate_entity.file_upload_config.image_config.detail
+                if (
+                    self.application_generate_entity.file_upload_config
+                    and self.application_generate_entity.file_upload_config.image_config
+                )
+                else None
+            )
+            image_detail_config = image_detail_config or ImagePromptMessageContent.DETAIL.LOW
+
+            prompt_message_contents: list[PromptMessageContentUnionTypes] = []
+            for file in self.files:
+                prompt_message_contents.append(
+                    file_manager.to_prompt_message_content(
+                        file,
+                        image_detail_config=image_detail_config,
+                    )
+                )
+            prompt_message_contents.append(TextPromptMessageContent(data=query))
+
+            prompt_messages.append(UserPromptMessage(content=prompt_message_contents))
+        else:
+            prompt_messages.append(UserPromptMessage(content=query))
+
+        return prompt_messages
+
+    def _clear_user_prompt_image_messages(self, prompt_messages: list[PromptMessage]) -> list[PromptMessage]:
+        """
+        As for now, gpt supports both fc and vision at the first iteration.
+        We need to remove the image messages from the prompt messages at the first iteration.
+        """
+        prompt_messages = deepcopy(prompt_messages)
+
+        for prompt_message in prompt_messages:
+            if isinstance(prompt_message, UserPromptMessage):
+                if isinstance(prompt_message.content, list):
+                    prompt_message.content = "\n".join(
+                        [
+                            content.data
+                            if content.type == PromptMessageContentType.TEXT
+                            else "[image]"
+                            if content.type == PromptMessageContentType.IMAGE
+                            else "[file]"
+                            for content in prompt_message.content
+                        ]
+                    )
+
+        return prompt_messages
+
+    def _organize_prompt_messages(self):
+        # For ReAct strategy, use the agent prompt template
+        if self.config.strategy == AgentEntity.Strategy.CHAIN_OF_THOUGHT and self.config.prompt:
+            prompt_template = self.config.prompt.first_prompt
+        else:
+            prompt_template = self.app_config.prompt_template.simple_prompt_template or ""
+
+        self.history_prompt_messages = self._init_system_message(prompt_template, self.history_prompt_messages)
+        query_prompt_messages = self._organize_user_query(self.query or "", [])
+
+        self.history_prompt_messages = AgentHistoryPromptTransform(
+            model_config=self.model_config,
+            prompt_messages=[*query_prompt_messages, *self._current_thoughts],
+            history_messages=self.history_prompt_messages,
+            memory=self.memory,
+        ).get_prompt()
+
+        prompt_messages = [*self.history_prompt_messages, *query_prompt_messages, *self._current_thoughts]
+        if len(self._current_thoughts) != 0:
+            # clear messages after the first iteration
+            prompt_messages = self._clear_user_prompt_image_messages(prompt_messages)
+        return prompt_messages

api/core/agent/base_agent_runner.py

@@ -6,7 +6,8 @@ from typing import Union, cast
 
 from sqlalchemy import select
 
-from core.agent.entities import AgentEntity, AgentToolEntity
+from core.agent.entities import AgentEntity, AgentToolEntity, ExecutionContext
+from core.agent.output_tools import build_agent_output_tools, select_output_tool_names
 from core.app.app_config.features.file_upload.manager import FileUploadConfigManager
 from core.app.apps.agent_chat.app_config_manager import AgentChatAppConfig
 from core.app.apps.base_app_queue_manager import AppQueueManager
@@ -36,6 +37,7 @@ from core.model_runtime.model_providers.__base.large_language_model import Large
 from core.prompt.utils.extract_thread_messages import extract_thread_messages
 from core.tools.__base.tool import Tool
 from core.tools.entities.tool_entities import (
+    ToolInvokeFrom,
     ToolParameter,
 )
 from core.tools.tool_manager import ToolManager
@@ -116,9 +118,20 @@ class BaseAgentRunner(AppRunner):
         features = model_schema.features if model_schema and model_schema.features else []
         self.stream_tool_call = ModelFeature.STREAM_TOOL_CALL in features
         self.files = application_generate_entity.files if ModelFeature.VISION in features else []
+        self.model_features = features
         self.query: str | None = ""
         self._current_thoughts: list[PromptMessage] = []
 
+    def build_execution_context(self) -> ExecutionContext:
+        """Build execution context."""
+        return ExecutionContext(
+            user_id=self.user_id,
+            app_id=self.app_config.app_id,
+            conversation_id=self.conversation.id,
+            message_id=self.message.id,
+            tenant_id=self.tenant_id,
+        )
+
     def _repack_app_generate_entity(
         self, app_generate_entity: AgentChatAppGenerateEntity
     ) -> AgentChatAppGenerateEntity:
@@ -240,6 +253,18 @@ class BaseAgentRunner(AppRunner):
             # save tool entity
             tool_instances[dataset_tool.entity.identity.name] = dataset_tool
 
+        output_tools = build_agent_output_tools(
+            tenant_id=self.tenant_id,
+            invoke_from=self.application_generate_entity.invoke_from,
+            tool_invoke_from=ToolInvokeFrom.AGENT,
+            output_tool_names=select_output_tool_names(
+                structured_output_enabled=False,
+                include_illegal_output=True,
+            ),
+        )
+        for tool in output_tools:
+            tool_instances[tool.entity.identity.name] = tool
+
         return tool_instances, prompt_messages_tools
 
     def update_prompt_message_tool(self, tool: Tool, prompt_tool: PromptMessageTool) -> PromptMessageTool:

api/core/agent/cot_agent_runner.py

@@ -1,437 +0,0 @@
-import json
-import logging
-from abc import ABC, abstractmethod
-from collections.abc import Generator, Mapping, Sequence
-from typing import Any
-
-from core.agent.base_agent_runner import BaseAgentRunner
-from core.agent.entities import AgentScratchpadUnit
-from core.agent.output_parser.cot_output_parser import CotAgentOutputParser
-from core.app.apps.base_app_queue_manager import PublishFrom
-from core.app.entities.queue_entities import QueueAgentThoughtEvent, QueueMessageEndEvent, QueueMessageFileEvent
-from core.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk, LLMResultChunkDelta, LLMUsage
-from core.model_runtime.entities.message_entities import (
-    AssistantPromptMessage,
-    PromptMessage,
-    PromptMessageTool,
-    ToolPromptMessage,
-    UserPromptMessage,
-)
-from core.ops.ops_trace_manager import TraceQueueManager
-from core.prompt.agent_history_prompt_transform import AgentHistoryPromptTransform
-from core.tools.__base.tool import Tool
-from core.tools.entities.tool_entities import ToolInvokeMeta
-from core.tools.tool_engine import ToolEngine
-from core.workflow.nodes.agent.exc import AgentMaxIterationError
-from models.model import Message
-
-logger = logging.getLogger(__name__)
-
-
-class CotAgentRunner(BaseAgentRunner, ABC):
-    _is_first_iteration = True
-    _ignore_observation_providers = ["wenxin"]
-    _historic_prompt_messages: list[PromptMessage]
-    _agent_scratchpad: list[AgentScratchpadUnit]
-    _instruction: str
-    _query: str
-    _prompt_messages_tools: Sequence[PromptMessageTool]
-
-    def run(
-        self,
-        message: Message,
-        query: str,
-        inputs: Mapping[str, str],
-    ) -> Generator:
-        """
-        Run Cot agent application
-        """
-
-        app_generate_entity = self.application_generate_entity
-        self._repack_app_generate_entity(app_generate_entity)
-        self._init_react_state(query)
-
-        trace_manager = app_generate_entity.trace_manager
-
-        # check model mode
-        if "Observation" not in app_generate_entity.model_conf.stop:
-            if app_generate_entity.model_conf.provider not in self._ignore_observation_providers:
-                app_generate_entity.model_conf.stop.append("Observation")
-
-        app_config = self.app_config
-        assert app_config.agent
-
-        # init instruction
-        inputs = inputs or {}
-        instruction = app_config.prompt_template.simple_prompt_template or ""
-        self._instruction = self._fill_in_inputs_from_external_data_tools(instruction, inputs)
-
-        iteration_step = 1
-        max_iteration_steps = min(app_config.agent.max_iteration, 99) + 1
-
-        # convert tools into ModelRuntime Tool format
-        tool_instances, prompt_messages_tools = self._init_prompt_tools()
-        self._prompt_messages_tools = prompt_messages_tools
-
-        function_call_state = True
-        llm_usage: dict[str, LLMUsage | None] = {"usage": None}
-        final_answer = ""
-        prompt_messages: list = []  # Initialize prompt_messages
-        agent_thought_id = ""  # Initialize agent_thought_id
-
-        def increase_usage(final_llm_usage_dict: dict[str, LLMUsage | None], usage: LLMUsage):
-            if not final_llm_usage_dict["usage"]:
-                final_llm_usage_dict["usage"] = usage
-            else:
-                llm_usage = final_llm_usage_dict["usage"]
-                llm_usage.prompt_tokens += usage.prompt_tokens
-                llm_usage.completion_tokens += usage.completion_tokens
-                llm_usage.total_tokens += usage.total_tokens
-                llm_usage.prompt_price += usage.prompt_price
-                llm_usage.completion_price += usage.completion_price
-                llm_usage.total_price += usage.total_price
-
-        model_instance = self.model_instance
-
-        while function_call_state and iteration_step <= max_iteration_steps:
-            # continue to run until there is not any tool call
-            function_call_state = False
-
-            if iteration_step == max_iteration_steps:
-                # the last iteration, remove all tools
-                self._prompt_messages_tools = []
-
-            message_file_ids: list[str] = []
-
-            agent_thought_id = self.create_agent_thought(
-                message_id=message.id, message="", tool_name="", tool_input="", messages_ids=message_file_ids
-            )
-
-            if iteration_step > 1:
-                self.queue_manager.publish(
-                    QueueAgentThoughtEvent(agent_thought_id=agent_thought_id), PublishFrom.APPLICATION_MANAGER
-                )
-
-            # recalc llm max tokens
-            prompt_messages = self._organize_prompt_messages()
-            self.recalc_llm_max_tokens(self.model_config, prompt_messages)
-            # invoke model
-            chunks = model_instance.invoke_llm(
-                prompt_messages=prompt_messages,
-                model_parameters=app_generate_entity.model_conf.parameters,
-                tools=[],
-                stop=app_generate_entity.model_conf.stop,
-                stream=True,
-                user=self.user_id,
-                callbacks=[],
-            )
-
-            usage_dict: dict[str, LLMUsage | None] = {}
-            react_chunks = CotAgentOutputParser.handle_react_stream_output(chunks, usage_dict)
-            scratchpad = AgentScratchpadUnit(
-                agent_response="",
-                thought="",
-                action_str="",
-                observation="",
-                action=None,
-            )
-
-            # publish agent thought if it's first iteration
-            if iteration_step == 1:
-                self.queue_manager.publish(
-                    QueueAgentThoughtEvent(agent_thought_id=agent_thought_id), PublishFrom.APPLICATION_MANAGER
-                )
-
-            for chunk in react_chunks:
-                if isinstance(chunk, AgentScratchpadUnit.Action):
-                    action = chunk
-                    # detect action
-                    assert scratchpad.agent_response is not None
-                    scratchpad.agent_response += json.dumps(chunk.model_dump())
-                    scratchpad.action_str = json.dumps(chunk.model_dump())
-                    scratchpad.action = action
-                else:
-                    assert scratchpad.agent_response is not None
-                    scratchpad.agent_response += chunk
-                    assert scratchpad.thought is not None
-                    scratchpad.thought += chunk
-                    yield LLMResultChunk(
-                        model=self.model_config.model,
-                        prompt_messages=prompt_messages,
-                        system_fingerprint="",
-                        delta=LLMResultChunkDelta(index=0, message=AssistantPromptMessage(content=chunk), usage=None),
-                    )
-
-            assert scratchpad.thought is not None
-            scratchpad.thought = scratchpad.thought.strip() or "I am thinking about how to help you"
-            self._agent_scratchpad.append(scratchpad)
-
-            # Check if max iteration is reached and model still wants to call tools
-            if iteration_step == max_iteration_steps and scratchpad.action:
-                if scratchpad.action.action_name.lower() != "final answer":
-                    raise AgentMaxIterationError(app_config.agent.max_iteration)
-
-            # get llm usage
-            if "usage" in usage_dict:
-                if usage_dict["usage"] is not None:
-                    increase_usage(llm_usage, usage_dict["usage"])
-            else:
-                usage_dict["usage"] = LLMUsage.empty_usage()
-
-            self.save_agent_thought(
-                agent_thought_id=agent_thought_id,
-                tool_name=(scratchpad.action.action_name if scratchpad.action and not scratchpad.is_final() else ""),
-                tool_input={scratchpad.action.action_name: scratchpad.action.action_input} if scratchpad.action else {},
-                tool_invoke_meta={},
-                thought=scratchpad.thought or "",
-                observation="",
-                answer=scratchpad.agent_response or "",
-                messages_ids=[],
-                llm_usage=usage_dict["usage"],
-            )
-
-            if not scratchpad.is_final():
-                self.queue_manager.publish(
-                    QueueAgentThoughtEvent(agent_thought_id=agent_thought_id), PublishFrom.APPLICATION_MANAGER
-                )
-
-            if not scratchpad.action:
-                # failed to extract action, return final answer directly
-                final_answer = ""
-            else:
-                if scratchpad.action.action_name.lower() == "final answer":
-                    # action is final answer, return final answer directly
-                    try:
-                        if isinstance(scratchpad.action.action_input, dict):
-                            final_answer = json.dumps(scratchpad.action.action_input, ensure_ascii=False)
-                        elif isinstance(scratchpad.action.action_input, str):
-                            final_answer = scratchpad.action.action_input
-                        else:
-                            final_answer = f"{scratchpad.action.action_input}"
-                    except TypeError:
-                        final_answer = f"{scratchpad.action.action_input}"
-                else:
-                    function_call_state = True
-                    # action is tool call, invoke tool
-                    tool_invoke_response, tool_invoke_meta = self._handle_invoke_action(
-                        action=scratchpad.action,
-                        tool_instances=tool_instances,
-                        message_file_ids=message_file_ids,
-                        trace_manager=trace_manager,
-                    )
-                    scratchpad.observation = tool_invoke_response
-                    scratchpad.agent_response = tool_invoke_response
-
-                    self.save_agent_thought(
-                        agent_thought_id=agent_thought_id,
-                        tool_name=scratchpad.action.action_name,
-                        tool_input={scratchpad.action.action_name: scratchpad.action.action_input},
-                        thought=scratchpad.thought or "",
-                        observation={scratchpad.action.action_name: tool_invoke_response},
-                        tool_invoke_meta={scratchpad.action.action_name: tool_invoke_meta.to_dict()},
-                        answer=scratchpad.agent_response,
-                        messages_ids=message_file_ids,
-                        llm_usage=usage_dict["usage"],
-                    )
-
-                    self.queue_manager.publish(
-                        QueueAgentThoughtEvent(agent_thought_id=agent_thought_id), PublishFrom.APPLICATION_MANAGER
-                    )
-
-                # update prompt tool message
-                for prompt_tool in self._prompt_messages_tools:
-                    self.update_prompt_message_tool(tool_instances[prompt_tool.name], prompt_tool)
-
-            iteration_step += 1
-
-        yield LLMResultChunk(
-            model=model_instance.model,
-            prompt_messages=prompt_messages,
-            delta=LLMResultChunkDelta(
-                index=0, message=AssistantPromptMessage(content=final_answer), usage=llm_usage["usage"]
-            ),
-            system_fingerprint="",
-        )
-
-        # save agent thought
-        self.save_agent_thought(
-            agent_thought_id=agent_thought_id,
-            tool_name="",
-            tool_input={},
-            tool_invoke_meta={},
-            thought=final_answer,
-            observation={},
-            answer=final_answer,
-            messages_ids=[],
-        )
-        # publish end event
-        self.queue_manager.publish(
-            QueueMessageEndEvent(
-                llm_result=LLMResult(
-                    model=model_instance.model,
-                    prompt_messages=prompt_messages,
-                    message=AssistantPromptMessage(content=final_answer),
-                    usage=llm_usage["usage"] or LLMUsage.empty_usage(),
-                    system_fingerprint="",
-                )
-            ),
-            PublishFrom.APPLICATION_MANAGER,
-        )
-
-    def _handle_invoke_action(
-        self,
-        action: AgentScratchpadUnit.Action,
-        tool_instances: Mapping[str, Tool],
-        message_file_ids: list[str],
-        trace_manager: TraceQueueManager | None = None,
-    ) -> tuple[str, ToolInvokeMeta]:
-        """
-        handle invoke action
-        :param action: action
-        :param tool_instances: tool instances
-        :param message_file_ids: message file ids
-        :param trace_manager: trace manager
-        :return: observation, meta
-        """
-        # action is tool call, invoke tool
-        tool_call_name = action.action_name
-        tool_call_args = action.action_input
-        tool_instance = tool_instances.get(tool_call_name)
-
-        if not tool_instance:
-            answer = f"there is not a tool named {tool_call_name}"
-            return answer, ToolInvokeMeta.error_instance(answer)
-
-        if isinstance(tool_call_args, str):
-            try:
-                tool_call_args = json.loads(tool_call_args)
-            except json.JSONDecodeError:
-                pass
-
-        # invoke tool
-        tool_invoke_response, message_files, tool_invoke_meta = ToolEngine.agent_invoke(
-            tool=tool_instance,
-            tool_parameters=tool_call_args,
-            user_id=self.user_id,
-            tenant_id=self.tenant_id,
-            message=self.message,
-            invoke_from=self.application_generate_entity.invoke_from,
-            agent_tool_callback=self.agent_callback,
-            trace_manager=trace_manager,
-        )
-
-        # publish files
-        for message_file_id in message_files:
-            # publish message file
-            self.queue_manager.publish(
-                QueueMessageFileEvent(message_file_id=message_file_id), PublishFrom.APPLICATION_MANAGER
-            )
-            # add message file ids
-            message_file_ids.append(message_file_id)
-
-        return tool_invoke_response, tool_invoke_meta
-
-    def _convert_dict_to_action(self, action: dict) -> AgentScratchpadUnit.Action:
-        """
-        convert dict to action
-        """
-        return AgentScratchpadUnit.Action(action_name=action["action"], action_input=action["action_input"])
-
-    def _fill_in_inputs_from_external_data_tools(self, instruction: str, inputs: Mapping[str, Any]) -> str:
-        """
-        fill in inputs from external data tools
-        """
-        for key, value in inputs.items():
-            try:
-                instruction = instruction.replace(f"{{{{{key}}}}}", str(value))
-            except Exception:
-                continue
-
-        return instruction
-
-    def _init_react_state(self, query):
-        """
-        init agent scratchpad
-        """
-        self._query = query
-        self._agent_scratchpad = []
-        self._historic_prompt_messages = self._organize_historic_prompt_messages()
-
-    @abstractmethod
-    def _organize_prompt_messages(self) -> list[PromptMessage]:
-        """
-        organize prompt messages
-        """
-
-    def _format_assistant_message(self, agent_scratchpad: list[AgentScratchpadUnit]) -> str:
-        """
-        format assistant message
-        """
-        message = ""
-        for scratchpad in agent_scratchpad:
-            if scratchpad.is_final():
-                message += f"Final Answer: {scratchpad.agent_response}"
-            else:
-                message += f"Thought: {scratchpad.thought}\n\n"
-                if scratchpad.action_str:
-                    message += f"Action: {scratchpad.action_str}\n\n"
-                if scratchpad.observation:
-                    message += f"Observation: {scratchpad.observation}\n\n"
-
-        return message
-
-    def _organize_historic_prompt_messages(
-        self, current_session_messages: list[PromptMessage] | None = None
-    ) -> list[PromptMessage]:
-        """
-        organize historic prompt messages
-        """
-        result: list[PromptMessage] = []
-        scratchpads: list[AgentScratchpadUnit] = []
-        current_scratchpad: AgentScratchpadUnit | None = None
-
-        for message in self.history_prompt_messages:
-            if isinstance(message, AssistantPromptMessage):
-                if not current_scratchpad:
-                    assert isinstance(message.content, str)
-                    current_scratchpad = AgentScratchpadUnit(
-                        agent_response=message.content,
-                        thought=message.content or "I am thinking about how to help you",
-                        action_str="",
-                        action=None,
-                        observation=None,
-                    )
-                    scratchpads.append(current_scratchpad)
-                if message.tool_calls:
-                    try:
-                        current_scratchpad.action = AgentScratchpadUnit.Action(
-                            action_name=message.tool_calls[0].function.name,
-                            action_input=json.loads(message.tool_calls[0].function.arguments),
-                        )
-                        current_scratchpad.action_str = json.dumps(current_scratchpad.action.to_dict())
-                    except Exception:
-                        logger.exception("Failed to parse tool call from assistant message")
-            elif isinstance(message, ToolPromptMessage):
-                if current_scratchpad:
-                    assert isinstance(message.content, str)
-                    current_scratchpad.observation = message.content
-                else:
-                    raise NotImplementedError("expected str type")
-            elif isinstance(message, UserPromptMessage):
-                if scratchpads:
-                    result.append(AssistantPromptMessage(content=self._format_assistant_message(scratchpads)))
-                    scratchpads = []
-                    current_scratchpad = None
-
-                result.append(message)
-
-        if scratchpads:
-            result.append(AssistantPromptMessage(content=self._format_assistant_message(scratchpads)))
-
-        historic_prompts = AgentHistoryPromptTransform(
-            model_config=self.model_config,
-            prompt_messages=current_session_messages or [],
-            history_messages=result,
-            memory=self.memory,
-        ).get_prompt()
-        return historic_prompts

api/core/agent/cot_chat_agent_runner.py

@@ -1,118 +0,0 @@
-import json
-
-from core.agent.cot_agent_runner import CotAgentRunner
-from core.file import file_manager
-from core.model_runtime.entities import (
-    AssistantPromptMessage,
-    PromptMessage,
-    SystemPromptMessage,
-    TextPromptMessageContent,
-    UserPromptMessage,
-)
-from core.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
-from core.model_runtime.utils.encoders import jsonable_encoder
-
-
-class CotChatAgentRunner(CotAgentRunner):
-    def _organize_system_prompt(self) -> SystemPromptMessage:
-        """
-        Organize system prompt
-        """
-        assert self.app_config.agent
-        assert self.app_config.agent.prompt
-
-        prompt_entity = self.app_config.agent.prompt
-        if not prompt_entity:
-            raise ValueError("Agent prompt configuration is not set")
-        first_prompt = prompt_entity.first_prompt
-
-        system_prompt = (
-            first_prompt.replace("{{instruction}}", self._instruction)
-            .replace("{{tools}}", json.dumps(jsonable_encoder(self._prompt_messages_tools)))
-            .replace("{{tool_names}}", ", ".join([tool.name for tool in self._prompt_messages_tools]))
-        )
-
-        return SystemPromptMessage(content=system_prompt)
-
-    def _organize_user_query(self, query, prompt_messages: list[PromptMessage]) -> list[PromptMessage]:
-        """
-        Organize user query
-        """
-        if self.files:
-            # get image detail config
-            image_detail_config = (
-                self.application_generate_entity.file_upload_config.image_config.detail
-                if (
-                    self.application_generate_entity.file_upload_config
-                    and self.application_generate_entity.file_upload_config.image_config
-                )
-                else None
-            )
-            image_detail_config = image_detail_config or ImagePromptMessageContent.DETAIL.LOW
-
-            prompt_message_contents: list[PromptMessageContentUnionTypes] = []
-            for file in self.files:
-                prompt_message_contents.append(
-                    file_manager.to_prompt_message_content(
-                        file,
-                        image_detail_config=image_detail_config,
-                    )
-                )
-            prompt_message_contents.append(TextPromptMessageContent(data=query))
-
-            prompt_messages.append(UserPromptMessage(content=prompt_message_contents))
-        else:
-            prompt_messages.append(UserPromptMessage(content=query))
-
-        return prompt_messages
-
-    def _organize_prompt_messages(self) -> list[PromptMessage]:
-        """
-        Organize
-        """
-        # organize system prompt
-        system_message = self._organize_system_prompt()
-
-        # organize current assistant messages
-        agent_scratchpad = self._agent_scratchpad
-        if not agent_scratchpad:
-            assistant_messages = []
-        else:
-            assistant_message = AssistantPromptMessage(content="")
-            assistant_message.content = ""  # FIXME: type check tell mypy that assistant_message.content is str
-            for unit in agent_scratchpad:
-                if unit.is_final():
-                    assert isinstance(assistant_message.content, str)
-                    assistant_message.content += f"Final Answer: {unit.agent_response}"
-                else:
-                    assert isinstance(assistant_message.content, str)
-                    assistant_message.content += f"Thought: {unit.thought}\n\n"
-                    if unit.action_str:
-                        assistant_message.content += f"Action: {unit.action_str}\n\n"
-                    if unit.observation:
-                        assistant_message.content += f"Observation: {unit.observation}\n\n"
-
-            assistant_messages = [assistant_message]
-
-        # query messages
-        query_messages = self._organize_user_query(self._query, [])
-
-        if assistant_messages:
-            # organize historic prompt messages
-            historic_messages = self._organize_historic_prompt_messages(
-                [system_message, *query_messages, *assistant_messages, UserPromptMessage(content="continue")]
-            )
-            messages = [
-                system_message,
-                *historic_messages,
-                *query_messages,
-                *assistant_messages,
-                UserPromptMessage(content="continue"),
-            ]
-        else:
-            # organize historic prompt messages
-            historic_messages = self._organize_historic_prompt_messages([system_message, *query_messages])
-            messages = [system_message, *historic_messages, *query_messages]
-
-        # join all messages
-        return messages

api/core/agent/cot_completion_agent_runner.py

@@ -1,87 +0,0 @@
-import json
-
-from core.agent.cot_agent_runner import CotAgentRunner
-from core.model_runtime.entities.message_entities import (
-    AssistantPromptMessage,
-    PromptMessage,
-    TextPromptMessageContent,
-    UserPromptMessage,
-)
-from core.model_runtime.utils.encoders import jsonable_encoder
-
-
-class CotCompletionAgentRunner(CotAgentRunner):
-    def _organize_instruction_prompt(self) -> str:
-        """
-        Organize instruction prompt
-        """
-        if self.app_config.agent is None:
-            raise ValueError("Agent configuration is not set")
-        prompt_entity = self.app_config.agent.prompt
-        if prompt_entity is None:
-            raise ValueError("prompt entity is not set")
-        first_prompt = prompt_entity.first_prompt
-
-        system_prompt = (
-            first_prompt.replace("{{instruction}}", self._instruction)
-            .replace("{{tools}}", json.dumps(jsonable_encoder(self._prompt_messages_tools)))
-            .replace("{{tool_names}}", ", ".join([tool.name for tool in self._prompt_messages_tools]))
-        )
-
-        return system_prompt
-
-    def _organize_historic_prompt(self, current_session_messages: list[PromptMessage] | None = None) -> str:
-        """
-        Organize historic prompt
-        """
-        historic_prompt_messages = self._organize_historic_prompt_messages(current_session_messages)
-        historic_prompt = ""
-
-        for message in historic_prompt_messages:
-            if isinstance(message, UserPromptMessage):
-                historic_prompt += f"Question: {message.content}\n\n"
-            elif isinstance(message, AssistantPromptMessage):
-                if isinstance(message.content, str):
-                    historic_prompt += message.content + "\n\n"
-                elif isinstance(message.content, list):
-                    for content in message.content:
-                        if not isinstance(content, TextPromptMessageContent):
-                            continue
-                        historic_prompt += content.data
-
-        return historic_prompt
-
-    def _organize_prompt_messages(self) -> list[PromptMessage]:
-        """
-        Organize prompt messages
-        """
-        # organize system prompt
-        system_prompt = self._organize_instruction_prompt()
-
-        # organize historic prompt messages
-        historic_prompt = self._organize_historic_prompt()
-
-        # organize current assistant messages
-        agent_scratchpad = self._agent_scratchpad
-        assistant_prompt = ""
-        for unit in agent_scratchpad or []:
-            if unit.is_final():
-                assistant_prompt += f"Final Answer: {unit.agent_response}"
-            else:
-                assistant_prompt += f"Thought: {unit.thought}\n\n"
-                if unit.action_str:
-                    assistant_prompt += f"Action: {unit.action_str}\n\n"
-                if unit.observation:
-                    assistant_prompt += f"Observation: {unit.observation}\n\n"
-
-        # query messages
-        query_prompt = f"Question: {self._query}"
-
-        # join all messages
-        prompt = (
-            system_prompt.replace("{{historic_messages}}", historic_prompt)
-            .replace("{{agent_scratchpad}}", assistant_prompt)
-            .replace("{{query}}", query_prompt)
-        )
-
-        return [UserPromptMessage(content=prompt)]

api/core/agent/entities.py

@@ -1,8 +1,11 @@
+import uuid
+from collections.abc import Mapping
 from enum import StrEnum
-from typing import Any, Union
+from typing import Any
 
 from pydantic import BaseModel, Field
 
+from core.agent.output_tools import FINAL_OUTPUT_TOOL
 from core.tools.entities.tool_entities import ToolInvokeMessage, ToolProviderType
 
 
@@ -39,9 +42,9 @@ class AgentScratchpadUnit(BaseModel):
         """
 
         action_name: str
-        action_input: Union[dict, str]
+        action_input: dict[str, Any] | str
 
-        def to_dict(self):
+        def to_dict(self) -> dict[str, Any]:
             """
             Convert to dictionary.
             """
@@ -60,9 +63,9 @@ class AgentScratchpadUnit(BaseModel):
         """
         Check if the scratchpad unit is final.
         """
-        return self.action is None or (
-            "final" in self.action.action_name.lower() and "answer" in self.action.action_name.lower()
-        )
+        if self.action is None:
+            return False
+        return self.action.action_name.lower() == FINAL_OUTPUT_TOOL
 
 
 class AgentEntity(BaseModel):
@@ -92,3 +95,119 @@ class AgentInvokeMessage(ToolInvokeMessage):
     """
 
     pass
+
+
+class ExecutionContext(BaseModel):
+    """Execution context containing trace and audit information.
+
+    This context carries all the IDs and metadata that are not part of
+    the core business logic but needed for tracing, auditing, and
+    correlation purposes.
+    """
+
+    user_id: str | None = None
+    app_id: str | None = None
+    conversation_id: str | None = None
+    message_id: str | None = None
+    tenant_id: str | None = None
+
+    @classmethod
+    def create_minimal(cls, user_id: str | None = None) -> "ExecutionContext":
+        """Create a minimal context with only essential fields."""
+        return cls(user_id=user_id)
+
+    def to_dict(self) -> dict[str, Any]:
+        """Convert to dictionary for passing to legacy code."""
+        return {
+            "user_id": self.user_id,
+            "app_id": self.app_id,
+            "conversation_id": self.conversation_id,
+            "message_id": self.message_id,
+            "tenant_id": self.tenant_id,
+        }
+
+    def with_updates(self, **kwargs: Any) -> "ExecutionContext":
+        """Create a new context with updated fields."""
+        data = self.to_dict()
+        data.update(kwargs)
+
+        return ExecutionContext(
+            user_id=data.get("user_id"),
+            app_id=data.get("app_id"),
+            conversation_id=data.get("conversation_id"),
+            message_id=data.get("message_id"),
+            tenant_id=data.get("tenant_id"),
+        )
+
+
+class AgentLog(BaseModel):
+    """
+    Agent Log.
+    """
+
+    class LogType(StrEnum):
+        """Type of agent log entry."""
+
+        ROUND = "round"  # A complete iteration round
+        THOUGHT = "thought"  # LLM thinking/reasoning
+        TOOL_CALL = "tool_call"  # Tool invocation
+
+    class LogMetadata(StrEnum):
+        STARTED_AT = "started_at"
+        FINISHED_AT = "finished_at"
+        ELAPSED_TIME = "elapsed_time"
+        TOTAL_PRICE = "total_price"
+        TOTAL_TOKENS = "total_tokens"
+        PROVIDER = "provider"
+        CURRENCY = "currency"
+        LLM_USAGE = "llm_usage"
+        ICON = "icon"
+        ICON_DARK = "icon_dark"
+
+    class LogStatus(StrEnum):
+        START = "start"
+        ERROR = "error"
+        SUCCESS = "success"
+
+    id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="The id of the log")
+    label: str = Field(..., description="The label of the log")
+    log_type: LogType = Field(..., description="The type of the log")
+    parent_id: str | None = Field(default=None, description="Leave empty for root log")
+    error: str | None = Field(default=None, description="The error message")
+    status: LogStatus = Field(..., description="The status of the log")
+    data: Mapping[str, Any] = Field(..., description="Detailed log data")
+    metadata: Mapping[LogMetadata, Any] = Field(default={}, description="The metadata of the log")
+
+
+class AgentOutputKind(StrEnum):
+    """
+    Agent output kind.
+    """
+
+    OUTPUT_TEXT = "output_text"
+    FINAL_OUTPUT_ANSWER = "final_output_answer"
+    FINAL_STRUCTURED_OUTPUT = "final_structured_output"
+    ILLEGAL_OUTPUT = "illegal_output"
+
+
+OutputKind = AgentOutputKind
+
+
+class AgentResult(BaseModel):
+    """
+    Agent execution result.
+    """
+
+    class StructuredOutput(BaseModel):
+        """
+        Structured output payload from output tools.
+        """
+
+        output_kind: AgentOutputKind
+        output_text: str | None = None
+        output_data: Mapping[str, Any] | None = None
+
+    output: str | StructuredOutput = Field(default="", description="The generated output")
+    files: list[Any] = Field(default_factory=list, description="Files produced during execution")
+    usage: Any | None = Field(default=None, description="LLM usage statistics")
+    finish_reason: str | None = Field(default=None, description="Reason for completion")

api/core/agent/fc_agent_runner.py

@@ -1,468 +0,0 @@
-import json
-import logging
-from collections.abc import Generator
-from copy import deepcopy
-from typing import Any, Union
-
-from core.agent.base_agent_runner import BaseAgentRunner
-from core.app.apps.base_app_queue_manager import PublishFrom
-from core.app.entities.queue_entities import QueueAgentThoughtEvent, QueueMessageEndEvent, QueueMessageFileEvent
-from core.file import file_manager
-from core.model_runtime.entities import (
-    AssistantPromptMessage,
-    LLMResult,
-    LLMResultChunk,
-    LLMResultChunkDelta,
-    LLMUsage,
-    PromptMessage,
-    PromptMessageContentType,
-    SystemPromptMessage,
-    TextPromptMessageContent,
-    ToolPromptMessage,
-    UserPromptMessage,
-)
-from core.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
-from core.prompt.agent_history_prompt_transform import AgentHistoryPromptTransform
-from core.tools.entities.tool_entities import ToolInvokeMeta
-from core.tools.tool_engine import ToolEngine
-from core.workflow.nodes.agent.exc import AgentMaxIterationError
-from models.model import Message
-
-logger = logging.getLogger(__name__)
-
-
-class FunctionCallAgentRunner(BaseAgentRunner):
-    def run(self, message: Message, query: str, **kwargs: Any) -> Generator[LLMResultChunk, None, None]:
-        """
-        Run FunctionCall agent application
-        """
-        self.query = query
-        app_generate_entity = self.application_generate_entity
-
-        app_config = self.app_config
-        assert app_config is not None, "app_config is required"
-        assert app_config.agent is not None, "app_config.agent is required"
-
-        # convert tools into ModelRuntime Tool format
-        tool_instances, prompt_messages_tools = self._init_prompt_tools()
-
-        assert app_config.agent
-
-        iteration_step = 1
-        max_iteration_steps = min(app_config.agent.max_iteration, 99) + 1
-
-        # continue to run until there is not any tool call
-        function_call_state = True
-        llm_usage: dict[str, LLMUsage | None] = {"usage": None}
-        final_answer = ""
-        prompt_messages: list = []  # Initialize prompt_messages
-
-        # get tracing instance
-        trace_manager = app_generate_entity.trace_manager
-
-        def increase_usage(final_llm_usage_dict: dict[str, LLMUsage | None], usage: LLMUsage):
-            if not final_llm_usage_dict["usage"]:
-                final_llm_usage_dict["usage"] = usage
-            else:
-                llm_usage = final_llm_usage_dict["usage"]
-                llm_usage.prompt_tokens += usage.prompt_tokens
-                llm_usage.completion_tokens += usage.completion_tokens
-                llm_usage.total_tokens += usage.total_tokens
-                llm_usage.prompt_price += usage.prompt_price
-                llm_usage.completion_price += usage.completion_price
-                llm_usage.total_price += usage.total_price
-
-        model_instance = self.model_instance
-
-        while function_call_state and iteration_step <= max_iteration_steps:
-            function_call_state = False
-
-            if iteration_step == max_iteration_steps:
-                # the last iteration, remove all tools
-                prompt_messages_tools = []
-
-            message_file_ids: list[str] = []
-            agent_thought_id = self.create_agent_thought(
-                message_id=message.id, message="", tool_name="", tool_input="", messages_ids=message_file_ids
-            )
-
-            # recalc llm max tokens
-            prompt_messages = self._organize_prompt_messages()
-            self.recalc_llm_max_tokens(self.model_config, prompt_messages)
-            # invoke model
-            chunks: Union[Generator[LLMResultChunk, None, None], LLMResult] = model_instance.invoke_llm(
-                prompt_messages=prompt_messages,
-                model_parameters=app_generate_entity.model_conf.parameters,
-                tools=prompt_messages_tools,
-                stop=app_generate_entity.model_conf.stop,
-                stream=self.stream_tool_call,
-                user=self.user_id,
-                callbacks=[],
-            )
-
-            tool_calls: list[tuple[str, str, dict[str, Any]]] = []
-
-            # save full response
-            response = ""
-
-            # save tool call names and inputs
-            tool_call_names = ""
-            tool_call_inputs = ""
-
-            current_llm_usage = None
-
-            if isinstance(chunks, Generator):
-                is_first_chunk = True
-                for chunk in chunks:
-                    if is_first_chunk:
-                        self.queue_manager.publish(
-                            QueueAgentThoughtEvent(agent_thought_id=agent_thought_id), PublishFrom.APPLICATION_MANAGER
-                        )
-                        is_first_chunk = False
-                    # check if there is any tool call
-                    if self.check_tool_calls(chunk):
-                        function_call_state = True
-                        tool_calls.extend(self.extract_tool_calls(chunk) or [])
-                        tool_call_names = ";".join([tool_call[1] for tool_call in tool_calls])
-                        try:
-                            tool_call_inputs = json.dumps(
-                                {tool_call[1]: tool_call[2] for tool_call in tool_calls}, ensure_ascii=False
-                            )
-                        except TypeError:
-                            # fallback: force ASCII to handle non-serializable objects
-                            tool_call_inputs = json.dumps({tool_call[1]: tool_call[2] for tool_call in tool_calls})
-
-                    if chunk.delta.message and chunk.delta.message.content:
-                        if isinstance(chunk.delta.message.content, list):
-                            for content in chunk.delta.message.content:
-                                response += content.data
-                        else:
-                            response += str(chunk.delta.message.content)
-
-                    if chunk.delta.usage:
-                        increase_usage(llm_usage, chunk.delta.usage)
-                        current_llm_usage = chunk.delta.usage
-
-                    yield chunk
-            else:
-                result = chunks
-                # check if there is any tool call
-                if self.check_blocking_tool_calls(result):
-                    function_call_state = True
-                    tool_calls.extend(self.extract_blocking_tool_calls(result) or [])
-                    tool_call_names = ";".join([tool_call[1] for tool_call in tool_calls])
-                    try:
-                        tool_call_inputs = json.dumps(
-                            {tool_call[1]: tool_call[2] for tool_call in tool_calls}, ensure_ascii=False
-                        )
-                    except TypeError:
-                        # fallback: force ASCII to handle non-serializable objects
-                        tool_call_inputs = json.dumps({tool_call[1]: tool_call[2] for tool_call in tool_calls})
-
-                if result.usage:
-                    increase_usage(llm_usage, result.usage)
-                    current_llm_usage = result.usage
-
-                if result.message and result.message.content:
-                    if isinstance(result.message.content, list):
-                        for content in result.message.content:
-                            response += content.data
-                    else:
-                        response += str(result.message.content)
-
-                if not result.message.content:
-                    result.message.content = ""
-
-                self.queue_manager.publish(
-                    QueueAgentThoughtEvent(agent_thought_id=agent_thought_id), PublishFrom.APPLICATION_MANAGER
-                )
-
-                yield LLMResultChunk(
-                    model=model_instance.model,
-                    prompt_messages=result.prompt_messages,
-                    system_fingerprint=result.system_fingerprint,
-                    delta=LLMResultChunkDelta(
-                        index=0,
-                        message=result.message,
-                        usage=result.usage,
-                    ),
-                )
-
-            assistant_message = AssistantPromptMessage(content=response, tool_calls=[])
-            if tool_calls:
-                assistant_message.tool_calls = [
-                    AssistantPromptMessage.ToolCall(
-                        id=tool_call[0],
-                        type="function",
-                        function=AssistantPromptMessage.ToolCall.ToolCallFunction(
-                            name=tool_call[1], arguments=json.dumps(tool_call[2], ensure_ascii=False)
-                        ),
-                    )
-                    for tool_call in tool_calls
-                ]
-
-            self._current_thoughts.append(assistant_message)
-
-            # save thought
-            self.save_agent_thought(
-                agent_thought_id=agent_thought_id,
-                tool_name=tool_call_names,
-                tool_input=tool_call_inputs,
-                thought=response,
-                tool_invoke_meta=None,
-                observation=None,
-                answer=response,
-                messages_ids=[],
-                llm_usage=current_llm_usage,
-            )
-            self.queue_manager.publish(
-                QueueAgentThoughtEvent(agent_thought_id=agent_thought_id), PublishFrom.APPLICATION_MANAGER
-            )
-
-            final_answer += response + "\n"
-
-            # Check if max iteration is reached and model still wants to call tools
-            if iteration_step == max_iteration_steps and tool_calls:
-                raise AgentMaxIterationError(app_config.agent.max_iteration)
-
-            # call tools
-            tool_responses = []
-            for tool_call_id, tool_call_name, tool_call_args in tool_calls:
-                tool_instance = tool_instances.get(tool_call_name)
-                if not tool_instance:
-                    tool_response = {
-                        "tool_call_id": tool_call_id,
-                        "tool_call_name": tool_call_name,
-                        "tool_response": f"there is not a tool named {tool_call_name}",
-                        "meta": ToolInvokeMeta.error_instance(f"there is not a tool named {tool_call_name}").to_dict(),
-                    }
-                else:
-                    # invoke tool
-                    tool_invoke_response, message_files, tool_invoke_meta = ToolEngine.agent_invoke(
-                        tool=tool_instance,
-                        tool_parameters=tool_call_args,
-                        user_id=self.user_id,
-                        tenant_id=self.tenant_id,
-                        message=self.message,
-                        invoke_from=self.application_generate_entity.invoke_from,
-                        agent_tool_callback=self.agent_callback,
-                        trace_manager=trace_manager,
-                        app_id=self.application_generate_entity.app_config.app_id,
-                        message_id=self.message.id,
-                        conversation_id=self.conversation.id,
-                    )
-                    # publish files
-                    for message_file_id in message_files:
-                        # publish message file
-                        self.queue_manager.publish(
-                            QueueMessageFileEvent(message_file_id=message_file_id), PublishFrom.APPLICATION_MANAGER
-                        )
-                        # add message file ids
-                        message_file_ids.append(message_file_id)
-
-                    tool_response = {
-                        "tool_call_id": tool_call_id,
-                        "tool_call_name": tool_call_name,
-                        "tool_response": tool_invoke_response,
-                        "meta": tool_invoke_meta.to_dict(),
-                    }
-
-                tool_responses.append(tool_response)
-                if tool_response["tool_response"] is not None:
-                    self._current_thoughts.append(
-                        ToolPromptMessage(
-                            content=str(tool_response["tool_response"]),
-                            tool_call_id=tool_call_id,
-                            name=tool_call_name,
-                        )
-                    )
-
-            if len(tool_responses) > 0:
-                # save agent thought
-                self.save_agent_thought(
-                    agent_thought_id=agent_thought_id,
-                    tool_name="",
-                    tool_input="",
-                    thought="",
-                    tool_invoke_meta={
-                        tool_response["tool_call_name"]: tool_response["meta"] for tool_response in tool_responses
-                    },
-                    observation={
-                        tool_response["tool_call_name"]: tool_response["tool_response"]
-                        for tool_response in tool_responses
-                    },
-                    answer="",
-                    messages_ids=message_file_ids,
-                )
-                self.queue_manager.publish(
-                    QueueAgentThoughtEvent(agent_thought_id=agent_thought_id), PublishFrom.APPLICATION_MANAGER
-                )
-
-            # update prompt tool
-            for prompt_tool in prompt_messages_tools:
-                self.update_prompt_message_tool(tool_instances[prompt_tool.name], prompt_tool)
-
-            iteration_step += 1
-
-        # publish end event
-        self.queue_manager.publish(
-            QueueMessageEndEvent(
-                llm_result=LLMResult(
-                    model=model_instance.model,
-                    prompt_messages=prompt_messages,
-                    message=AssistantPromptMessage(content=final_answer),
-                    usage=llm_usage["usage"] or LLMUsage.empty_usage(),
-                    system_fingerprint="",
-                )
-            ),
-            PublishFrom.APPLICATION_MANAGER,
-        )
-
-    def check_tool_calls(self, llm_result_chunk: LLMResultChunk) -> bool:
-        """
-        Check if there is any tool call in llm result chunk
-        """
-        if llm_result_chunk.delta.message.tool_calls:
-            return True
-        return False
-
-    def check_blocking_tool_calls(self, llm_result: LLMResult) -> bool:
-        """
-        Check if there is any blocking tool call in llm result
-        """
-        if llm_result.message.tool_calls:
-            return True
-        return False
-
-    def extract_tool_calls(self, llm_result_chunk: LLMResultChunk) -> list[tuple[str, str, dict[str, Any]]]:
-        """
-        Extract tool calls from llm result chunk
-
-        Returns:
-            List[Tuple[str, str, Dict[str, Any]]]: [(tool_call_id, tool_call_name, tool_call_args)]
-        """
-        tool_calls = []
-        for prompt_message in llm_result_chunk.delta.message.tool_calls:
-            args = {}
-            if prompt_message.function.arguments != "":
-                args = json.loads(prompt_message.function.arguments)
-
-            tool_calls.append(
-                (
-                    prompt_message.id,
-                    prompt_message.function.name,
-                    args,
-                )
-            )
-
-        return tool_calls
-
-    def extract_blocking_tool_calls(self, llm_result: LLMResult) -> list[tuple[str, str, dict[str, Any]]]:
-        """
-        Extract blocking tool calls from llm result
-
-        Returns:
-            List[Tuple[str, str, Dict[str, Any]]]: [(tool_call_id, tool_call_name, tool_call_args)]
-        """
-        tool_calls = []
-        for prompt_message in llm_result.message.tool_calls:
-            args = {}
-            if prompt_message.function.arguments != "":
-                args = json.loads(prompt_message.function.arguments)
-
-            tool_calls.append(
-                (
-                    prompt_message.id,
-                    prompt_message.function.name,
-                    args,
-                )
-            )
-
-        return tool_calls
-
-    def _init_system_message(self, prompt_template: str, prompt_messages: list[PromptMessage]) -> list[PromptMessage]:
-        """
-        Initialize system message
-        """
-        if not prompt_messages and prompt_template:
-            return [
-                SystemPromptMessage(content=prompt_template),
-            ]
-
-        if prompt_messages and not isinstance(prompt_messages[0], SystemPromptMessage) and prompt_template:
-            prompt_messages.insert(0, SystemPromptMessage(content=prompt_template))
-
-        return prompt_messages or []
-
-    def _organize_user_query(self, query: str, prompt_messages: list[PromptMessage]) -> list[PromptMessage]:
-        """
-        Organize user query
-        """
-        if self.files:
-            # get image detail config
-            image_detail_config = (
-                self.application_generate_entity.file_upload_config.image_config.detail
-                if (
-                    self.application_generate_entity.file_upload_config
-                    and self.application_generate_entity.file_upload_config.image_config
-                )
-                else None
-            )
-            image_detail_config = image_detail_config or ImagePromptMessageContent.DETAIL.LOW
-
-            prompt_message_contents: list[PromptMessageContentUnionTypes] = []
-            for file in self.files:
-                prompt_message_contents.append(
-                    file_manager.to_prompt_message_content(
-                        file,
-                        image_detail_config=image_detail_config,
-                    )
-                )
-            prompt_message_contents.append(TextPromptMessageContent(data=query))
-
-            prompt_messages.append(UserPromptMessage(content=prompt_message_contents))
-        else:
-            prompt_messages.append(UserPromptMessage(content=query))
-
-        return prompt_messages
-
-    def _clear_user_prompt_image_messages(self, prompt_messages: list[PromptMessage]) -> list[PromptMessage]:
-        """
-        As for now, gpt supports both fc and vision at the first iteration.
-        We need to remove the image messages from the prompt messages at the first iteration.
-        """
-        prompt_messages = deepcopy(prompt_messages)
-
-        for prompt_message in prompt_messages:
-            if isinstance(prompt_message, UserPromptMessage):
-                if isinstance(prompt_message.content, list):
-                    prompt_message.content = "\n".join(
-                        [
-                            content.data
-                            if content.type == PromptMessageContentType.TEXT
-                            else "[image]"
-                            if content.type == PromptMessageContentType.IMAGE
-                            else "[file]"
-                            for content in prompt_message.content
-                        ]
-                    )
-
-        return prompt_messages
-
-    def _organize_prompt_messages(self):
-        prompt_template = self.app_config.prompt_template.simple_prompt_template or ""
-        self.history_prompt_messages = self._init_system_message(prompt_template, self.history_prompt_messages)
-        query_prompt_messages = self._organize_user_query(self.query or "", [])
-
-        self.history_prompt_messages = AgentHistoryPromptTransform(
-            model_config=self.model_config,
-            prompt_messages=[*query_prompt_messages, *self._current_thoughts],
-            history_messages=self.history_prompt_messages,
-            memory=self.memory,
-        ).get_prompt()
-
-        prompt_messages = [*self.history_prompt_messages, *query_prompt_messages, *self._current_thoughts]
-        if len(self._current_thoughts) != 0:
-            # clear messages after the first iteration
-            prompt_messages = self._clear_user_prompt_image_messages(prompt_messages)
-        return prompt_messages

api/core/agent/output_parser/cot_output_parser.py

@@ -1,7 +1,7 @@
 import json
 import re
 from collections.abc import Generator
-from typing import Union
+from typing import Any, Union, cast
 
 from core.agent.entities import AgentScratchpadUnit
 from core.model_runtime.entities.llm_entities import LLMResultChunk
@@ -10,46 +10,52 @@ from core.model_runtime.entities.llm_entities import LLMResultChunk
 class CotAgentOutputParser:
     @classmethod
     def handle_react_stream_output(
-        cls, llm_response: Generator[LLMResultChunk, None, None], usage_dict: dict
+        cls, llm_response: Generator[LLMResultChunk, None, None], usage_dict: dict[str, Any]
     ) -> Generator[Union[str, AgentScratchpadUnit.Action], None, None]:
-        def parse_action(action) -> Union[str, AgentScratchpadUnit.Action]:
-            action_name = None
-            action_input = None
-            if isinstance(action, str):
+        def parse_action(action: Any) -> Union[str, AgentScratchpadUnit.Action]:
+            action_name: str | None = None
+            action_input: Any | None = None
+            parsed_action: Any = action
+            if isinstance(parsed_action, str):
                 try:
-                    action = json.loads(action, strict=False)
+                    parsed_action = json.loads(parsed_action, strict=False)
                 except json.JSONDecodeError:
-                    return action or ""
+                    return parsed_action or ""
 
             # cohere always returns a list
-            if isinstance(action, list) and len(action) == 1:
-                action = action[0]
+            if isinstance(parsed_action, list):
+                action_list: list[Any] = cast(list[Any], parsed_action)
+                if len(action_list) == 1:
+                    parsed_action = action_list[0]
 
-            for key, value in action.items():
-                if "input" in key.lower():
-                    action_input = value
-                else:
-                    action_name = value
+            if isinstance(parsed_action, dict):
+                action_dict: dict[str, Any] = cast(dict[str, Any], parsed_action)
+                for key, value in action_dict.items():
+                    if "input" in key.lower():
+                        action_input = value
+                    elif isinstance(value, str):
+                        action_name = value
+            else:
+                return json.dumps(parsed_action)
 
             if action_name is not None and action_input is not None:
                 return AgentScratchpadUnit.Action(
                     action_name=action_name,
                     action_input=action_input,
                 )
-            else:
-                return json.dumps(action)
+            return json.dumps(parsed_action)
 
-        def extra_json_from_code_block(code_block) -> list[Union[list, dict]]:
+        def extra_json_from_code_block(code_block: str) -> list[dict[str, Any] | list[Any]]:
             blocks = re.findall(r"```[json]*\s*([\[{].*[]}])\s*```", code_block, re.DOTALL | re.IGNORECASE)
             if not blocks:
                 return []
             try:
-                json_blocks = []
+                json_blocks: list[dict[str, Any] | list[Any]] = []
                 for block in blocks:
                     json_text = re.sub(r"^[a-zA-Z]+\n", "", block.strip(), flags=re.MULTILINE)
                     json_blocks.append(json.loads(json_text, strict=False))
                 return json_blocks
-            except:
+            except Exception:
                 return []
 
         code_block_cache = ""

api/core/agent/output_tools.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+from collections.abc import Sequence
+from typing import Any
+
+from core.app.entities.app_invoke_entities import InvokeFrom
+from core.tools.__base.tool import Tool
+from core.tools.entities.tool_entities import ToolInvokeFrom, ToolParameter, ToolProviderType
+from core.tools.tool_manager import ToolManager
+
+OUTPUT_TOOL_PROVIDER = "agent_output"
+
+OUTPUT_TEXT_TOOL = "output_text"
+FINAL_OUTPUT_TOOL = "final_output_answer"
+FINAL_STRUCTURED_OUTPUT_TOOL = "final_structured_output"
+ILLEGAL_OUTPUT_TOOL = "illegal_output"
+
+OUTPUT_TOOL_NAMES: Sequence[str] = (
+    OUTPUT_TEXT_TOOL,
+    FINAL_OUTPUT_TOOL,
+    FINAL_STRUCTURED_OUTPUT_TOOL,
+    ILLEGAL_OUTPUT_TOOL,
+)
+
+OUTPUT_TOOL_NAME_SET = set(OUTPUT_TOOL_NAMES)
+
+
+def select_output_tool_names(
+    *,
+    structured_output_enabled: bool,
+    include_illegal_output: bool = False,
+) -> list[str]:
+    tool_names = [OUTPUT_TEXT_TOOL]
+    if structured_output_enabled:
+        tool_names.append(FINAL_STRUCTURED_OUTPUT_TOOL)
+    else:
+        tool_names.append(FINAL_OUTPUT_TOOL)
+    if include_illegal_output:
+        tool_names.append(ILLEGAL_OUTPUT_TOOL)
+    return tool_names
+
+
+def select_terminal_tool_name(*, structured_output_enabled: bool) -> str:
+    return FINAL_STRUCTURED_OUTPUT_TOOL if structured_output_enabled else FINAL_OUTPUT_TOOL
+
+
+def build_agent_output_tools(
+    *,
+    tenant_id: str,
+    invoke_from: InvokeFrom,
+    tool_invoke_from: ToolInvokeFrom,
+    output_tool_names: Sequence[str],
+    structured_output_schema: dict[str, Any] | None = None,
+) -> list[Tool]:
+    tools: list[Tool] = []
+    tool_names: list[str] = []
+    for tool_name in output_tool_names:
+        if tool_name not in OUTPUT_TOOL_NAME_SET:
+            raise ValueError(f"Unknown output tool name: {tool_name}")
+        if tool_name not in tool_names:
+            tool_names.append(tool_name)
+
+    for tool_name in tool_names:
+        tool = ToolManager.get_tool_runtime(
+            provider_type=ToolProviderType.BUILT_IN,
+            provider_id=OUTPUT_TOOL_PROVIDER,
+            tool_name=tool_name,
+            tenant_id=tenant_id,
+            invoke_from=invoke_from,
+            tool_invoke_from=tool_invoke_from,
+        )
+
+        if tool_name == FINAL_STRUCTURED_OUTPUT_TOOL and structured_output_schema:
+            tool.entity = tool.entity.model_copy(deep=True)
+            for parameter in tool.entity.parameters:
+                if parameter.name != "data":
+                    continue
+                parameter.type = ToolParameter.ToolParameterType.OBJECT
+                parameter.form = ToolParameter.ToolParameterForm.LLM
+                parameter.required = True
+                parameter.input_schema = structured_output_schema
+        tools.append(tool)
+
+    return tools

api/core/agent/patterns/README.md

@@ -0,0 +1,55 @@
+# Agent Patterns
+
+A unified agent pattern module that powers both Agent V2 workflow nodes and agent applications. Strategies share a common execution contract while adapting to model capabilities and tool availability.
+
+## Overview
+
+The module applies a strategy pattern around LLM/tool orchestration. `StrategyFactory` auto-selects the best implementation based on model features or an explicit agent strategy, and each strategy streams logs and usage consistently.
+
+## Key Features
+
+- **Dual strategies**
+  - `FunctionCallStrategy`: uses native LLM function/tool calling when the model exposes `TOOL_CALL`, `MULTI_TOOL_CALL`, or `STREAM_TOOL_CALL`.
+  - `ReActStrategy`: ReAct (reasoning + acting) flow driven by `CotAgentOutputParser`, used when function calling is unavailable or explicitly requested.
+- **Explicit or auto selection**
+  - `StrategyFactory.create_strategy` prefers an explicit `AgentEntity.Strategy` (FUNCTION_CALLING or CHAIN_OF_THOUGHT).
+  - Otherwise it falls back to function calling when tool-call features exist, or ReAct when they do not.
+- **Unified execution contract**
+  - `AgentPattern.run` yields streaming `AgentLog` entries and `LLMResultChunk` data, returning an `AgentResult` with text, files, usage, and `finish_reason`.
+  - Iterations are configurable and hard-capped at 99 rounds; the last round forces a final answer by withholding tools.
+- **Tool handling and hooks**
+  - Tools convert to `PromptMessageTool` objects before invocation.
+  - Optional `tool_invoke_hook` lets callers override tool execution (e.g., agent apps) while workflow runs use `ToolEngine.generic_invoke`.
+  - Tool outputs support text, links, JSON, variables, blobs, retriever resources, and file attachments; `target=="self"` files are reloaded into model context, others are returned as outputs.
+- **File-aware arguments**
+  - Tool args accept `[File: <id>]` or `[Files: <id1, id2>]` placeholders that resolve to `File` objects before invocation, enabling models to reference uploaded files safely.
+- **ReAct prompt shaping**
+  - System prompts replace `{{instruction}}`, `{{tools}}`, and `{{tool_names}}` placeholders.
+  - Adds `Observation` to stop sequences and appends scratchpad text so the model sees prior Thought/Action/Observation history.
+- **Observability and accounting**
+  - Standardized `AgentLog` entries for rounds, model thoughts, and tool calls, including usage aggregation (`LLMUsage`) across streaming and non-streaming paths.
+
+## Architecture
+
+```
+agent/patterns/
+├── base.py              # Shared utilities: logging, usage, tool invocation, file handling
+├── function_call.py     # Native function-calling loop with tool execution
+├── react.py             # ReAct loop with CoT parsing and scratchpad wiring
+└── strategy_factory.py  # Strategy selection by model features or explicit override
+```
+
+## Usage
+
+- For auto-selection:
+  - Call `StrategyFactory.create_strategy(model_features, model_instance, context, tools, files, ...)` and run the returned strategy with prompt messages and model params.
+- For explicit behavior:
+  - Pass `agent_strategy=AgentEntity.Strategy.FUNCTION_CALLING` to force native calls (falls back to ReAct if unsupported), or `CHAIN_OF_THOUGHT` to force ReAct.
+- Both strategies stream chunks and logs; collect the generator output until it returns an `AgentResult`.
+
+## Integration Points
+
+- **Model runtime**: delegates to `ModelInstance.invoke_llm` for both streaming and non-streaming calls.
+- **Tool system**: defaults to `ToolEngine.generic_invoke`, with `tool_invoke_hook` for custom callers.
+- **Files**: flows through `File` objects for tool inputs/outputs and model-context attachments.
+- **Execution context**: `ExecutionContext` fields (user/app/conversation/message) propagate to tool invocations and logging.

api/core/agent/patterns/__init__.py

@@ -0,0 +1,19 @@
+"""Agent patterns module.
+
+This module provides different strategies for agent execution:
+- FunctionCallStrategy: Uses native function/tool calling
+- ReActStrategy: Uses ReAct (Reasoning + Acting) approach
+- StrategyFactory: Factory for creating strategies based on model features
+"""
+
+from .base import AgentPattern
+from .function_call import FunctionCallStrategy
+from .react import ReActStrategy
+from .strategy_factory import StrategyFactory
+
+__all__ = [
+    "AgentPattern",
+    "FunctionCallStrategy",
+    "ReActStrategy",
+    "StrategyFactory",
+]