Merge remote-tracking branch 'origin/main' into refactor/datasets-components

- Add comprehensive tests for handleExportDSL and exportCheck functions
- Test file download, error handling, and secret variable detection

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

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

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

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

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

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

.github/workflows/api-tests.yml

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

.github/workflows/build-push.yml

@@ -75,9 +75,7 @@ jobs:
         with:
           context: "{{defaultContext}}:${{ matrix.context }}"
           platforms: ${{ matrix.platform }}
-          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' }}
+          build-args: COMMIT_SHA=${{ fromJSON(steps.meta.outputs.json).labels['org.opencontainers.image.revision'] }}
           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,7 +209,6 @@ api/.vscode
 .history
 
 .idea/
-web/migration/
 
 # pnpm
 /.pnpm-store

AGENTS.md

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

CONTRIBUTING.md

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

Makefile

@@ -80,7 +80,7 @@ test:
 		echo "Target: $(TARGET_TESTS)"; \
 		uv run --project api --dev pytest $(TARGET_TESTS); \
 	else \
-		PYTEST_XDIST_ARGS="-n auto" uv run --project api --dev dev/pytest/pytest_unit_tests.sh; \
+		uv run --project api --dev dev/pytest/pytest_unit_tests.sh; \
 	fi
 	@echo "✅ Tests complete"
 

api/.env.example

@@ -33,9 +33,6 @@ 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
 
@@ -720,14 +717,3 @@ 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/agent-notes/configs/feature/__init__.py.md

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

@@ -1,6 +0,0 @@
-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,4 +1,3 @@
-import os
 import sys
 
 
@@ -9,15 +8,10 @@ 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.
@@ -28,15 +22,8 @@ else:
 
     from app_factory import create_app
 
-    socketio_app, flask_app = create_app()
-    app = flask_app
-    celery = flask_app.extensions["celery"]
+    app = create_app()
+    celery = app.extensions["celery"]
 
 if __name__ == "__main__":
-    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()
+    app.run(host="0.0.0.0", port=5001)

api/app_factory.py

@@ -1,7 +1,6 @@
 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
 
@@ -9,7 +8,6 @@ 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__)
 
@@ -62,18 +60,14 @@ def create_flask_app_with_configs() -> DifyApp:
     return dify_app
 
 
-def create_app() -> tuple[socketio.WSGIApp, DifyApp]:
+def create_app() -> 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 socketio_app, app
+    return app
 
 
 def initialize_extensions(app: DifyApp):

api/commands.py

@@ -23,8 +23,7 @@ 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.sandbox import SandboxBuilder, SandboxType
-from core.tools.utils.system_encryption import encrypt_system_params
+from core.tools.utils.system_oauth_encryption import encrypt_system_oauth_params
 from events.app_event import app_was_created
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
@@ -1608,7 +1607,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:
+        except FileNotFoundError as e:
             click.echo(click.style(f"  -> Skipping path {storage_path} as it does not exist.", fg="yellow"))
             continue
         except Exception as e:
@@ -1856,57 +1855,6 @@ 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")
@@ -1926,7 +1874,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_params(client_params_dict)
+        oauth_client_params = encrypt_system_oauth_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"))
@@ -1975,7 +1923,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_params(client_params_dict)
+        oauth_client_params = encrypt_system_oauth_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,7 +2,6 @@ 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
 
@@ -83,17 +82,6 @@ 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,17 +249,6 @@ 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
@@ -1245,13 +1234,6 @@ 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",
@@ -1346,7 +1328,6 @@ class FeatureConfig(
     TriggerConfig,
     AsyncWorkflowConfig,
     PluginConfig,
-    CliApiConfig,
     MarketplaceConfig,
     DataSetConfig,
     EndpointConfig,
@@ -1371,7 +1352,6 @@ class FeatureConfig(
     WorkflowConfig,
     WorkflowNodeExecutionConfig,
     WorkspaceConfig,
-    CollaborationConfig,
     LoginConfig,
     AccountConfig,
     SwaggerUIConfig,

api/controllers/cli_api/__init__.py

@@ -1,27 +0,0 @@
-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 .plugin import plugin as _plugin
-
-api.add_namespace(cli_api_ns)
-
-__all__ = [
-    "_plugin",
-    "api",
-    "bp",
-    "cli_api_ns",
-]

api/controllers/cli_api/plugin/plugin.py

@@ -1,157 +0,0 @@
-from flask import abort
-from flask_restx import Resource
-
-from controllers.cli_api import cli_api_ns
-from controllers.cli_api.plugin.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.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.session.cli_api import CliContext
-from core.skill.entities import ToolInvocationRequest
-from core.tools.entities.tool_entities import ToolProviderType
-from libs.helper import length_prefixed_response
-from models import Account, Tenant
-from models.model import EndUser
-
-
-@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):
-        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)
-
-        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):
-        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):
-        # generate signed url
-        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/list")
-class CliFetchToolsListApi(Resource):
-    @cli_api_only
-    @get_cli_user_tenant
-    @setup_required
-    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/cli_api/plugin/wraps.py

@@ -1,136 +0,0 @@
-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]):
-        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

@@ -1,56 +0,0 @@
-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,7 +32,6 @@ 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,
@@ -40,7 +39,6 @@ from . import (
     feature,
     init_validate,
     ping,
-    sandbox_files,
     setup,
     spec,
     version,
@@ -52,7 +50,6 @@ from .app import (
     agent,
     annotation,
     app,
-    app_asset,
     audio,
     completion,
     conversation,
@@ -63,11 +60,9 @@ from .app import (
     model_config,
     ops_trace,
     site,
-    skills,
     statistic,
     workflow,
     workflow_app_log,
-    workflow_comment,
     workflow_draft_variable,
     workflow_run,
     workflow_statistic,
@@ -119,7 +114,6 @@ from .explore import (
     saved_message,
     trial,
 )
-from .socketio import workflow as socketio_workflow  # pyright: ignore[reportUnusedImport]
 
 # Import tag controllers
 from .tag import tags
@@ -134,7 +128,6 @@ from .workspace import (
     model_providers,
     models,
     plugin,
-    sandbox_providers,
     tool_providers,
     trigger_providers,
     workspace,
@@ -153,7 +146,6 @@ __all__ = [
     "api",
     "apikey",
     "app",
-    "app_asset",
     "audio",
     "banner",
     "billing",
@@ -202,12 +194,9 @@ __all__ = [
     "rag_pipeline_import",
     "rag_pipeline_workflow",
     "recommended_app",
-    "sandbox_files",
-    "sandbox_providers",
     "saved_message",
     "setup",
     "site",
-    "skills",
     "spec",
     "statistic",
     "tags",
@@ -218,7 +207,6 @@ __all__ = [
     "website",
     "workflow",
     "workflow_app_log",
-    "workflow_comment",
     "workflow_draft_variable",
     "workflow_run",
     "workflow_statistic",

api/controllers/console/app/app.py

@@ -1,6 +1,5 @@
 import uuid
 from datetime import datetime
-from enum import StrEnum
 from typing import Any, Literal, TypeAlias
 
 from flask import request
@@ -31,7 +30,6 @@ 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
@@ -57,11 +55,6 @@ 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)")
@@ -126,11 +119,6 @@ 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")
 
@@ -356,7 +344,6 @@ 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
@@ -506,7 +493,6 @@ 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(
@@ -524,10 +510,6 @@ 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:
@@ -538,7 +520,6 @@ 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
@@ -707,29 +688,6 @@ 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

@@ -1,321 +0,0 @@
-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,14 +51,6 @@ 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)
 )
@@ -147,68 +139,3 @@ 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/error.py

@@ -110,6 +110,8 @@ 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
@@ -119,21 +121,3 @@ 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

@@ -16,11 +16,6 @@ from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotIni
 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.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
@@ -60,34 +55,6 @@ 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))
 
@@ -97,8 +64,6 @@ reg(RuleCodeGeneratePayload)
 reg(RuleStructuredOutputPayload)
 reg(InstructionGeneratePayload)
 reg(InstructionTemplatePayload)
-reg(ContextGeneratePayload)
-reg(SuggestedQuestionsPayload)
 
 
 @console_ns.route("/rule-generate")
@@ -313,70 +278,3 @@ 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

@@ -202,7 +202,6 @@ message_detail_model = console_ns.model(
         "status": fields.String,
         "error": fields.String,
         "parent_message_id": fields.String,
-        "generation_detail": fields.Raw,
     },
 )
 

api/controllers/console/app/skills.py

@@ -1,83 +0,0 @@
-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,10 +33,8 @@ 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
@@ -45,12 +43,9 @@ 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__)
@@ -165,14 +160,6 @@ 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
 
@@ -181,15 +168,6 @@ 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))
 
@@ -205,11 +183,8 @@ 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
@@ -679,14 +654,13 @@ 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 = AppBundleService.publish(
+            workflow = workflow_service.publish_workflow(
                 session=session,
                 app_model=app_model,
                 account=current_user,
@@ -797,31 +771,6 @@ 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__])
@@ -1199,83 +1148,3 @@ 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

@@ -1,317 +0,0 @@
-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,16 +17,15 @@ 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, ArrayPromptMessageSegment, FileSegment, Segment
+from core.variables.segments import ArrayFileSegment, 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 libs.login import current_account_with_tenant, login_required
+from factories.variable_factory import build_segment_with_type
+from libs.login import 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
 
@@ -44,16 +43,6 @@ 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),
@@ -62,14 +51,6 @@ 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):
@@ -77,8 +58,6 @@ 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:
@@ -268,8 +247,6 @@ 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(),
         )
@@ -406,7 +383,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 = variable_factory.build_segment_with_type(variable.value_type, raw_value)
+            new_value = 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
@@ -499,34 +476,6 @@ 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):
@@ -578,31 +527,3 @@ 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/sandbox_files.py

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

@@ -1 +0,0 @@
-

api/controllers/console/socketio/workflow.py

@@ -1,119 +0,0 @@
-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/workspace/account.py

@@ -36,7 +36,6 @@ 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_fields
 from libs.datetime_utils import naive_utc_now
@@ -74,10 +73,6 @@ class AccountAvatarPayload(BaseModel):
     avatar: str
 
 
-class AccountAvatarQuery(BaseModel):
-    avatar: str = Field(..., description="Avatar file ID")
-
-
 class AccountInterfaceLanguagePayload(BaseModel):
     interface_language: str
 
@@ -163,7 +158,6 @@ def reg(cls: type[BaseModel]):
 reg(AccountInitPayload)
 reg(AccountNamePayload)
 reg(AccountAvatarPayload)
-reg(AccountAvatarQuery)
 reg(AccountInterfaceLanguagePayload)
 reg(AccountInterfaceThemePayload)
 reg(AccountTimezonePayload)
@@ -267,18 +261,6 @@ class AccountNameApi(Resource):
 
 @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

api/controllers/console/workspace/dsl.py

@@ -1,65 +0,0 @@
-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/sandbox_providers.py

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

api/controllers/files/storage_files.py

@@ -1,80 +0,0 @@
-"""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,53 +448,3 @@ 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,6 +75,7 @@ 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,15 +5,14 @@ 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)
@@ -89,11 +88,11 @@ def plugin_inner_api_only(view: Callable[P, R]):
         if not dify_config.PLUGIN_DAEMON_KEY:
             abort(404)
 
-        # validate using inner api key
+        # get header 'X-Inner-Api-Key'
         inner_api_key = request.headers.get("X-Inner-Api-Key")
-        if inner_api_key and inner_api_key == dify_config.INNER_API_KEY_FOR_PLUGIN:
-            return view(*args, **kwargs)
+        if not inner_api_key or inner_api_key != dify_config.INNER_API_KEY_FOR_PLUGIN:
+            abort(404)
 
-        abort(401)
+        return view(*args, **kwargs)
 
     return decorated

api/core/agent/agent_app_runner.py

@@ -1,380 +0,0 @@
-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, 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,
-    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
-        has_published_thought = False
-        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=True,
-        )
-
-        # 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):
-                    # Handle LLM chunk
-                    if current_agent_thought_id and not has_published_thought:
-                        self.queue_manager.publish(
-                            QueueAgentThoughtEvent(agent_thought_id=current_agent_thought_id),
-                            PublishFrom.APPLICATION_MANAGER,
-                        )
-                        has_published_thought = True
-
-                    yield output
-
-                elif isinstance(output, AgentLog):
-                    # 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,
-                            )
-                            has_published_thought = False
-
-                        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):
-            final_answer = result.text
-            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,
-            )
-
-    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,7 @@ from typing import Union, cast
 
 from sqlalchemy import select
 
-from core.agent.entities import AgentEntity, AgentToolEntity, ExecutionContext
+from core.agent.entities import AgentEntity, AgentToolEntity
 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
@@ -116,20 +116,9 @@ 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:

api/core/agent/cot_agent_runner.py

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

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

@@ -0,0 +1,87 @@
+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,5 +1,3 @@
-import uuid
-from collections.abc import Mapping
 from enum import StrEnum
 from typing import Any, Union
 
@@ -94,96 +92,3 @@ 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) -> "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 AgentResult(BaseModel):
-    """
-    Agent execution result.
-    """
-
-    text: str = Field(default="", description="The generated text")
-    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

@@ -0,0 +1,468 @@
+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/patterns/README.md

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

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

api/core/agent/patterns/base.py

@@ -1,474 +0,0 @@
-"""Base class for agent strategies."""
-
-from __future__ import annotations
-
-import json
-import re
-import time
-from abc import ABC, abstractmethod
-from collections.abc import Callable, Generator
-from typing import TYPE_CHECKING, Any
-
-from core.agent.entities import AgentLog, AgentResult, ExecutionContext
-from core.file import File
-from core.model_manager import ModelInstance
-from core.model_runtime.entities import (
-    AssistantPromptMessage,
-    LLMResult,
-    LLMResultChunk,
-    LLMResultChunkDelta,
-    PromptMessage,
-    PromptMessageTool,
-)
-from core.model_runtime.entities.llm_entities import LLMUsage
-from core.model_runtime.entities.message_entities import TextPromptMessageContent
-from core.tools.entities.tool_entities import ToolInvokeMessage, ToolInvokeMeta
-
-if TYPE_CHECKING:
-    from core.tools.__base.tool import Tool
-
-# Type alias for tool invoke hook
-# Returns: (response_content, message_file_ids, tool_invoke_meta)
-ToolInvokeHook = Callable[["Tool", dict[str, Any], str], tuple[str, list[str], ToolInvokeMeta]]
-
-
-class AgentPattern(ABC):
-    """Base class for agent execution strategies."""
-
-    def __init__(
-        self,
-        model_instance: ModelInstance,
-        tools: list[Tool],
-        context: ExecutionContext,
-        max_iterations: int = 10,
-        workflow_call_depth: int = 0,
-        files: list[File] = [],
-        tool_invoke_hook: ToolInvokeHook | None = None,
-    ):
-        """Initialize the agent strategy."""
-        self.model_instance = model_instance
-        self.tools = tools
-        self.context = context
-        self.max_iterations = min(max_iterations, 99)  # Cap at 99 iterations
-        self.workflow_call_depth = workflow_call_depth
-        self.files: list[File] = files
-        self.tool_invoke_hook = tool_invoke_hook
-
-    @abstractmethod
-    def run(
-        self,
-        prompt_messages: list[PromptMessage],
-        model_parameters: dict[str, Any],
-        stop: list[str] = [],
-        stream: bool = True,
-    ) -> Generator[LLMResultChunk | AgentLog, None, AgentResult]:
-        """Execute the agent strategy."""
-        pass
-
-    def _accumulate_usage(self, total_usage: dict[str, Any], delta_usage: LLMUsage) -> None:
-        """Accumulate LLM usage statistics."""
-        if not total_usage.get("usage"):
-            # Create a copy to avoid modifying the original
-            total_usage["usage"] = LLMUsage(
-                prompt_tokens=delta_usage.prompt_tokens,
-                prompt_unit_price=delta_usage.prompt_unit_price,
-                prompt_price_unit=delta_usage.prompt_price_unit,
-                prompt_price=delta_usage.prompt_price,
-                completion_tokens=delta_usage.completion_tokens,
-                completion_unit_price=delta_usage.completion_unit_price,
-                completion_price_unit=delta_usage.completion_price_unit,
-                completion_price=delta_usage.completion_price,
-                total_tokens=delta_usage.total_tokens,
-                total_price=delta_usage.total_price,
-                currency=delta_usage.currency,
-                latency=delta_usage.latency,
-            )
-        else:
-            current: LLMUsage = total_usage["usage"]
-            current.prompt_tokens += delta_usage.prompt_tokens
-            current.completion_tokens += delta_usage.completion_tokens
-            current.total_tokens += delta_usage.total_tokens
-            current.prompt_price += delta_usage.prompt_price
-            current.completion_price += delta_usage.completion_price
-            current.total_price += delta_usage.total_price
-
-    def _extract_content(self, content: Any) -> str:
-        """Extract text content from message content."""
-        if isinstance(content, list):
-            # Content items are PromptMessageContentUnionTypes
-            text_parts = []
-            for c in content:
-                # Check if it's a TextPromptMessageContent (which has data attribute)
-                if isinstance(c, TextPromptMessageContent):
-                    text_parts.append(c.data)
-            return "".join(text_parts)
-        return str(content)
-
-    def _has_tool_calls(self, chunk: LLMResultChunk) -> bool:
-        """Check if chunk contains tool calls."""
-        # LLMResultChunk always has delta attribute
-        return bool(chunk.delta.message and chunk.delta.message.tool_calls)
-
-    def _has_tool_calls_result(self, result: LLMResult) -> bool:
-        """Check if result contains tool calls (non-streaming)."""
-        # LLMResult always has message attribute
-        return bool(result.message and result.message.tool_calls)
-
-    def _extract_tool_calls(self, chunk: LLMResultChunk) -> list[tuple[str, str, dict[str, Any]]]:
-        """Extract tool calls from streaming chunk."""
-        tool_calls: list[tuple[str, str, dict[str, Any]]] = []
-        if chunk.delta.message and chunk.delta.message.tool_calls:
-            for tool_call in chunk.delta.message.tool_calls:
-                if tool_call.function:
-                    try:
-                        args = json.loads(tool_call.function.arguments) if tool_call.function.arguments else {}
-                    except json.JSONDecodeError:
-                        args = {}
-                    tool_calls.append((tool_call.id or "", tool_call.function.name, args))
-        return tool_calls
-
-    def _extract_tool_calls_result(self, result: LLMResult) -> list[tuple[str, str, dict[str, Any]]]:
-        """Extract tool calls from non-streaming result."""
-        tool_calls = []
-        if result.message and result.message.tool_calls:
-            for tool_call in result.message.tool_calls:
-                if tool_call.function:
-                    try:
-                        args = json.loads(tool_call.function.arguments) if tool_call.function.arguments else {}
-                    except json.JSONDecodeError:
-                        args = {}
-                    tool_calls.append((tool_call.id or "", tool_call.function.name, args))
-        return tool_calls
-
-    def _extract_text_from_message(self, message: PromptMessage) -> str:
-        """Extract text content from a prompt message."""
-        # PromptMessage always has content attribute
-        content = message.content
-        if isinstance(content, str):
-            return content
-        elif isinstance(content, list):
-            # Extract text from content list
-            text_parts = []
-            for item in content:
-                if isinstance(item, TextPromptMessageContent):
-                    text_parts.append(item.data)
-            return " ".join(text_parts)
-        return ""
-
-    def _get_tool_metadata(self, tool_instance: Tool) -> dict[AgentLog.LogMetadata, Any]:
-        """Get metadata for a tool including provider and icon info."""
-        from core.tools.tool_manager import ToolManager
-
-        metadata: dict[AgentLog.LogMetadata, Any] = {}
-        if tool_instance.entity and tool_instance.entity.identity:
-            identity = tool_instance.entity.identity
-            if identity.provider:
-                metadata[AgentLog.LogMetadata.PROVIDER] = identity.provider
-
-            # Get icon using ToolManager for proper URL generation
-            tenant_id = self.context.tenant_id
-            if tenant_id and identity.provider:
-                try:
-                    provider_type = tool_instance.tool_provider_type()
-                    icon = ToolManager.get_tool_icon(tenant_id, provider_type, identity.provider)
-                    if isinstance(icon, str):
-                        metadata[AgentLog.LogMetadata.ICON] = icon
-                    elif isinstance(icon, dict):
-                        # Handle icon dict with background/content or light/dark variants
-                        metadata[AgentLog.LogMetadata.ICON] = icon
-                except Exception:
-                    # Fallback to identity.icon if ToolManager fails
-                    if identity.icon:
-                        metadata[AgentLog.LogMetadata.ICON] = identity.icon
-            elif identity.icon:
-                metadata[AgentLog.LogMetadata.ICON] = identity.icon
-        return metadata
-
-    def _create_log(
-        self,
-        label: str,
-        log_type: AgentLog.LogType,
-        status: AgentLog.LogStatus,
-        data: dict[str, Any] | None = None,
-        parent_id: str | None = None,
-        extra_metadata: dict[AgentLog.LogMetadata, Any] | None = None,
-    ) -> AgentLog:
-        """Create a new AgentLog with standard metadata."""
-        metadata: dict[AgentLog.LogMetadata, Any] = {
-            AgentLog.LogMetadata.STARTED_AT: time.perf_counter(),
-        }
-        if extra_metadata:
-            metadata.update(extra_metadata)
-
-        return AgentLog(
-            label=label,
-            log_type=log_type,
-            status=status,
-            data=data or {},
-            parent_id=parent_id,
-            metadata=metadata,
-        )
-
-    def _finish_log(
-        self,
-        log: AgentLog,
-        data: dict[str, Any] | None = None,
-        usage: LLMUsage | None = None,
-    ) -> AgentLog:
-        """Finish an AgentLog by updating its status and metadata."""
-        log.status = AgentLog.LogStatus.SUCCESS
-
-        if data is not None:
-            log.data = data
-
-        # Calculate elapsed time
-        started_at = log.metadata.get(AgentLog.LogMetadata.STARTED_AT, time.perf_counter())
-        finished_at = time.perf_counter()
-
-        # Update metadata
-        log.metadata = {
-            **log.metadata,
-            AgentLog.LogMetadata.FINISHED_AT: finished_at,
-            # Calculate elapsed time in seconds
-            AgentLog.LogMetadata.ELAPSED_TIME: round(finished_at - started_at, 4),
-        }
-
-        # Add usage information if provided
-        if usage:
-            log.metadata.update(
-                {
-                    AgentLog.LogMetadata.TOTAL_PRICE: usage.total_price,
-                    AgentLog.LogMetadata.CURRENCY: usage.currency,
-                    AgentLog.LogMetadata.TOTAL_TOKENS: usage.total_tokens,
-                    AgentLog.LogMetadata.LLM_USAGE: usage,
-                }
-            )
-
-        return log
-
-    def _replace_file_references(self, tool_args: dict[str, Any]) -> dict[str, Any]:
-        """
-        Replace file references in tool arguments with actual File objects.
-
-        Args:
-            tool_args: Dictionary of tool arguments
-
-        Returns:
-            Updated tool arguments with file references replaced
-        """
-        # Process each argument in the dictionary
-        processed_args: dict[str, Any] = {}
-        for key, value in tool_args.items():
-            processed_args[key] = self._process_file_reference(value)
-        return processed_args
-
-    def _process_file_reference(self, data: Any) -> Any:
-        """
-        Recursively process data to replace file references.
-        Supports both single file [File: file_id] and multiple files [Files: file_id1, file_id2, ...].
-
-        Args:
-            data: The data to process (can be dict, list, str, or other types)
-
-        Returns:
-            Processed data with file references replaced
-        """
-        single_file_pattern = re.compile(r"^\[File:\s*([^\]]+)\]$")
-        multiple_files_pattern = re.compile(r"^\[Files:\s*([^\]]+)\]$")
-
-        if isinstance(data, dict):
-            # Process dictionary recursively
-            return {key: self._process_file_reference(value) for key, value in data.items()}
-        elif isinstance(data, list):
-            # Process list recursively
-            return [self._process_file_reference(item) for item in data]
-        elif isinstance(data, str):
-            # Check for single file pattern [File: file_id]
-            single_match = single_file_pattern.match(data.strip())
-            if single_match:
-                file_id = single_match.group(1).strip()
-                # Find the file in self.files
-                for file in self.files:
-                    if file.id and str(file.id) == file_id:
-                        return file
-                # If file not found, return original value
-                return data
-
-            # Check for multiple files pattern [Files: file_id1, file_id2, ...]
-            multiple_match = multiple_files_pattern.match(data.strip())
-            if multiple_match:
-                file_ids_str = multiple_match.group(1).strip()
-                # Split by comma and strip whitespace
-                file_ids = [fid.strip() for fid in file_ids_str.split(",")]
-
-                # Find all matching files
-                matched_files: list[File] = []
-                for file_id in file_ids:
-                    for file in self.files:
-                        if file.id and str(file.id) == file_id:
-                            matched_files.append(file)
-                            break
-
-                # Return list of files if any were found, otherwise return original
-                return matched_files or data
-
-            return data
-        else:
-            # Return other types as-is
-            return data
-
-    def _create_text_chunk(self, text: str, prompt_messages: list[PromptMessage]) -> LLMResultChunk:
-        """Create a text chunk for streaming."""
-        return LLMResultChunk(
-            model=self.model_instance.model,
-            prompt_messages=prompt_messages,
-            delta=LLMResultChunkDelta(
-                index=0,
-                message=AssistantPromptMessage(content=text),
-                usage=None,
-            ),
-            system_fingerprint="",
-        )
-
-    def _invoke_tool(
-        self,
-        tool_instance: Tool,
-        tool_args: dict[str, Any],
-        tool_name: str,
-    ) -> tuple[str, list[File], ToolInvokeMeta | None]:
-        """
-        Invoke a tool and collect its response.
-
-        Args:
-            tool_instance: The tool instance to invoke
-            tool_args: Tool arguments
-            tool_name: Name of the tool
-
-        Returns:
-            Tuple of (response_content, tool_files, tool_invoke_meta)
-        """
-        # Process tool_args to replace file references with actual File objects
-        tool_args = self._replace_file_references(tool_args)
-
-        # If a tool invoke hook is set, use it instead of generic_invoke
-        if self.tool_invoke_hook:
-            response_content, _, tool_invoke_meta = self.tool_invoke_hook(tool_instance, tool_args, tool_name)
-            # Note: message_file_ids are stored in DB, we don't convert them to File objects here
-            # The caller (AgentAppRunner) handles file publishing
-            return response_content, [], tool_invoke_meta
-
-        # Default: use generic_invoke for workflow scenarios
-        # Import here to avoid circular import
-        from core.tools.tool_engine import DifyWorkflowCallbackHandler, ToolEngine
-
-        tool_response = ToolEngine().generic_invoke(
-            tool=tool_instance,
-            tool_parameters=tool_args,
-            user_id=self.context.user_id or "",
-            workflow_tool_callback=DifyWorkflowCallbackHandler(),
-            workflow_call_depth=self.workflow_call_depth,
-            app_id=self.context.app_id,
-            conversation_id=self.context.conversation_id,
-            message_id=self.context.message_id,
-        )
-
-        # Collect response and files
-        response_content = ""
-        tool_files: list[File] = []
-
-        for response in tool_response:
-            if response.type == ToolInvokeMessage.MessageType.TEXT:
-                assert isinstance(response.message, ToolInvokeMessage.TextMessage)
-                response_content += response.message.text
-
-            elif response.type == ToolInvokeMessage.MessageType.LINK:
-                # Handle link messages
-                if isinstance(response.message, ToolInvokeMessage.TextMessage):
-                    response_content += f"[Link: {response.message.text}]"
-
-            elif response.type == ToolInvokeMessage.MessageType.IMAGE:
-                # Handle image URL messages
-                if isinstance(response.message, ToolInvokeMessage.TextMessage):
-                    response_content += f"[Image: {response.message.text}]"
-
-            elif response.type == ToolInvokeMessage.MessageType.IMAGE_LINK:
-                # Handle image link messages
-                if isinstance(response.message, ToolInvokeMessage.TextMessage):
-                    response_content += f"[Image: {response.message.text}]"
-
-            elif response.type == ToolInvokeMessage.MessageType.BINARY_LINK:
-                # Handle binary file link messages
-                if isinstance(response.message, ToolInvokeMessage.TextMessage):
-                    filename = response.meta.get("filename", "file") if response.meta else "file"
-                    response_content += f"[File: {filename} - {response.message.text}]"
-
-            elif response.type == ToolInvokeMessage.MessageType.JSON:
-                # Handle JSON messages
-                if isinstance(response.message, ToolInvokeMessage.JsonMessage):
-                    response_content += json.dumps(response.message.json_object, ensure_ascii=False, indent=2)
-
-            elif response.type == ToolInvokeMessage.MessageType.BLOB:
-                # Handle blob messages - convert to text representation
-                if isinstance(response.message, ToolInvokeMessage.BlobMessage):
-                    mime_type = (
-                        response.meta.get("mime_type", "application/octet-stream")
-                        if response.meta
-                        else "application/octet-stream"
-                    )
-                    size = len(response.message.blob)
-                    response_content += f"[Binary data: {mime_type}, size: {size} bytes]"
-
-            elif response.type == ToolInvokeMessage.MessageType.VARIABLE:
-                # Handle variable messages
-                if isinstance(response.message, ToolInvokeMessage.VariableMessage):
-                    var_name = response.message.variable_name
-                    var_value = response.message.variable_value
-                    if isinstance(var_value, str):
-                        response_content += var_value
-                    else:
-                        response_content += f"[Variable {var_name}: {json.dumps(var_value, ensure_ascii=False)}]"
-
-            elif response.type == ToolInvokeMessage.MessageType.BLOB_CHUNK:
-                # Handle blob chunk messages - these are parts of a larger blob
-                if isinstance(response.message, ToolInvokeMessage.BlobChunkMessage):
-                    response_content += f"[Blob chunk {response.message.sequence}: {len(response.message.blob)} bytes]"
-
-            elif response.type == ToolInvokeMessage.MessageType.RETRIEVER_RESOURCES:
-                # Handle retriever resources messages
-                if isinstance(response.message, ToolInvokeMessage.RetrieverResourceMessage):
-                    response_content += response.message.context
-
-            elif response.type == ToolInvokeMessage.MessageType.FILE:
-                # Extract file from meta
-                if response.meta and "file" in response.meta:
-                    file = response.meta["file"]
-                    if isinstance(file, File):
-                        # Check if file is for model or tool output
-                        if response.meta.get("target") == "self":
-                            # File is for model - add to files for next prompt
-                            self.files.append(file)
-                            response_content += f"File '{file.filename}' has been loaded into your context."
-                        else:
-                            # File is tool output
-                            tool_files.append(file)
-
-        return response_content, tool_files, None
-
-    def _find_tool_by_name(self, tool_name: str) -> Tool | None:
-        """Find a tool instance by its name."""
-        for tool in self.tools:
-            if tool.entity.identity.name == tool_name:
-                return tool
-        return None
-
-    def _convert_tools_to_prompt_format(self) -> list[PromptMessageTool]:
-        """Convert tools to prompt message format."""
-        prompt_tools: list[PromptMessageTool] = []
-        for tool in self.tools:
-            prompt_tools.append(tool.to_prompt_message_tool())
-        return prompt_tools
-
-    def _update_usage_with_empty(self, llm_usage: dict[str, Any]) -> None:
-        """Initialize usage tracking with empty usage if not set."""
-        if "usage" not in llm_usage or llm_usage["usage"] is None:
-            llm_usage["usage"] = LLMUsage.empty_usage()

api/core/agent/patterns/function_call.py

@@ -1,299 +0,0 @@
-"""Function Call strategy implementation."""
-
-import json
-from collections.abc import Generator
-from typing import Any, Union
-
-from core.agent.entities import AgentLog, AgentResult
-from core.file import File
-from core.model_runtime.entities import (
-    AssistantPromptMessage,
-    LLMResult,
-    LLMResultChunk,
-    LLMResultChunkDelta,
-    LLMUsage,
-    PromptMessage,
-    PromptMessageTool,
-    ToolPromptMessage,
-)
-from core.tools.entities.tool_entities import ToolInvokeMeta
-
-from .base import AgentPattern
-
-
-class FunctionCallStrategy(AgentPattern):
-    """Function Call strategy using model's native tool calling capability."""
-
-    def run(
-        self,
-        prompt_messages: list[PromptMessage],
-        model_parameters: dict[str, Any],
-        stop: list[str] = [],
-        stream: bool = True,
-    ) -> Generator[LLMResultChunk | AgentLog, None, AgentResult]:
-        """Execute the function call agent strategy."""
-        # Convert tools to prompt format
-        prompt_tools: list[PromptMessageTool] = self._convert_tools_to_prompt_format()
-
-        # Initialize tracking
-        iteration_step: int = 1
-        max_iterations: int = self.max_iterations + 1
-        function_call_state: bool = True
-        total_usage: dict[str, LLMUsage | None] = {"usage": None}
-        messages: list[PromptMessage] = list(prompt_messages)  # Create mutable copy
-        final_text: str = ""
-        finish_reason: str | None = None
-        output_files: list[File] = []  # Track files produced by tools
-
-        while function_call_state and iteration_step <= max_iterations:
-            function_call_state = False
-            round_log = self._create_log(
-                label=f"ROUND {iteration_step}",
-                log_type=AgentLog.LogType.ROUND,
-                status=AgentLog.LogStatus.START,
-                data={},
-            )
-            yield round_log
-            # On last iteration, remove tools to force final answer
-            current_tools: list[PromptMessageTool] = [] if iteration_step == max_iterations else prompt_tools
-            model_log = self._create_log(
-                label=f"{self.model_instance.model} Thought",
-                log_type=AgentLog.LogType.THOUGHT,
-                status=AgentLog.LogStatus.START,
-                data={},
-                parent_id=round_log.id,
-                extra_metadata={
-                    AgentLog.LogMetadata.PROVIDER: self.model_instance.provider,
-                },
-            )
-            yield model_log
-
-            # Track usage for this round only
-            round_usage: dict[str, LLMUsage | None] = {"usage": None}
-
-            # Invoke model
-            chunks: Union[Generator[LLMResultChunk, None, None], LLMResult] = self.model_instance.invoke_llm(
-                prompt_messages=messages,
-                model_parameters=model_parameters,
-                tools=current_tools,
-                stop=stop,
-                stream=stream,
-                user=self.context.user_id,
-                callbacks=[],
-            )
-
-            # Process response
-            tool_calls, response_content, chunk_finish_reason = yield from self._handle_chunks(
-                chunks, round_usage, model_log
-            )
-            messages.append(self._create_assistant_message(response_content, tool_calls))
-
-            # Accumulate to total usage
-            round_usage_value = round_usage.get("usage")
-            if round_usage_value:
-                self._accumulate_usage(total_usage, round_usage_value)
-
-            # Update final text if no tool calls (this is likely the final answer)
-            if not tool_calls:
-                final_text = response_content
-
-            # Update finish reason
-            if chunk_finish_reason:
-                finish_reason = chunk_finish_reason
-
-            # Process tool calls
-            tool_outputs: dict[str, str] = {}
-            if tool_calls:
-                function_call_state = True
-                # Execute tools
-                for tool_call_id, tool_name, tool_args in tool_calls:
-                    tool_response, tool_files, _ = yield from self._handle_tool_call(
-                        tool_name, tool_args, tool_call_id, messages, round_log
-                    )
-                    tool_outputs[tool_name] = tool_response
-                    # Track files produced by tools
-                    output_files.extend(tool_files)
-            yield self._finish_log(
-                round_log,
-                data={
-                    "llm_result": response_content,
-                    "tool_calls": [
-                        {"name": tc[1], "args": tc[2], "output": tool_outputs.get(tc[1], "")} for tc in tool_calls
-                    ]
-                    if tool_calls
-                    else [],
-                    "final_answer": final_text if not function_call_state else None,
-                },
-                usage=round_usage.get("usage"),
-            )
-            iteration_step += 1
-
-        # Return final result
-        from core.agent.entities import AgentResult
-
-        return AgentResult(
-            text=final_text,
-            files=output_files,
-            usage=total_usage.get("usage") or LLMUsage.empty_usage(),
-            finish_reason=finish_reason,
-        )
-
-    def _handle_chunks(
-        self,
-        chunks: Union[Generator[LLMResultChunk, None, None], LLMResult],
-        llm_usage: dict[str, LLMUsage | None],
-        start_log: AgentLog,
-    ) -> Generator[
-        LLMResultChunk | AgentLog,
-        None,
-        tuple[list[tuple[str, str, dict[str, Any]]], str, str | None],
-    ]:
-        """Handle LLM response chunks and extract tool calls and content.
-
-        Returns a tuple of (tool_calls, response_content, finish_reason).
-        """
-        tool_calls: list[tuple[str, str, dict[str, Any]]] = []
-        response_content: str = ""
-        finish_reason: str | None = None
-        if isinstance(chunks, Generator):
-            # Streaming response
-            for chunk in chunks:
-                # Extract tool calls
-                if self._has_tool_calls(chunk):
-                    tool_calls.extend(self._extract_tool_calls(chunk))
-
-                # Extract content
-                if chunk.delta.message and chunk.delta.message.content:
-                    response_content += self._extract_content(chunk.delta.message.content)
-
-                # Track usage
-                if chunk.delta.usage:
-                    self._accumulate_usage(llm_usage, chunk.delta.usage)
-
-                # Capture finish reason
-                if chunk.delta.finish_reason:
-                    finish_reason = chunk.delta.finish_reason
-
-                yield chunk
-        else:
-            # Non-streaming response
-            result: LLMResult = chunks
-
-            if self._has_tool_calls_result(result):
-                tool_calls.extend(self._extract_tool_calls_result(result))
-
-            if result.message and result.message.content:
-                response_content += self._extract_content(result.message.content)
-
-            if result.usage:
-                self._accumulate_usage(llm_usage, result.usage)
-
-            # Convert to streaming format
-            yield LLMResultChunk(
-                model=result.model,
-                prompt_messages=result.prompt_messages,
-                delta=LLMResultChunkDelta(index=0, message=result.message, usage=result.usage),
-            )
-        yield self._finish_log(
-            start_log,
-            data={
-                "result": response_content,
-            },
-            usage=llm_usage.get("usage"),
-        )
-        return tool_calls, response_content, finish_reason
-
-    def _create_assistant_message(
-        self, content: str, tool_calls: list[tuple[str, str, dict[str, Any]]] | None = None
-    ) -> AssistantPromptMessage:
-        """Create assistant message with tool calls."""
-        if tool_calls is None:
-            return AssistantPromptMessage(content=content)
-        return AssistantPromptMessage(
-            content=content or "",
-            tool_calls=[
-                AssistantPromptMessage.ToolCall(
-                    id=tc[0],
-                    type="function",
-                    function=AssistantPromptMessage.ToolCall.ToolCallFunction(name=tc[1], arguments=json.dumps(tc[2])),
-                )
-                for tc in tool_calls
-            ],
-        )
-
-    def _handle_tool_call(
-        self,
-        tool_name: str,
-        tool_args: dict[str, Any],
-        tool_call_id: str,
-        messages: list[PromptMessage],
-        round_log: AgentLog,
-    ) -> Generator[AgentLog, None, tuple[str, list[File], ToolInvokeMeta | None]]:
-        """Handle a single tool call and return response with files and meta."""
-        # Find tool
-        tool_instance = self._find_tool_by_name(tool_name)
-        if not tool_instance:
-            raise ValueError(f"Tool {tool_name} not found")
-
-        # Get tool metadata (provider, icon, etc.)
-        tool_metadata = self._get_tool_metadata(tool_instance)
-
-        # Create tool call log
-        tool_call_log = self._create_log(
-            label=f"CALL {tool_name}",
-            log_type=AgentLog.LogType.TOOL_CALL,
-            status=AgentLog.LogStatus.START,
-            data={
-                "tool_call_id": tool_call_id,
-                "tool_name": tool_name,
-                "tool_args": tool_args,
-            },
-            parent_id=round_log.id,
-            extra_metadata=tool_metadata,
-        )
-        yield tool_call_log
-
-        # Invoke tool using base class method with error handling
-        try:
-            response_content, tool_files, tool_invoke_meta = self._invoke_tool(tool_instance, tool_args, tool_name)
-
-            yield self._finish_log(
-                tool_call_log,
-                data={
-                    **tool_call_log.data,
-                    "output": response_content,
-                    "files": len(tool_files),
-                    "meta": tool_invoke_meta.to_dict() if tool_invoke_meta else None,
-                },
-            )
-            final_content = response_content or "Tool executed successfully"
-            # Add tool response to messages
-            messages.append(
-                ToolPromptMessage(
-                    content=final_content,
-                    tool_call_id=tool_call_id,
-                    name=tool_name,
-                )
-            )
-            return response_content, tool_files, tool_invoke_meta
-        except Exception as e:
-            # Tool invocation failed, yield error log
-            error_message = str(e)
-            tool_call_log.status = AgentLog.LogStatus.ERROR
-            tool_call_log.error = error_message
-            tool_call_log.data = {
-                **tool_call_log.data,
-                "error": error_message,
-            }
-            yield tool_call_log
-
-            # Add error message to conversation
-            error_content = f"Tool execution failed: {error_message}"
-            messages.append(
-                ToolPromptMessage(
-                    content=error_content,
-                    tool_call_id=tool_call_id,
-                    name=tool_name,
-                )
-            )
-            return error_content, [], None

api/core/agent/patterns/react.py

@@ -1,418 +0,0 @@
-"""ReAct strategy implementation."""
-
-from __future__ import annotations
-
-import json
-from collections.abc import Generator
-from typing import TYPE_CHECKING, Any, Union
-
-from core.agent.entities import AgentLog, AgentResult, AgentScratchpadUnit, ExecutionContext
-from core.agent.output_parser.cot_output_parser import CotAgentOutputParser
-from core.file import File
-from core.model_manager import ModelInstance
-from core.model_runtime.entities import (
-    AssistantPromptMessage,
-    LLMResult,
-    LLMResultChunk,
-    LLMResultChunkDelta,
-    PromptMessage,
-    SystemPromptMessage,
-)
-
-from .base import AgentPattern, ToolInvokeHook
-
-if TYPE_CHECKING:
-    from core.tools.__base.tool import Tool
-
-
-class ReActStrategy(AgentPattern):
-    """ReAct strategy using reasoning and acting approach."""
-
-    def __init__(
-        self,
-        model_instance: ModelInstance,
-        tools: list[Tool],
-        context: ExecutionContext,
-        max_iterations: int = 10,
-        workflow_call_depth: int = 0,
-        files: list[File] = [],
-        tool_invoke_hook: ToolInvokeHook | None = None,
-        instruction: str = "",
-    ):
-        """Initialize the ReAct strategy with instruction support."""
-        super().__init__(
-            model_instance=model_instance,
-            tools=tools,
-            context=context,
-            max_iterations=max_iterations,
-            workflow_call_depth=workflow_call_depth,
-            files=files,
-            tool_invoke_hook=tool_invoke_hook,
-        )
-        self.instruction = instruction
-
-    def run(
-        self,
-        prompt_messages: list[PromptMessage],
-        model_parameters: dict[str, Any],
-        stop: list[str] = [],
-        stream: bool = True,
-    ) -> Generator[LLMResultChunk | AgentLog, None, AgentResult]:
-        """Execute the ReAct agent strategy."""
-        # Initialize tracking
-        agent_scratchpad: list[AgentScratchpadUnit] = []
-        iteration_step: int = 1
-        max_iterations: int = self.max_iterations + 1
-        react_state: bool = True
-        total_usage: dict[str, Any] = {"usage": None}
-        output_files: list[File] = []  # Track files produced by tools
-        final_text: str = ""
-        finish_reason: str | None = None
-
-        # Add "Observation" to stop sequences
-        if "Observation" not in stop:
-            stop = stop.copy()
-            stop.append("Observation")
-
-        while react_state and iteration_step <= max_iterations:
-            react_state = False
-            round_log = self._create_log(
-                label=f"ROUND {iteration_step}",
-                log_type=AgentLog.LogType.ROUND,
-                status=AgentLog.LogStatus.START,
-                data={},
-            )
-            yield round_log
-
-            # Build prompt with/without tools based on iteration
-            include_tools = iteration_step < max_iterations
-            current_messages = self._build_prompt_with_react_format(
-                prompt_messages, agent_scratchpad, include_tools, self.instruction
-            )
-
-            model_log = self._create_log(
-                label=f"{self.model_instance.model} Thought",
-                log_type=AgentLog.LogType.THOUGHT,
-                status=AgentLog.LogStatus.START,
-                data={},
-                parent_id=round_log.id,
-                extra_metadata={
-                    AgentLog.LogMetadata.PROVIDER: self.model_instance.provider,
-                },
-            )
-            yield model_log
-
-            # Track usage for this round only
-            round_usage: dict[str, Any] = {"usage": None}
-
-            # Use current messages directly (files are handled by base class if needed)
-            messages_to_use = current_messages
-
-            # Invoke model
-            chunks: Union[Generator[LLMResultChunk, None, None], LLMResult] = self.model_instance.invoke_llm(
-                prompt_messages=messages_to_use,
-                model_parameters=model_parameters,
-                stop=stop,
-                stream=stream,
-                user=self.context.user_id or "",
-                callbacks=[],
-            )
-
-            # Process response
-            scratchpad, chunk_finish_reason = yield from self._handle_chunks(
-                chunks, round_usage, model_log, current_messages
-            )
-            agent_scratchpad.append(scratchpad)
-
-            # Accumulate to total usage
-            round_usage_value = round_usage.get("usage")
-            if round_usage_value:
-                self._accumulate_usage(total_usage, round_usage_value)
-
-            # Update finish reason
-            if chunk_finish_reason:
-                finish_reason = chunk_finish_reason
-
-            # Check if we have an action to execute
-            if scratchpad.action and scratchpad.action.action_name.lower() != "final answer":
-                react_state = True
-                # Execute tool
-                observation, tool_files = yield from self._handle_tool_call(
-                    scratchpad.action, current_messages, round_log
-                )
-                scratchpad.observation = observation
-                # Track files produced by tools
-                output_files.extend(tool_files)
-
-                # Add observation to scratchpad for display
-                yield self._create_text_chunk(f"\nObservation: {observation}\n", current_messages)
-            else:
-                # Extract final answer
-                if scratchpad.action and scratchpad.action.action_input:
-                    final_answer = scratchpad.action.action_input
-                    if isinstance(final_answer, dict):
-                        final_answer = json.dumps(final_answer, ensure_ascii=False)
-                    final_text = str(final_answer)
-                elif scratchpad.thought:
-                    # If no action but we have thought, use thought as final answer
-                    final_text = scratchpad.thought
-
-            yield self._finish_log(
-                round_log,
-                data={
-                    "thought": scratchpad.thought,
-                    "action": scratchpad.action_str if scratchpad.action else None,
-                    "observation": scratchpad.observation or None,
-                    "final_answer": final_text if not react_state else None,
-                },
-                usage=round_usage.get("usage"),
-            )
-            iteration_step += 1
-
-        # Return final result
-
-        from core.agent.entities import AgentResult
-
-        return AgentResult(
-            text=final_text, files=output_files, usage=total_usage.get("usage"), finish_reason=finish_reason
-        )
-
-    def _build_prompt_with_react_format(
-        self,
-        original_messages: list[PromptMessage],
-        agent_scratchpad: list[AgentScratchpadUnit],
-        include_tools: bool = True,
-        instruction: str = "",
-    ) -> list[PromptMessage]:
-        """Build prompt messages with ReAct format."""
-        # Copy messages to avoid modifying original
-        messages = list(original_messages)
-
-        # Find and update the system prompt that should already exist
-        system_prompt_found = False
-        for i, msg in enumerate(messages):
-            if isinstance(msg, SystemPromptMessage):
-                system_prompt_found = True
-                # The system prompt from frontend already has the template, just replace placeholders
-
-                # Format tools
-                tools_str = ""
-                tool_names = []
-                if include_tools and self.tools:
-                    # Convert tools to prompt message tools format
-                    prompt_tools = [tool.to_prompt_message_tool() for tool in self.tools]
-                    tool_names = [tool.name for tool in prompt_tools]
-
-                    # Format tools as JSON for comprehensive information
-                    from core.model_runtime.utils.encoders import jsonable_encoder
-
-                    tools_str = json.dumps(jsonable_encoder(prompt_tools), indent=2)
-                    tool_names_str = ", ".join(f'"{name}"' for name in tool_names)
-                else:
-                    tools_str = "No tools available"
-                    tool_names_str = ""
-
-                # Replace placeholders in the existing system prompt
-                updated_content = msg.content
-                assert isinstance(updated_content, str)
-                updated_content = updated_content.replace("{{instruction}}", instruction)
-                updated_content = updated_content.replace("{{tools}}", tools_str)
-                updated_content = updated_content.replace("{{tool_names}}", tool_names_str)
-
-                # Create new SystemPromptMessage with updated content
-                messages[i] = SystemPromptMessage(content=updated_content)
-                break
-
-        # If no system prompt found, that's unexpected but add scratchpad anyway
-        if not system_prompt_found:
-            # This shouldn't happen if frontend is working correctly
-            pass
-
-        # Format agent scratchpad
-        scratchpad_str = ""
-        if agent_scratchpad:
-            scratchpad_parts: list[str] = []
-            for unit in agent_scratchpad:
-                if unit.thought:
-                    scratchpad_parts.append(f"Thought: {unit.thought}")
-                if unit.action_str:
-                    scratchpad_parts.append(f"Action:\n```\n{unit.action_str}\n```")
-                if unit.observation:
-                    scratchpad_parts.append(f"Observation: {unit.observation}")
-            scratchpad_str = "\n".join(scratchpad_parts)
-
-        # If there's a scratchpad, append it to the last message
-        if scratchpad_str:
-            messages.append(AssistantPromptMessage(content=scratchpad_str))
-
-        return messages
-
-    def _handle_chunks(
-        self,
-        chunks: Union[Generator[LLMResultChunk, None, None], LLMResult],
-        llm_usage: dict[str, Any],
-        model_log: AgentLog,
-        current_messages: list[PromptMessage],
-    ) -> Generator[
-        LLMResultChunk | AgentLog,
-        None,
-        tuple[AgentScratchpadUnit, str | None],
-    ]:
-        """Handle LLM response chunks and extract action/thought.
-
-        Returns a tuple of (scratchpad_unit, finish_reason).
-        """
-        usage_dict: dict[str, Any] = {}
-
-        # Convert non-streaming to streaming format if needed
-        if isinstance(chunks, LLMResult):
-            # Create a generator from the LLMResult
-            def result_to_chunks() -> Generator[LLMResultChunk, None, None]:
-                yield LLMResultChunk(
-                    model=chunks.model,
-                    prompt_messages=chunks.prompt_messages,
-                    delta=LLMResultChunkDelta(
-                        index=0,
-                        message=chunks.message,
-                        usage=chunks.usage,
-                        finish_reason=None,  # LLMResult doesn't have finish_reason, only streaming chunks do
-                    ),
-                    system_fingerprint=chunks.system_fingerprint or "",
-                )
-
-            streaming_chunks = result_to_chunks()
-        else:
-            streaming_chunks = chunks
-
-        react_chunks = CotAgentOutputParser.handle_react_stream_output(streaming_chunks, usage_dict)
-
-        # Initialize scratchpad unit
-        scratchpad = AgentScratchpadUnit(
-            agent_response="",
-            thought="",
-            action_str="",
-            observation="",
-            action=None,
-        )
-
-        finish_reason: str | None = None
-
-        # Process chunks
-        for chunk in react_chunks:
-            if isinstance(chunk, AgentScratchpadUnit.Action):
-                # Action detected
-                action_str = json.dumps(chunk.model_dump())
-                scratchpad.agent_response = (scratchpad.agent_response or "") + action_str
-                scratchpad.action_str = action_str
-                scratchpad.action = chunk
-
-                yield self._create_text_chunk(json.dumps(chunk.model_dump()), current_messages)
-            else:
-                # Text chunk
-                chunk_text = str(chunk)
-                scratchpad.agent_response = (scratchpad.agent_response or "") + chunk_text
-                scratchpad.thought = (scratchpad.thought or "") + chunk_text
-
-                yield self._create_text_chunk(chunk_text, current_messages)
-
-        # Update usage
-        if usage_dict.get("usage"):
-            if llm_usage.get("usage"):
-                self._accumulate_usage(llm_usage, usage_dict["usage"])
-            else:
-                llm_usage["usage"] = usage_dict["usage"]
-
-        # Clean up thought
-        scratchpad.thought = (scratchpad.thought or "").strip() or "I am thinking about how to help you"
-
-        # Finish model log
-        yield self._finish_log(
-            model_log,
-            data={
-                "thought": scratchpad.thought,
-                "action": scratchpad.action_str if scratchpad.action else None,
-            },
-            usage=llm_usage.get("usage"),
-        )
-
-        return scratchpad, finish_reason
-
-    def _handle_tool_call(
-        self,
-        action: AgentScratchpadUnit.Action,
-        prompt_messages: list[PromptMessage],
-        round_log: AgentLog,
-    ) -> Generator[AgentLog, None, tuple[str, list[File]]]:
-        """Handle tool call and return observation with files."""
-        tool_name = action.action_name
-        tool_args: dict[str, Any] | str = action.action_input
-
-        # Find tool instance first to get metadata
-        tool_instance = self._find_tool_by_name(tool_name)
-        tool_metadata = self._get_tool_metadata(tool_instance) if tool_instance else {}
-
-        # Start tool log with tool metadata
-        tool_log = self._create_log(
-            label=f"CALL {tool_name}",
-            log_type=AgentLog.LogType.TOOL_CALL,
-            status=AgentLog.LogStatus.START,
-            data={
-                "tool_name": tool_name,
-                "tool_args": tool_args,
-            },
-            parent_id=round_log.id,
-            extra_metadata=tool_metadata,
-        )
-        yield tool_log
-
-        if not tool_instance:
-            # Finish tool log with error
-            yield self._finish_log(
-                tool_log,
-                data={
-                    **tool_log.data,
-                    "error": f"Tool {tool_name} not found",
-                },
-            )
-            return f"Tool {tool_name} not found", []
-
-        # Ensure tool_args is a dict
-        tool_args_dict: dict[str, Any]
-        if isinstance(tool_args, str):
-            try:
-                tool_args_dict = json.loads(tool_args)
-            except json.JSONDecodeError:
-                tool_args_dict = {"input": tool_args}
-        elif not isinstance(tool_args, dict):
-            tool_args_dict = {"input": str(tool_args)}
-        else:
-            tool_args_dict = tool_args
-
-        # Invoke tool using base class method with error handling
-        try:
-            response_content, tool_files, tool_invoke_meta = self._invoke_tool(tool_instance, tool_args_dict, tool_name)
-
-            # Finish tool log
-            yield self._finish_log(
-                tool_log,
-                data={
-                    **tool_log.data,
-                    "output": response_content,
-                    "files": len(tool_files),
-                    "meta": tool_invoke_meta.to_dict() if tool_invoke_meta else None,
-                },
-            )
-
-            return response_content or "Tool executed successfully", tool_files
-        except Exception as e:
-            # Tool invocation failed, yield error log
-            error_message = str(e)
-            tool_log.status = AgentLog.LogStatus.ERROR
-            tool_log.error = error_message
-            tool_log.data = {
-                **tool_log.data,
-                "error": error_message,
-            }
-            yield tool_log
-
-            return f"Tool execution failed: {error_message}", []

api/core/agent/patterns/strategy_factory.py

@@ -1,107 +0,0 @@
-"""Strategy factory for creating agent strategies."""
-
-from __future__ import annotations
-
-from typing import TYPE_CHECKING
-
-from core.agent.entities import AgentEntity, ExecutionContext
-from core.file.models import File
-from core.model_manager import ModelInstance
-from core.model_runtime.entities.model_entities import ModelFeature
-
-from .base import AgentPattern, ToolInvokeHook
-from .function_call import FunctionCallStrategy
-from .react import ReActStrategy
-
-if TYPE_CHECKING:
-    from core.tools.__base.tool import Tool
-
-
-class StrategyFactory:
-    """Factory for creating agent strategies based on model features."""
-
-    # Tool calling related features
-    TOOL_CALL_FEATURES = {ModelFeature.TOOL_CALL, ModelFeature.MULTI_TOOL_CALL, ModelFeature.STREAM_TOOL_CALL}
-
-    @staticmethod
-    def create_strategy(
-        model_features: list[ModelFeature],
-        model_instance: ModelInstance,
-        context: ExecutionContext,
-        tools: list[Tool],
-        files: list[File],
-        max_iterations: int = 10,
-        workflow_call_depth: int = 0,
-        agent_strategy: AgentEntity.Strategy | None = None,
-        tool_invoke_hook: ToolInvokeHook | None = None,
-        instruction: str = "",
-    ) -> AgentPattern:
-        """
-        Create an appropriate strategy based on model features.
-
-        Args:
-            model_features: List of model features/capabilities
-            model_instance: Model instance to use
-            context: Execution context containing trace/audit information
-            tools: Available tools
-            files: Available files
-            max_iterations: Maximum iterations for the strategy
-            workflow_call_depth: Depth of workflow calls
-            agent_strategy: Optional explicit strategy override
-            tool_invoke_hook: Optional hook for custom tool invocation (e.g., agent_invoke)
-            instruction: Optional instruction for ReAct strategy
-
-        Returns:
-            AgentStrategy instance
-        """
-        # If explicit strategy is provided and it's Function Calling, try to use it if supported
-        if agent_strategy == AgentEntity.Strategy.FUNCTION_CALLING:
-            if set(model_features) & StrategyFactory.TOOL_CALL_FEATURES:
-                return FunctionCallStrategy(
-                    model_instance=model_instance,
-                    context=context,
-                    tools=tools,
-                    files=files,
-                    max_iterations=max_iterations,
-                    workflow_call_depth=workflow_call_depth,
-                    tool_invoke_hook=tool_invoke_hook,
-                )
-            # Fallback to ReAct if FC is requested but not supported
-
-        # If explicit strategy is Chain of Thought (ReAct)
-        if agent_strategy == AgentEntity.Strategy.CHAIN_OF_THOUGHT:
-            return ReActStrategy(
-                model_instance=model_instance,
-                context=context,
-                tools=tools,
-                files=files,
-                max_iterations=max_iterations,
-                workflow_call_depth=workflow_call_depth,
-                tool_invoke_hook=tool_invoke_hook,
-                instruction=instruction,
-            )
-
-        # Default auto-selection logic
-        if set(model_features) & StrategyFactory.TOOL_CALL_FEATURES:
-            # Model supports native function calling
-            return FunctionCallStrategy(
-                model_instance=model_instance,
-                context=context,
-                tools=tools,
-                files=files,
-                max_iterations=max_iterations,
-                workflow_call_depth=workflow_call_depth,
-                tool_invoke_hook=tool_invoke_hook,
-            )
-        else:
-            # Use ReAct strategy for models without function calling
-            return ReActStrategy(
-                model_instance=model_instance,
-                context=context,
-                tools=tools,
-                files=files,
-                max_iterations=max_iterations,
-                workflow_call_depth=workflow_call_depth,
-                tool_invoke_hook=tool_invoke_hook,
-                instruction=instruction,
-            )

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

@@ -29,13 +29,11 @@ from core.app.apps.message_based_app_generator import MessageBasedAppGenerator
 from core.app.apps.message_based_app_queue_manager import MessageBasedAppQueueManager
 from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, InvokeFrom
 from core.app.entities.task_entities import ChatbotAppBlockingResponse, ChatbotAppStreamResponse
-from core.app.layers.sandbox_layer import SandboxLayer
 from core.helper.trace_id_helper import extract_external_trace_id_from_args
 from core.model_runtime.errors.invoke import InvokeAuthorizationError
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.prompt.utils.get_thread_messages_length import get_thread_messages_length
 from core.repositories import DifyCoreRepositoryFactory
-from core.sandbox import Sandbox
 from core.workflow.repositories.draft_variable_repository import (
     DraftVariableSaverFactory,
 )
@@ -47,10 +45,7 @@ from factories import file_factory
 from libs.flask_utils import preserve_flask_contexts
 from models import Account, App, Conversation, EndUser, Message, Workflow, WorkflowNodeExecutionTriggeredFrom
 from models.enums import WorkflowRunTriggeredFrom
-from models.workflow_features import WorkflowFeatures
 from services.conversation_service import ConversationService
-from services.sandbox.sandbox_provider_service import SandboxProviderService
-from services.sandbox.sandbox_service import SandboxService
 from services.workflow_draft_variable_service import (
     DraftVarLoader,
     WorkflowDraftVariableService,
@@ -522,31 +517,6 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
                 if workflow is None:
                     raise ValueError("Workflow not found")
 
-                sandbox: Sandbox | None = None
-                graph_engine_layers: tuple = ()
-                if workflow.get_feature(WorkflowFeatures.SANDBOX).enabled:
-                    sandbox_provider = SandboxProviderService.get_sandbox_provider(
-                        application_generate_entity.app_config.tenant_id
-                    )
-                    if workflow.version == Workflow.VERSION_DRAFT:
-                        sandbox = SandboxService.create_draft(
-                            tenant_id=application_generate_entity.app_config.tenant_id,
-                            app_id=application_generate_entity.app_config.app_id,
-                            user_id=application_generate_entity.user_id,
-                            sandbox_provider=sandbox_provider,
-                        )
-                    else:
-                        if application_generate_entity.workflow_run_id is None:
-                            raise ValueError("workflow_run_id is required when sandbox is enabled")
-                        sandbox = SandboxService.create(
-                            tenant_id=application_generate_entity.app_config.tenant_id,
-                            app_id=application_generate_entity.app_config.app_id,
-                            user_id=application_generate_entity.user_id,
-                            workflow_execution_id=application_generate_entity.workflow_run_id,
-                            sandbox_provider=sandbox_provider,
-                        )
-                    graph_engine_layers = (SandboxLayer(sandbox=sandbox),)
-
                 # Determine system_user_id based on invocation source
                 is_external_api_call = application_generate_entity.invoke_from in {
                     InvokeFrom.WEB_APP,
@@ -577,8 +547,6 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
                 app=app,
                 workflow_execution_repository=workflow_execution_repository,
                 workflow_node_execution_repository=workflow_node_execution_repository,
-                graph_engine_layers=graph_engine_layers,
-                sandbox=sandbox,
             )
 
             try:

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

@@ -25,7 +25,6 @@ from core.app.workflow.layers.persistence import PersistenceWorkflowInfo, Workfl
 from core.db.session_factory import session_factory
 from core.moderation.base import ModerationError
 from core.moderation.input_moderation import InputModeration
-from core.sandbox import Sandbox
 from core.variables.variables import Variable
 from core.workflow.enums import WorkflowType
 from core.workflow.graph_engine.command_channels.redis_channel import RedisChannel
@@ -67,7 +66,6 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
         workflow_execution_repository: WorkflowExecutionRepository,
         workflow_node_execution_repository: WorkflowNodeExecutionRepository,
         graph_engine_layers: Sequence[GraphEngineLayer] = (),
-        sandbox: Sandbox | None = None,
     ):
         super().__init__(
             queue_manager=queue_manager,
@@ -84,7 +82,6 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
         self._app = app
         self._workflow_execution_repository = workflow_execution_repository
         self._workflow_node_execution_repository = workflow_node_execution_repository
-        self._sandbox = sandbox
 
     @trace_span(WorkflowAppRunnerHandler)
     def run(self):
@@ -159,10 +156,6 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
 
             # init graph
             graph_runtime_state = GraphRuntimeState(variable_pool=variable_pool, start_at=time.time())
-
-            if self._sandbox:
-                graph_runtime_state.set_sandbox(self._sandbox)
-
             graph = self._init_graph(
                 graph_config=self._workflow.graph_dict,
                 graph_runtime_state=graph_runtime_state,

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

@@ -82,7 +82,7 @@ class AdvancedChatAppGenerateResponseConverter(AppGenerateResponseConverter):
                 data = cls._error_to_stream_response(sub_stream_response.err)
                 response_chunk.update(data)
             else:
-                response_chunk.update(sub_stream_response.model_dump(mode="json", exclude_none=True))
+                response_chunk.update(sub_stream_response.model_dump(mode="json"))
             yield response_chunk
 
     @classmethod
@@ -110,7 +110,7 @@ class AdvancedChatAppGenerateResponseConverter(AppGenerateResponseConverter):
             }
 
             if isinstance(sub_stream_response, MessageEndStreamResponse):
-                sub_stream_response_dict = sub_stream_response.model_dump(mode="json", exclude_none=True)
+                sub_stream_response_dict = sub_stream_response.model_dump(mode="json")
                 metadata = sub_stream_response_dict.get("metadata", {})
                 sub_stream_response_dict["metadata"] = cls._get_simple_metadata(metadata)
                 response_chunk.update(sub_stream_response_dict)
@@ -120,6 +120,6 @@ class AdvancedChatAppGenerateResponseConverter(AppGenerateResponseConverter):
             elif isinstance(sub_stream_response, NodeStartStreamResponse | NodeFinishStreamResponse):
                 response_chunk.update(sub_stream_response.to_ignore_detail_dict())
             else:
-                response_chunk.update(sub_stream_response.model_dump(mode="json", exclude_none=True))
+                response_chunk.update(sub_stream_response.model_dump(mode="json"))
 
             yield response_chunk

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

@@ -4,7 +4,6 @@ import re
 import time
 from collections.abc import Callable, Generator, Mapping
 from contextlib import contextmanager
-from dataclasses import dataclass, field
 from threading import Thread
 from typing import Any, Union
 
@@ -20,7 +19,6 @@ from core.app.entities.app_invoke_entities import (
     InvokeFrom,
 )
 from core.app.entities.queue_entities import (
-    ChunkType,
     MessageQueueMessage,
     QueueAdvancedChatMessageEndEvent,
     QueueAgentLogEvent,
@@ -72,134 +70,13 @@ from core.workflow.runtime import GraphRuntimeState
 from core.workflow.system_variable import SystemVariable
 from extensions.ext_database import db
 from libs.datetime_utils import naive_utc_now
-from models import Account, Conversation, EndUser, LLMGenerationDetail, Message, MessageFile
+from models import Account, Conversation, EndUser, Message, MessageFile
 from models.enums import CreatorUserRole
 from models.workflow import Workflow
 
 logger = logging.getLogger(__name__)
 
 
-@dataclass
-class StreamEventBuffer:
-    """
-    Buffer for recording stream events in order to reconstruct the generation sequence.
-    Records the exact order of text chunks, thoughts, and tool calls as they stream.
-    """
-
-    # Accumulated reasoning content (each thought block is a separate element)
-    reasoning_content: list[str] = field(default_factory=list)
-    # Current reasoning buffer (accumulates until we see a different event type)
-    _current_reasoning: str = ""
-    # Tool calls with their details
-    tool_calls: list[dict] = field(default_factory=list)
-    # Tool call ID to index mapping for updating results
-    _tool_call_id_map: dict[str, int] = field(default_factory=dict)
-    # Sequence of events in stream order
-    sequence: list[dict] = field(default_factory=list)
-    # Current position in answer text
-    _content_position: int = 0
-    # Track last event type to detect transitions
-    _last_event_type: str | None = None
-
-    def _flush_current_reasoning(self) -> None:
-        """Flush accumulated reasoning to the list and add to sequence."""
-        if self._current_reasoning.strip():
-            self.reasoning_content.append(self._current_reasoning.strip())
-            self.sequence.append({"type": "reasoning", "index": len(self.reasoning_content) - 1})
-            self._current_reasoning = ""
-
-    def record_text_chunk(self, text: str) -> None:
-        """Record a text chunk event."""
-        if not text:
-            return
-
-        # Flush any pending reasoning first
-        if self._last_event_type == "thought":
-            self._flush_current_reasoning()
-
-        text_len = len(text)
-        start_pos = self._content_position
-
-        # If last event was also content, extend it; otherwise create new
-        if self.sequence and self.sequence[-1].get("type") == "content":
-            self.sequence[-1]["end"] = start_pos + text_len
-        else:
-            self.sequence.append({"type": "content", "start": start_pos, "end": start_pos + text_len})
-
-        self._content_position += text_len
-        self._last_event_type = "content"
-
-    def record_thought_chunk(self, text: str) -> None:
-        """Record a thought/reasoning chunk event."""
-        if not text:
-            return
-
-        # Accumulate thought content
-        self._current_reasoning += text
-        self._last_event_type = "thought"
-
-    def record_tool_call(
-        self,
-        tool_call_id: str,
-        tool_name: str,
-        tool_arguments: str,
-        tool_icon: str | dict | None = None,
-        tool_icon_dark: str | dict | None = None,
-    ) -> None:
-        """Record a tool call event."""
-        if not tool_call_id:
-            return
-
-        # Flush any pending reasoning first
-        if self._last_event_type == "thought":
-            self._flush_current_reasoning()
-
-        # Check if this tool call already exists (we might get multiple chunks)
-        if tool_call_id in self._tool_call_id_map:
-            idx = self._tool_call_id_map[tool_call_id]
-            # Update arguments if provided
-            if tool_arguments:
-                self.tool_calls[idx]["arguments"] = tool_arguments
-        else:
-            # New tool call
-            tool_call = {
-                "id": tool_call_id or "",
-                "name": tool_name or "",
-                "arguments": tool_arguments or "",
-                "result": "",
-                "elapsed_time": None,
-                "icon": tool_icon,
-                "icon_dark": tool_icon_dark,
-            }
-            self.tool_calls.append(tool_call)
-            idx = len(self.tool_calls) - 1
-            self._tool_call_id_map[tool_call_id] = idx
-            self.sequence.append({"type": "tool_call", "index": idx})
-
-        self._last_event_type = "tool_call"
-
-    def record_tool_result(self, tool_call_id: str, result: str, tool_elapsed_time: float | None = None) -> None:
-        """Record a tool result event (update existing tool call)."""
-        if not tool_call_id:
-            return
-        if tool_call_id in self._tool_call_id_map:
-            idx = self._tool_call_id_map[tool_call_id]
-            self.tool_calls[idx]["result"] = result
-            self.tool_calls[idx]["elapsed_time"] = tool_elapsed_time
-            # Remove from map after result is recorded, so that subsequent calls
-            # with the same tool_call_id are treated as new tool calls
-            del self._tool_call_id_map[tool_call_id]
-
-    def finalize(self) -> None:
-        """Finalize the buffer, flushing any pending data."""
-        if self._last_event_type == "thought":
-            self._flush_current_reasoning()
-
-    def has_data(self) -> bool:
-        """Check if there's any meaningful data recorded."""
-        return bool(self.reasoning_content or self.tool_calls or self.sequence)
-
-
 class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
     """
     AdvancedChatAppGenerateTaskPipeline is a class that generate stream output and state management for Application.
@@ -267,8 +144,6 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         self._workflow_run_id: str = ""
         self._draft_var_saver_factory = draft_var_saver_factory
         self._graph_runtime_state: GraphRuntimeState | None = None
-        # Stream event buffer for recording generation sequence
-        self._stream_buffer = StreamEventBuffer()
         self._seed_graph_runtime_state_from_queue_manager()
 
     def process(self) -> Union[ChatbotAppBlockingResponse, Generator[ChatbotAppStreamResponse, None, None]]:
@@ -508,7 +383,7 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         queue_message: Union[WorkflowQueueMessage, MessageQueueMessage] | None = None,
         **kwargs,
     ) -> Generator[StreamResponse, None, None]:
-        """Handle text chunk events and record to stream buffer for sequence reconstruction."""
+        """Handle text chunk events."""
         delta_text = event.text
         if delta_text is None:
             return
@@ -530,53 +405,9 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         if tts_publisher and queue_message:
             tts_publisher.publish(queue_message)
 
-        tool_call = event.tool_call
-        tool_result = event.tool_result
-        tool_payload = tool_call or tool_result
-        tool_call_id = tool_payload.id if tool_payload and tool_payload.id else ""
-        tool_name = tool_payload.name if tool_payload and tool_payload.name else ""
-        tool_arguments = tool_call.arguments if tool_call and tool_call.arguments else ""
-        tool_files = tool_result.files if tool_result else []
-        tool_elapsed_time = tool_result.elapsed_time if tool_result else None
-        tool_icon = tool_payload.icon if tool_payload else None
-        tool_icon_dark = tool_payload.icon_dark if tool_payload else None
-        # Record stream event based on chunk type
-        chunk_type = event.chunk_type or ChunkType.TEXT
-        match chunk_type:
-            case ChunkType.TEXT:
-                self._stream_buffer.record_text_chunk(delta_text)
-                self._task_state.answer += delta_text
-            case ChunkType.THOUGHT:
-                # Reasoning should not be part of final answer text
-                self._stream_buffer.record_thought_chunk(delta_text)
-            case ChunkType.TOOL_CALL:
-                self._stream_buffer.record_tool_call(
-                    tool_call_id=tool_call_id,
-                    tool_name=tool_name,
-                    tool_arguments=tool_arguments,
-                    tool_icon=tool_icon,
-                    tool_icon_dark=tool_icon_dark,
-                )
-            case ChunkType.TOOL_RESULT:
-                self._stream_buffer.record_tool_result(
-                    tool_call_id=tool_call_id,
-                    result=delta_text,
-                    tool_elapsed_time=tool_elapsed_time,
-                )
-            case _:
-                pass
+        self._task_state.answer += delta_text
         yield self._message_cycle_manager.message_to_stream_response(
-            answer=delta_text,
-            message_id=self._message_id,
-            from_variable_selector=event.from_variable_selector,
-            chunk_type=event.chunk_type.value if event.chunk_type else None,
-            tool_call_id=tool_call_id or None,
-            tool_name=tool_name or None,
-            tool_arguments=tool_arguments or None,
-            tool_files=tool_files,
-            tool_elapsed_time=tool_elapsed_time,
-            tool_icon=tool_icon,
-            tool_icon_dark=tool_icon_dark,
+            answer=delta_text, message_id=self._message_id, from_variable_selector=event.from_variable_selector
         )
 
     def _handle_iteration_start_event(
@@ -944,7 +775,6 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
 
         # If there are assistant files, remove markdown image links from answer
         answer_text = self._task_state.answer
-        answer_text = self._strip_think_blocks(answer_text)
         if self._recorded_files:
             # Remove markdown image links since we're storing files separately
             answer_text = re.sub(r"!\[.*?\]\(.*?\)", "", answer_text).strip()
@@ -996,54 +826,6 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         ]
         session.add_all(message_files)
 
-        # Save generation detail (reasoning/tool calls/sequence) from stream buffer
-        self._save_generation_detail(session=session, message=message)
-
-    @staticmethod
-    def _strip_think_blocks(text: str) -> str:
-        """Remove <think>...</think> blocks (including their content) from text."""
-        if not text or "<think" not in text.lower():
-            return text
-
-        clean_text = re.sub(r"<think[^>]*>.*?</think>", "", text, flags=re.IGNORECASE | re.DOTALL)
-        clean_text = re.sub(r"\n\s*\n", "\n\n", clean_text).strip()
-        return clean_text
-
-    def _save_generation_detail(self, *, session: Session, message: Message) -> None:
-        """
-        Save LLM generation detail for Chatflow using stream event buffer.
-        The buffer records the exact order of events as they streamed,
-        allowing accurate reconstruction of the generation sequence.
-        """
-        # Finalize the stream buffer to flush any pending data
-        self._stream_buffer.finalize()
-
-        # Only save if there's meaningful data
-        if not self._stream_buffer.has_data():
-            return
-
-        reasoning_content = self._stream_buffer.reasoning_content
-        tool_calls = self._stream_buffer.tool_calls
-        sequence = self._stream_buffer.sequence
-
-        # Check if generation detail already exists for this message
-        existing = session.query(LLMGenerationDetail).filter_by(message_id=message.id).first()
-
-        if existing:
-            existing.reasoning_content = json.dumps(reasoning_content) if reasoning_content else None
-            existing.tool_calls = json.dumps(tool_calls) if tool_calls else None
-            existing.sequence = json.dumps(sequence) if sequence else None
-        else:
-            generation_detail = LLMGenerationDetail(
-                tenant_id=self._application_generate_entity.app_config.tenant_id,
-                app_id=self._application_generate_entity.app_config.app_id,
-                message_id=message.id,
-                reasoning_content=json.dumps(reasoning_content) if reasoning_content else None,
-                tool_calls=json.dumps(tool_calls) if tool_calls else None,
-                sequence=json.dumps(sequence) if sequence else None,
-            )
-            session.add(generation_detail)
-
     def _seed_graph_runtime_state_from_queue_manager(self) -> None:
         """Bootstrap the cached runtime state from the queue manager when present."""
         candidate = self._base_task_pipeline.queue_manager.graph_runtime_state

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

@@ -3,8 +3,10 @@ from typing import cast
 
 from sqlalchemy import select
 
-from core.agent.agent_app_runner import AgentAppRunner
+from core.agent.cot_chat_agent_runner import CotChatAgentRunner
+from core.agent.cot_completion_agent_runner import CotCompletionAgentRunner
 from core.agent.entities import AgentEntity
+from core.agent.fc_agent_runner import FunctionCallAgentRunner
 from core.app.apps.agent_chat.app_config_manager import AgentChatAppConfig
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
 from core.app.apps.base_app_runner import AppRunner
@@ -12,7 +14,8 @@ from core.app.entities.app_invoke_entities import AgentChatAppGenerateEntity
 from core.app.entities.queue_entities import QueueAnnotationReplyEvent
 from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance
-from core.model_runtime.entities.model_entities import ModelFeature
+from core.model_runtime.entities.llm_entities import LLMMode
+from core.model_runtime.entities.model_entities import ModelFeature, ModelPropertyKey
 from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
 from core.moderation.base import ModerationError
 from extensions.ext_database import db
@@ -191,7 +194,22 @@ class AgentChatAppRunner(AppRunner):
             raise ValueError("Message not found")
         db.session.close()
 
-        runner = AgentAppRunner(
+        runner_cls: type[FunctionCallAgentRunner] | type[CotChatAgentRunner] | type[CotCompletionAgentRunner]
+        # start agent runner
+        if agent_entity.strategy == AgentEntity.Strategy.CHAIN_OF_THOUGHT:
+            # check LLM mode
+            if model_schema.model_properties.get(ModelPropertyKey.MODE) == LLMMode.CHAT:
+                runner_cls = CotChatAgentRunner
+            elif model_schema.model_properties.get(ModelPropertyKey.MODE) == LLMMode.COMPLETION:
+                runner_cls = CotCompletionAgentRunner
+            else:
+                raise ValueError(f"Invalid LLM mode: {model_schema.model_properties.get(ModelPropertyKey.MODE)}")
+        elif agent_entity.strategy == AgentEntity.Strategy.FUNCTION_CALLING:
+            runner_cls = FunctionCallAgentRunner
+        else:
+            raise ValueError(f"Invalid agent strategy: {agent_entity.strategy}")
+
+        runner = runner_cls(
             tenant_id=app_config.tenant_id,
             application_generate_entity=application_generate_entity,
             conversation=conversation_result,

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

@@ -81,7 +81,7 @@ class AgentChatAppGenerateResponseConverter(AppGenerateResponseConverter):
                 data = cls._error_to_stream_response(sub_stream_response.err)
                 response_chunk.update(data)
             else:
-                response_chunk.update(sub_stream_response.model_dump(mode="json", exclude_none=True))
+                response_chunk.update(sub_stream_response.model_dump(mode="json"))
             yield response_chunk
 
     @classmethod
@@ -109,7 +109,7 @@ class AgentChatAppGenerateResponseConverter(AppGenerateResponseConverter):
             }
 
             if isinstance(sub_stream_response, MessageEndStreamResponse):
-                sub_stream_response_dict = sub_stream_response.model_dump(mode="json", exclude_none=True)
+                sub_stream_response_dict = sub_stream_response.model_dump(mode="json")
                 metadata = sub_stream_response_dict.get("metadata", {})
                 sub_stream_response_dict["metadata"] = cls._get_simple_metadata(metadata)
                 response_chunk.update(sub_stream_response_dict)
@@ -117,6 +117,6 @@ class AgentChatAppGenerateResponseConverter(AppGenerateResponseConverter):
                 data = cls._error_to_stream_response(sub_stream_response.err)
                 response_chunk.update(data)
             else:
-                response_chunk.update(sub_stream_response.model_dump(mode="json", exclude_none=True))
+                response_chunk.update(sub_stream_response.model_dump(mode="json"))
 
             yield response_chunk

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

@@ -81,7 +81,7 @@ class ChatAppGenerateResponseConverter(AppGenerateResponseConverter):
                 data = cls._error_to_stream_response(sub_stream_response.err)
                 response_chunk.update(data)
             else:
-                response_chunk.update(sub_stream_response.model_dump(mode="json", exclude_none=True))
+                response_chunk.update(sub_stream_response.model_dump(mode="json"))
             yield response_chunk
 
     @classmethod
@@ -109,7 +109,7 @@ class ChatAppGenerateResponseConverter(AppGenerateResponseConverter):
             }
 
             if isinstance(sub_stream_response, MessageEndStreamResponse):
-                sub_stream_response_dict = sub_stream_response.model_dump(mode="json", exclude_none=True)
+                sub_stream_response_dict = sub_stream_response.model_dump(mode="json")
                 metadata = sub_stream_response_dict.get("metadata", {})
                 sub_stream_response_dict["metadata"] = cls._get_simple_metadata(metadata)
                 response_chunk.update(sub_stream_response_dict)
@@ -117,6 +117,6 @@ class ChatAppGenerateResponseConverter(AppGenerateResponseConverter):
                 data = cls._error_to_stream_response(sub_stream_response.err)
                 response_chunk.update(data)
             else:
-                response_chunk.update(sub_stream_response.model_dump(mode="json", exclude_none=True))
+                response_chunk.update(sub_stream_response.model_dump(mode="json"))
 
             yield response_chunk

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

@@ -70,8 +70,6 @@ class _NodeSnapshot:
     """Empty string means the node is not executing inside an iteration."""
     loop_id: str = ""
     """Empty string means the node is not executing inside a loop."""
-    parent_node_id: str = ""
-    """Empty string means the node is not an nested node (extractor node)."""
 
 
 class WorkflowResponseConverter:
@@ -133,7 +131,6 @@ class WorkflowResponseConverter:
             start_at=event.start_at,
             iteration_id=event.in_iteration_id or "",
             loop_id=event.in_loop_id or "",
-            parent_node_id=event.in_parent_node_id or "",
         )
         node_execution_id = NodeExecutionId(event.node_execution_id)
         self._node_snapshots[node_execution_id] = snapshot
@@ -290,7 +287,6 @@ class WorkflowResponseConverter:
                 created_at=int(snapshot.start_at.timestamp()),
                 iteration_id=event.in_iteration_id,
                 loop_id=event.in_loop_id,
-                parent_node_id=event.in_parent_node_id,
                 agent_strategy=event.agent_strategy,
             ),
         )
@@ -377,7 +373,6 @@ class WorkflowResponseConverter:
                 files=self.fetch_files_from_node_outputs(event.outputs or {}),
                 iteration_id=event.in_iteration_id,
                 loop_id=event.in_loop_id,
-                parent_node_id=event.in_parent_node_id,
             ),
         )
 
@@ -427,7 +422,6 @@ class WorkflowResponseConverter:
                 files=self.fetch_files_from_node_outputs(event.outputs or {}),
                 iteration_id=event.in_iteration_id,
                 loop_id=event.in_loop_id,
-                parent_node_id=event.in_parent_node_id,
                 retry_index=event.retry_index,
             ),
         )
@@ -677,7 +671,7 @@ class WorkflowResponseConverter:
             task_id=task_id,
             data=AgentLogStreamResponse.Data(
                 node_execution_id=event.node_execution_id,
-                message_id=event.id,
+                id=event.id,
                 parent_id=event.parent_id,
                 label=event.label,
                 error=event.error,

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

@@ -79,7 +79,7 @@ class CompletionAppGenerateResponseConverter(AppGenerateResponseConverter):
                 data = cls._error_to_stream_response(sub_stream_response.err)
                 response_chunk.update(data)
             else:
-                response_chunk.update(sub_stream_response.model_dump(mode="json", exclude_none=True))
+                response_chunk.update(sub_stream_response.model_dump(mode="json"))
             yield response_chunk
 
     @classmethod
@@ -106,7 +106,7 @@ class CompletionAppGenerateResponseConverter(AppGenerateResponseConverter):
             }
 
             if isinstance(sub_stream_response, MessageEndStreamResponse):
-                sub_stream_response_dict = sub_stream_response.model_dump(mode="json", exclude_none=True)
+                sub_stream_response_dict = sub_stream_response.model_dump(mode="json")
                 metadata = sub_stream_response_dict.get("metadata", {})
                 if not isinstance(metadata, dict):
                     metadata = {}
@@ -116,6 +116,6 @@ class CompletionAppGenerateResponseConverter(AppGenerateResponseConverter):
                 data = cls._error_to_stream_response(sub_stream_response.err)
                 response_chunk.update(data)
             else:
-                response_chunk.update(sub_stream_response.model_dump(mode="json", exclude_none=True))
+                response_chunk.update(sub_stream_response.model_dump(mode="json"))
 
             yield response_chunk

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

@@ -60,7 +60,7 @@ class WorkflowAppGenerateResponseConverter(AppGenerateResponseConverter):
                 data = cls._error_to_stream_response(sub_stream_response.err)
                 response_chunk.update(cast(dict, data))
             else:
-                response_chunk.update(sub_stream_response.model_dump(exclude_none=True))
+                response_chunk.update(sub_stream_response.model_dump())
             yield response_chunk
 
     @classmethod
@@ -91,5 +91,5 @@ class WorkflowAppGenerateResponseConverter(AppGenerateResponseConverter):
             elif isinstance(sub_stream_response, NodeStartStreamResponse | NodeFinishStreamResponse):
                 response_chunk.update(cast(dict, sub_stream_response.to_ignore_detail_dict()))
             else:
-                response_chunk.update(sub_stream_response.model_dump(exclude_none=True))
+                response_chunk.update(sub_stream_response.model_dump())
             yield response_chunk

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

@@ -25,13 +25,11 @@ from core.app.apps.workflow.generate_response_converter import WorkflowAppGenera
 from core.app.apps.workflow.generate_task_pipeline import WorkflowAppGenerateTaskPipeline
 from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity
 from core.app.entities.task_entities import WorkflowAppBlockingResponse, WorkflowAppStreamResponse
-from core.app.layers.sandbox_layer import SandboxLayer
 from core.db.session_factory import session_factory
 from core.helper.trace_id_helper import extract_external_trace_id_from_args
 from core.model_runtime.errors.invoke import InvokeAuthorizationError
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.repositories import DifyCoreRepositoryFactory
-from core.sandbox import Sandbox
 from core.workflow.graph_engine.layers.base import GraphEngineLayer
 from core.workflow.repositories.draft_variable_repository import DraftVariableSaverFactory
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
@@ -42,9 +40,6 @@ from factories import file_factory
 from libs.flask_utils import preserve_flask_contexts
 from models import Account, App, EndUser, Workflow, WorkflowNodeExecutionTriggeredFrom
 from models.enums import WorkflowRunTriggeredFrom
-from models.workflow_features import WorkflowFeatures
-from services.sandbox.sandbox_provider_service import SandboxProviderService
-from services.sandbox.sandbox_service import SandboxService
 from services.workflow_draft_variable_service import DraftVarLoader, WorkflowDraftVariableService
 
 if TYPE_CHECKING:
@@ -498,31 +493,6 @@ class WorkflowAppGenerator(BaseAppGenerator):
                 if workflow is None:
                     raise ValueError("Workflow not found")
 
-                sandbox: Sandbox | None = None
-                if workflow.get_feature(WorkflowFeatures.SANDBOX).enabled:
-                    sandbox_provider = SandboxProviderService.get_sandbox_provider(
-                        application_generate_entity.app_config.tenant_id
-                    )
-                    if workflow.version == Workflow.VERSION_DRAFT:
-                        sandbox = SandboxService.create_draft(
-                            tenant_id=application_generate_entity.app_config.tenant_id,
-                            app_id=application_generate_entity.app_config.app_id,
-                            user_id=application_generate_entity.user_id,
-                            sandbox_provider=sandbox_provider,
-                        )
-                    else:
-                        sandbox = SandboxService.create(
-                            tenant_id=application_generate_entity.app_config.tenant_id,
-                            app_id=application_generate_entity.app_config.app_id,
-                            user_id=application_generate_entity.user_id,
-                            workflow_execution_id=application_generate_entity.workflow_execution_id,
-                            sandbox_provider=sandbox_provider,
-                        )
-                    graph_engine_layers = (
-                        *graph_engine_layers,
-                        SandboxLayer(sandbox=sandbox),
-                    )
-
                 # Determine system_user_id based on invocation source
                 is_external_api_call = application_generate_entity.invoke_from in {
                     InvokeFrom.WEB_APP,
@@ -547,7 +517,6 @@ class WorkflowAppGenerator(BaseAppGenerator):
                 workflow_node_execution_repository=workflow_node_execution_repository,
                 root_node_id=root_node_id,
                 graph_engine_layers=graph_engine_layers,
-                sandbox=sandbox,
             )
 
             try:

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

@@ -8,7 +8,6 @@ from core.app.apps.workflow.app_config_manager import WorkflowAppConfig
 from core.app.apps.workflow_app_runner import WorkflowBasedAppRunner
 from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity
 from core.app.workflow.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
-from core.sandbox import Sandbox
 from core.workflow.enums import WorkflowType
 from core.workflow.graph_engine.command_channels.redis_channel import RedisChannel
 from core.workflow.graph_engine.layers.base import GraphEngineLayer
@@ -43,7 +42,6 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
         workflow_execution_repository: WorkflowExecutionRepository,
         workflow_node_execution_repository: WorkflowNodeExecutionRepository,
         graph_engine_layers: Sequence[GraphEngineLayer] = (),
-        sandbox: Sandbox | None = None,
     ):
         super().__init__(
             queue_manager=queue_manager,
@@ -57,7 +55,6 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
         self._root_node_id = root_node_id
         self._workflow_execution_repository = workflow_execution_repository
         self._workflow_node_execution_repository = workflow_node_execution_repository
-        self._sandbox = sandbox
 
     @trace_span(WorkflowAppRunnerHandler)
     def run(self):
@@ -102,9 +99,6 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
 
             graph_runtime_state = GraphRuntimeState(variable_pool=variable_pool, start_at=time.perf_counter())
 
-            if self._sandbox:
-                graph_runtime_state.set_sandbox(self._sandbox)
-
             # init graph
             graph = self._init_graph(
                 graph_config=self._workflow.graph_dict,

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

@@ -60,7 +60,7 @@ class WorkflowAppGenerateResponseConverter(AppGenerateResponseConverter):
                 data = cls._error_to_stream_response(sub_stream_response.err)
                 response_chunk.update(data)
             else:
-                response_chunk.update(sub_stream_response.model_dump(mode="json", exclude_none=True))
+                response_chunk.update(sub_stream_response.model_dump(mode="json"))
             yield response_chunk
 
     @classmethod
@@ -91,5 +91,5 @@ class WorkflowAppGenerateResponseConverter(AppGenerateResponseConverter):
             elif isinstance(sub_stream_response, NodeStartStreamResponse | NodeFinishStreamResponse):
                 response_chunk.update(sub_stream_response.to_ignore_detail_dict())
             else:
-                response_chunk.update(sub_stream_response.model_dump(mode="json", exclude_none=True))
+                response_chunk.update(sub_stream_response.model_dump(mode="json"))
             yield response_chunk

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

@@ -13,7 +13,6 @@ from core.app.apps.common.workflow_response_converter import WorkflowResponseCon
 from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity
 from core.app.entities.queue_entities import (
     AppQueueEvent,
-    ChunkType,
     MessageQueueMessage,
     QueueAgentLogEvent,
     QueueErrorEvent,
@@ -484,33 +483,11 @@ class WorkflowAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         if delta_text is None:
             return
 
-        tool_call = event.tool_call
-        tool_result = event.tool_result
-        tool_payload = tool_call or tool_result
-        tool_call_id = tool_payload.id if tool_payload and tool_payload.id else None
-        tool_name = tool_payload.name if tool_payload and tool_payload.name else None
-        tool_arguments = tool_call.arguments if tool_call else None
-        tool_elapsed_time = tool_result.elapsed_time if tool_result else None
-        tool_files = tool_result.files if tool_result else []
-        tool_icon = tool_payload.icon if tool_payload else None
-        tool_icon_dark = tool_payload.icon_dark if tool_payload else None
-
         # only publish tts message at text chunk streaming
         if tts_publisher and queue_message:
             tts_publisher.publish(queue_message)
 
-        yield self._text_chunk_to_stream_response(
-            text=delta_text,
-            from_variable_selector=event.from_variable_selector,
-            chunk_type=event.chunk_type,
-            tool_call_id=tool_call_id,
-            tool_name=tool_name,
-            tool_arguments=tool_arguments,
-            tool_files=tool_files,
-            tool_elapsed_time=tool_elapsed_time,
-            tool_icon=tool_icon,
-            tool_icon_dark=tool_icon_dark,
-        )
+        yield self._text_chunk_to_stream_response(delta_text, from_variable_selector=event.from_variable_selector)
 
     def _handle_agent_log_event(self, event: QueueAgentLogEvent, **kwargs) -> Generator[StreamResponse, None, None]:
         """Handle agent log events."""
@@ -673,61 +650,16 @@ class WorkflowAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         session.add(workflow_app_log)
 
     def _text_chunk_to_stream_response(
-        self,
-        text: str,
-        from_variable_selector: list[str] | None = None,
-        chunk_type: ChunkType | None = None,
-        tool_call_id: str | None = None,
-        tool_name: str | None = None,
-        tool_arguments: str | None = None,
-        tool_files: list[str] | None = None,
-        tool_error: str | None = None,
-        tool_elapsed_time: float | None = None,
-        tool_icon: str | dict | None = None,
-        tool_icon_dark: str | dict | None = None,
+        self, text: str, from_variable_selector: list[str] | None = None
     ) -> TextChunkStreamResponse:
         """
         Handle completed event.
         :param text: text
         :return:
         """
-        from core.app.entities.task_entities import ChunkType as ResponseChunkType
-
-        response_chunk_type = ResponseChunkType(chunk_type.value) if chunk_type else ResponseChunkType.TEXT
-
-        data = TextChunkStreamResponse.Data(
-            text=text,
-            from_variable_selector=from_variable_selector,
-            chunk_type=response_chunk_type,
-        )
-
-        if response_chunk_type == ResponseChunkType.TOOL_CALL:
-            data = data.model_copy(
-                update={
-                    "tool_call_id": tool_call_id,
-                    "tool_name": tool_name,
-                    "tool_arguments": tool_arguments,
-                    "tool_icon": tool_icon,
-                    "tool_icon_dark": tool_icon_dark,
-                }
-            )
-        elif response_chunk_type == ResponseChunkType.TOOL_RESULT:
-            data = data.model_copy(
-                update={
-                    "tool_call_id": tool_call_id,
-                    "tool_name": tool_name,
-                    "tool_arguments": tool_arguments,
-                    "tool_files": tool_files,
-                    "tool_error": tool_error,
-                    "tool_elapsed_time": tool_elapsed_time,
-                    "tool_icon": tool_icon,
-                    "tool_icon_dark": tool_icon_dark,
-                }
-            )
-
         response = TextChunkStreamResponse(
             task_id=self._application_generate_entity.task_id,
-            data=data,
+            data=TextChunkStreamResponse.Data(text=text, from_variable_selector=from_variable_selector),
         )
 
         return response

api/core/app/apps/workflow_app_runner.py

@@ -353,7 +353,6 @@ class WorkflowBasedAppRunner:
                     start_at=event.start_at,
                     in_iteration_id=event.in_iteration_id,
                     in_loop_id=event.in_loop_id,
-                    in_parent_node_id=event.in_parent_node_id,
                     inputs=inputs,
                     process_data=process_data,
                     outputs=outputs,
@@ -374,7 +373,6 @@ class WorkflowBasedAppRunner:
                     start_at=event.start_at,
                     in_iteration_id=event.in_iteration_id,
                     in_loop_id=event.in_loop_id,
-                    in_parent_node_id=event.in_parent_node_id,
                     agent_strategy=event.agent_strategy,
                     provider_type=event.provider_type,
                     provider_id=event.provider_id,
@@ -398,7 +396,6 @@ class WorkflowBasedAppRunner:
                     execution_metadata=execution_metadata,
                     in_iteration_id=event.in_iteration_id,
                     in_loop_id=event.in_loop_id,
-                    in_parent_node_id=event.in_parent_node_id,
                 )
             )
         elif isinstance(event, NodeRunFailedEvent):
@@ -415,7 +412,6 @@ class WorkflowBasedAppRunner:
                     execution_metadata=event.node_run_result.metadata,
                     in_iteration_id=event.in_iteration_id,
                     in_loop_id=event.in_loop_id,
-                    in_parent_node_id=event.in_parent_node_id,
                 )
             )
         elif isinstance(event, NodeRunExceptionEvent):
@@ -432,25 +428,15 @@ class WorkflowBasedAppRunner:
                     execution_metadata=event.node_run_result.metadata,
                     in_iteration_id=event.in_iteration_id,
                     in_loop_id=event.in_loop_id,
-                    in_parent_node_id=event.in_parent_node_id,
                 )
             )
         elif isinstance(event, NodeRunStreamChunkEvent):
-            from core.app.entities.queue_entities import ChunkType as QueueChunkType
-
-            if event.is_final and not event.chunk:
-                return
-
             self._publish_event(
                 QueueTextChunkEvent(
                     text=event.chunk,
                     from_variable_selector=list(event.selector),
                     in_iteration_id=event.in_iteration_id,
                     in_loop_id=event.in_loop_id,
-                    chunk_type=QueueChunkType(event.chunk_type.value),
-                    tool_call=event.tool_call,
-                    tool_result=event.tool_result,
-                    in_parent_node_id=event.in_parent_node_id,
                 )
             )
         elif isinstance(event, NodeRunRetrieverResourceEvent):
@@ -459,7 +445,6 @@ class WorkflowBasedAppRunner:
                     retriever_resources=event.retriever_resources,
                     in_iteration_id=event.in_iteration_id,
                     in_loop_id=event.in_loop_id,
-                    in_parent_node_id=event.in_parent_node_id,
                 )
             )
         elif isinstance(event, NodeRunAgentLogEvent):

api/core/app/entities/app_asset_entities.py

@@ -1,317 +0,0 @@
-from __future__ import annotations
-
-from collections import defaultdict
-from collections.abc import Generator
-from enum import StrEnum
-
-from pydantic import BaseModel, Field
-
-
-class AssetNodeType(StrEnum):
-    FILE = "file"
-    FOLDER = "folder"
-
-
-class AppAssetNode(BaseModel):
-    id: str = Field(description="Unique identifier for the node")
-    node_type: AssetNodeType = Field(description="Type of node: file or folder")
-    name: str = Field(description="Name of the file or folder")
-    parent_id: str | None = Field(default=None, description="Parent folder ID, None for root level")
-    order: int = Field(default=0, description="Sort order within parent folder, lower values first")
-    extension: str = Field(default="", description="File extension without dot, empty for folders")
-    size: int = Field(default=0, description="File size in bytes, 0 for folders")
-
-    @classmethod
-    def create_folder(cls, node_id: str, name: str, parent_id: str | None = None) -> AppAssetNode:
-        return cls(id=node_id, node_type=AssetNodeType.FOLDER, name=name, parent_id=parent_id)
-
-    @classmethod
-    def create_file(cls, node_id: str, name: str, parent_id: str | None = None, size: int = 0) -> AppAssetNode:
-        return cls(
-            id=node_id,
-            node_type=AssetNodeType.FILE,
-            name=name,
-            parent_id=parent_id,
-            extension=name.rsplit(".", 1)[-1] if "." in name else "",
-            size=size,
-        )
-
-
-class AppAssetNodeView(BaseModel):
-    id: str = Field(description="Unique identifier for the node")
-    node_type: str = Field(description="Type of node: 'file' or 'folder'")
-    name: str = Field(description="Name of the file or folder")
-    path: str = Field(description="Full path from root, e.g. '/folder/file.txt'")
-    extension: str = Field(default="", description="File extension without dot")
-    size: int = Field(default=0, description="File size in bytes")
-    children: list[AppAssetNodeView] = Field(default_factory=list, description="Child nodes for folders")
-
-
-class BatchUploadNode(BaseModel):
-    """Structure for batch upload_url tree nodes, used for both input and output."""
-
-    name: str
-    node_type: AssetNodeType
-    size: int = 0
-    children: list[BatchUploadNode] = []
-    id: str | None = None
-    upload_url: str | None = None
-
-    def to_app_asset_nodes(self, parent_id: str | None = None) -> list[AppAssetNode]:
-        """
-        Generate IDs and convert to AppAssetNode list.
-        Mutates self to set id field.
-        """
-        from uuid import uuid4
-
-        self.id = str(uuid4())
-        nodes: list[AppAssetNode] = []
-
-        if self.node_type == AssetNodeType.FOLDER:
-            nodes.append(AppAssetNode.create_folder(self.id, self.name, parent_id))
-            for child in self.children:
-                nodes.extend(child.to_app_asset_nodes(self.id))
-        else:
-            nodes.append(AppAssetNode.create_file(self.id, self.name, parent_id, self.size))
-
-        return nodes
-
-
-class TreeNodeNotFoundError(Exception):
-    """Tree internal: node not found"""
-
-    pass
-
-
-class TreeParentNotFoundError(Exception):
-    """Tree internal: parent folder not found"""
-
-    pass
-
-
-class TreePathConflictError(Exception):
-    """Tree internal: path already exists"""
-
-    pass
-
-
-class AppAssetFileTree(BaseModel):
-    """
-    File tree structure for app assets using adjacency list pattern.
-
-    Design:
-    - Storage: Flat list with parent_id references (adjacency list)
-    - Path: Computed dynamically via get_path(), not stored
-    - Order: Integer field for user-defined sorting within each folder
-    - API response: transform() builds nested tree with computed paths
-
-    Why adjacency list over nested tree or materialized path:
-    - Simpler CRUD: move/rename only updates one node's parent_id
-    - No path cascade: renaming parent doesn't require updating all descendants
-    - JSON-friendly: flat list serializes cleanly to database JSON column
-    - Trade-off: path lookup is O(depth), acceptable for typical file trees
-    """
-
-    nodes: list[AppAssetNode] = Field(default_factory=list, description="Flat list of all nodes in the tree")
-
-    def get(self, node_id: str) -> AppAssetNode | None:
-        return next((n for n in self.nodes if n.id == node_id), None)
-
-    def get_children(self, parent_id: str | None) -> list[AppAssetNode]:
-        return [n for n in self.nodes if n.parent_id == parent_id]
-
-    def has_child_named(self, parent_id: str | None, name: str) -> bool:
-        return any(n.name == name and n.parent_id == parent_id for n in self.nodes)
-
-    def get_path(self, node_id: str) -> str:
-        node = self.get(node_id)
-        if not node:
-            raise TreeNodeNotFoundError(node_id)
-        parts: list[str] = []
-        current: AppAssetNode | None = node
-        while current:
-            parts.append(current.name)
-            current = self.get(current.parent_id) if current.parent_id else None
-        return "/".join(reversed(parts))
-
-    def relative_path(self, a: AppAssetNode, b: AppAssetNode) -> str:
-        """
-        Calculate relative path from node a to node b for Markdown references.
-        Path is computed from a's parent directory (where the file resides).
-
-        Examples:
-            /foo/a.md -> /foo/b.md         => ./b.md
-            /foo/a.md -> /foo/sub/b.md     => ./sub/b.md
-            /foo/sub/a.md -> /foo/b.md     => ../b.md
-            /foo/sub/deep/a.md -> /foo/b.md => ../../b.md
-        """
-
-        def get_ancestor_ids(node_id: str | None) -> list[str]:
-            chain: list[str] = []
-            current_id = node_id
-            while current_id:
-                chain.append(current_id)
-                node = self.get(current_id)
-                current_id = node.parent_id if node else None
-            return chain
-
-        a_dir_ancestors = get_ancestor_ids(a.parent_id)
-        b_ancestors = [b.id] + get_ancestor_ids(b.parent_id)
-        a_dir_set = set(a_dir_ancestors)
-
-        lca_id: str | None = None
-        lca_index_in_b = -1
-        for idx, ancestor_id in enumerate(b_ancestors):
-            if ancestor_id in a_dir_set or (a.parent_id is None and b_ancestors[idx:] == []):
-                lca_id = ancestor_id
-                lca_index_in_b = idx
-                break
-
-        if a.parent_id is None:
-            steps_up = 0
-            lca_index_in_b = len(b_ancestors)
-        elif lca_id is None:
-            steps_up = len(a_dir_ancestors)
-            lca_index_in_b = len(b_ancestors)
-        else:
-            steps_up = 0
-            for ancestor_id in a_dir_ancestors:
-                if ancestor_id == lca_id:
-                    break
-                steps_up += 1
-
-        path_down: list[str] = []
-        for i in range(lca_index_in_b - 1, -1, -1):
-            node = self.get(b_ancestors[i])
-            if node:
-                path_down.append(node.name)
-
-        if steps_up == 0:
-            return "./" + "/".join(path_down)
-
-        parts: list[str] = [".."] * steps_up + path_down
-        return "/".join(parts)
-
-    def get_descendant_ids(self, node_id: str) -> list[str]:
-        result: list[str] = []
-        stack = [node_id]
-        while stack:
-            current_id = stack.pop()
-            for child in self.nodes:
-                if child.parent_id == current_id:
-                    result.append(child.id)
-                    stack.append(child.id)
-        return result
-
-    def add(self, node: AppAssetNode) -> AppAssetNode:
-        if self.get(node.id):
-            raise TreePathConflictError(node.id)
-        if self.has_child_named(node.parent_id, node.name):
-            raise TreePathConflictError(node.name)
-        if node.parent_id:
-            parent = self.get(node.parent_id)
-            if not parent or parent.node_type != AssetNodeType.FOLDER:
-                raise TreeParentNotFoundError(node.parent_id)
-        siblings = self.get_children(node.parent_id)
-        node.order = max((s.order for s in siblings), default=-1) + 1
-        self.nodes.append(node)
-        return node
-
-    def update(self, node_id: str, size: int) -> AppAssetNode:
-        node = self.get(node_id)
-        if not node or node.node_type != AssetNodeType.FILE:
-            raise TreeNodeNotFoundError(node_id)
-        node.size = size
-        return node
-
-    def rename(self, node_id: str, new_name: str) -> AppAssetNode:
-        node = self.get(node_id)
-        if not node:
-            raise TreeNodeNotFoundError(node_id)
-        if node.name != new_name and self.has_child_named(node.parent_id, new_name):
-            raise TreePathConflictError(new_name)
-        node.name = new_name
-        if node.node_type == AssetNodeType.FILE:
-            node.extension = new_name.rsplit(".", 1)[-1] if "." in new_name else ""
-        return node
-
-    def move(self, node_id: str, new_parent_id: str | None) -> AppAssetNode:
-        node = self.get(node_id)
-        if not node:
-            raise TreeNodeNotFoundError(node_id)
-        if new_parent_id:
-            parent = self.get(new_parent_id)
-            if not parent or parent.node_type != AssetNodeType.FOLDER:
-                raise TreeParentNotFoundError(new_parent_id)
-        if self.has_child_named(new_parent_id, node.name):
-            raise TreePathConflictError(node.name)
-        node.parent_id = new_parent_id
-        siblings = self.get_children(new_parent_id)
-        node.order = max((s.order for s in siblings if s.id != node_id), default=-1) + 1
-        return node
-
-    def reorder(self, node_id: str, after_node_id: str | None) -> AppAssetNode:
-        node = self.get(node_id)
-        if not node:
-            raise TreeNodeNotFoundError(node_id)
-
-        siblings = sorted(self.get_children(node.parent_id), key=lambda x: x.order)
-        siblings = [s for s in siblings if s.id != node_id]
-
-        if after_node_id is None:
-            insert_idx = 0
-        else:
-            after_node = self.get(after_node_id)
-            if not after_node or after_node.parent_id != node.parent_id:
-                raise TreeNodeNotFoundError(after_node_id)
-            insert_idx = next((i for i, s in enumerate(siblings) if s.id == after_node_id), -1) + 1
-
-        siblings.insert(insert_idx, node)
-        for idx, sibling in enumerate(siblings):
-            sibling.order = idx
-
-        return node
-
-    def remove(self, node_id: str) -> list[str]:
-        node = self.get(node_id)
-        if not node:
-            raise TreeNodeNotFoundError(node_id)
-        ids_to_remove = [node_id] + self.get_descendant_ids(node_id)
-        self.nodes = [n for n in self.nodes if n.id not in ids_to_remove]
-        return ids_to_remove
-
-    def walk_files(self) -> Generator[AppAssetNode, None, None]:
-        return (n for n in self.nodes if n.node_type == AssetNodeType.FILE)
-
-    def transform(self) -> list[AppAssetNodeView]:
-        by_parent: dict[str | None, list[AppAssetNode]] = defaultdict(list)
-        for n in self.nodes:
-            by_parent[n.parent_id].append(n)
-
-        for children in by_parent.values():
-            children.sort(key=lambda x: x.order)
-
-        paths: dict[str, str] = {}
-        tree_views: dict[str, AppAssetNodeView] = {}
-
-        def build_view(node: AppAssetNode, parent_path: str) -> None:
-            path = f"{parent_path}/{node.name}"
-            paths[node.id] = path
-            child_views: list[AppAssetNodeView] = []
-            for child in by_parent.get(node.id, []):
-                build_view(child, path)
-                child_views.append(tree_views[child.id])
-            tree_views[node.id] = AppAssetNodeView(
-                id=node.id,
-                node_type=node.node_type.value,
-                name=node.name,
-                path=path,
-                extension=node.extension,
-                size=node.size,
-                children=child_views,
-            )
-
-        for root_node in by_parent.get(None, []):
-            build_view(root_node, "")
-
-        return [tree_views[n.id] for n in by_parent.get(None, [])]

api/core/app/entities/app_bundle_entities.py

@@ -1,96 +0,0 @@
-from __future__ import annotations
-
-import re
-from datetime import UTC, datetime
-
-from pydantic import BaseModel, ConfigDict, Field
-
-from core.app.entities.app_asset_entities import AppAssetFileTree
-
-# Constants
-BUNDLE_DSL_FILENAME_PATTERN = re.compile(r"^[^/]+\.ya?ml$")
-BUNDLE_MAX_SIZE = 50 * 1024 * 1024  # 50MB
-MANIFEST_FILENAME = "manifest.json"
-MANIFEST_SCHEMA_VERSION = "1.0"
-
-
-# Exceptions
-class BundleFormatError(Exception):
-    """Raised when bundle format is invalid."""
-
-    pass
-
-
-class ZipSecurityError(Exception):
-    """Raised when zip file contains security violations."""
-
-    pass
-
-
-# Manifest DTOs
-class ManifestFileEntry(BaseModel):
-    """Maps node_id to file path in the bundle."""
-
-    model_config = ConfigDict(extra="forbid")
-
-    node_id: str
-    path: str
-
-
-class ManifestIntegrity(BaseModel):
-    """Basic integrity check fields."""
-
-    model_config = ConfigDict(extra="forbid")
-
-    file_count: int
-
-
-class ManifestAppAssets(BaseModel):
-    """App assets section containing the full tree."""
-
-    model_config = ConfigDict(extra="forbid")
-
-    tree: AppAssetFileTree
-
-
-class BundleManifest(BaseModel):
-    """
-    Bundle manifest for app asset import/export.
-
-    Schema version 1.0:
-    - dsl_filename: DSL file name in bundle root (e.g. "my_app.yml")
-    - tree: Full AppAssetFileTree (files + folders) for 100% restoration including node IDs
-    - files: Explicit node_id -> path mapping for file nodes only
-    - integrity: Basic file_count validation
-    """
-
-    model_config = ConfigDict(extra="forbid")
-
-    schema_version: str = Field(default=MANIFEST_SCHEMA_VERSION)
-    generated_at: datetime = Field(default_factory=lambda: datetime.now(tz=UTC))
-    dsl_filename: str = Field(description="DSL file name in bundle root")
-    app_assets: ManifestAppAssets
-    files: list[ManifestFileEntry]
-    integrity: ManifestIntegrity
-
-    @property
-    def assets_prefix(self) -> str:
-        """Assets directory name (DSL filename without extension)."""
-        return self.dsl_filename.rsplit(".", 1)[0]
-
-    @classmethod
-    def from_tree(cls, tree: AppAssetFileTree, dsl_filename: str) -> BundleManifest:
-        """Build manifest from an AppAssetFileTree."""
-        files = [ManifestFileEntry(node_id=n.id, path=tree.get_path(n.id)) for n in tree.walk_files()]
-        return cls(
-            dsl_filename=dsl_filename,
-            app_assets=ManifestAppAssets(tree=tree),
-            files=files,
-            integrity=ManifestIntegrity(file_count=len(files)),
-        )
-
-
-# Export result
-class BundleExportResult(BaseModel):
-    download_url: str = Field(description="Temporary download URL for the ZIP")
-    filename: str = Field(description="Suggested filename for the ZIP")

api/core/app/entities/app_invoke_entities.py

@@ -36,9 +36,6 @@ class InvokeFrom(StrEnum):
     # this is used for plugin trigger and webhook trigger.
     TRIGGER = "trigger"
 
-    # AGENT indicates that this invocation is from an agent.
-    AGENT = "agent"
-
     # EXPLORE indicates that this invocation is from
     # the workflow (or chatflow) explore page.
     EXPLORE = "explore"