chore: fix test cases

Merge branch 'main' into mcp-conditional-capability
2025-10-29 14:51:19 +08:00 · 2025-10-29 14:15:32 +08:00 · 2025-09-29 12:26:26 +08:00 · 2025-09-25 09:14:44 +00:00 · 2025-09-25 16:46:00 +08:00
780 changed files with 5482 additions and 44134 deletions
@@ -6,7 +6,7 @@ cd web && pnpm install
 pipx install uv

 echo "alias start-api=\"cd $WORKSPACE_ROOT/api && uv run python -m flask run --host 0.0.0.0 --port=5001 --debug\"" >> ~/.bashrc
-echo "alias start-worker=\"cd $WORKSPACE_ROOT/api && uv run python -m celery -A app.celery worker -P threads -c 1 --loglevel INFO -Q dataset,generation,mail,ops_trace,app_deletion,plugin,workflow_storage\"" >> ~/.bashrc
+echo "alias start-worker=\"cd $WORKSPACE_ROOT/api && uv run python -m celery -A app.celery worker -P gevent -c 1 --loglevel INFO -Q dataset,generation,mail,ops_trace,app_deletion,plugin,workflow_storage\"" >> ~/.bashrc
 echo "alias start-web=\"cd $WORKSPACE_ROOT/web && pnpm dev\"" >> ~/.bashrc
 echo "alias start-web-prod=\"cd $WORKSPACE_ROOT/web && pnpm build && pnpm start\"" >> ~/.bashrc
 echo "alias start-containers=\"cd $WORKSPACE_ROOT/docker && docker-compose -f docker-compose.middleware.yaml -p dify --env-file middleware.env up -d\"" >> ~/.bashrc
@@ -8,7 +8,8 @@
            "module": "flask",
            "env": {
                "FLASK_APP": "app.py",
-                "FLASK_ENV": "development"
+                "FLASK_ENV": "development",
+                "GEVENT_SUPPORT": "True"
            },
            "args": [
                "run",
@@ -27,7 +28,9 @@
            "type": "debugpy",
            "request": "launch",
            "module": "celery",
-            "env": {},
+            "env": {
+                "GEVENT_SUPPORT": "True"
+            },
            "args": [
                "-A",
                "app.celery",
@@ -30,9 +30,6 @@ INTERNAL_FILES_URL=http://127.0.0.1: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

@@ -374,12 +371,6 @@ UPLOAD_IMAGE_FILE_SIZE_LIMIT=10
 UPLOAD_VIDEO_FILE_SIZE_LIMIT=100
 UPLOAD_AUDIO_FILE_SIZE_LIMIT=50

-# Comma-separated list of file extensions blocked from upload for security reasons.
-# Extensions should be lowercase without dots (e.g., exe,bat,sh,dll).
-# Empty by default to allow all file types.
-# Recommended: exe,bat,cmd,com,scr,vbs,ps1,msi,dll
-UPLOAD_FILE_EXTENSION_BLACKLIST=
-
 # Model configuration
 MULTIMODAL_SEND_FORMAT=base64
 PROMPT_GENERATION_MAX_TOKENS=512
@@ -617,6 +608,3 @@ SWAGGER_UI_PATH=/swagger-ui.html
 # Whether to encrypt dataset IDs when exporting DSL files (default: true)
 # Set to false to export dataset IDs as plain text for easier cross-environment import
 DSL_EXPORT_ENCRYPT_DATASET_ID=true
-
-# Maximum number of segments for dataset segments API (0 for unlimited)
-DATASET_MAX_SEGMENTS_PER_REQUEST=0
@@ -15,11 +15,7 @@ FROM base AS packages
 # RUN sed -i 's@deb.debian.org@mirrors.aliyun.com@g' /etc/apt/sources.list.d/debian.sources

 RUN apt-get update \
-    && apt-get install -y --no-install-recommends \
-          # basic environment
-          g++ \
-          # for building gmpy2
-          libmpfr-dev libmpc-dev
+    && apt-get install -y --no-install-recommends gcc g++ libc-dev libffi-dev libgmp-dev libmpfr-dev libmpc-dev

 # Install Python dependencies
 COPY pyproject.toml uv.lock ./
@@ -53,9 +49,7 @@ RUN \
    # Install dependencies
    && apt-get install -y --no-install-recommends \
        # basic environment
-        curl nodejs \
-        # for gmpy2 \
-        libgmp-dev libmpfr-dev libmpc-dev \
+        curl nodejs libgmp-dev libmpfr-dev libmpc-dev \
        # For Security
        expat libldap-2.5-0 perl libsqlite3-0 zlib1g \
        # install fonts to support the use of tools like pypdfium2
@@ -80,7 +80,7 @@
 1. If you need to handle and debug the async tasks (e.g. dataset importing and documents indexing), please start the worker service.

 ```bash
-uv run celery -A app.celery worker -P threads -c 2 --loglevel INFO -Q dataset,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,priority_pipeline,pipeline
+uv run celery -A app.celery worker -P gevent -c 2 --loglevel INFO -Q dataset,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,priority_pipeline,pipeline
 ```

 Additionally, if you want to debug the celery scheduled tasks, you can run the following command in another terminal to start the beat service:
@@ -1,4 +1,3 @@
-import os
 import sys


@@ -9,35 +8,33 @@ def is_db_command():


 # create app
-celery = None
-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.
+    # It seems that JetBrains Python debugger does not work well with gevent,
+    # so we need to disable gevent in debug mode.
+    # If you are using debugpy and set GEVENT_SUPPORT=True, you can debug with gevent.
+    # if (flask_debug := os.environ.get("FLASK_DEBUG", "0")) and flask_debug.lower() in {"false", "0", "no"}:
+    # from gevent import monkey
    #
-    # See `api/docker/entrypoint.sh` (lines 33 and 47) for details.
+    # # gevent
+    # monkey.patch_all()
    #
-    # For third-party library patching, refer to `gunicorn.conf.py` and `celery_entrypoint.py`.
+    # from grpc.experimental import gevent as grpc_gevent  # type: ignore
+    #
+    # # grpc gevent
+    # grpc_gevent.init_gevent()
+
+    # import psycogreen.gevent  # type: ignore
+    #
+    # psycogreen.gevent.patch_psycopg()

    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
-
-    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)
@@ -31,22 +31,14 @@ def create_flask_app_with_configs() -> DifyApp:
    return dify_app


-def create_app() -> tuple[any, DifyApp]:
+def create_app() -> DifyApp:
    start_time = time.perf_counter()
    app = create_flask_app_with_configs()
    initialize_extensions(app)
-
-    import socketio
-
-    from extensions.ext_socketio import sio
-
-    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):
@@ -1601,7 +1601,7 @@ def transform_datasource_credentials():
                            "integration_secret": api_key,
                        }
                        datasource_provider = DatasourceProvider(
-                            provider="jinareader",
+                            provider="jina",
                            tenant_id=tenant_id,
                            plugin_id=jina_plugin_id,
                            auth_type=api_key_credential_type.value,
@@ -331,31 +331,6 @@ class FileUploadConfig(BaseSettings):
        default=10,
    )

-    inner_UPLOAD_FILE_EXTENSION_BLACKLIST: str = Field(
-        description=(
-            "Comma-separated list of file extensions that are blocked from upload. "
-            "Extensions should be lowercase without dots (e.g., 'exe,bat,sh,dll'). "
-            "Empty by default to allow all file types."
-        ),
-        validation_alias=AliasChoices("UPLOAD_FILE_EXTENSION_BLACKLIST"),
-        default="",
-    )
-
-    @computed_field  # type: ignore[misc]
-    @property
-    def UPLOAD_FILE_EXTENSION_BLACKLIST(self) -> set[str]:
-        """
-        Parse and return the blacklist as a set of lowercase extensions.
-        Returns an empty set if no blacklist is configured.
-        """
-        if not self.inner_UPLOAD_FILE_EXTENSION_BLACKLIST:
-            return set()
-        return {
-            ext.strip().lower().strip(".")
-            for ext in self.inner_UPLOAD_FILE_EXTENSION_BLACKLIST.split(",")
-            if ext.strip()
-        }
-

 class HttpConfig(BaseSettings):
    """
@@ -871,16 +846,6 @@ class MailConfig(BaseSettings):
        default=None,
    )

-    ENABLE_TRIAL_APP: bool = Field(
-        description="Enable trial app",
-        default=False,
-    )
-
-    ENABLE_EXPLORE_BANNER: bool = Field(
-        description="Enable explore banner",
-        default=False,
-    )
-

 class RagEtlConfig(BaseSettings):
    """
@@ -955,11 +920,6 @@ class DataSetConfig(BaseSettings):
        default=True,
    )

-    DATASET_MAX_SEGMENTS_PER_REQUEST: NonNegativeInt = Field(
-        description="Maximum number of segments for dataset segments API (0 for unlimited)",
-        default=0,
-    )
-

 class WorkspaceConfig(BaseSettings):
    """
@@ -1093,13 +1053,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",
@@ -1159,13 +1112,6 @@ class SwaggerUIConfig(BaseSettings):
    )


-class TenantSelfTaskQueueConfig(BaseSettings):
-    TENANT_SELF_TASK_QUEUE_PULL_SIZE: int = Field(
-        description="Default batch size for tenant self task queue pull operations",
-        default=1,
-    )
-
-
 class FeatureConfig(
    # place the configs in alphabet order
    AppExecutionConfig,
@@ -1190,13 +1136,11 @@ class FeatureConfig(
    RagEtlConfig,
    RepositoryConfig,
    SecurityConfig,
-    TenantSelfTaskQueueConfig,
    ToolConfig,
    UpdateConfig,
    WorkflowConfig,
    WorkflowNodeExecutionConfig,
    WorkspaceConfig,
-    CollaborationConfig,
    LoginConfig,
    AccountConfig,
    SwaggerUIConfig,
@@ -8,11 +8,6 @@ class HostedCreditConfig(BaseSettings):
        default="",
    )

-    HOSTED_POOL_CREDITS: int = Field(
-        description="Pool credits for hosted service",
-        default=200,
-    )
-
    def get_model_credits(self, model_name: str) -> int:
        """
        Get credit value for a specific model name.
@@ -65,46 +60,19 @@ class HostedOpenAiConfig(BaseSettings):

    HOSTED_OPENAI_TRIAL_MODELS: str = Field(
        description="Comma-separated list of available models for trial access",
-        default="gpt-4,"
-        "gpt-4-turbo-preview,"
-        "gpt-4-turbo-2024-04-09,"
-        "gpt-4-1106-preview,"
-        "gpt-4-0125-preview,"
-        "gpt-4-turbo,"
-        "gpt-4.1,"
-        "gpt-4.1-2025-04-14,"
-        "gpt-4.1-mini,"
-        "gpt-4.1-mini-2025-04-14,"
-        "gpt-4.1-nano,"
-        "gpt-4.1-nano-2025-04-14,"
-        "gpt-3.5-turbo,"
+        default="gpt-3.5-turbo,"
+        "gpt-3.5-turbo-1106,"
+        "gpt-3.5-turbo-instruct,"
        "gpt-3.5-turbo-16k,"
        "gpt-3.5-turbo-16k-0613,"
-        "gpt-3.5-turbo-1106,"
        "gpt-3.5-turbo-0613,"
        "gpt-3.5-turbo-0125,"
-        "gpt-3.5-turbo-instruct,"
-        "text-davinci-003,"
-        "chatgpt-4o-latest,"
-        "gpt-4o,"
-        "gpt-4o-2024-05-13,"
-        "gpt-4o-2024-08-06,"
-        "gpt-4o-2024-11-20,"
-        "gpt-4o-audio-preview,"
-        "gpt-4o-audio-preview-2025-06-03,"
-        "gpt-4o-mini,"
-        "gpt-4o-mini-2024-07-18,"
-        "o3-mini,"
-        "o3-mini-2025-01-31,"
-        "gpt-5-mini-2025-08-07,"
-        "gpt-5-mini,"
-        "o4-mini,"
-        "o4-mini-2025-04-16,"
-        "gpt-5-chat-latest,"
-        "gpt-5,"
-        "gpt-5-2025-08-07,"
-        "gpt-5-nano,"
-        "gpt-5-nano-2025-08-07",
+        "text-davinci-003",
+    )
+
+    HOSTED_OPENAI_QUOTA_LIMIT: NonNegativeInt = Field(
+        description="Quota limit for hosted OpenAI service usage",
+        default=200,
    )

    HOSTED_OPENAI_PAID_ENABLED: bool = Field(
@@ -119,13 +87,6 @@ class HostedOpenAiConfig(BaseSettings):
        "gpt-4-turbo-2024-04-09,"
        "gpt-4-1106-preview,"
        "gpt-4-0125-preview,"
-        "gpt-4-turbo,"
-        "gpt-4.1,"
-        "gpt-4.1-2025-04-14,"
-        "gpt-4.1-mini,"
-        "gpt-4.1-mini-2025-04-14,"
-        "gpt-4.1-nano,"
-        "gpt-4.1-nano-2025-04-14,"
        "gpt-3.5-turbo,"
        "gpt-3.5-turbo-16k,"
        "gpt-3.5-turbo-16k-0613,"
@@ -133,150 +94,7 @@ class HostedOpenAiConfig(BaseSettings):
        "gpt-3.5-turbo-0613,"
        "gpt-3.5-turbo-0125,"
        "gpt-3.5-turbo-instruct,"
-        "text-davinci-003,"
-        "chatgpt-4o-latest,"
-        "gpt-4o,"
-        "gpt-4o-2024-05-13,"
-        "gpt-4o-2024-08-06,"
-        "gpt-4o-2024-11-20,"
-        "gpt-4o-audio-preview,"
-        "gpt-4o-audio-preview-2025-06-03,"
-        "gpt-4o-mini,"
-        "gpt-4o-mini-2024-07-18,"
-        "o3-mini,"
-        "o3-mini-2025-01-31,"
-        "gpt-5-mini-2025-08-07,"
-        "gpt-5-mini,"
-        "o4-mini,"
-        "o4-mini-2025-04-16,"
-        "gpt-5-chat-latest,"
-        "gpt-5,"
-        "gpt-5-2025-08-07,"
-        "gpt-5-nano,"
-        "gpt-5-nano-2025-08-07",
-    )
-
-
-class HostedGeminiConfig(BaseSettings):
-    """
-    Configuration for fetching Gemini service
-    """
-
-    HOSTED_GEMINI_API_KEY: str | None = Field(
-        description="API key for hosted Gemini service",
-        default=None,
-    )
-
-    HOSTED_GEMINI_API_BASE: str | None = Field(
-        description="Base URL for hosted Gemini API",
-        default=None,
-    )
-
-    HOSTED_GEMINI_API_ORGANIZATION: str | None = Field(
-        description="Organization ID for hosted Gemini service",
-        default=None,
-    )
-
-    HOSTED_GEMINI_TRIAL_ENABLED: bool = Field(
-        description="Enable trial access to hosted Gemini service",
-        default=False,
-    )
-
-    HOSTED_GEMINI_TRIAL_MODELS: str = Field(
-        description="Comma-separated list of available models for trial access",
-        default="gemini-2.5-flash,gemini-2.0-flash,gemini-2.0-flash-lite,",
-    )
-
-    HOSTED_GEMINI_PAID_ENABLED: bool = Field(
-        description="Enable paid access to hosted gemini service",
-        default=False,
-    )
-
-    HOSTED_GEMINI_PAID_MODELS: str = Field(
-        description="Comma-separated list of available models for paid access",
-        default="gemini-2.5-flash,gemini-2.0-flash,gemini-2.0-flash-lite,",
-    )
-
-
-class HostedXAIConfig(BaseSettings):
-    """
-    Configuration for fetching XAI service
-    """
-
-    HOSTED_XAI_API_KEY: str | None = Field(
-        description="API key for hosted XAI service",
-        default=None,
-    )
-
-    HOSTED_XAI_API_BASE: str | None = Field(
-        description="Base URL for hosted XAI API",
-        default=None,
-    )
-
-    HOSTED_XAI_API_ORGANIZATION: str | None = Field(
-        description="Organization ID for hosted XAI service",
-        default=None,
-    )
-
-    HOSTED_XAI_TRIAL_ENABLED: bool = Field(
-        description="Enable trial access to hosted XAI service",
-        default=False,
-    )
-
-    HOSTED_XAI_TRIAL_MODELS: str = Field(
-        description="Comma-separated list of available models for trial access",
-        default="grok-3,grok-3-mini,grok-3-mini-fast",
-    )
-
-    HOSTED_XAI_PAID_ENABLED: bool = Field(
-        description="Enable paid access to hosted XAI service",
-        default=False,
-    )
-
-    HOSTED_XAI_PAID_MODELS: str = Field(
-        description="Comma-separated list of available models for paid access",
-        default="grok-3,grok-3-mini,grok-3-mini-fast",
-    )
-
-
-class HostedDeepseekConfig(BaseSettings):
-    """
-    Configuration for fetching Deepseek service
-    """
-
-    HOSTED_DEEPSEEK_API_KEY: str | None = Field(
-        description="API key for hosted Deepseek service",
-        default=None,
-    )
-
-    HOSTED_DEEPSEEK_API_BASE: str | None = Field(
-        description="Base URL for hosted Deepseek API",
-        default=None,
-    )
-
-    HOSTED_DEEPSEEK_API_ORGANIZATION: str | None = Field(
-        description="Organization ID for hosted Deepseek service",
-        default=None,
-    )
-
-    HOSTED_DEEPSEEK_TRIAL_ENABLED: bool = Field(
-        description="Enable trial access to hosted Deepseek service",
-        default=False,
-    )
-
-    HOSTED_DEEPSEEK_TRIAL_MODELS: str = Field(
-        description="Comma-separated list of available models for trial access",
-        default="deepseek-chat,deepseek-reasoner",
-    )
-
-    HOSTED_DEEPSEEK_PAID_ENABLED: bool = Field(
-        description="Enable paid access to hosted XAI service",
-        default=False,
-    )
-
-    HOSTED_DEEPSEEK_PAID_MODELS: str = Field(
-        description="Comma-separated list of available models for paid access",
-        default="deepseek-chat,deepseek-reasoner",
+        "text-davinci-003",
    )


@@ -326,30 +144,16 @@ class HostedAnthropicConfig(BaseSettings):
        default=False,
    )

+    HOSTED_ANTHROPIC_QUOTA_LIMIT: NonNegativeInt = Field(
+        description="Quota limit for hosted Anthropic service usage",
+        default=600000,
+    )
+
    HOSTED_ANTHROPIC_PAID_ENABLED: bool = Field(
        description="Enable paid access to hosted Anthropic service",
        default=False,
    )

-    HOSTED_ANTHROPIC_TRIAL_MODELS: str = Field(
-        description="Comma-separated list of available models for paid access",
-        default="claude-opus-4-20250514,"
-        "claude-sonnet-4-20250514,"
-        "claude-3-5-haiku-20241022,"
-        "claude-3-opus-20240229,"
-        "claude-3-7-sonnet-20250219,"
-        "claude-3-haiku-20240307",
-    )
-    HOSTED_ANTHROPIC_PAID_MODELS: str = Field(
-        description="Comma-separated list of available models for paid access",
-        default="claude-opus-4-20250514,"
-        "claude-sonnet-4-20250514,"
-        "claude-3-5-haiku-20241022,"
-        "claude-3-opus-20240229,"
-        "claude-3-7-sonnet-20250219,"
-        "claude-3-haiku-20240307",
-    )
-

 class HostedMinmaxConfig(BaseSettings):
    """
@@ -446,8 +250,5 @@ class HostedServiceConfig(
    HostedModerationConfig,
    # credit config
    HostedCreditConfig,
-    HostedGeminiConfig,
-    HostedXAIConfig,
-    HostedDeepseekConfig,
 ):
    pass
@@ -22,11 +22,6 @@ class WeaviateConfig(BaseSettings):
        default=True,
    )

-    WEAVIATE_GRPC_ENDPOINT: str | None = Field(
-        description="URL of the Weaviate gRPC server (e.g., 'grpc://localhost:50051' or 'grpcs://weaviate.example.com:443')",
-        default=None,
-    )
-
    WEAVIATE_BATCH_SIZE: PositiveInt = Field(
        description="Number of objects to be processed in a single batch operation (default is 100)",
        default=100,
@@ -25,12 +25,6 @@ class UnsupportedFileTypeError(BaseHTTPException):
    code = 415


-class BlockedFileExtensionError(BaseHTTPException):
-    error_code = "file_extension_blocked"
-    description = "The file extension is blocked for security reasons."
-    code = 400
-
-
 class TooManyFilesError(BaseHTTPException):
    error_code = "too_many_files"
    description = "Only one file is allowed."
@@ -58,13 +58,11 @@ from .app import (
    mcp_server,
    message,
    model_config,
-    online_user,
    ops_trace,
    site,
    statistic,
    workflow,
    workflow_app_log,
-    workflow_comment,
    workflow_draft_variable,
    workflow_run,
    workflow_statistic,
@@ -108,12 +106,10 @@ from .datasets.rag_pipeline import (

 # Import explore controllers
 from .explore import (
-    banner,
    installed_app,
    parameter,
    recommended_app,
    saved_message,
-    trial,
 )

 # Import tag controllers
@@ -147,7 +143,6 @@ __all__ = [
    "apikey",
    "app",
    "audio",
-    "banner",
    "billing",
    "bp",
    "completion",
@@ -201,7 +196,6 @@ __all__ = [
    "statistic",
    "tags",
    "tool_providers",
-    "trial",
    "version",
    "website",
    "workflow",
@@ -16,7 +16,7 @@ from controllers.console import api, console_ns
 from controllers.console.wraps import only_edition_cloud
 from extensions.ext_database import db
 from libs.token import extract_access_token
-from models.model import App, ExporleBanner, InstalledApp, RecommendedApp, TrialApp
+from models.model import App, InstalledApp, RecommendedApp


 def admin_required(view: Callable[P, R]):
@@ -52,8 +52,6 @@ class InsertExploreAppListApi(Resource):
                "language": fields.String(required=True, description="Language code"),
                "category": fields.String(required=True, description="App category"),
                "position": fields.Integer(required=True, description="Display position"),
-                "can_trial": fields.Boolean(required=True, description="Can trial"),
-                "trial_limit": fields.Integer(required=True, description="Trial limit"),
            },
        )
    )
@@ -73,8 +71,6 @@ class InsertExploreAppListApi(Resource):
            .add_argument("language", type=supported_language, required=True, nullable=False, location="json")
            .add_argument("category", type=str, required=True, nullable=False, location="json")
            .add_argument("position", type=int, required=True, nullable=False, location="json")
-            .add_argument("can_trial", type=bool, required=True, nullable=False, location="json")
-            .add_argument("trial_limit", type=int, required=True, nullable=False, location="json")
        )
        args = parser.parse_args()

@@ -112,20 +108,6 @@ class InsertExploreAppListApi(Resource):
                )

                db.session.add(recommended_app)
-                if args["can_trial"]:
-                    trial_app = db.session.execute(
-                        select(TrialApp).where(TrialApp.app_id == args["app_id"])
-                    ).scalar_one_or_none()
-                    if not trial_app:
-                        db.session.add(
-                            TrialApp(
-                                app_id=args["app_id"],
-                                tenant_id=app.tenant_id,
-                                trial_limit=args["trial_limit"],
-                            )
-                        )
-                    else:
-                        trial_app.trial_limit = args["trial_limit"]

                app.is_public = True
                db.session.commit()
@@ -140,20 +122,6 @@ class InsertExploreAppListApi(Resource):
                recommended_app.category = args["category"]
                recommended_app.position = args["position"]

-                if args["can_trial"]:
-                    trial_app = db.session.execute(
-                        select(TrialApp).where(TrialApp.app_id == args["app_id"])
-                    ).scalar_one_or_none()
-                    if not trial_app:
-                        db.session.add(
-                            TrialApp(
-                                app_id=args["app_id"],
-                                tenant_id=app.tenant_id,
-                                trial_limit=args["trial_limit"],
-                            )
-                        )
-                    else:
-                        trial_app.trial_limit = args["trial_limit"]
                app.is_public = True

                db.session.commit()
@@ -199,83 +167,7 @@ class InsertExploreAppApi(Resource):
            for installed_app in installed_apps:
                session.delete(installed_app)

-            trial_app = session.execute(
-                select(TrialApp).where(TrialApp.app_id == recommended_app.app_id)
-            ).scalar_one_or_none()
-            if trial_app:
-                session.delete(trial_app)
-
        db.session.delete(recommended_app)
        db.session.commit()

        return {"result": "success"}, 204
-
-
-@console_ns.route("/admin/insert-explore-banner")
-class InsertExploreBanner(Resource):
-    @api.doc("insert_explore_banner")
-    @api.doc(description="Insert an explore banner")
-    @api.expect(
-        api.model(
-            "InsertExploreBannerRequest",
-            {
-                "content": fields.String(required=True, description="Banner content"),
-                "link": fields.String(required=True, description="Banner link"),
-                "sort": fields.Integer(required=True, description="Banner sort"),
-            },
-        )
-    )
-    @api.response(200, "Banner inserted successfully")
-    @admin_required
-    @only_edition_cloud
-    def post(self):
-        parser = reqparse.RequestParser()
-        parser.add_argument("category", type=str, required=True, nullable=False, location="json")
-        parser.add_argument("title", type=str, required=True, nullable=False, location="json")
-        parser.add_argument("description", type=str, required=True, nullable=False, location="json")
-        parser.add_argument("img-src", type=str, required=True, nullable=False, location="json")
-
-        parser.add_argument("language", type=str, required=True, nullable=False, location="json")
-        parser.add_argument("link", type=str, required=True, nullable=False, location="json")
-        parser.add_argument("sort", type=int, required=True, nullable=False, location="json")
-
-        args = parser.parse_args()
-
-        content = {
-            "category": args["category"],
-            "title": args["title"],
-            "description": args["description"],
-            "img-src": args["img-src"],
-        }
-
-        if not args["language"]:
-            args["language"] = "en-US"
-
-        banner = ExporleBanner(
-            content=content,
-            link=args["link"],
-            sort=args["sort"],
-            language=args["language"],
-        )
-        db.session.add(banner)
-        db.session.commit()
-
-        return {"result": "success"}, 200
-
-
-@console_ns.route("/admin/delete-explore-banner/<uuid:banner_id>")
-class DeleteExploreBanner(Resource):
-    @api.doc("delete_explore_banner")
-    @api.doc(description="Delete an explore banner")
-    @api.response(204, "Banner deleted successfully")
-    @admin_required
-    @only_edition_cloud
-    def delete(self, banner_id):
-        banner = db.session.execute(select(ExporleBanner).where(ExporleBanner.id == banner_id)).scalar_one_or_none()
-        if not banner:
-            raise NotFound(f"Banner '{banner_id}' is not found")
-
-        db.session.delete(banner)
-        db.session.commit()
-
-        return {"result": "success"}, 204
@@ -16,7 +16,6 @@ from fields.annotation_fields import (
    annotation_fields,
    annotation_hit_history_fields,
 )
-from libs.helper import uuid_value
 from libs.login import login_required
 from services.annotation_service import AppAnnotationService

@@ -176,10 +175,8 @@ class AnnotationApi(Resource):
        api.model(
            "CreateAnnotationRequest",
            {
-                "message_id": fields.String(description="Message ID (optional)"),
-                "question": fields.String(description="Question text (required when message_id not provided)"),
-                "answer": fields.String(description="Answer text (use 'answer' or 'content')"),
-                "content": fields.String(description="Content text (use 'answer' or 'content')"),
+                "question": fields.String(required=True, description="Question text"),
+                "answer": fields.String(required=True, description="Answer text"),
                "annotation_reply": fields.Raw(description="Annotation reply data"),
            },
        )
@@ -196,14 +193,11 @@ class AnnotationApi(Resource):
        app_id = str(app_id)
        parser = (
            reqparse.RequestParser()
-            .add_argument("message_id", required=False, type=uuid_value, location="json")
-            .add_argument("question", required=False, type=str, location="json")
-            .add_argument("answer", required=False, type=str, location="json")
-            .add_argument("content", required=False, type=str, location="json")
-            .add_argument("annotation_reply", required=False, type=dict, location="json")
+            .add_argument("question", required=True, type=str, location="json")
+            .add_argument("answer", required=True, type=str, location="json")
        )
        args = parser.parse_args()
-        annotation = AppAnnotationService.up_insert_app_annotation_from_message(args, app_id)
+        annotation = AppAnnotationService.insert_app_annotation_directly(args, app_id)
        return annotation

    @setup_required
@@ -1,5 +1,7 @@
+from datetime import datetime
+
+import pytz
 import sqlalchemy as sa
-from flask import abort
 from flask_restx import Resource, marshal_with, reqparse
 from flask_restx.inputs import int_range
 from sqlalchemy import func, or_
@@ -17,7 +19,7 @@ from fields.conversation_fields import (
    conversation_pagination_fields,
    conversation_with_summary_pagination_fields,
 )
-from libs.datetime_utils import naive_utc_now, parse_time_range
+from libs.datetime_utils import naive_utc_now
 from libs.helper import DatetimeString
 from libs.login import current_account_with_tenant, login_required
 from models import Conversation, EndUser, Message, MessageAnnotation
@@ -88,17 +90,25 @@ class CompletionConversationApi(Resource):

        account = current_user
        assert account.timezone is not None
+        timezone = pytz.timezone(account.timezone)
+        utc_timezone = pytz.utc

-        try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
-        except ValueError as e:
-            abort(400, description=str(e))
+        if args["start"]:
+            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
+            start_datetime = start_datetime.replace(second=0)
+
+            start_datetime_timezone = timezone.localize(start_datetime)
+            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)

-        if start_datetime_utc:
            query = query.where(Conversation.created_at >= start_datetime_utc)

-        if end_datetime_utc:
-            end_datetime_utc = end_datetime_utc.replace(second=59)
+        if args["end"]:
+            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
+            end_datetime = end_datetime.replace(second=59)
+
+            end_datetime_timezone = timezone.localize(end_datetime)
+            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
+
            query = query.where(Conversation.created_at < end_datetime_utc)

        # FIXME, the type ignore in this file
@@ -260,21 +270,29 @@ class ChatConversationApi(Resource):

        account = current_user
        assert account.timezone is not None
+        timezone = pytz.timezone(account.timezone)
+        utc_timezone = pytz.utc

-        try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
-        except ValueError as e:
-            abort(400, description=str(e))
+        if args["start"]:
+            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
+            start_datetime = start_datetime.replace(second=0)
+
+            start_datetime_timezone = timezone.localize(start_datetime)
+            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)

-        if start_datetime_utc:
            match args["sort_by"]:
                case "updated_at" | "-updated_at":
                    query = query.where(Conversation.updated_at >= start_datetime_utc)
                case "created_at" | "-created_at" | _:
                    query = query.where(Conversation.created_at >= start_datetime_utc)

-        if end_datetime_utc:
-            end_datetime_utc = end_datetime_utc.replace(second=59)
+        if args["end"]:
+            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
+            end_datetime = end_datetime.replace(second=59)
+
+            end_datetime_timezone = timezone.localize(end_datetime)
+            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
+
            match args["sort_by"]:
                case "updated_at" | "-updated_at":
                    query = query.where(Conversation.updated_at <= end_datetime_utc)
@@ -115,9 +115,3 @@ class InvokeRateLimitError(BaseHTTPException):
    error_code = "rate_limit_error"
    description = "Rate Limit Error"
    code = 429
-
-
-class NeedAddIdsError(BaseHTTPException):
-    error_code = "need_add_ids"
-    description = "Need to add ids."
-    code = 400
@@ -16,6 +16,7 @@ from controllers.console.app.wraps import get_app_model
 from controllers.console.explore.error import AppSuggestedQuestionsAfterAnswerDisabledError
 from controllers.console.wraps import (
    account_initialization_required,
+    cloud_edition_billing_resource_check,
    edit_permission_required,
    setup_required,
 )
@@ -23,11 +24,12 @@ from core.app.entities.app_invoke_entities import InvokeFrom
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
 from core.model_runtime.errors.invoke import InvokeError
 from extensions.ext_database import db
-from fields.conversation_fields import message_detail_fields
+from fields.conversation_fields import annotation_fields, message_detail_fields
 from libs.helper import uuid_value
 from libs.infinite_scroll_pagination import InfiniteScrollPagination
 from libs.login import current_account_with_tenant, login_required
 from models.model import AppMode, Conversation, Message, MessageAnnotation, MessageFeedback
+from services.annotation_service import AppAnnotationService
 from services.errors.conversation import ConversationNotExistsError
 from services.errors.message import MessageNotExistsError, SuggestedQuestionsAfterAnswerDisabledError
 from services.message_service import MessageService
@@ -192,6 +194,45 @@ class MessageFeedbackApi(Resource):
        return {"result": "success"}


+@console_ns.route("/apps/<uuid:app_id>/annotations")
+class MessageAnnotationApi(Resource):
+    @api.doc("create_message_annotation")
+    @api.doc(description="Create message annotation")
+    @api.doc(params={"app_id": "Application ID"})
+    @api.expect(
+        api.model(
+            "MessageAnnotationRequest",
+            {
+                "message_id": fields.String(description="Message ID"),
+                "question": fields.String(required=True, description="Question text"),
+                "answer": fields.String(required=True, description="Answer text"),
+                "annotation_reply": fields.Raw(description="Annotation reply"),
+            },
+        )
+    )
+    @api.response(200, "Annotation created successfully", annotation_fields)
+    @api.response(403, "Insufficient permissions")
+    @marshal_with(annotation_fields)
+    @get_app_model
+    @setup_required
+    @login_required
+    @cloud_edition_billing_resource_check("annotation")
+    @account_initialization_required
+    @edit_permission_required
+    def post(self, app_model):
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("message_id", required=False, type=uuid_value, location="json")
+            .add_argument("question", required=True, type=str, location="json")
+            .add_argument("answer", required=True, type=str, location="json")
+            .add_argument("annotation_reply", required=False, type=dict, location="json")
+        )
+        args = parser.parse_args()
+        annotation = AppAnnotationService.up_insert_app_annotation_from_message(args, app_model.id)
+
+        return annotation
+
+
@console_ns.route("/apps/<uuid:app_id>/annotations/count")
 class MessageAnnotationCountApi(Resource):
    @api.doc("get_annotation_count")
@@ -1,339 +0,0 @@
-import json
-import time
-
-from werkzeug.wrappers import Request as WerkzeugRequest
-
-from extensions.ext_redis import redis_client
-from extensions.ext_socketio import sio
-from libs.passport import PassportService
-from libs.token import extract_access_token
-from services.account_service import AccountService
-
-SESSION_STATE_TTL_SECONDS = 3600
-WORKFLOW_ONLINE_USERS_PREFIX = "workflow_online_users:"
-WORKFLOW_LEADER_PREFIX = "workflow_leader:"
-WS_SID_MAP_PREFIX = "ws_sid_map:"
-
-
-def _workflow_key(workflow_id: str) -> str:
-    return f"{WORKFLOW_ONLINE_USERS_PREFIX}{workflow_id}"
-
-
-def _leader_key(workflow_id: str) -> str:
-    return f"{WORKFLOW_LEADER_PREFIX}{workflow_id}"
-
-
-def _sid_key(sid: str) -> str:
-    return f"{WS_SID_MAP_PREFIX}{sid}"
-
-
-def _refresh_session_state(workflow_id: str, sid: str) -> None:
-    """
-    Refresh TTLs for workflow + session keys so healthy sessions do not linger forever after crashes.
-    """
-    workflow_key = _workflow_key(workflow_id)
-    sid_key = _sid_key(sid)
-    if redis_client.exists(workflow_key):
-        redis_client.expire(workflow_key, SESSION_STATE_TTL_SECONDS)
-    if redis_client.exists(sid_key):
-        redis_client.expire(sid_key, SESSION_STATE_TTL_SECONDS)
-
-
-@sio.on("connect")
-def socket_connect(sid, environ, auth):
-    """
-    WebSocket connect event, do authentication here.
-    """
-    token = None
-    if auth and isinstance(auth, dict):
-        token = auth.get("token")
-
-    if not token:
-        try:
-            request_environ = WerkzeugRequest(environ)
-            token = extract_access_token(request_environ)
-        except Exception:
-            token = None
-
-    if not token:
-        return False
-
-    try:
-        decoded = PassportService().verify(token)
-        user_id = decoded.get("user_id")
-        if not user_id:
-            return False
-
-        with sio.app.app_context():
-            user = AccountService.load_logged_in_account(account_id=user_id)
-            if not user:
-                return False
-
-            sio.save_session(sid, {"user_id": user.id, "username": user.name, "avatar": user.avatar})
-
-            return True
-
-    except Exception:
-        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
-
-    session = sio.get_session(sid)
-    user_id = session.get("user_id")
-
-    if not user_id:
-        return {"msg": "unauthorized"}, 401
-
-    # Each session is stored independently with sid as key
-    session_info = {
-        "user_id": user_id,
-        "username": session.get("username", "Unknown"),
-        "avatar": session.get("avatar", None),
-        "sid": sid,
-        "connected_at": int(time.time()),  # Add timestamp to differentiate tabs
-    }
-
-    workflow_key = _workflow_key(workflow_id)
-    # Store session info with sid as key
-    redis_client.hset(workflow_key, sid, json.dumps(session_info))
-    redis_client.set(
-        _sid_key(sid),
-        json.dumps({"workflow_id": workflow_id, "user_id": user_id}),
-        ex=SESSION_STATE_TTL_SECONDS,
-    )
-    _refresh_session_state(workflow_id, sid)
-
-    # Leader election: first session becomes the leader
-    leader_sid = get_or_set_leader(workflow_id, sid)
-    is_leader = leader_sid == sid
-
-    sio.enter_room(sid, workflow_id)
-    broadcast_online_users(workflow_id)
-
-    # Notify this session of their leader status
-    sio.emit("status", {"isLeader": is_leader}, room=sid)
-
-    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.
-    """
-    mapping = redis_client.get(_sid_key(sid))
-    if mapping:
-        data = json.loads(mapping)
-        workflow_id = data["workflow_id"]
-
-        # Remove this specific session
-        redis_client.hdel(_workflow_key(workflow_id), sid)
-        redis_client.delete(_sid_key(sid))
-
-        # Handle leader re-election if the leader session disconnected
-        handle_leader_disconnect(workflow_id, sid)
-
-        broadcast_online_users(workflow_id)
-
-
-def _clear_session_state(workflow_id: str, sid: str) -> None:
-    redis_client.hdel(_workflow_key(workflow_id), sid)
-    redis_client.delete(_sid_key(sid))
-
-
-def _is_session_active(workflow_id: str, sid: str) -> bool:
-    if not sid:
-        return False
-
-    try:
-        if not sio.manager.is_connected(sid, "/"):
-            return False
-    except AttributeError:
-        return False
-
-    if not redis_client.hexists(_workflow_key(workflow_id), sid):
-        return False
-
-    if not redis_client.exists(_sid_key(sid)):
-        return False
-
-    return True
-
-
-def get_or_set_leader(workflow_id: str, sid: str) -> str:
-    """
-    Get current leader session or set this session as leader if no valid leader exists.
-    Returns the leader session id (sid).
-    """
-    raw_leader = redis_client.get(_leader_key(workflow_id))
-    current_leader = raw_leader.decode("utf-8") if isinstance(raw_leader, bytes) else raw_leader
-    leader_replaced = False
-
-    if current_leader and not _is_session_active(workflow_id, current_leader):
-        _clear_session_state(workflow_id, current_leader)
-        redis_client.delete(_leader_key(workflow_id))
-        current_leader = None
-        leader_replaced = True
-
-    if not current_leader:
-        redis_client.set(_leader_key(workflow_id), sid, ex=SESSION_STATE_TTL_SECONDS)  # Expire in 1 hour
-        if leader_replaced:
-            broadcast_leader_change(workflow_id, sid)
-        return sid
-
-    return current_leader
-
-
-def handle_leader_disconnect(workflow_id, disconnected_sid):
-    """
-    Handle leader re-election when a session disconnects.
-    If the disconnected session was the leader, elect a new leader from remaining sessions.
-    """
-    current_leader = redis_client.get(_leader_key(workflow_id))
-
-    if current_leader:
-        current_leader = current_leader.decode("utf-8") if isinstance(current_leader, bytes) else current_leader
-
-        if current_leader == disconnected_sid:
-            # Leader session disconnected, elect a new leader
-            sessions_json = redis_client.hgetall(_workflow_key(workflow_id))
-
-            if sessions_json:
-                # Get the first remaining session as new leader
-                new_leader_sid = list(sessions_json.keys())[0]
-                if isinstance(new_leader_sid, bytes):
-                    new_leader_sid = new_leader_sid.decode("utf-8")
-
-                redis_client.set(_leader_key(workflow_id), new_leader_sid, ex=SESSION_STATE_TTL_SECONDS)
-
-                # Notify all sessions about the new leader
-                broadcast_leader_change(workflow_id, new_leader_sid)
-            else:
-                # No sessions left, remove leader
-                redis_client.delete(_leader_key(workflow_id))
-
-
-def broadcast_leader_change(workflow_id, new_leader_sid):
-    """
-    Broadcast leader change to all sessions in the workflow.
-    """
-    sessions_json = redis_client.hgetall(_workflow_key(workflow_id))
-
-    for sid, session_info_json in sessions_json.items():
-        try:
-            sid_str = sid.decode("utf-8") if isinstance(sid, bytes) else sid
-            is_leader = sid_str == new_leader_sid
-            # Emit to each session whether they are the new leader
-            sio.emit("status", {"isLeader": is_leader}, room=sid_str)
-        except Exception:
-            continue
-
-
-def get_current_leader(workflow_id):
-    """
-    Get the current leader for a workflow.
-    """
-    leader = redis_client.get(_leader_key(workflow_id))
-    return leader.decode("utf-8") if leader and isinstance(leader, bytes) else leader
-
-
-def broadcast_online_users(workflow_id):
-    """
-    Broadcast online users to the workflow room.
-    Each session is shown as a separate user (even if same person has multiple tabs).
-    """
-    sessions_json = redis_client.hgetall(_workflow_key(workflow_id))
-    users = []
-
-    for sid, session_info_json in sessions_json.items():
-        try:
-            session_info = json.loads(session_info_json)
-            # Each session appears as a separate "user" in the UI
-            users.append(
-                {
-                    "user_id": session_info["user_id"],
-                    "username": session_info["username"],
-                    "avatar": session_info.get("avatar"),
-                    "sid": session_info["sid"],
-                    "connected_at": session_info.get("connected_at"),
-                }
-            )
-        except Exception:
-            continue
-
-    # Sort by connection time to maintain consistent order
-    users.sort(key=lambda x: x.get("connected_at") or 0)
-
-    # Get current leader session
-    leader_sid = get_current_leader(workflow_id)
-
-    sio.emit("online_users", {"workflow_id": workflow_id, "users": users, "leader": leader_sid}, room=workflow_id)
-
-
-@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
-
-    """
-    mapping = redis_client.get(_sid_key(sid))
-
-    if not mapping:
-        return {"msg": "unauthorized"}, 401
-
-    mapping_data = json.loads(mapping)
-    workflow_id = mapping_data["workflow_id"]
-    user_id = mapping_data["user_id"]
-    _refresh_session_state(workflow_id, sid)
-
-    event_type = data.get("type")
-    event_data = data.get("data")
-    timestamp = data.get("timestamp", int(time.time()))
-
-    if not event_type:
-        return {"msg": "invalid event type"}, 400
-
-    sio.emit(
-        "collaboration_update",
-        {"type": event_type, "userId": user_id, "data": event_data, "timestamp": timestamp},
-        room=workflow_id,
-        skip_sid=sid,
-    )
-
-    return {"msg": "event_broadcasted"}
-
-
-@sio.on("graph_event")
-def handle_graph_event(sid, data):
-    """
-    Handle graph events - simple broadcast relay.
-    """
-    mapping = redis_client.get(_sid_key(sid))
-
-    if not mapping:
-        return {"msg": "unauthorized"}, 401
-
-    mapping_data = json.loads(mapping)
-    workflow_id = mapping_data["workflow_id"]
-    _refresh_session_state(workflow_id, sid)
-
-    sio.emit("graph_update", data, room=workflow_id, skip_sid=sid)
-
-    return {"msg": "graph_update_broadcasted"}
@@ -1,7 +1,9 @@
+from datetime import datetime
 from decimal import Decimal

+import pytz
 import sqlalchemy as sa
-from flask import abort, jsonify
+from flask import jsonify
 from flask_restx import Resource, fields, reqparse

 from controllers.console import api, console_ns
@@ -9,7 +11,6 @@ from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
 from core.app.entities.app_invoke_entities import InvokeFrom
 from extensions.ext_database import db
-from libs.datetime_utils import parse_time_range
 from libs.helper import DatetimeString
 from libs.login import current_account_with_tenant, login_required
 from models import AppMode, Message
@@ -55,16 +56,26 @@ WHERE
        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
        assert account.timezone is not None

-        try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
-        except ValueError as e:
-            abort(400, description=str(e))
+        timezone = pytz.timezone(account.timezone)
+        utc_timezone = pytz.utc
+
+        if args["start"]:
+            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
+            start_datetime = start_datetime.replace(second=0)
+
+            start_datetime_timezone = timezone.localize(start_datetime)
+            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)

-        if start_datetime_utc:
            sql_query += " AND created_at >= :start"
            arg_dict["start"] = start_datetime_utc

-        if end_datetime_utc:
+        if args["end"]:
+            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
+            end_datetime = end_datetime.replace(second=0)
+
+            end_datetime_timezone = timezone.localize(end_datetime)
+            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
+
            sql_query += " AND created_at < :end"
            arg_dict["end"] = end_datetime_utc

@@ -109,11 +120,8 @@ class DailyConversationStatistic(Resource):
        )
        args = parser.parse_args()
        assert account.timezone is not None
-
-        try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
-        except ValueError as e:
-            abort(400, description=str(e))
+        timezone = pytz.timezone(account.timezone)
+        utc_timezone = pytz.utc

        stmt = (
            sa.select(
@@ -126,10 +134,18 @@ class DailyConversationStatistic(Resource):
            .where(Message.app_id == app_model.id, Message.invoke_from != InvokeFrom.DEBUGGER)
        )

-        if start_datetime_utc:
+        if args["start"]:
+            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
+            start_datetime = start_datetime.replace(second=0)
+            start_datetime_timezone = timezone.localize(start_datetime)
+            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
            stmt = stmt.where(Message.created_at >= start_datetime_utc)

-        if end_datetime_utc:
+        if args["end"]:
+            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
+            end_datetime = end_datetime.replace(second=0)
+            end_datetime_timezone = timezone.localize(end_datetime)
+            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
            stmt = stmt.where(Message.created_at < end_datetime_utc)

        stmt = stmt.group_by("date").order_by("date")
@@ -182,17 +198,26 @@ WHERE
    AND invoke_from != :invoke_from"""
        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
        assert account.timezone is not None
+        timezone = pytz.timezone(account.timezone)
+        utc_timezone = pytz.utc

-        try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
-        except ValueError as e:
-            abort(400, description=str(e))
+        if args["start"]:
+            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
+            start_datetime = start_datetime.replace(second=0)
+
+            start_datetime_timezone = timezone.localize(start_datetime)
+            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)

-        if start_datetime_utc:
            sql_query += " AND created_at >= :start"
            arg_dict["start"] = start_datetime_utc

-        if end_datetime_utc:
+        if args["end"]:
+            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
+            end_datetime = end_datetime.replace(second=0)
+
+            end_datetime_timezone = timezone.localize(end_datetime)
+            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
+
            sql_query += " AND created_at < :end"
            arg_dict["end"] = end_datetime_utc

@@ -248,17 +273,26 @@ WHERE
    AND invoke_from != :invoke_from"""
        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
        assert account.timezone is not None
+        timezone = pytz.timezone(account.timezone)
+        utc_timezone = pytz.utc

-        try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
-        except ValueError as e:
-            abort(400, description=str(e))
+        if args["start"]:
+            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
+            start_datetime = start_datetime.replace(second=0)
+
+            start_datetime_timezone = timezone.localize(start_datetime)
+            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)

-        if start_datetime_utc:
            sql_query += " AND created_at >= :start"
            arg_dict["start"] = start_datetime_utc

-        if end_datetime_utc:
+        if args["end"]:
+            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
+            end_datetime = end_datetime.replace(second=0)
+
+            end_datetime_timezone = timezone.localize(end_datetime)
+            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
+
            sql_query += " AND created_at < :end"
            arg_dict["end"] = end_datetime_utc

@@ -323,17 +357,26 @@ FROM
            AND m.invoke_from != :invoke_from"""
        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
        assert account.timezone is not None
+        timezone = pytz.timezone(account.timezone)
+        utc_timezone = pytz.utc

-        try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
-        except ValueError as e:
-            abort(400, description=str(e))
+        if args["start"]:
+            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
+            start_datetime = start_datetime.replace(second=0)
+
+            start_datetime_timezone = timezone.localize(start_datetime)
+            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)

-        if start_datetime_utc:
            sql_query += " AND c.created_at >= :start"
            arg_dict["start"] = start_datetime_utc

-        if end_datetime_utc:
+        if args["end"]:
+            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
+            end_datetime = end_datetime.replace(second=0)
+
+            end_datetime_timezone = timezone.localize(end_datetime)
+            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
+
            sql_query += " AND c.created_at < :end"
            arg_dict["end"] = end_datetime_utc

@@ -403,17 +446,26 @@ WHERE
    AND m.invoke_from != :invoke_from"""
        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
        assert account.timezone is not None
+        timezone = pytz.timezone(account.timezone)
+        utc_timezone = pytz.utc

-        try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
-        except ValueError as e:
-            abort(400, description=str(e))
+        if args["start"]:
+            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
+            start_datetime = start_datetime.replace(second=0)
+
+            start_datetime_timezone = timezone.localize(start_datetime)
+            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)

-        if start_datetime_utc:
            sql_query += " AND m.created_at >= :start"
            arg_dict["start"] = start_datetime_utc

-        if end_datetime_utc:
+        if args["end"]:
+            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
+            end_datetime = end_datetime.replace(second=0)
+
+            end_datetime_timezone = timezone.localize(end_datetime)
+            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
+
            sql_query += " AND m.created_at < :end"
            arg_dict["end"] = end_datetime_utc

@@ -473,17 +525,26 @@ WHERE
    AND invoke_from != :invoke_from"""
        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
        assert account.timezone is not None
+        timezone = pytz.timezone(account.timezone)
+        utc_timezone = pytz.utc

-        try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
-        except ValueError as e:
-            abort(400, description=str(e))
+        if args["start"]:
+            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
+            start_datetime = start_datetime.replace(second=0)
+
+            start_datetime_timezone = timezone.localize(start_datetime)
+            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)

-        if start_datetime_utc:
            sql_query += " AND created_at >= :start"
            arg_dict["start"] = start_datetime_utc

-        if end_datetime_utc:
+        if args["end"]:
+            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
+            end_datetime = end_datetime.replace(second=0)
+
+            end_datetime_timezone = timezone.localize(end_datetime)
+            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
+
            sql_query += " AND created_at < :end"
            arg_dict["end"] = end_datetime_utc

@@ -541,17 +602,26 @@ WHERE
    AND invoke_from != :invoke_from"""
        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
        assert account.timezone is not None
+        timezone = pytz.timezone(account.timezone)
+        utc_timezone = pytz.utc

-        try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
-        except ValueError as e:
-            abort(400, description=str(e))
+        if args["start"]:
+            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
+            start_datetime = start_datetime.replace(second=0)
+
+            start_datetime_timezone = timezone.localize(start_datetime)
+            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)

-        if start_datetime_utc:
            sql_query += " AND created_at >= :start"
            arg_dict["start"] = start_datetime_utc

-        if end_datetime_utc:
+        if args["end"]:
+            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
+            end_datetime = end_datetime.replace(second=0)
+
+            end_datetime_timezone = timezone.localize(end_datetime)
+            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
+
            sql_query += " AND created_at < :end"
            arg_dict["end"] = end_datetime_utc

@@ -5,12 +5,10 @@ from typing import cast

 from flask import abort, request
 from flask_restx import Resource, fields, inputs, marshal_with, reqparse
-from pydantic_core import ValidationError
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound

 import services
-from configs import dify_config
 from controllers.console import api, console_ns
 from controllers.console.app.error import ConversationCompletedError, DraftWorkflowNotExist, DraftWorkflowNotSync
 from controllers.console.app.wraps import get_app_model
@@ -23,9 +21,7 @@ from core.file.models import File
 from core.helper.trace_id_helper import get_external_trace_id
 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.online_user_fields import online_user_list_fields
 from fields.workflow_fields import workflow_fields, workflow_pagination_fields
 from fields.workflow_run_fields import workflow_run_node_execution_fields
 from libs import helper
@@ -103,22 +99,10 @@ class DraftWorkflowApi(Resource):
                "hash": fields.String(description="Workflow hash for validation"),
                "environment_variables": fields.List(fields.Raw, required=True, description="Environment variables"),
                "conversation_variables": fields.List(fields.Raw, description="Conversation variables"),
-                "memory_blocks": fields.List(fields.Raw, description="Memory blocks"),
            },
        )
    )
-    @api.response(
-        200,
-        "Draft workflow synced successfully",
-        api.model(
-            "SyncDraftWorkflowResponse",
-            {
-                "result": fields.String,
-                "hash": fields.String,
-                "updated_at": fields.String,
-            },
-        ),
-    )
+    @api.response(200, "Draft workflow synced successfully", workflow_fields)
    @api.response(400, "Invalid workflow configuration")
    @api.response(403, "Permission denied")
    @edit_permission_required
@@ -138,8 +122,6 @@ class DraftWorkflowApi(Resource):
                .add_argument("hash", type=str, required=False, location="json")
                .add_argument("environment_variables", type=list, required=True, location="json")
                .add_argument("conversation_variables", type=list, required=False, location="json")
-                .add_argument("force_upload", type=bool, required=False, default=False, location="json")
-                .add_argument("memory_blocks", type=list, required=False, location="json")
            )
            args = parser.parse_args()
        elif "text/plain" in content_type:
@@ -157,8 +139,6 @@ class DraftWorkflowApi(Resource):
                    "hash": data.get("hash"),
                    "environment_variables": data.get("environment_variables"),
                    "conversation_variables": data.get("conversation_variables"),
-                    "memory_blocks": data.get("memory_blocks"),
-                    "force_upload": data.get("force_upload", False),
                }
            except json.JSONDecodeError:
                return {"message": "Invalid JSON data"}, 400
@@ -175,10 +155,6 @@ class DraftWorkflowApi(Resource):
            conversation_variables = [
                variable_factory.build_conversation_variable_from_mapping(obj) for obj in conversation_variables_list
            ]
-            memory_blocks_list = args.get("memory_blocks") or []
-            from core.memory.entities import MemoryBlockSpec
-
-            memory_blocks = [MemoryBlockSpec.model_validate(obj) for obj in memory_blocks_list]
            workflow = workflow_service.sync_draft_workflow(
                app_model=app_model,
                graph=args["graph"],
@@ -187,13 +163,9 @@ class DraftWorkflowApi(Resource):
                account=current_user,
                environment_variables=environment_variables,
                conversation_variables=conversation_variables,
-                force_upload=args.get("force_upload", False),
-                memory_blocks=memory_blocks,
            )
        except WorkflowHashNotEqualError:
            raise DraftWorkflowNotSync()
-        except ValidationError as e:
-            return {"message": str(e)}, 400

        return {
            "result": "success",
@@ -762,45 +734,6 @@ class ConvertToWorkflowApi(Resource):
        }


-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/config")
-class WorkflowConfigApi(Resource):
-    """Resource for workflow configuration."""
-
-    @api.doc("get_workflow_config")
-    @api.doc(description="Get workflow configuration")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.response(200, "Workflow configuration retrieved successfully")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def get(self, app_model: App):
-        return {
-            "parallel_depth_limit": dify_config.WORKFLOW_PARALLEL_DEPTH_LIMIT,
-        }
-
-
-class WorkflowFeaturesApi(Resource):
-    """Update draft workflow features."""
-
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def post(self, app_model: App):
-        parser = reqparse.RequestParser()
-        parser.add_argument("features", type=dict, required=True, location="json")
-        args = parser.parse_args()
-
-        features = args.get("features")
-
-        # Update draft workflow 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):
    @api.doc("get_all_published_workflows")
@@ -982,105 +915,3 @@ class DraftWorkflowNodeLastRunApi(Resource):
        if node_exec is None:
            raise NotFound("last run not found")
        return node_exec
-
-
-class WorkflowOnlineUsersApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @marshal_with(online_user_list_fields)
-    def get(self):
-        parser = reqparse.RequestParser()
-        parser.add_argument("workflow_ids", type=str, required=True, location="args")
-        args = parser.parse_args()
-
-        workflow_ids = [id.strip() for id in args["workflow_ids"].split(",")]
-
-        results = []
-        for workflow_id in workflow_ids:
-            users_json = redis_client.hgetall(f"workflow_online_users:{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.add_resource(
-    DraftWorkflowApi,
-    "/apps/<uuid:app_id>/workflows/draft",
-)
-api.add_resource(
-    WorkflowConfigApi,
-    "/apps/<uuid:app_id>/workflows/draft/config",
-)
-api.add_resource(
-    WorkflowFeaturesApi,
-    "/apps/<uuid:app_id>/workflows/draft/features",
-)
-api.add_resource(
-    AdvancedChatDraftWorkflowRunApi,
-    "/apps/<uuid:app_id>/advanced-chat/workflows/draft/run",
-)
-api.add_resource(
-    DraftWorkflowRunApi,
-    "/apps/<uuid:app_id>/workflows/draft/run",
-)
-api.add_resource(
-    WorkflowTaskStopApi,
-    "/apps/<uuid:app_id>/workflow-runs/tasks/<string:task_id>/stop",
-)
-api.add_resource(
-    DraftWorkflowNodeRunApi,
-    "/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/run",
-)
-api.add_resource(
-    AdvancedChatDraftRunIterationNodeApi,
-    "/apps/<uuid:app_id>/advanced-chat/workflows/draft/iteration/nodes/<string:node_id>/run",
-)
-api.add_resource(
-    WorkflowDraftRunIterationNodeApi,
-    "/apps/<uuid:app_id>/workflows/draft/iteration/nodes/<string:node_id>/run",
-)
-api.add_resource(
-    AdvancedChatDraftRunLoopNodeApi,
-    "/apps/<uuid:app_id>/advanced-chat/workflows/draft/loop/nodes/<string:node_id>/run",
-)
-api.add_resource(
-    WorkflowDraftRunLoopNodeApi,
-    "/apps/<uuid:app_id>/workflows/draft/loop/nodes/<string:node_id>/run",
-)
-api.add_resource(
-    PublishedWorkflowApi,
-    "/apps/<uuid:app_id>/workflows/publish",
-)
-api.add_resource(
-    PublishedAllWorkflowApi,
-    "/apps/<uuid:app_id>/workflows",
-)
-api.add_resource(
-    DefaultBlockConfigsApi,
-    "/apps/<uuid:app_id>/workflows/default-workflow-block-configs",
-)
-api.add_resource(
-    DefaultBlockConfigApi,
-    "/apps/<uuid:app_id>/workflows/default-workflow-block-configs/<string:block_type>",
-)
-api.add_resource(
-    ConvertToWorkflowApi,
-    "/apps/<uuid:app_id>/convert-to-workflow",
-)
-api.add_resource(
-    WorkflowByIdApi,
-    "/apps/<uuid:app_id>/workflows/<string:workflow_id>",
-)
-api.add_resource(
-    DraftWorkflowNodeLastRunApi,
-    "/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/last-run",
-)
-api.add_resource(WorkflowOnlineUsersApi, "/apps/workflows/online-users")
@@ -1,240 +0,0 @@
-import logging
-
-from flask_restx import Resource, fields, marshal_with, reqparse
-
-from controllers.console import api
-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__)
-
-
-class WorkflowCommentListApi(Resource):
-    """API for listing and creating workflow comments."""
-
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model
-    @marshal_with(workflow_comment_basic_fields, 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
-
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model
-    @marshal_with(workflow_comment_create_fields)
-    def post(self, app_model: App):
-        """Create a new workflow comment."""
-        parser = reqparse.RequestParser()
-        parser.add_argument("position_x", type=float, required=True, location="json")
-        parser.add_argument("position_y", type=float, required=True, location="json")
-        parser.add_argument("content", type=str, required=True, location="json")
-        parser.add_argument("mentioned_user_ids", type=list, location="json", default=[])
-        args = parser.parse_args()
-
-        result = WorkflowCommentService.create_comment(
-            tenant_id=current_user.current_tenant_id,
-            app_id=app_model.id,
-            created_by=current_user.id,
-            content=args.content,
-            position_x=args.position_x,
-            position_y=args.position_y,
-            mentioned_user_ids=args.mentioned_user_ids,
-        )
-
-        return result, 201
-
-
-class WorkflowCommentDetailApi(Resource):
-    """API for managing individual workflow comments."""
-
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model
-    @marshal_with(workflow_comment_detail_fields)
-    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
-
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model
-    @marshal_with(workflow_comment_update_fields)
-    def put(self, app_model: App, comment_id: str):
-        """Update a workflow comment."""
-        parser = reqparse.RequestParser()
-        parser.add_argument("content", type=str, required=True, location="json")
-        parser.add_argument("position_x", type=float, required=False, location="json")
-        parser.add_argument("position_y", type=float, required=False, location="json")
-        parser.add_argument("mentioned_user_ids", type=list, location="json", default=[])
-        args = parser.parse_args()
-
-        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=args.content,
-            position_x=args.position_x,
-            position_y=args.position_y,
-            mentioned_user_ids=args.mentioned_user_ids,
-        )
-
-        return result
-
-    @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
-
-
-class WorkflowCommentResolveApi(Resource):
-    """API for resolving and reopening workflow comments."""
-
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model
-    @marshal_with(workflow_comment_resolve_fields)
-    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
-
-
-class WorkflowCommentReplyApi(Resource):
-    """API for managing comment replies."""
-
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model
-    @marshal_with(workflow_comment_reply_create_fields)
-    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
-        )
-
-        parser = reqparse.RequestParser()
-        parser.add_argument("content", type=str, required=True, location="json")
-        parser.add_argument("mentioned_user_ids", type=list, location="json", default=[])
-        args = parser.parse_args()
-
-        result = WorkflowCommentService.create_reply(
-            comment_id=comment_id,
-            content=args.content,
-            created_by=current_user.id,
-            mentioned_user_ids=args.mentioned_user_ids,
-        )
-
-        return result, 201
-
-
-class WorkflowCommentReplyDetailApi(Resource):
-    """API for managing individual comment replies."""
-
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model
-    @marshal_with(workflow_comment_reply_update_fields)
-    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
-        )
-
-        parser = reqparse.RequestParser()
-        parser.add_argument("content", type=str, required=True, location="json")
-        parser.add_argument("mentioned_user_ids", type=list, location="json", default=[])
-        args = parser.parse_args()
-
-        reply = WorkflowCommentService.update_reply(
-            reply_id=reply_id, user_id=current_user.id, content=args.content, mentioned_user_ids=args.mentioned_user_ids
-        )
-
-        return reply
-
-    @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
-
-
-class WorkflowCommentMentionUsersApi(Resource):
-    """API for getting mentionable users for workflow comments."""
-
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model
-    @marshal_with({"users": fields.List(fields.Nested(account_with_role_fields))})
-    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}
-
-
-# Register API routes
-api.add_resource(WorkflowCommentListApi, "/apps/<uuid:app_id>/workflow/comments")
-api.add_resource(WorkflowCommentDetailApi, "/apps/<uuid:app_id>/workflow/comments/<string:comment_id>")
-api.add_resource(WorkflowCommentResolveApi, "/apps/<uuid:app_id>/workflow/comments/<string:comment_id>/resolve")
-api.add_resource(WorkflowCommentReplyApi, "/apps/<uuid:app_id>/workflow/comments/<string:comment_id>/replies")
-api.add_resource(
-    WorkflowCommentReplyDetailApi, "/apps/<uuid:app_id>/workflow/comments/<string:comment_id>/replies/<string:reply_id>"
-)
-api.add_resource(WorkflowCommentMentionUsersApi, "/apps/<uuid:app_id>/workflow/comments/mention-users")
@@ -19,8 +19,8 @@ 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 factories.variable_factory import build_segment_with_type
 from libs.login import current_user, login_required
 from models import Account, App, AppMode
 from models.workflow import WorkflowDraftVariable
@@ -355,7 +355,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
@@ -448,35 +448,8 @@ class ConversationVariableCollectionApi(Resource):
        db.session.commit()
        return _get_variable_list(app_model, CONVERSATION_VARIABLE_NODE_ID)

-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=AppMode.ADVANCED_CHAT)
-    def post(self, app_model: App):
-        # The role of the current user in the ta table must be admin, owner, or editor
-        if not current_user.is_editor:
-            raise Forbidden()
-
-        parser = reqparse.RequestParser()
-        parser.add_argument("conversation_variables", type=list, required=True, location="json")
-        args = parser.parse_args()
-
-        workflow_service = WorkflowService()
-
-        conversation_variables_list = args.get("conversation_variables") or []
-        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):
    @api.doc("get_system_variables")
    @api.doc(description="Get system variables for workflow")
@@ -526,44 +499,3 @@ class EnvironmentVariableCollectionApi(Resource):
            )

        return {"items": env_vars_list}
-
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def post(self, app_model: App):
-        # The role of the current user in the ta table must be admin, owner, or editor
-        if not current_user.is_editor:
-            raise Forbidden()
-
-        parser = reqparse.RequestParser()
-        parser.add_argument("environment_variables", type=list, required=True, location="json")
-        args = parser.parse_args()
-
-        workflow_service = WorkflowService()
-
-        environment_variables_list = args.get("environment_variables") or []
-        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.add_resource(
-    WorkflowVariableCollectionApi,
-    "/apps/<uuid:app_id>/workflows/draft/variables",
-)
-api.add_resource(NodeVariableCollectionApi, "/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/variables")
-api.add_resource(VariableApi, "/apps/<uuid:app_id>/workflows/draft/variables/<uuid:variable_id>")
-api.add_resource(VariableResetApi, "/apps/<uuid:app_id>/workflows/draft/variables/<uuid:variable_id>/reset")
-
-api.add_resource(ConversationVariableCollectionApi, "/apps/<uuid:app_id>/workflows/draft/conversation-variables")
-api.add_resource(SystemVariableCollectionApi, "/apps/<uuid:app_id>/workflows/draft/system-variables")
-api.add_resource(EnvironmentVariableCollectionApi, "/apps/<uuid:app_id>/workflows/draft/environment-variables")
@@ -1,4 +1,7 @@
-from flask import abort, jsonify
+from datetime import datetime
+
+import pytz
+from flask import jsonify
 from flask_restx import Resource, reqparse
 from sqlalchemy.orm import sessionmaker

@@ -6,7 +9,6 @@ from controllers.console import api, console_ns
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
 from extensions.ext_database import db
-from libs.datetime_utils import parse_time_range
 from libs.helper import DatetimeString
 from libs.login import current_account_with_tenant, login_required
 from models.enums import WorkflowRunTriggeredFrom
@@ -41,11 +43,23 @@ class WorkflowDailyRunsStatistic(Resource):
        args = parser.parse_args()

        assert account.timezone is not None
+        timezone = pytz.timezone(account.timezone)
+        utc_timezone = pytz.utc

-        try:
-            start_date, end_date = parse_time_range(args["start"], args["end"], account.timezone)
-        except ValueError as e:
-            abort(400, description=str(e))
+        start_date = None
+        end_date = None
+
+        if args["start"]:
+            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
+            start_datetime = start_datetime.replace(second=0)
+            start_datetime_timezone = timezone.localize(start_datetime)
+            start_date = start_datetime_timezone.astimezone(utc_timezone)
+
+        if args["end"]:
+            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
+            end_datetime = end_datetime.replace(second=0)
+            end_datetime_timezone = timezone.localize(end_datetime)
+            end_date = end_datetime_timezone.astimezone(utc_timezone)

        response_data = self._workflow_run_repo.get_daily_runs_statistics(
            tenant_id=app_model.tenant_id,
@@ -86,11 +100,23 @@ class WorkflowDailyTerminalsStatistic(Resource):
        args = parser.parse_args()

        assert account.timezone is not None
+        timezone = pytz.timezone(account.timezone)
+        utc_timezone = pytz.utc

-        try:
-            start_date, end_date = parse_time_range(args["start"], args["end"], account.timezone)
-        except ValueError as e:
-            abort(400, description=str(e))
+        start_date = None
+        end_date = None
+
+        if args["start"]:
+            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
+            start_datetime = start_datetime.replace(second=0)
+            start_datetime_timezone = timezone.localize(start_datetime)
+            start_date = start_datetime_timezone.astimezone(utc_timezone)
+
+        if args["end"]:
+            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
+            end_datetime = end_datetime.replace(second=0)
+            end_datetime_timezone = timezone.localize(end_datetime)
+            end_date = end_datetime_timezone.astimezone(utc_timezone)

        response_data = self._workflow_run_repo.get_daily_terminals_statistics(
            tenant_id=app_model.tenant_id,
@@ -131,11 +157,23 @@ class WorkflowDailyTokenCostStatistic(Resource):
        args = parser.parse_args()

        assert account.timezone is not None
+        timezone = pytz.timezone(account.timezone)
+        utc_timezone = pytz.utc

-        try:
-            start_date, end_date = parse_time_range(args["start"], args["end"], account.timezone)
-        except ValueError as e:
-            abort(400, description=str(e))
+        start_date = None
+        end_date = None
+
+        if args["start"]:
+            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
+            start_datetime = start_datetime.replace(second=0)
+            start_datetime_timezone = timezone.localize(start_datetime)
+            start_date = start_datetime_timezone.astimezone(utc_timezone)
+
+        if args["end"]:
+            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
+            end_datetime = end_datetime.replace(second=0)
+            end_datetime_timezone = timezone.localize(end_datetime)
+            end_date = end_datetime_timezone.astimezone(utc_timezone)

        response_data = self._workflow_run_repo.get_daily_token_cost_statistics(
            tenant_id=app_model.tenant_id,
@@ -176,11 +214,23 @@ class WorkflowAverageAppInteractionStatistic(Resource):
        args = parser.parse_args()

        assert account.timezone is not None
+        timezone = pytz.timezone(account.timezone)
+        utc_timezone = pytz.utc

-        try:
-            start_date, end_date = parse_time_range(args["start"], args["end"], account.timezone)
-        except ValueError as e:
-            abort(400, description=str(e))
+        start_date = None
+        end_date = None
+
+        if args["start"]:
+            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
+            start_datetime = start_datetime.replace(second=0)
+            start_datetime_timezone = timezone.localize(start_datetime)
+            start_date = start_datetime_timezone.astimezone(utc_timezone)
+
+        if args["end"]:
+            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
+            end_datetime = end_datetime.replace(second=0)
+            end_datetime_timezone = timezone.localize(end_datetime)
+            end_date = end_datetime_timezone.astimezone(utc_timezone)

        response_data = self._workflow_run_repo.get_average_app_interaction_statistics(
            tenant_id=app_model.tenant_id,
@@ -23,15 +23,6 @@ def _load_app_model(app_id: str) -> App | None:
    return app_model


-def _load_app_model_with_trial(app_id: str) -> App | None:
-    app_model = (
-        db.session.query(App)
-        .where(App.id == app_id, App.status == "normal")
-        .first()
-    )
-    return app_model
-
-
 def get_app_model(view: Callable[P, R] | None = None, *, mode: Union[AppMode, list[AppMode], None] = None):
    def decorator(view_func: Callable[P1, R1]):
        @wraps(view_func)
@@ -71,44 +62,3 @@ def get_app_model(view: Callable[P, R] | None = None, *, mode: Union[AppMode, li
        return decorator
    else:
        return decorator(view)
-
-
-def get_app_model_with_trial(view: Callable[P, R] | None = None, *, mode: Union[AppMode, list[AppMode], None] = None):
-    def decorator(view_func: Callable[P, R]):
-        @wraps(view_func)
-        def decorated_view(*args: P.args, **kwargs: P.kwargs):
-            if not kwargs.get("app_id"):
-                raise ValueError("missing app_id in path parameters")
-
-            app_id = kwargs.get("app_id")
-            app_id = str(app_id)
-
-            del kwargs["app_id"]
-
-            app_model = _load_app_model_with_trial(app_id)
-
-            if not app_model:
-                raise AppNotFoundError()
-
-            app_mode = AppMode.value_of(app_model.mode)
-
-            if mode is not None:
-                if isinstance(mode, list):
-                    modes = mode
-                else:
-                    modes = [mode]
-
-                if app_mode not in modes:
-                    mode_values = {m.value for m in modes}
-                    raise AppNotFoundError(f"App mode is not in the supported list: {mode_values}")
-
-            kwargs["app_model"] = app_model
-
-            return view_func(*args, **kwargs)
-
-        return decorated_view
-
-    if view is None:
-        return decorator
-    else:
-        return decorator(view)
@@ -25,13 +25,10 @@ from controllers.console.wraps import email_password_login_enabled, setup_requir
 from events.tenant_event import tenant_was_created
 from libs.helper import email, extract_remote_ip
 from libs.login import current_account_with_tenant
-from libs.passport import PassportService
 from libs.token import (
-    check_csrf_token,
    clear_access_token_from_cookie,
    clear_csrf_token_from_cookie,
    clear_refresh_token_from_cookie,
-    extract_access_token,
    extract_refresh_token,
    set_access_token_to_cookie,
    set_csrf_token_to_cookie,
@@ -292,18 +289,3 @@ class RefreshTokenApi(Resource):
            return response
        except Exception as e:
            return {"result": "fail", "message": str(e)}, 401
-
-
-# this api helps frontend to check whether user is authenticated
-# TODO: remove in the future. frontend should redirect to login page by catching 401 status
-@console_ns.route("/login/status")
-class LoginStatus(Resource):
-    def get(self):
-        token = extract_access_token(request)
-        res = True
-        try:
-            validated = PassportService().verify(token=token)
-            check_csrf_token(request=request, user_id=validated.get("user_id"))
-        except:
-            res = False
-        return {"logged_in": res}
@@ -2,7 +2,6 @@ from flask_restx import Resource, reqparse

 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, only_edition_cloud, setup_required
-from enums.cloud_plan import CloudPlan
 from libs.login import current_account_with_tenant, login_required
 from services.billing_service import BillingService

@@ -17,13 +16,7 @@ class Subscription(Resource):
        current_user, current_tenant_id = current_account_with_tenant()
        parser = (
            reqparse.RequestParser()
-            .add_argument(
-                "plan",
-                type=str,
-                required=True,
-                location="args",
-                choices=[CloudPlan.PROFESSIONAL, CloudPlan.TEAM],
-            )
+            .add_argument("plan", type=str, required=True, location="args", choices=["professional", "team"])
            .add_argument("interval", type=str, required=True, location="args", choices=["month", "year"])
        )
        args = parser.parse_args()
@@ -746,7 +746,7 @@ class DocumentApi(DocumentResource):
                "name": document.name,
                "created_from": document.created_from,
                "created_by": document.created_by,
-                "created_at": int(document.created_at.timestamp()),
+                "created_at": document.created_at.timestamp(),
                "tokens": document.tokens,
                "indexing_status": document.indexing_status,
                "completed_at": int(document.completed_at.timestamp()) if document.completed_at else None,
@@ -779,7 +779,7 @@ class DocumentApi(DocumentResource):
                "name": document.name,
                "created_from": document.created_from,
                "created_by": document.created_by,
-                "created_at": int(document.created_at.timestamp()),
+                "created_at": document.created_at.timestamp(),
                "tokens": document.tokens,
                "indexing_status": document.indexing_status,
                "completed_at": int(document.completed_at.timestamp()) if document.completed_at else None,
@@ -1,43 +0,0 @@
-from flask import request
-from flask_restx import Resource
-
-from controllers.console import api
-from controllers.console.explore.wraps import explore_banner_enabled
-from extensions.ext_database import db
-from models.model import ExporleBanner
-
-
-class BannerApi(Resource):
-    """Resource for banner list."""
-
-    @explore_banner_enabled
-    def get(self):
-        """Get banner list."""
-        language = request.args.get("language", "en-US")
-
-        # Build base query for enabled banners
-        base_query = db.session.query(ExporleBanner).where(ExporleBanner.status == "enabled")
-        
-        # Try to get banners in the requested language
-        banners = base_query.where(ExporleBanner.language == language).order_by(ExporleBanner.sort).all()
-        
-        # Fallback to en-US if no banners found and language is not en-US
-        if not banners and language != "en-US":
-            banners = base_query.where(ExporleBanner.language == "en-US").order_by(ExporleBanner.sort).all()
-        # Convert banners to serializable format
-        result = []
-        for banner in banners:
-            banner_data = {
-                "id": banner.id,
-                "content": banner.content,  # Already parsed as JSON by SQLAlchemy
-                "link": banner.link,
-                "sort": banner.sort,
-                "status": banner.status,
-                "created_at": banner.created_at.isoformat() if banner.created_at else None,
-            }
-            result.append(banner_data)
-
-        return result
-
-
-api.add_resource(BannerApi, "/explore/banners")
@@ -29,25 +29,3 @@ class AppAccessDeniedError(BaseHTTPException):
    error_code = "access_denied"
    description = "App access denied."
    code = 403
-
-
-class TrialAppNotAllowed(BaseHTTPException):
-    """*403* `Trial App Not Allowed`
-
-    Raise if the user has reached the trial app limit.
-    """
-
-    error_code = "trial_app_not_allowed"
-    code = 403
-    description = "the app is not allowed to be trial."
-
-
-class TrialAppLimitExceeded(BaseHTTPException):
-    """*403* `Trial App Limit Exceeded`
-
-    Raise if the user has exceeded the trial app limit.
-    """
-
-    error_code = "trial_app_limit_exceeded"
-    code = 403
-    description = "The user has exceeded the trial app limit."
@@ -27,7 +27,6 @@ recommended_app_fields = {
    "category": fields.String,
    "position": fields.Integer,
    "is_listed": fields.Boolean,
-    "can_trial": fields.Boolean,
 }

 recommended_app_list_fields = {
@@ -1,514 +0,0 @@
-import logging
-from typing import Any, cast
-
-from flask import request
-from flask_restx import Resource, marshal, marshal_with, reqparse
-from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
-
-import services
-from controllers.common import fields
-from controllers.common.fields import build_site_model
-from controllers.console import api
-from controllers.console.app.error import (
-    AppUnavailableError,
-    AudioTooLargeError,
-    CompletionRequestError,
-    ConversationCompletedError,
-    NeedAddIdsError,
-    NoAudioUploadedError,
-    ProviderModelCurrentlyNotSupportError,
-    ProviderNotInitializeError,
-    ProviderNotSupportSpeechToTextError,
-    ProviderQuotaExceededError,
-    UnsupportedAudioTypeError,
-)
-from controllers.console.app.wraps import get_app_model_with_trial
-from controllers.console.explore.error import (
-    AppSuggestedQuestionsAfterAnswerDisabledError,
-    NotChatAppError,
-    NotCompletionAppError,
-    NotWorkflowAppError,
-)
-from controllers.console.explore.wraps import TrialAppResource, trial_feature_enable
-from controllers.service_api import service_api_ns
-from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
-from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
-from core.app.apps.base_app_queue_manager import AppQueueManager
-from core.app.entities.app_invoke_entities import InvokeFrom
-from core.errors.error import (
-    ModelCurrentlyNotSupportError,
-    ProviderTokenNotInitError,
-    QuotaExceededError,
-)
-from core.model_runtime.errors.invoke import InvokeError
-from core.workflow.graph_engine.manager import GraphEngineManager
-from extensions.ext_database import db
-from fields.app_fields import app_detail_fields_with_site
-from fields.dataset_fields import dataset_fields
-from fields.workflow_fields import workflow_fields
-from libs import helper
-from libs.helper import uuid_value
-from libs.login import current_user
-from models import Account
-from models.account import TenantStatus
-from models.model import AppMode, Site
-from models.workflow import Workflow
-from services.app_generate_service import AppGenerateService
-from services.app_service import AppService
-from services.audio_service import AudioService
-from services.dataset_service import DatasetService
-from services.errors.audio import (
-    AudioTooLargeServiceError,
-    NoAudioUploadedServiceError,
-    ProviderNotSupportSpeechToTextServiceError,
-    UnsupportedAudioTypeServiceError,
-)
-from services.errors.conversation import ConversationNotExistsError
-from services.errors.llm import InvokeRateLimitError
-from services.errors.message import (
-    MessageNotExistsError,
-    SuggestedQuestionsAfterAnswerDisabledError,
-)
-from services.message_service import MessageService
-from services.recommended_app_service import RecommendedAppService
-
-logger = logging.getLogger(__name__)
-
-
-class TrialAppWorkflowRunApi(TrialAppResource):
-    def post(self, trial_app):
-        """
-        Run workflow
-        """
-        app_model = trial_app
-        if not app_model:
-            raise NotWorkflowAppError()
-        app_mode = AppMode.value_of(app_model.mode)
-        if app_mode != AppMode.WORKFLOW:
-            raise NotWorkflowAppError()
-
-        parser = reqparse.RequestParser()
-        parser.add_argument("inputs", type=dict, required=True, nullable=False, location="json")
-        parser.add_argument("files", type=list, required=False, location="json")
-        args = parser.parse_args()
-        assert current_user is not None
-        try:
-            app_id = app_model.id
-            user_id = current_user.id
-            response = AppGenerateService.generate(
-                app_model=app_model, user=current_user, args=args, invoke_from=InvokeFrom.EXPLORE, streaming=True
-            )
-            RecommendedAppService.add_trial_app_record(app_id, user_id)
-            return helper.compact_generate_response(response)
-        except ProviderTokenNotInitError as ex:
-            raise ProviderNotInitializeError(ex.description)
-        except QuotaExceededError:
-            raise ProviderQuotaExceededError()
-        except ModelCurrentlyNotSupportError:
-            raise ProviderModelCurrentlyNotSupportError()
-        except InvokeError as e:
-            raise CompletionRequestError(e.description)
-        except InvokeRateLimitError as ex:
-            raise InvokeRateLimitHttpError(ex.description)
-        except ValueError as e:
-            raise e
-        except Exception:
-            logger.exception("internal server error.")
-            raise InternalServerError()
-
-
-class TrialAppWorkflowTaskStopApi(TrialAppResource):
-    def post(self, trial_app, task_id: str):
-        """
-        Stop workflow task
-        """
-        app_model = trial_app
-        if not app_model:
-            raise NotWorkflowAppError()
-        app_mode = AppMode.value_of(app_model.mode)
-        if app_mode != AppMode.WORKFLOW:
-            raise NotWorkflowAppError()
-        assert current_user is not None
-
-        # Stop using both mechanisms for backward compatibility
-        # Legacy stop flag mechanism (without user check)
-        AppQueueManager.set_stop_flag_no_user_check(task_id)
-
-        # New graph engine command channel mechanism
-        GraphEngineManager.send_stop_command(task_id)
-
-        return {"result": "success"}
-
-
-class TrialChatApi(TrialAppResource):
-    @trial_feature_enable
-    def post(self, trial_app):
-        app_model = trial_app
-        app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
-            raise NotChatAppError()
-
-        parser = reqparse.RequestParser()
-        parser.add_argument("inputs", type=dict, required=True, location="json")
-        parser.add_argument("query", type=str, required=True, location="json")
-        parser.add_argument("files", type=list, required=False, location="json")
-        parser.add_argument("conversation_id", type=uuid_value, location="json")
-        parser.add_argument("parent_message_id", type=uuid_value, required=False, location="json")
-        parser.add_argument("retriever_from", type=str, required=False, default="explore_app", location="json")
-        args = parser.parse_args()
-
-        args["auto_generate_name"] = False
-
-        try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account instance")
-            
-            # Get IDs before they might be detached from session
-            app_id = app_model.id
-            user_id = current_user.id
-            
-            response = AppGenerateService.generate(
-                app_model=app_model, user=current_user, args=args, invoke_from=InvokeFrom.EXPLORE, streaming=True
-            )
-            RecommendedAppService.add_trial_app_record(app_id, user_id)
-            return helper.compact_generate_response(response)
-        except services.errors.conversation.ConversationNotExistsError:
-            raise NotFound("Conversation Not Exists.")
-        except services.errors.conversation.ConversationCompletedError:
-            raise ConversationCompletedError()
-        except services.errors.app_model_config.AppModelConfigBrokenError:
-            logger.exception("App model config broken.")
-            raise AppUnavailableError()
-        except ProviderTokenNotInitError as ex:
-            raise ProviderNotInitializeError(ex.description)
-        except QuotaExceededError:
-            raise ProviderQuotaExceededError()
-        except ModelCurrentlyNotSupportError:
-            raise ProviderModelCurrentlyNotSupportError()
-        except InvokeError as e:
-            raise CompletionRequestError(e.description)
-        except InvokeRateLimitError as ex:
-            raise InvokeRateLimitHttpError(ex.description)
-        except ValueError as e:
-            raise e
-        except Exception:
-            logger.exception("internal server error.")
-            raise InternalServerError()
-
-
-class TrialMessageSuggestedQuestionApi(TrialAppResource):
-    @trial_feature_enable
-    def get(self, trial_app, message_id):
-        app_model = trial_app
-        app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
-            raise NotChatAppError()
-
-        message_id = str(message_id)
-
-        try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account instance")
-            questions = MessageService.get_suggested_questions_after_answer(
-                app_model=app_model, user=current_user, message_id=message_id, invoke_from=InvokeFrom.EXPLORE
-            )
-        except MessageNotExistsError:
-            raise NotFound("Message not found")
-        except ConversationNotExistsError:
-            raise NotFound("Conversation not found")
-        except SuggestedQuestionsAfterAnswerDisabledError:
-            raise AppSuggestedQuestionsAfterAnswerDisabledError()
-        except ProviderTokenNotInitError as ex:
-            raise ProviderNotInitializeError(ex.description)
-        except QuotaExceededError:
-            raise ProviderQuotaExceededError()
-        except ModelCurrentlyNotSupportError:
-            raise ProviderModelCurrentlyNotSupportError()
-        except InvokeError as e:
-            raise CompletionRequestError(e.description)
-        except Exception:
-            logger.exception("internal server error.")
-            raise InternalServerError()
-
-        return {"data": questions}
-
-
-class TrialChatAudioApi(TrialAppResource):
-    @trial_feature_enable
-    def post(self, trial_app):
-        app_model = trial_app
-
-        file = request.files["file"]
-
-        try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account instance")
-            
-            # Get IDs before they might be detached from session
-            app_id = app_model.id
-            user_id = current_user.id
-            
-            response = AudioService.transcript_asr(app_model=app_model, file=file, end_user=None)
-            RecommendedAppService.add_trial_app_record(app_id, user_id)
-            return response
-        except services.errors.app_model_config.AppModelConfigBrokenError:
-            logger.exception("App model config broken.")
-            raise AppUnavailableError()
-        except NoAudioUploadedServiceError:
-            raise NoAudioUploadedError()
-        except AudioTooLargeServiceError as e:
-            raise AudioTooLargeError(str(e))
-        except UnsupportedAudioTypeServiceError:
-            raise UnsupportedAudioTypeError()
-        except ProviderNotSupportSpeechToTextServiceError:
-            raise ProviderNotSupportSpeechToTextError()
-        except ProviderTokenNotInitError as ex:
-            raise ProviderNotInitializeError(ex.description)
-        except QuotaExceededError:
-            raise ProviderQuotaExceededError()
-        except ModelCurrentlyNotSupportError:
-            raise ProviderModelCurrentlyNotSupportError()
-        except InvokeError as e:
-            raise CompletionRequestError(e.description)
-        except ValueError as e:
-            raise e
-        except Exception as e:
-            logger.exception("internal server error.")
-            raise InternalServerError()
-
-
-class TrialChatTextApi(TrialAppResource):
-    @trial_feature_enable
-    def post(self, trial_app):
-        app_model = trial_app
-        try:
-            parser = reqparse.RequestParser()
-            parser.add_argument("message_id", type=str, required=False, location="json")
-            parser.add_argument("voice", type=str, location="json")
-            parser.add_argument("text", type=str, location="json")
-            parser.add_argument("streaming", type=bool, location="json")
-            args = parser.parse_args()
-
-            message_id = args.get("message_id", None)
-            text = args.get("text", None)
-            voice = args.get("voice", None)
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account instance")
-            
-            # Get IDs before they might be detached from session
-            app_id = app_model.id
-            user_id = current_user.id
-            
-            response = AudioService.transcript_tts(app_model=app_model, text=text, voice=voice, message_id=message_id)
-            RecommendedAppService.add_trial_app_record(app_id, user_id)
-            return response
-        except services.errors.app_model_config.AppModelConfigBrokenError:
-            logger.exception("App model config broken.")
-            raise AppUnavailableError()
-        except NoAudioUploadedServiceError:
-            raise NoAudioUploadedError()
-        except AudioTooLargeServiceError as e:
-            raise AudioTooLargeError(str(e))
-        except UnsupportedAudioTypeServiceError:
-            raise UnsupportedAudioTypeError()
-        except ProviderNotSupportSpeechToTextServiceError:
-            raise ProviderNotSupportSpeechToTextError()
-        except ProviderTokenNotInitError as ex:
-            raise ProviderNotInitializeError(ex.description)
-        except QuotaExceededError:
-            raise ProviderQuotaExceededError()
-        except ModelCurrentlyNotSupportError:
-            raise ProviderModelCurrentlyNotSupportError()
-        except InvokeError as e:
-            raise CompletionRequestError(e.description)
-        except ValueError as e:
-            raise e
-        except Exception as e:
-            logger.exception("internal server error.")
-            raise InternalServerError()
-
-
-class TrialCompletionApi(TrialAppResource):
-    @trial_feature_enable
-    def post(self, trial_app):
-        app_model = trial_app
-        if app_model.mode != "completion":
-            raise NotCompletionAppError()
-
-        parser = reqparse.RequestParser()
-        parser.add_argument("inputs", type=dict, required=True, location="json")
-        parser.add_argument("query", type=str, location="json", default="")
-        parser.add_argument("files", type=list, required=False, location="json")
-        parser.add_argument("response_mode", type=str, choices=["blocking", "streaming"], location="json")
-        parser.add_argument("retriever_from", type=str, required=False, default="explore_app", location="json")
-        args = parser.parse_args()
-
-        streaming = args["response_mode"] == "streaming"
-        args["auto_generate_name"] = False
-
-        try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account instance")
-            
-            # Get IDs before they might be detached from session
-            app_id = app_model.id
-            user_id = current_user.id
-            
-            response = AppGenerateService.generate(
-                app_model=app_model, user=current_user, args=args, invoke_from=InvokeFrom.EXPLORE, streaming=streaming
-            )
-
-            RecommendedAppService.add_trial_app_record(app_id, user_id)
-            return helper.compact_generate_response(response)
-        except services.errors.conversation.ConversationNotExistsError:
-            raise NotFound("Conversation Not Exists.")
-        except services.errors.conversation.ConversationCompletedError:
-            raise ConversationCompletedError()
-        except services.errors.app_model_config.AppModelConfigBrokenError:
-            logger.exception("App model config broken.")
-            raise AppUnavailableError()
-        except ProviderTokenNotInitError as ex:
-            raise ProviderNotInitializeError(ex.description)
-        except QuotaExceededError:
-            raise ProviderQuotaExceededError()
-        except ModelCurrentlyNotSupportError:
-            raise ProviderModelCurrentlyNotSupportError()
-        except InvokeError as e:
-            raise CompletionRequestError(e.description)
-        except ValueError as e:
-            raise e
-        except Exception:
-            logger.exception("internal server error.")
-            raise InternalServerError()
-
-
-class TrialSitApi(Resource):
-    """Resource for trial app sites."""
-
-    @trial_feature_enable
-    @get_app_model_with_trial
-    @service_api_ns.marshal_with(build_site_model(service_api_ns))
-    def get(self, app_model):
-        """Retrieve app site info.
-
-        Returns the site configuration for the application including theme, icons, and text.
-        """
-        site = db.session.query(Site).where(Site.app_id == app_model.id).first()
-
-        if not site:
-            raise Forbidden()
-
-        assert app_model.tenant
-        if app_model.tenant.status == TenantStatus.ARCHIVE:
-            raise Forbidden()
-
-        return site
-
-
-class TrialAppParameterApi(Resource):
-    """Resource for app variables."""
-
-    @trial_feature_enable
-    @get_app_model_with_trial
-    @marshal_with(fields.parameters_fields)
-    def get(self, app_model):
-        """Retrieve app parameters."""
-
-        if app_model is None:
-            raise AppUnavailableError()
-
-        if app_model.mode in {AppMode.ADVANCED_CHAT, AppMode.WORKFLOW}:
-            workflow = app_model.workflow
-            if workflow is None:
-                raise AppUnavailableError()
-
-            features_dict = workflow.features_dict
-            user_input_form = workflow.user_input_form(to_old_structure=True)
-        else:
-            app_model_config = app_model.app_model_config
-            if app_model_config is None:
-                raise AppUnavailableError()
-
-            features_dict = app_model_config.to_dict()
-
-            user_input_form = features_dict.get("user_input_form", [])
-
-        return get_parameters_from_feature_dict(features_dict=features_dict, user_input_form=user_input_form)
-
-
-class AppApi(Resource):
-    @trial_feature_enable
-    @get_app_model_with_trial
-    @marshal_with(app_detail_fields_with_site)
-    def get(self, app_model):
-        """Get app detail"""
-
-        app_service = AppService()
-        app_model = app_service.get_app(app_model)
-
-        return app_model
-
-
-class AppWorkflowApi(Resource):
-    @trial_feature_enable
-    @get_app_model_with_trial
-    @marshal_with(workflow_fields)
-    def get(self, app_model):
-        """Get workflow detail"""
-        if not app_model.workflow_id:
-            raise AppUnavailableError()
-
-        workflow = (
-            db.session.query(Workflow)
-            .where(
-                Workflow.id == app_model.workflow_id,
-            )
-            .first()
-        )
-        return workflow
-
-
-class DatasetListApi(Resource):
-    @trial_feature_enable
-    @get_app_model_with_trial
-    def get(self, app_model):
-        page = request.args.get("page", default=1, type=int)
-        limit = request.args.get("limit", default=20, type=int)
-        ids = request.args.getlist("ids")
-      
-        tenant_id = app_model.tenant_id
-        if ids:
-            datasets, total = DatasetService.get_datasets_by_ids(ids, tenant_id)
-        else:
-            raise NeedAddIdsError()
-
-        data = cast(list[dict[str, Any]], marshal(datasets, dataset_fields))
-
-        response = {"data": data, "has_more": len(datasets) == limit, "limit": limit, "total": total, "page": page}
-        return response
-
-
-api.add_resource(TrialChatApi, "/trial-apps/<uuid:app_id>/chat-messages", endpoint="trial_app_chat_completion")
-
-api.add_resource(
-    TrialMessageSuggestedQuestionApi,
-    "/trial-apps/<uuid:app_id>/messages/<uuid:message_id>/suggested-questions",
-    endpoint="trial_app_suggested_question",
-)
-
-api.add_resource(TrialChatAudioApi, "/trial-apps/<uuid:app_id>/audio-to-text", endpoint="trial_app_audio")
-api.add_resource(TrialChatTextApi, "/trial-apps/<uuid:app_id>/text-to-audio", endpoint="trial_app_text")
-
-api.add_resource(TrialCompletionApi, "/trial-apps/<uuid:app_id>/completion-messages", endpoint="trial_app_completion")
-
-api.add_resource(TrialSitApi, "/trial-apps/<uuid:app_id>/site")
-
-api.add_resource(TrialAppParameterApi, "/trial-apps/<uuid:app_id>/parameters", endpoint="trial_app_parameters")
-
-api.add_resource(AppApi, "/trial-apps/<uuid:app_id>", endpoint="trial_app")
-
-api.add_resource(TrialAppWorkflowRunApi, "/trial-apps/<uuid:app_id>/workflows/run", endpoint="trial_app_workflow_run")
-api.add_resource(TrialAppWorkflowTaskStopApi, "/trial-apps/<uuid:app_id>/workflows/tasks/<string:task_id>/stop")
-
-api.add_resource(AppWorkflowApi, "/trial-apps/<uuid:app_id>/workflows", endpoint="trial_app_workflow")
-api.add_resource(DatasetListApi, "/trial-apps/<uuid:app_id>/datasets", endpoint="trial_app_datasets")
@@ -2,16 +2,14 @@ from collections.abc import Callable
 from functools import wraps
 from typing import Concatenate, ParamSpec, TypeVar

-from flask import abort
-from flask_login import current_user
 from flask_restx import Resource
 from werkzeug.exceptions import NotFound

-from controllers.console.explore.error import AppAccessDeniedError, TrialAppLimitExceeded, TrialAppNotAllowed
+from controllers.console.explore.error import AppAccessDeniedError
 from controllers.console.wraps import account_initialization_required
 from extensions.ext_database import db
 from libs.login import current_account_with_tenant, login_required
-from models import AccountTrialAppRecord, App, InstalledApp, TrialApp
+from models import InstalledApp
 from services.enterprise.enterprise_service import EnterpriseService
 from services.feature_service import FeatureService

@@ -73,59 +71,6 @@ def user_allowed_to_access_app(view: Callable[Concatenate[InstalledApp, P], R] |
    return decorator


-def trial_app_required(view: Callable[Concatenate[App, P], R] | None = None):
-    def decorator(view: Callable[Concatenate[App, P], R]):
-        @wraps(view)
-        def decorated(app_id: str, *args: P.args, **kwargs: P.kwargs):
-            trial_app = db.session.query(TrialApp).where(TrialApp.app_id == str(app_id)).first()
-
-            if trial_app is None:
-                raise TrialAppNotAllowed()
-            app = trial_app.app
-
-            if app is None:
-                raise TrialAppNotAllowed()
-
-            account_trial_app_record = (
-                db.session.query(AccountTrialAppRecord)
-                .where(AccountTrialAppRecord.account_id == current_user.id, AccountTrialAppRecord.app_id == app_id)
-                .first()
-            )
-            if account_trial_app_record:
-                if account_trial_app_record.count >= trial_app.trial_limit:
-                    raise TrialAppLimitExceeded()
-
-            return view(app, *args, **kwargs)
-
-        return decorated
-
-    if view:
-        return decorator(view)
-    return decorator
-
-
-def trial_feature_enable(view: Callable[..., R]) -> Callable[..., R]:
-    @wraps(view)
-    def decorated(*args, **kwargs):
-        features = FeatureService.get_system_features()
-        if not features.enable_trial_app:
-            abort(403, "Trial app feature is not enabled.")
-        return view(*args, **kwargs)
-
-    return decorated
-
-
-def explore_banner_enabled(view: Callable[..., R]) -> Callable[..., R]:
-    @wraps(view)
-    def decorated(*args, **kwargs):
-        features = FeatureService.get_system_features()
-        if not features.enable_explore_banner:
-            abort(403, "Explore banner feature is not enabled.")
-        return view(*args, **kwargs)
-
-    return decorated
-
-
 class InstalledAppResource(Resource):
    # must be reversed if there are multiple decorators

@@ -135,13 +80,3 @@ class InstalledAppResource(Resource):
        account_initialization_required,
        login_required,
    ]
-
-
-class TrialAppResource(Resource):
-    # must be reversed if there are multiple decorators
-
-    method_decorators = [
-        trial_app_required,
-        account_initialization_required,
-        login_required,
-    ]
@@ -8,7 +8,6 @@ import services
 from configs import dify_config
 from constants import DOCUMENT_EXTENSIONS
 from controllers.common.errors import (
-    BlockedFileExtensionError,
    FilenameNotExistsError,
    FileTooLargeError,
    NoFileUploadedError,
@@ -84,8 +83,6 @@ class FileApi(Resource):
            raise FileTooLargeError(file_too_large_error.description)
        except services.errors.file.UnsupportedFileTypeError:
            raise UnsupportedFileTypeError()
-        except services.errors.file.BlockedFileExtensionError as blocked_extension_error:
-            raise BlockedFileExtensionError(blocked_extension_error.description)

        return upload_file, 201

@@ -32,7 +32,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
@@ -129,17 +128,6 @@ class AccountNameApi(Resource):

@console_ns.route("/account/avatar")
 class AccountAvatarApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def get(self):
-        parser = reqparse.RequestParser()
-        parser.add_argument("avatar", type=str, required=True, location="args")
-        args = parser.parse_args()
-
-        avatar_url = file_helpers.get_signed_file_url(args["avatar"])
-        return {"avatar_url": avatar_url}
-
    @setup_required
    @login_required
    @account_initialization_required
@@ -21,7 +21,6 @@ from controllers.console.wraps import (
    cloud_edition_billing_resource_check,
    setup_required,
 )
-from enums.cloud_plan import CloudPlan
 from extensions.ext_database import db
 from libs.helper import TimestampField
 from libs.login import current_account_with_tenant, login_required
@@ -51,9 +50,6 @@ tenant_fields = {
    "in_trial": fields.Boolean,
    "trial_end_reason": fields.String,
    "custom_config": fields.Raw(attribute="custom_config"),
-    "trial_credits": fields.Integer,
-    "trial_credits_used": fields.Integer,
-    "next_credit_reset_date": fields.Integer,
 }

 tenants_fields = {
@@ -87,7 +83,7 @@ class TenantListApi(Resource):
                "name": tenant.name,
                "status": tenant.status,
                "created_at": tenant.created_at,
-                "plan": features.billing.subscription.plan if features.billing.enabled else CloudPlan.SANDBOX,
+                "plan": features.billing.subscription.plan if features.billing.enabled else "sandbox",
                "current": tenant.id == current_tenant_id if current_tenant_id else False,
            }

@@ -10,7 +10,6 @@ from flask import abort, request

 from configs import dify_config
 from controllers.console.workspace.error import AccountNotInitializedError
-from enums.cloud_plan import CloudPlan
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from libs.login import current_account_with_tenant
@@ -134,7 +133,7 @@ def cloud_edition_billing_knowledge_limit_check(resource: str):
            features = FeatureService.get_features(current_tenant_id)
            if features.billing.enabled:
                if resource == "add_segment":
-                    if features.billing.subscription.plan == CloudPlan.SANDBOX:
+                    if features.billing.subscription.plan == "sandbox":
                        abort(
                            403,
                            "To unlock this feature and elevate your Dify experience, please upgrade to a paid plan.",
@@ -19,7 +19,6 @@ from .app import (
    annotation,
    app,
    audio,
-    chatflow_memory,
    completion,
    conversation,
    file,
@@ -41,7 +40,6 @@ __all__ = [
    "annotation",
    "app",
    "audio",
-    "chatflow_memory",
    "completion",
    "conversation",
    "dataset",
@@ -1,109 +0,0 @@
-from flask_restx import Resource, reqparse
-
-from controllers.service_api import api
-from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate_app_token
-from core.memory.entities import MemoryBlock, MemoryCreatedBy
-from core.workflow.runtime.variable_pool import VariablePool
-from models import App, EndUser
-from services.chatflow_memory_service import ChatflowMemoryService
-from services.workflow_service import WorkflowService
-
-
-class MemoryListApi(Resource):
-    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))
-    def get(self, app_model: App, end_user: EndUser):
-        parser = reqparse.RequestParser()
-        parser.add_argument("conversation_id", required=False, type=str | None, default=None)
-        parser.add_argument("memory_id", required=False, type=str | None, default=None)
-        parser.add_argument("version", required=False, type=int | None, default=None)
-        args = parser.parse_args()
-        conversation_id: str | None = args.get("conversation_id")
-        memory_id = args.get("memory_id")
-        version = args.get("version")
-
-        if conversation_id:
-            result = ChatflowMemoryService.get_persistent_memories_with_conversation(
-                app_model, MemoryCreatedBy(end_user_id=end_user.id), conversation_id, version
-            )
-            session_memories = ChatflowMemoryService.get_session_memories_with_conversation(
-                app_model, MemoryCreatedBy(end_user_id=end_user.id), conversation_id, version
-            )
-            result = [*result, *session_memories]
-        else:
-            result = ChatflowMemoryService.get_persistent_memories(
-                app_model, MemoryCreatedBy(end_user_id=end_user.id), version
-            )
-
-        if memory_id:
-            result = [it for it in result if it.spec.id == memory_id]
-        return [it for it in result if it.spec.end_user_visible]
-
-
-class MemoryEditApi(Resource):
-    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))
-    def put(self, app_model: App, end_user: EndUser):
-        parser = reqparse.RequestParser()
-        parser.add_argument("id", type=str, required=True)
-        parser.add_argument("conversation_id", type=str | None, required=False, default=None)
-        parser.add_argument("node_id", type=str | None, required=False, default=None)
-        parser.add_argument("update", type=str, required=True)
-        args = parser.parse_args()
-        workflow = WorkflowService().get_published_workflow(app_model)
-        update = args.get("update")
-        conversation_id = args.get("conversation_id")
-        node_id = args.get("node_id")
-        if not isinstance(update, str):
-            return {"error": "Invalid update"}, 400
-        if not workflow:
-            return {"error": "Workflow not found"}, 404
-        memory_spec = next((it for it in workflow.memory_blocks if it.id == args["id"]), None)
-        if not memory_spec:
-            return {"error": "Memory not found"}, 404
-
-        # First get existing memory
-        existing_memory = ChatflowMemoryService.get_memory_by_spec(
-            spec=memory_spec,
-            tenant_id=app_model.tenant_id,
-            app_id=app_model.id,
-            created_by=MemoryCreatedBy(end_user_id=end_user.id),
-            conversation_id=conversation_id,
-            node_id=node_id,
-            is_draft=False,
-        )
-
-        # Create updated memory instance with incremented version
-        updated_memory = MemoryBlock(
-            spec=existing_memory.spec,
-            tenant_id=existing_memory.tenant_id,
-            app_id=existing_memory.app_id,
-            conversation_id=existing_memory.conversation_id,
-            node_id=existing_memory.node_id,
-            value=update,  # New value
-            version=existing_memory.version + 1,  # Increment version for update
-            edited_by_user=True,
-            created_by=existing_memory.created_by,
-        )
-
-        ChatflowMemoryService.save_memory(updated_memory, VariablePool(), False)
-        return "", 204
-
-
-class MemoryDeleteApi(Resource):
-    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))
-    def delete(self, app_model: App, end_user: EndUser):
-        parser = reqparse.RequestParser()
-        parser.add_argument("id", type=str, required=False, default=None)
-        args = parser.parse_args()
-        memory_id = args.get("id")
-
-        if memory_id:
-            ChatflowMemoryService.delete_memory(app_model, memory_id, MemoryCreatedBy(end_user_id=end_user.id))
-            return "", 204
-        else:
-            ChatflowMemoryService.delete_all_user_memories(app_model, MemoryCreatedBy(end_user_id=end_user.id))
-            return "", 200
-
-
-api.add_resource(MemoryListApi, "/memories")
-api.add_resource(MemoryEditApi, "/memory-edit")
-api.add_resource(MemoryDeleteApi, "/memories")
@@ -592,7 +592,7 @@ class DocumentApi(DatasetApiResource):
                "name": document.name,
                "created_from": document.created_from,
                "created_by": document.created_by,
-                "created_at": int(document.created_at.timestamp()),
+                "created_at": document.created_at.timestamp(),
                "tokens": document.tokens,
                "indexing_status": document.indexing_status,
                "completed_at": int(document.completed_at.timestamp()) if document.completed_at else None,
@@ -625,7 +625,7 @@ class DocumentApi(DatasetApiResource):
                "name": document.name,
                "created_from": document.created_from,
                "created_by": document.created_by,
-                "created_at": int(document.created_at.timestamp()),
+                "created_at": document.created_at.timestamp(),
                "tokens": document.tokens,
                "indexing_status": document.indexing_status,
                "completed_at": int(document.completed_at.timestamp()) if document.completed_at else None,
@@ -2,7 +2,6 @@ from flask import request
 from flask_restx import marshal, reqparse
 from werkzeug.exceptions import NotFound

-from configs import dify_config
 from controllers.service_api import service_api_ns
 from controllers.service_api.app.error import ProviderNotInitializeError
 from controllers.service_api.wraps import (
@@ -108,10 +107,6 @@ class SegmentApi(DatasetApiResource):
        # validate args
        args = segment_create_parser.parse_args()
        if args["segments"] is not None:
-            segments_limit = dify_config.DATASET_MAX_SEGMENTS_PER_REQUEST
-            if segments_limit > 0 and len(args["segments"]) > segments_limit:
-                raise ValueError(f"Exceeded maximum segments limit of {segments_limit}.")
-
            for args_item in args["segments"]:
                SegmentService.segment_create_args_validate(args_item, document)
            segments = SegmentService.multi_create_segment(args["segments"], document, dataset)
@@ -13,7 +13,6 @@ from sqlalchemy import select, update
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden, NotFound, Unauthorized

-from enums.cloud_plan import CloudPlan
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from libs.datetime_utils import naive_utc_now
@@ -67,7 +66,6 @@ def validate_app_token(view: Callable[P, R] | None = None, *, fetch_user_arg: Fe

            kwargs["app_model"] = app_model

-            # If caller needs end-user context, attach EndUser to current_user
            if fetch_user_arg:
                if fetch_user_arg.fetch_from == WhereisUserArg.QUERY:
                    user_id = request.args.get("user")
@@ -76,6 +74,7 @@ def validate_app_token(view: Callable[P, R] | None = None, *, fetch_user_arg: Fe
                elif fetch_user_arg.fetch_from == WhereisUserArg.FORM:
                    user_id = request.form.get("user")
                else:
+                    # use default-user
                    user_id = None

                if not user_id and fetch_user_arg.required:
@@ -90,28 +89,6 @@ def validate_app_token(view: Callable[P, R] | None = None, *, fetch_user_arg: Fe
                # Set EndUser as current logged-in user for flask_login.current_user
                current_app.login_manager._update_request_context_with_user(end_user)  # type: ignore
                user_logged_in.send(current_app._get_current_object(), user=end_user)  # type: ignore
-            else:
-                # For service API without end-user context, ensure an Account is logged in
-                # so services relying on current_account_with_tenant() work correctly.
-                tenant_owner_info = (
-                    db.session.query(Tenant, Account)
-                    .join(TenantAccountJoin, Tenant.id == TenantAccountJoin.tenant_id)
-                    .join(Account, TenantAccountJoin.account_id == Account.id)
-                    .where(
-                        Tenant.id == app_model.tenant_id,
-                        TenantAccountJoin.role == "owner",
-                        Tenant.status == TenantStatus.NORMAL,
-                    )
-                    .one_or_none()
-                )
-
-                if tenant_owner_info:
-                    tenant_model, account = tenant_owner_info
-                    account.current_tenant = tenant_model
-                    current_app.login_manager._update_request_context_with_user(account)  # type: ignore
-                    user_logged_in.send(current_app._get_current_object(), user=current_user)  # type: ignore
-                else:
-                    raise Unauthorized("Tenant owner account not found or tenant is not active.")

            return view_func(*args, **kwargs)

@@ -161,7 +138,7 @@ def cloud_edition_billing_knowledge_limit_check(resource: str, api_token_type: s
            features = FeatureService.get_features(api_token.tenant_id)
            if features.billing.enabled:
                if resource == "add_segment":
-                    if features.billing.subscription.plan == CloudPlan.SANDBOX:
+                    if features.billing.subscription.plan == "sandbox":
                        raise Forbidden(
                            "To unlock this feature and elevate your Dify experience, please upgrade to a paid plan."
                        )
@@ -18,7 +18,6 @@ web_ns = Namespace("web", description="Web application API operations", path="/"
 from . import (
    app,
    audio,
-    chatflow_memory,
    completion,
    conversation,
    feature,
@@ -40,7 +39,6 @@ __all__ = [
    "app",
    "audio",
    "bp",
-    "chatflow_memory",
    "completion",
    "conversation",
    "feature",
@@ -1,108 +0,0 @@
-from flask_restx import reqparse
-
-from controllers.web import api
-from controllers.web.wraps import WebApiResource
-from core.memory.entities import MemoryBlock, MemoryCreatedBy
-from core.workflow.runtime.variable_pool import VariablePool
-from models import App, EndUser
-from services.chatflow_memory_service import ChatflowMemoryService
-from services.workflow_service import WorkflowService
-
-
-class MemoryListApi(WebApiResource):
-    def get(self, app_model: App, end_user: EndUser):
-        parser = reqparse.RequestParser()
-        parser.add_argument("conversation_id", required=False, type=str | None, default=None)
-        parser.add_argument("memory_id", required=False, type=str | None, default=None)
-        parser.add_argument("version", required=False, type=int | None, default=None)
-        args = parser.parse_args()
-        conversation_id: str | None = args.get("conversation_id")
-        memory_id = args.get("memory_id")
-        version = args.get("version")
-
-        if conversation_id:
-            result = ChatflowMemoryService.get_persistent_memories_with_conversation(
-                app_model, MemoryCreatedBy(end_user_id=end_user.id), conversation_id, version
-            )
-            session_memories = ChatflowMemoryService.get_session_memories_with_conversation(
-                app_model, MemoryCreatedBy(end_user_id=end_user.id), conversation_id, version
-            )
-            result = [*result, *session_memories]
-        else:
-            result = ChatflowMemoryService.get_persistent_memories(
-                app_model, MemoryCreatedBy(end_user_id=end_user.id), version
-            )
-
-        if memory_id:
-            result = [it for it in result if it.spec.id == memory_id]
-        return [it for it in result if it.spec.end_user_visible]
-
-
-class MemoryEditApi(WebApiResource):
-    def put(self, app_model: App, end_user: EndUser):
-        parser = reqparse.RequestParser()
-        parser.add_argument("id", type=str, required=True)
-        parser.add_argument("conversation_id", type=str | None, required=False, default=None)
-        parser.add_argument("node_id", type=str | None, required=False, default=None)
-        parser.add_argument("update", type=str, required=True)
-        args = parser.parse_args()
-        workflow = WorkflowService().get_published_workflow(app_model)
-        update = args.get("update")
-        conversation_id = args.get("conversation_id")
-        node_id = args.get("node_id")
-        if not isinstance(update, str):
-            return {"error": "Update must be a string"}, 400
-        if not workflow:
-            return {"error": "Workflow not found"}, 404
-        memory_spec = next((it for it in workflow.memory_blocks if it.id == args["id"]), None)
-        if not memory_spec:
-            return {"error": "Memory not found"}, 404
-        if not memory_spec.end_user_editable:
-            return {"error": "Memory not editable"}, 403
-
-        # First get existing memory
-        existing_memory = ChatflowMemoryService.get_memory_by_spec(
-            spec=memory_spec,
-            tenant_id=app_model.tenant_id,
-            app_id=app_model.id,
-            created_by=MemoryCreatedBy(end_user_id=end_user.id),
-            conversation_id=conversation_id,
-            node_id=node_id,
-            is_draft=False,
-        )
-
-        # Create updated memory instance with incremented version
-        updated_memory = MemoryBlock(
-            spec=existing_memory.spec,
-            tenant_id=existing_memory.tenant_id,
-            app_id=existing_memory.app_id,
-            conversation_id=existing_memory.conversation_id,
-            node_id=existing_memory.node_id,
-            value=update,  # New value
-            version=existing_memory.version + 1,  # Increment version for update
-            edited_by_user=True,
-            created_by=existing_memory.created_by,
-        )
-
-        ChatflowMemoryService.save_memory(updated_memory, VariablePool(), False)
-        return "", 204
-
-
-class MemoryDeleteApi(WebApiResource):
-    def delete(self, app_model: App, end_user: EndUser):
-        parser = reqparse.RequestParser()
-        parser.add_argument("id", type=str, required=False, default=None)
-        args = parser.parse_args()
-        memory_id = args.get("id")
-
-        if memory_id:
-            ChatflowMemoryService.delete_memory(app_model, memory_id, MemoryCreatedBy(end_user_id=end_user.id))
-            return "", 204
-        else:
-            ChatflowMemoryService.delete_all_user_memories(app_model, MemoryCreatedBy(end_user_id=end_user.id))
-            return "", 200
-
-
-api.add_resource(MemoryListApi, "/memories")
-api.add_resource(MemoryEditApi, "/memory-edit")
-api.add_resource(MemoryDeleteApi, "/memories")
@@ -1,11 +1,10 @@
 import logging
 import time
-from collections.abc import Mapping, MutableMapping, Sequence
+from collections.abc import Mapping
 from typing import Any, cast

 from sqlalchemy import select
 from sqlalchemy.orm import Session
-from typing_extensions import override

 from core.app.apps.advanced_chat.app_config_manager import AdvancedChatAppConfig
 from core.app.apps.base_app_queue_manager import AppQueueManager
@@ -21,16 +20,12 @@ from core.app.entities.queue_entities import (
    QueueTextChunkEvent,
 )
 from core.app.features.annotation_reply.annotation_reply import AnnotationReplyFeature
-from core.memory.entities import MemoryCreatedBy, MemoryScope
-from core.model_runtime.entities import AssistantPromptMessage, UserPromptMessage
 from core.moderation.base import ModerationError
 from core.moderation.input_moderation import InputModeration
 from core.variables.variables import VariableUnion
 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
 from core.workflow.graph_engine.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
-from core.workflow.graph_events import GraphRunSucceededEvent
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 from core.workflow.runtime import GraphRuntimeState, VariablePool
@@ -43,8 +38,6 @@ from models import Workflow
 from models.enums import UserFrom
 from models.model import App, Conversation, Message, MessageAnnotation
 from models.workflow import ConversationVariable
-from services.chatflow_history_service import ChatflowHistoryService
-from services.chatflow_memory_service import ChatflowMemoryService

 logger = logging.getLogger(__name__)

@@ -68,13 +61,11 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
        app: App,
        workflow_execution_repository: WorkflowExecutionRepository,
        workflow_node_execution_repository: WorkflowNodeExecutionRepository,
-        graph_engine_layers: Sequence[GraphEngineLayer] = (),
    ):
        super().__init__(
            queue_manager=queue_manager,
            variable_loader=variable_loader,
            app_id=application_generate_entity.app_config.app_id,
-            graph_engine_layers=graph_engine_layers,
        )
        self.application_generate_entity = application_generate_entity
        self.conversation = conversation
@@ -87,10 +78,6 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
        self._workflow_node_execution_repository = workflow_node_execution_repository

    def run(self):
-        ChatflowMemoryService.wait_for_sync_memory_completion(
-            workflow=self._workflow, conversation_id=self.conversation.id
-        )
-
        app_config = self.application_generate_entity.app_config
        app_config = cast(AdvancedChatAppConfig, app_config)

@@ -153,7 +140,6 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
                # Based on the definition of `VariableUnion`,
                # `list[Variable]` can be safely used as `list[VariableUnion]` since they are compatible.
                conversation_variables=conversation_variables,
-                memory_blocks=self._fetch_memory_blocks(),
            )

            # init graph
@@ -209,39 +195,12 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
        )

        workflow_entry.graph_engine.layer(persistence_layer)
-        for layer in self._graph_engine_layers:
-            workflow_entry.graph_engine.layer(layer)

        generator = workflow_entry.run()

        for event in generator:
            self._handle_event(workflow_entry, event)

-        try:
-            self._check_app_memory_updates(variable_pool)
-        except Exception as e:
-            logger.exception("Failed to check app memory updates", exc_info=e)
-
-    @override
-    def _handle_event(self, workflow_entry: WorkflowEntry, event: Any) -> None:
-        super()._handle_event(workflow_entry, event)
-        if isinstance(event, GraphRunSucceededEvent):
-            workflow_outputs = event.outputs
-            if not workflow_outputs:
-                logger.warning("Chatflow output is empty.")
-                return
-            assistant_message = workflow_outputs.get("answer")
-            if not assistant_message:
-                logger.warning("Chatflow output does not contain 'answer'.")
-                return
-            if not isinstance(assistant_message, str):
-                logger.warning("Chatflow output 'answer' is not a string.")
-                return
-            try:
-                self._sync_conversation_to_chatflow_tables(assistant_message)
-            except Exception as e:
-                logger.exception("Failed to sync conversation to memory tables", exc_info=e)
-
    def handle_input_moderation(
        self,
        app_record: App,
@@ -439,67 +398,3 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):

        # Return combined list
        return existing_variables + new_variables
-
-    def _fetch_memory_blocks(self) -> Mapping[str, str]:
-        """fetch all memory blocks for current app"""
-
-        memory_blocks_dict: MutableMapping[str, str] = {}
-        is_draft = self.application_generate_entity.invoke_from == InvokeFrom.DEBUGGER
-        conversation_id = self.conversation.id
-        memory_block_specs = self._workflow.memory_blocks
-        # Get runtime memory values
-        memories = ChatflowMemoryService.get_memories_by_specs(
-            memory_block_specs=memory_block_specs,
-            tenant_id=self._workflow.tenant_id,
-            app_id=self._workflow.app_id,
-            node_id=None,
-            conversation_id=conversation_id,
-            is_draft=is_draft,
-            created_by=self._get_created_by(),
-        )
-
-        # Build memory_id -> value mapping
-        for memory in memories:
-            if memory.spec.scope == MemoryScope.APP:
-                # App level: use memory_id directly
-                memory_blocks_dict[memory.spec.id] = memory.value
-            else:  # NODE scope
-                node_id = memory.node_id
-                if not node_id:
-                    logger.warning("Memory block %s has no node_id, skip.", memory.spec.id)
-                    continue
-                key = f"{node_id}.{memory.spec.id}"
-                memory_blocks_dict[key] = memory.value
-
-        return memory_blocks_dict
-
-    def _sync_conversation_to_chatflow_tables(self, assistant_message: str):
-        ChatflowHistoryService.save_app_message(
-            prompt_message=UserPromptMessage(content=(self.application_generate_entity.query)),
-            conversation_id=self.conversation.id,
-            app_id=self._workflow.app_id,
-            tenant_id=self._workflow.tenant_id,
-        )
-        ChatflowHistoryService.save_app_message(
-            prompt_message=AssistantPromptMessage(content=assistant_message),
-            conversation_id=self.conversation.id,
-            app_id=self._workflow.app_id,
-            tenant_id=self._workflow.tenant_id,
-        )
-
-    def _check_app_memory_updates(self, variable_pool: VariablePool):
-        is_draft = self.application_generate_entity.invoke_from == InvokeFrom.DEBUGGER
-
-        ChatflowMemoryService.update_app_memory_if_needed(
-            workflow=self._workflow,
-            conversation_id=self.conversation.id,
-            variable_pool=variable_pool,
-            is_draft=is_draft,
-            created_by=self._get_created_by(),
-        )
-
-    def _get_created_by(self) -> MemoryCreatedBy:
-        if self.application_generate_entity.invoke_from in {InvokeFrom.DEBUGGER, InvokeFrom.EXPLORE}:
-            return MemoryCreatedBy(account_id=self.application_generate_entity.user_id)
-        else:
-            return MemoryCreatedBy(end_user_id=self.application_generate_entity.user_id)
@@ -1,4 +1,3 @@
-import json
 import logging
 import re
 import time
@@ -61,7 +60,6 @@ from core.app.task_pipeline.based_generate_task_pipeline import BasedGenerateTas
 from core.app.task_pipeline.message_cycle_manager import MessageCycleManager
 from core.base.tts import AppGeneratorTTSPublisher, AudioTrunk
 from core.model_runtime.entities.llm_entities import LLMUsage
-from core.model_runtime.utils.encoders import jsonable_encoder
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.workflow.enums import WorkflowExecutionStatus
 from core.workflow.nodes import NodeType
@@ -393,14 +391,6 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
        if should_direct_answer:
            return

-        current_time = time.perf_counter()
-        if self._task_state.first_token_time is None and delta_text.strip():
-            self._task_state.first_token_time = current_time
-            self._task_state.is_streaming_response = True
-
-        if delta_text.strip():
-            self._task_state.last_token_time = current_time
-
        # Only publish tts message at text chunk streaming
        if tts_publisher and queue_message:
            tts_publisher.publish(queue_message)
@@ -782,33 +772,7 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
        message.answer = answer_text
        message.updated_at = naive_utc_now()
        message.provider_response_latency = time.perf_counter() - self._base_task_pipeline.start_at
-
-        # Set usage first before dumping metadata
-        if graph_runtime_state and graph_runtime_state.llm_usage:
-            usage = graph_runtime_state.llm_usage
-            message.message_tokens = usage.prompt_tokens
-            message.message_unit_price = usage.prompt_unit_price
-            message.message_price_unit = usage.prompt_price_unit
-            message.answer_tokens = usage.completion_tokens
-            message.answer_unit_price = usage.completion_unit_price
-            message.answer_price_unit = usage.completion_price_unit
-            message.total_price = usage.total_price
-            message.currency = usage.currency
-            self._task_state.metadata.usage = usage
-        else:
-            usage = LLMUsage.empty_usage()
-            self._task_state.metadata.usage = usage
-
-        # Add streaming metrics to usage if available
-        if self._task_state.is_streaming_response and self._task_state.first_token_time:
-            start_time = self._base_task_pipeline.start_at
-            first_token_time = self._task_state.first_token_time
-            last_token_time = self._task_state.last_token_time or first_token_time
-            usage.time_to_first_token = round(first_token_time - start_time, 3)
-            usage.time_to_generate = round(last_token_time - first_token_time, 3)
-
-        metadata = self._task_state.metadata.model_dump()
-        message.message_metadata = json.dumps(jsonable_encoder(metadata))
+        message.message_metadata = self._task_state.metadata.model_dump_json()
        message_files = [
            MessageFile(
                message_id=message.id,
@@ -826,6 +790,20 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
        ]
        session.add_all(message_files)

+        if graph_runtime_state and graph_runtime_state.llm_usage:
+            usage = graph_runtime_state.llm_usage
+            message.message_tokens = usage.prompt_tokens
+            message.message_unit_price = usage.prompt_unit_price
+            message.message_price_unit = usage.prompt_price_unit
+            message.answer_tokens = usage.completion_tokens
+            message.answer_unit_price = usage.completion_unit_price
+            message.answer_price_unit = usage.completion_price_unit
+            message.total_price = usage.total_price
+            message.currency = usage.currency
+            self._task_state.metadata.usage = usage
+        else:
+            self._task_state.metadata.usage = LLMUsage.empty_usage()
+
    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
@@ -144,7 +144,7 @@ class AgentChatAppRunner(AppRunner):
            prompt_template_entity=app_config.prompt_template,
            inputs=dict(inputs),
            files=list(files),
-            query=query,
+            query=query or "",
            memory=memory,
        )

@@ -172,7 +172,7 @@ class AgentChatAppRunner(AppRunner):
            prompt_template_entity=app_config.prompt_template,
            inputs=dict(inputs),
            files=list(files),
-            query=query,
+            query=query or "",
            memory=memory,
        )

@@ -79,7 +79,7 @@ class AppRunner:
        prompt_template_entity: PromptTemplateEntity,
        inputs: Mapping[str, str],
        files: Sequence["File"],
-        query: str = "",
+        query: str | None = None,
        context: str | None = None,
        memory: TokenBufferMemory | None = None,
        image_detail_config: ImagePromptMessageContent.DETAIL | None = None,
@@ -105,7 +105,7 @@ class AppRunner:
                app_mode=AppMode.value_of(app_record.mode),
                prompt_template_entity=prompt_template_entity,
                inputs=inputs,
-                query=query,
+                query=query or "",
                files=files,
                context=context,
                memory=memory,
@@ -190,7 +190,7 @@ class MessageBasedAppGenerator(BaseAppGenerator):
            override_model_configs=json.dumps(override_model_configs) if override_model_configs else None,
            conversation_id=conversation.id,
            inputs=application_generate_entity.inputs,
-            query=application_generate_entity.query,
+            query=application_generate_entity.query or "",
            message="",
            message_tokens=0,
            message_unit_price=0,
@@ -40,15 +40,19 @@ from core.workflow.repositories.draft_variable_repository import DraftVariableSa
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 from core.workflow.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader
-from enums.cloud_plan import CloudPlan
 from extensions.ext_database import db
+from extensions.ext_redis import redis_client
 from libs.flask_utils import preserve_flask_contexts
 from models import Account, EndUser, Workflow, WorkflowNodeExecutionTriggeredFrom
 from models.dataset import Document, DocumentPipelineExecutionLog, Pipeline
 from models.enums import WorkflowRunTriggeredFrom
 from models.model import AppMode
 from services.datasource_provider_service import DatasourceProviderService
+from services.feature_service import FeatureService
+from services.file_service import FileService
 from services.workflow_draft_variable_service import DraftVarLoader, WorkflowDraftVariableService
+from tasks.rag_pipeline.priority_rag_pipeline_run_task import priority_rag_pipeline_run_task
+from tasks.rag_pipeline.rag_pipeline_run_task import rag_pipeline_run_task

 logger = logging.getLogger(__name__)

@@ -251,7 +255,7 @@ class PipelineGenerator(BaseAppGenerator):
            json_text = json.dumps(text)
            upload_file = FileService(db.engine).upload_text(json_text, name, user.id, dataset.tenant_id)
            features = FeatureService.get_features(dataset.tenant_id)
-            if features.billing.enabled and features.billing.subscription.plan == CloudPlan.SANDBOX:
+            if features.billing.enabled and features.billing.subscription.plan == "sandbox":
                tenant_pipeline_task_key = f"tenant_pipeline_task:{dataset.tenant_id}"
                tenant_self_pipeline_task_queue = f"tenant_self_pipeline_task_queue:{dataset.tenant_id}"

@@ -135,8 +135,6 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
        )

        workflow_entry.graph_engine.layer(persistence_layer)
-        for layer in self._graph_engine_layers:
-            workflow_entry.graph_engine.layer(layer)

        generator = workflow_entry.run()

@@ -1,5 +1,5 @@
 import time
-from collections.abc import Mapping, Sequence
+from collections.abc import Mapping
 from typing import Any, cast

 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
@@ -27,7 +27,6 @@ from core.app.entities.queue_entities import (
 )
 from core.workflow.entities import GraphInitParams
 from core.workflow.graph import Graph
-from core.workflow.graph_engine.layers.base import GraphEngineLayer
 from core.workflow.graph_events import (
    GraphEngineEvent,
    GraphRunFailedEvent,
@@ -70,12 +69,10 @@ class WorkflowBasedAppRunner:
        queue_manager: AppQueueManager,
        variable_loader: VariableLoader = DUMMY_VARIABLE_LOADER,
        app_id: str,
-        graph_engine_layers: Sequence[GraphEngineLayer] = (),
    ):
        self._queue_manager = queue_manager
        self._variable_loader = variable_loader
        self._app_id = app_id
-        self._graph_engine_layers = graph_engine_layers

    def _init_graph(
        self,
@@ -129,7 +129,7 @@ class EasyUIBasedAppGenerateEntity(AppGenerateEntity):
    app_config: EasyUIBasedAppConfig = None  # type: ignore
    model_conf: ModelConfigWithCredentialsEntity

-    query: str = ""
+    query: str | None = None

    # pydantic configs
    model_config = ConfigDict(protected_namespaces=())
@@ -48,9 +48,6 @@ class WorkflowTaskState(TaskState):
    """

    answer: str = ""
-    first_token_time: float | None = None
-    last_token_time: float | None = None
-    is_streaming_response: bool = False


 class StreamEvent(StrEnum):
@@ -1,71 +0,0 @@
-from sqlalchemy import Engine
-from sqlalchemy.orm import sessionmaker
-
-from core.workflow.graph_engine.layers.base import GraphEngineLayer
-from core.workflow.graph_events.base import GraphEngineEvent
-from core.workflow.graph_events.graph import GraphRunPausedEvent
-from repositories.api_workflow_run_repository import APIWorkflowRunRepository
-from repositories.factory import DifyAPIRepositoryFactory
-
-
-class PauseStatePersistenceLayer(GraphEngineLayer):
-    def __init__(self, session_factory: Engine | sessionmaker, state_owner_user_id: str):
-        """Create a PauseStatePersistenceLayer.
-
-        The `state_owner_user_id` is used when creating state file for pause.
-        It generally should id of the creator of workflow.
-        """
-        if isinstance(session_factory, Engine):
-            session_factory = sessionmaker(session_factory)
-        self._session_maker = session_factory
-        self._state_owner_user_id = state_owner_user_id
-
-    def _get_repo(self) -> APIWorkflowRunRepository:
-        return DifyAPIRepositoryFactory.create_api_workflow_run_repository(self._session_maker)
-
-    def on_graph_start(self) -> None:
-        """
-        Called when graph execution starts.
-
-        This is called after the engine has been initialized but before any nodes
-        are executed. Layers can use this to set up resources or log start information.
-        """
-        pass
-
-    def on_event(self, event: GraphEngineEvent) -> None:
-        """
-        Called for every event emitted by the engine.
-
-        This method receives all events generated during graph execution, including:
-        - Graph lifecycle events (start, success, failure)
-        - Node execution events (start, success, failure, retry)
-        - Stream events for response nodes
-        - Container events (iteration, loop)
-
-        Args:
-            event: The event emitted by the engine
-        """
-        if not isinstance(event, GraphRunPausedEvent):
-            return
-
-        assert self.graph_runtime_state is not None
-        workflow_run_id: str | None = self.graph_runtime_state.system_variable.workflow_execution_id
-        assert workflow_run_id is not None
-        repo = self._get_repo()
-        repo.create_workflow_pause(
-            workflow_run_id=workflow_run_id,
-            state_owner_user_id=self._state_owner_user_id,
-            state=self.graph_runtime_state.dumps(),
-        )
-
-    def on_graph_end(self, error: Exception | None) -> None:
-        """
-        Called when graph execution ends.
-
-        This is called after all nodes have been executed or when execution is
-        aborted. Layers can use this to clean up resources or log final state.
-
-        Args:
-            error: The exception that caused execution to fail, or None if successful
-        """
-        pass
@@ -121,7 +121,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
        if self._application_generate_entity.app_config.app_mode != AppMode.COMPLETION:
            # start generate conversation name thread
            self._conversation_name_generate_thread = self._message_cycle_manager.generate_conversation_name(
-                conversation_id=self._conversation_id, query=self._application_generate_entity.query
+                conversation_id=self._conversation_id, query=self._application_generate_entity.query or ""
            )

        generator = self._wrapper_process_stream_response(trace_manager=self._application_generate_entity.trace_manager)
@@ -140,27 +140,7 @@ class MessageCycleManager:
        if not self._application_generate_entity.app_config.additional_features:
            raise ValueError("Additional features not found")
        if self._application_generate_entity.app_config.additional_features.show_retrieve_source:
-            merged_resources = [r for r in self._task_state.metadata.retriever_resources or [] if r]
-            existing_ids = {(r.dataset_id, r.document_id) for r in merged_resources if r.dataset_id and r.document_id}
-
-            # Add new unique resources from the event
-            for resource in event.retriever_resources or []:
-                if not resource:
-                    continue
-
-                is_duplicate = (
-                    resource.dataset_id
-                    and resource.document_id
-                    and (resource.dataset_id, resource.document_id) in existing_ids
-                )
-
-                if not is_duplicate:
-                    merged_resources.append(resource)
-
-            for i, resource in enumerate(merged_resources, 1):
-                resource.position = i
-
-            self._task_state.metadata.retriever_resources = merged_resources
+            self._task_state.metadata.retriever_resources = event.retriever_resources

    def message_file_to_stream_response(self, event: QueueMessageFileEvent) -> MessageFileStreamResponse | None:
        """
@@ -1,13 +0,0 @@
-from dataclasses import dataclass
-
-
-@dataclass
-class DocumentTask:
-    """Document task entity for document indexing operations.
-    
-    This class represents a document indexing task that can be queued
-    and processed by the document indexing system.
-    """
-    tenant_id: str
-    dataset_id: str
-    document_ids: list[str]
@@ -14,7 +14,6 @@ from core.helper.provider_cache import NoOpProviderCredentialCache
 from core.mcp.types import OAuthClientInformation, OAuthClientMetadata, OAuthTokens
 from core.tools.entities.common_entities import I18nObject
 from core.tools.entities.tool_entities import ToolProviderType
-from core.tools.utils.encryption import create_provider_encrypter

 if TYPE_CHECKING:
    from models.tools import MCPToolProvider
@@ -272,6 +271,8 @@ class MCPProviderEntity(BaseModel):

    def _decrypt_dict(self, data: dict[str, Any]) -> dict[str, Any]:
        """Generic method to decrypt dictionary fields"""
+        # Lazy import to avoid circular dependency
+        from core.tools.utils.encryption import create_provider_encrypter

        if not data:
            return {}
@@ -74,10 +74,6 @@ class File(BaseModel):
        storage_key: str | None = None,
        dify_model_identity: str | None = FILE_MODEL_IDENTITY,
        url: str | None = None,
-        # Legacy compatibility fields - explicitly handle known extra fields
-        tool_file_id: str | None = None,
-        upload_file_id: str | None = None,
-        datasource_file_id: str | None = None,
    ):
        super().__init__(
            id=id,
@@ -6,7 +6,10 @@ from core.helper.code_executor.template_transformer import TemplateTransformer
 class NodeJsTemplateTransformer(TemplateTransformer):
    @classmethod
    def get_runner_script(cls) -> str:
-        runner_script = dedent(f"""            {cls._code_placeholder}
+        runner_script = dedent(
+            f"""
+            // declare main function
+            {cls._code_placeholder}

            // decode and prepare input object
            var inputs_obj = JSON.parse(Buffer.from('{cls._inputs_placeholder}', 'base64').toString('utf-8'))
@@ -18,5 +21,6 @@ class NodeJsTemplateTransformer(TemplateTransformer):
            var output_json = JSON.stringify(output_obj)
            var result = `<<RESULT>>${{output_json}}<<RESULT>>`
            console.log(result)
-            """)
+            """
+        )
        return runner_script
@@ -6,7 +6,9 @@ from core.helper.code_executor.template_transformer import TemplateTransformer
 class Python3TemplateTransformer(TemplateTransformer):
    @classmethod
    def get_runner_script(cls) -> str:
-        runner_script = dedent(f"""            {cls._code_placeholder}
+        runner_script = dedent(f"""
+            # declare main function
+            {cls._code_placeholder}

            import json
            from base64 import b64decode
@@ -29,18 +29,6 @@ def batch_fetch_plugin_manifests(plugin_ids: list[str]) -> Sequence[MarketplaceP
    return [MarketplacePluginDeclaration.model_validate(plugin) for plugin in response.json()["data"]["plugins"]]


-def batch_fetch_plugin_by_ids(plugin_ids: list[str]) -> list[dict]:
-    if not plugin_ids:
-        return []
-
-    url = str(marketplace_api_url / "api/v1/plugins/batch")
-    response = httpx.post(url, json={"plugin_ids": plugin_ids}, headers={"X-Dify-Version": dify_config.project.version})
-    response.raise_for_status()
-
-    data = response.json()
-    return data.get("data", {}).get("plugins", [])
-
-
 def batch_fetch_plugin_manifests_ignore_deserialization_error(
    plugin_ids: list[str],
 ) -> Sequence[MarketplacePluginDeclaration]:
@@ -56,9 +56,6 @@ class HostingConfiguration:
        self.provider_map[f"{DEFAULT_PLUGIN_ID}/minimax/minimax"] = self.init_minimax()
        self.provider_map[f"{DEFAULT_PLUGIN_ID}/spark/spark"] = self.init_spark()
        self.provider_map[f"{DEFAULT_PLUGIN_ID}/zhipuai/zhipuai"] = self.init_zhipuai()
-        self.provider_map[f"{DEFAULT_PLUGIN_ID}/gemini/google"] = self.init_gemini()
-        self.provider_map[f"{DEFAULT_PLUGIN_ID}/x/x"] = self.init_xai()
-        self.provider_map[f"{DEFAULT_PLUGIN_ID}/deepseek/deepseek"] = self.init_deepseek()

        self.moderation_config = self.init_moderation_config()

@@ -131,7 +128,7 @@ class HostingConfiguration:
        quotas: list[HostingQuota] = []

        if dify_config.HOSTED_OPENAI_TRIAL_ENABLED:
-            hosted_quota_limit = 0
+            hosted_quota_limit = dify_config.HOSTED_OPENAI_QUOTA_LIMIT
            trial_models = self.parse_restrict_models_from_env("HOSTED_OPENAI_TRIAL_MODELS")
            trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit, restrict_models=trial_models)
            quotas.append(trial_quota)
@@ -159,49 +156,18 @@ class HostingConfiguration:
            quota_unit=quota_unit,
        )

-    def init_gemini(self) -> HostingProvider:
-        quota_unit = QuotaUnit.CREDITS
-        quotas: list[HostingQuota] = []
-
-        if dify_config.HOSTED_GEMINI_TRIAL_ENABLED:
-            hosted_quota_limit = 0
-            trial_models = self.parse_restrict_models_from_env("HOSTED_GEMINI_TRIAL_MODELS")
-            trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit, restrict_models=trial_models)
-            quotas.append(trial_quota)
-
-        if dify_config.HOSTED_GEMINI_PAID_ENABLED:
-            paid_models = self.parse_restrict_models_from_env("HOSTED_GEMINI_PAID_MODELS")
-            paid_quota = PaidHostingQuota(restrict_models=paid_models)
-            quotas.append(paid_quota)
-
-        if len(quotas) > 0:
-            credentials = {
-                "google_api_key": dify_config.HOSTED_GEMINI_API_KEY,
-            }
-
-            if dify_config.HOSTED_GEMINI_API_BASE:
-                credentials["google_base_url"] = dify_config.HOSTED_GEMINI_API_BASE
-
-            return HostingProvider(enabled=True, credentials=credentials, quota_unit=quota_unit, quotas=quotas)
-
-        return HostingProvider(
-            enabled=False,
-            quota_unit=quota_unit,
-        )
-
-    def init_anthropic(self) -> HostingProvider:
-        quota_unit = QuotaUnit.CREDITS
+    @staticmethod
+    def init_anthropic() -> HostingProvider:
+        quota_unit = QuotaUnit.TOKENS
        quotas: list[HostingQuota] = []

        if dify_config.HOSTED_ANTHROPIC_TRIAL_ENABLED:
-            hosted_quota_limit = 0
-            trail_models = self.parse_restrict_models_from_env("HOSTED_ANTHROPIC_TRIAL_MODELS")
-            trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit, restrict_models=trail_models)
+            hosted_quota_limit = dify_config.HOSTED_ANTHROPIC_QUOTA_LIMIT
+            trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit)
            quotas.append(trial_quota)

        if dify_config.HOSTED_ANTHROPIC_PAID_ENABLED:
-            paid_models = self.parse_restrict_models_from_env("HOSTED_ANTHROPIC_PAID_MODELS")
-            paid_quota = PaidHostingQuota(restrict_models=paid_models)
+            paid_quota = PaidHostingQuota()
            quotas.append(paid_quota)

        if len(quotas) > 0:
@@ -219,66 +185,6 @@ class HostingConfiguration:
            quota_unit=quota_unit,
        )

-    def init_xai(self) -> HostingProvider:
-        quota_unit = QuotaUnit.CREDITS
-        quotas: list[HostingQuota] = []
-
-        if dify_config.HOSTED_XAI_TRIAL_ENABLED:
-            hosted_quota_limit = 0
-            trail_models = self.parse_restrict_models_from_env("HOSTED_XAI_TRIAL_MODELS")
-            trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit, restrict_models=trail_models)
-            quotas.append(trial_quota)
-
-        if dify_config.HOSTED_XAI_PAID_ENABLED:
-            paid_models = self.parse_restrict_models_from_env("HOSTED_XAI_PAID_MODELS")
-            paid_quota = PaidHostingQuota(restrict_models=paid_models)
-            quotas.append(paid_quota)
-
-        if len(quotas) > 0:
-            credentials = {
-                "api_key": dify_config.HOSTED_XAI_API_KEY,
-            }
-
-            if dify_config.HOSTED_XAI_API_BASE:
-                credentials["endpoint_url"] = dify_config.HOSTED_XAI_API_BASE
-
-            return HostingProvider(enabled=True, credentials=credentials, quota_unit=quota_unit, quotas=quotas)
-
-        return HostingProvider(
-            enabled=False,
-            quota_unit=quota_unit,
-        )
-
-    def init_deepseek(self) -> HostingProvider:
-        quota_unit = QuotaUnit.CREDITS
-        quotas: list[HostingQuota] = []
-
-        if dify_config.HOSTED_DEEPSEEK_TRIAL_ENABLED:
-            hosted_quota_limit = 0
-            trail_models = self.parse_restrict_models_from_env("HOSTED_DEEPSEEK_TRIAL_MODELS")
-            trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit, restrict_models=trail_models)
-            quotas.append(trial_quota)
-
-        if dify_config.HOSTED_DEEPSEEK_PAID_ENABLED:
-            paid_models = self.parse_restrict_models_from_env("HOSTED_DEEPSEEK_PAID_MODELS")
-            paid_quota = PaidHostingQuota(restrict_models=paid_models)
-            quotas.append(paid_quota)
-
-        if len(quotas) > 0:
-            credentials = {
-                "api_key": dify_config.HOSTED_DEEPSEEK_API_KEY,
-            }
-
-            if dify_config.HOSTED_DEEPSEEK_API_BASE:
-                credentials["endpoint_url"] = dify_config.HOSTED_DEEPSEEK_API_BASE
-
-            return HostingProvider(enabled=True, credentials=credentials, quota_unit=quota_unit, quotas=quotas)
-
-        return HostingProvider(
-            enabled=False,
-            quota_unit=quota_unit,
-        )
-
    @staticmethod
    def init_minimax() -> HostingProvider:
        quota_unit = QuotaUnit.TOKENS
@@ -14,12 +14,10 @@ from core.llm_generator.prompts import (
    JAVASCRIPT_CODE_GENERATOR_PROMPT_TEMPLATE,
    LLM_MODIFY_CODE_SYSTEM,
    LLM_MODIFY_PROMPT_SYSTEM,
-    MEMORY_UPDATE_PROMPT,
    PYTHON_CODE_GENERATOR_PROMPT_TEMPLATE,
    SYSTEM_STRUCTURED_OUTPUT_GENERATE,
    WORKFLOW_RULE_CONFIG_PROMPT_GENERATE_TEMPLATE,
 )
-from core.memory.entities import MemoryBlock, MemoryBlockSpec
 from core.model_manager import ModelManager
 from core.model_runtime.entities.llm_entities import LLMResult
 from core.model_runtime.entities.message_entities import PromptMessage, SystemPromptMessage, UserPromptMessage
@@ -30,7 +28,6 @@ from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
 from core.ops.utils import measure_time
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
 from core.workflow.entities.workflow_node_execution import WorkflowNodeExecutionMetadataKey
-from core.workflow.runtime.variable_pool import VariablePool
 from extensions.ext_database import db
 from extensions.ext_storage import storage
 from models import App, Message, WorkflowNodeExecutionModel
@@ -565,35 +562,3 @@ class LLMGenerator:
                "Failed to invoke LLM model, model: %s", json.dumps(model_config.get("name")), exc_info=True
            )
            return {"error": f"An unexpected error occurred: {str(e)}"}
-
-    @staticmethod
-    def update_memory_block(
-        tenant_id: str,
-        visible_history: Sequence[tuple[str, str]],
-        variable_pool: VariablePool,
-        memory_block: MemoryBlock,
-        memory_spec: MemoryBlockSpec,
-    ) -> str:
-        model_instance = ModelManager().get_model_instance(
-            tenant_id=tenant_id,
-            provider=memory_spec.model.provider,
-            model=memory_spec.model.name,
-            model_type=ModelType.LLM,
-        )
-        formatted_history = ""
-        for sender, message in visible_history:
-            formatted_history += f"{sender}: {message}\n"
-        filled_instruction = variable_pool.convert_template(memory_spec.instruction).text
-        formatted_prompt = PromptTemplateParser(MEMORY_UPDATE_PROMPT).format(
-            inputs={
-                "formatted_history": formatted_history,
-                "current_value": memory_block.value,
-                "instruction": filled_instruction,
-            }
-        )
-        llm_result = model_instance.invoke_llm(
-            prompt_messages=[UserPromptMessage(content=formatted_prompt)],
-            model_parameters=memory_spec.model.completion_params,
-            stream=False,
-        )
-        return llm_result.message.get_text_content()
@@ -422,18 +422,3 @@ INSTRUCTION_GENERATE_TEMPLATE_PROMPT = """The output of this prompt is not as ex
 You should edit the prompt according to the IDEAL OUTPUT."""

 INSTRUCTION_GENERATE_TEMPLATE_CODE = """Please fix the errors in the {{#error_message#}}."""
-
-MEMORY_UPDATE_PROMPT = """
-Based on the following conversation history, update the memory content:
-
-Conversation history:
-{{formatted_history}}
-
-Current memory:
-{{current_value}}
-
-Update instruction:
-{{instruction}}
-
-Please output only the updated memory content, no other text like greeting:
-"""
@@ -1,119 +0,0 @@
-from __future__ import annotations
-
-from enum import StrEnum
-from typing import TYPE_CHECKING, Optional
-from uuid import uuid4
-
-from pydantic import BaseModel, Field
-
-if TYPE_CHECKING:
-    from core.app.app_config.entities import ModelConfig
-
-
-class MemoryScope(StrEnum):
-    """Memory scope determined by node_id field"""
-    APP = "app"  # node_id is None
-    NODE = "node"  # node_id is not None
-
-
-class MemoryTerm(StrEnum):
-    """Memory term determined by conversation_id field"""
-    SESSION = "session"  # conversation_id is not None
-    PERSISTENT = "persistent"  # conversation_id is None
-
-
-class MemoryStrategy(StrEnum):
-    ON_TURNS = "on_turns"
-
-
-class MemoryScheduleMode(StrEnum):
-    SYNC = "sync"
-    ASYNC = "async"
-
-
-class MemoryBlockSpec(BaseModel):
-    """Memory block specification for workflow configuration"""
-    id: str = Field(
-        default_factory=lambda: str(uuid4()),
-        description="Unique identifier for the memory block",
-    )
-    name: str = Field(description="Display name of the memory block")
-    description: str = Field(default="", description="Description of the memory block")
-    template: str = Field(description="Initial template content for the memory")
-    instruction: str = Field(description="Instructions for updating the memory")
-    scope: MemoryScope = Field(description="Scope of the memory (app or node level)")
-    term: MemoryTerm = Field(description="Term of the memory (session or persistent)")
-    strategy: MemoryStrategy = Field(description="Update strategy for the memory")
-    update_turns: int = Field(gt=0, description="Number of turns between updates")
-    preserved_turns: int = Field(gt=0, description="Number of conversation turns to preserve")
-    schedule_mode: MemoryScheduleMode = Field(description="Synchronous or asynchronous update mode")
-    model: ModelConfig = Field(description="Model configuration for memory updates")
-    end_user_visible: bool = Field(default=False, description="Whether memory is visible to end users")
-    end_user_editable: bool = Field(default=False, description="Whether memory is editable by end users")
-
-
-class MemoryCreatedBy(BaseModel):
-    end_user_id: str | None = None
-    account_id: str | None = None
-
-
-class MemoryBlock(BaseModel):
-    """Runtime memory block instance
-
-    Design Rules:
-    - app_id = None: Global memory (future feature, not implemented yet)
-    - app_id = str: App-specific memory
-    - conversation_id = None: Persistent memory (cross-conversation)
-    - conversation_id = str: Session memory (conversation-specific)
-    - node_id = None: App-level scope
-    - node_id = str: Node-level scope
-
-    These rules implicitly determine scope and term without redundant storage.
-    """
-    spec: MemoryBlockSpec
-    tenant_id: str
-    value: str
-    app_id: str
-    conversation_id: Optional[str] = None
-    node_id: Optional[str] = None
-    edited_by_user: bool = False
-    created_by: MemoryCreatedBy
-    version: int = Field(description="Memory block version number")
-
-
-class MemoryValueData(BaseModel):
-    value: str
-    edited_by_user: bool = False
-
-
-class ChatflowConversationMetadata(BaseModel):
-    """Metadata for chatflow conversation with visible message count"""
-    type: str = "mutable_visible_window"
-    visible_count: int = Field(gt=0, description="Number of visible messages to keep")
-
-
-class MemoryBlockWithConversation(MemoryBlock):
-    """MemoryBlock with optional conversation metadata for session memories"""
-    conversation_metadata: ChatflowConversationMetadata = Field(
-        description="Conversation metadata, only present for session memories"
-    )
-
-    @classmethod
-    def from_memory_block(
-        cls,
-        memory_block: MemoryBlock,
-        conversation_metadata: ChatflowConversationMetadata
-    ) -> MemoryBlockWithConversation:
-        """Create MemoryBlockWithConversation from MemoryBlock"""
-        return cls(
-            spec=memory_block.spec,
-            tenant_id=memory_block.tenant_id,
-            value=memory_block.value,
-            app_id=memory_block.app_id,
-            conversation_id=memory_block.conversation_id,
-            node_id=memory_block.node_id,
-            edited_by_user=memory_block.edited_by_user,
-            created_by=memory_block.created_by,
-            version=memory_block.version,
-            conversation_metadata=conversation_metadata
-        )
@@ -1,6 +0,0 @@
-class MemorySyncTimeoutError(Exception):
-    def __init__(self, app_id: str, conversation_id: str):
-        self.app_id = app_id
-        self.conversation_id = conversation_id
-        self.message = "Memory synchronization timeout after 50 seconds"
-        super().__init__(self.message)
@@ -38,8 +38,6 @@ class LLMUsageMetadata(TypedDict, total=False):
    prompt_price: Union[float, str]
    completion_price: Union[float, str]
    latency: float
-    time_to_first_token: float
-    time_to_generate: float


 class LLMUsage(ModelUsage):
@@ -59,8 +57,6 @@ class LLMUsage(ModelUsage):
    total_price: Decimal
    currency: str
    latency: float
-    time_to_first_token: float | None = None
-    time_to_generate: float | None = None

    @classmethod
    def empty_usage(cls):
@@ -77,8 +73,6 @@ class LLMUsage(ModelUsage):
            total_price=Decimal("0.0"),
            currency="USD",
            latency=0.0,
-            time_to_first_token=None,
-            time_to_generate=None,
        )

    @classmethod
@@ -114,8 +108,6 @@ class LLMUsage(ModelUsage):
            prompt_price=Decimal(str(metadata.get("prompt_price", 0))),
            completion_price=Decimal(str(metadata.get("completion_price", 0))),
            latency=metadata.get("latency", 0.0),
-            time_to_first_token=metadata.get("time_to_first_token"),
-            time_to_generate=metadata.get("time_to_generate"),
        )

    def plus(self, other: LLMUsage) -> LLMUsage:
@@ -141,8 +133,6 @@ class LLMUsage(ModelUsage):
                total_price=self.total_price + other.total_price,
                currency=other.currency,
                latency=self.latency + other.latency,
-                time_to_first_token=other.time_to_first_token,
-                time_to_generate=other.time_to_generate,
            )

    def __add__(self, other: LLMUsage) -> LLMUsage:
@@ -62,9 +62,6 @@ class MessageTraceInfo(BaseTraceInfo):
    file_list: Union[str, dict[str, Any], list] | None = None
    message_file_data: Any | None = None
    conversation_mode: str
-    gen_ai_server_time_to_first_token: float | None = None
-    llm_streaming_time_to_generate: float | None = None
-    is_streaming_request: bool = False


 class ModerationTraceInfo(BaseTraceInfo):
@@ -14,7 +14,7 @@ from flask import current_app
 from sqlalchemy import select
 from sqlalchemy.orm import Session, sessionmaker

-from core.helper.encrypter import batch_decrypt_token, encrypt_token, obfuscated_token
+from core.helper.encrypter import decrypt_token, encrypt_token, obfuscated_token
 from core.ops.entities.config_entity import (
    OPS_FILE_PATH,
    TracingProviderEnum,
@@ -141,8 +141,6 @@ provider_config_map = OpsTraceProviderConfigMap()

 class OpsTraceManager:
    ops_trace_instances_cache: LRUCache = LRUCache(maxsize=128)
-    decrypted_configs_cache: LRUCache = LRUCache(maxsize=128)
-    _decryption_cache_lock = threading.RLock()

    @classmethod
    def encrypt_tracing_config(
@@ -163,7 +161,7 @@ class OpsTraceManager:
            provider_config_map[tracing_provider]["other_keys"],
        )

-        new_config: dict[str, Any] = {}
+        new_config = {}
        # Encrypt necessary keys
        for key in secret_keys:
            if key in tracing_config:
@@ -193,41 +191,20 @@ class OpsTraceManager:
        :param tracing_config: tracing config
        :return:
        """
-        config_json = json.dumps(tracing_config, sort_keys=True)
-        decrypted_config_key = (
-            tenant_id,
-            tracing_provider,
-            config_json,
+        config_class, secret_keys, other_keys = (
+            provider_config_map[tracing_provider]["config_class"],
+            provider_config_map[tracing_provider]["secret_keys"],
+            provider_config_map[tracing_provider]["other_keys"],
        )
+        new_config = {}
+        for key in secret_keys:
+            if key in tracing_config:
+                new_config[key] = decrypt_token(tenant_id, tracing_config[key])

-        # First check without lock for performance
-        cached_config = cls.decrypted_configs_cache.get(decrypted_config_key)
-        if cached_config is not None:
-            return dict(cached_config)
+        for key in other_keys:
+            new_config[key] = tracing_config.get(key, "")

-        with cls._decryption_cache_lock:
-            # Second check (double-checked locking) to prevent race conditions
-            cached_config = cls.decrypted_configs_cache.get(decrypted_config_key)
-            if cached_config is not None:
-                return dict(cached_config)
-
-            config_class, secret_keys, other_keys = (
-                provider_config_map[tracing_provider]["config_class"],
-                provider_config_map[tracing_provider]["secret_keys"],
-                provider_config_map[tracing_provider]["other_keys"],
-            )
-            new_config: dict[str, Any] = {}
-            keys_to_decrypt = [key for key in secret_keys if key in tracing_config]
-            if keys_to_decrypt:
-                decrypted_values = batch_decrypt_token(tenant_id, [tracing_config[key] for key in keys_to_decrypt])
-                new_config.update(zip(keys_to_decrypt, decrypted_values))
-
-            for key in other_keys:
-                new_config[key] = tracing_config.get(key, "")
-
-            decrypted_config = config_class(**new_config).model_dump()
-            cls.decrypted_configs_cache[decrypted_config_key] = decrypted_config
-            return dict(decrypted_config)
+        return config_class(**new_config).model_dump()

    @classmethod
    def obfuscated_decrypt_token(cls, tracing_provider: str, decrypt_tracing_config: dict):
@@ -242,7 +219,7 @@ class OpsTraceManager:
            provider_config_map[tracing_provider]["secret_keys"],
            provider_config_map[tracing_provider]["other_keys"],
        )
-        new_config: dict[str, Any] = {}
+        new_config = {}
        for key in secret_keys:
            if key in decrypt_tracing_config:
                new_config[key] = obfuscated_token(decrypt_tracing_config[key])
@@ -619,8 +596,6 @@ class TraceTask:
            file_url = f"{self.file_base_url}/{message_file_data.url}" if message_file_data else ""
            file_list.append(file_url)

-        streaming_metrics = self._extract_streaming_metrics(message_data)
-
        metadata = {
            "conversation_id": message_data.conversation_id,
            "ls_provider": message_data.model_provider,
@@ -653,9 +628,6 @@ class TraceTask:
            metadata=metadata,
            message_file_data=message_file_data,
            conversation_mode=conversation_mode,
-            gen_ai_server_time_to_first_token=streaming_metrics.get("gen_ai_server_time_to_first_token"),
-            llm_streaming_time_to_generate=streaming_metrics.get("llm_streaming_time_to_generate"),
-            is_streaming_request=streaming_metrics.get("is_streaming_request", False),
        )

        return message_trace_info
@@ -881,24 +853,6 @@ class TraceTask:

        return generate_name_trace_info

-    def _extract_streaming_metrics(self, message_data) -> dict:
-        if not message_data.message_metadata:
-            return {}
-
-        try:
-            metadata = json.loads(message_data.message_metadata)
-            usage = metadata.get("usage", {})
-            time_to_first_token = usage.get("time_to_first_token")
-            time_to_generate = usage.get("time_to_generate")
-
-            return {
-                "gen_ai_server_time_to_first_token": time_to_first_token,
-                "llm_streaming_time_to_generate": time_to_generate,
-                "is_streaming_request": time_to_first_token is not None,
-            }
-        except (json.JSONDecodeError, AttributeError):
-            return {}
-

 trace_manager_timer: threading.Timer | None = None
 trace_manager_queue: queue.Queue = queue.Queue()
@@ -5,18 +5,12 @@ Tencent APM Trace Client - handles network operations, metrics, and API communic
 from __future__ import annotations

 import importlib
-import json
 import logging
 import os
 import socket
 from typing import TYPE_CHECKING
 from urllib.parse import urlparse

-try:
-    from importlib.metadata import version
-except ImportError:
-    from importlib_metadata import version  # type: ignore[import-not-found]
-
 if TYPE_CHECKING:
    from opentelemetry.metrics import Meter
    from opentelemetry.metrics._internal.instrument import Histogram
@@ -33,27 +27,12 @@ from opentelemetry.util.types import AttributeValue

 from configs import dify_config

-from .entities.semconv import (
-    GEN_AI_SERVER_TIME_TO_FIRST_TOKEN,
-    GEN_AI_STREAMING_TIME_TO_GENERATE,
-    GEN_AI_TOKEN_USAGE,
-    GEN_AI_TRACE_DURATION,
-    LLM_OPERATION_DURATION,
-)
+from .entities.tencent_semconv import LLM_OPERATION_DURATION
 from .entities.tencent_trace_entity import SpanData

 logger = logging.getLogger(__name__)


-def _get_opentelemetry_sdk_version() -> str:
-    """Get OpenTelemetry SDK version dynamically."""
-    try:
-        return version("opentelemetry-sdk")
-    except Exception:
-        logger.debug("Failed to get opentelemetry-sdk version, using default")
-        return "1.27.0"  # fallback version
-
-
 class TencentTraceClient:
    """Tencent APM trace client using OpenTelemetry OTLP exporter"""

@@ -78,9 +57,6 @@ class TencentTraceClient:
                ResourceAttributes.SERVICE_VERSION: f"dify-{dify_config.project.version}-{dify_config.COMMIT_SHA}",
                ResourceAttributes.DEPLOYMENT_ENVIRONMENT: f"{dify_config.DEPLOY_ENV}-{dify_config.EDITION}",
                ResourceAttributes.HOST_NAME: socket.gethostname(),
-                ResourceAttributes.TELEMETRY_SDK_LANGUAGE: "python",
-                ResourceAttributes.TELEMETRY_SDK_NAME: "opentelemetry",
-                ResourceAttributes.TELEMETRY_SDK_VERSION: _get_opentelemetry_sdk_version(),
            }
        )
        # Prepare gRPC endpoint/metadata
@@ -104,23 +80,18 @@ class TencentTraceClient:
        )
        self.tracer_provider.add_span_processor(self.span_processor)

-        # use dify api version as tracer version
-        self.tracer = self.tracer_provider.get_tracer("dify-sdk", dify_config.project.version)
+        self.tracer = self.tracer_provider.get_tracer("dify.tencent_apm")

        # Store span contexts for parent-child relationships
        self.span_contexts: dict[int, trace_api.SpanContext] = {}

        self.meter: Meter | None = None
-        self.meter_provider: MeterProvider | None = None
        self.hist_llm_duration: Histogram | None = None
-        self.hist_token_usage: Histogram | None = None
-        self.hist_time_to_first_token: Histogram | None = None
-        self.hist_time_to_generate: Histogram | None = None
-        self.hist_trace_duration: Histogram | None = None
        self.metric_reader: MetricReader | None = None

        # Metrics exporter and instruments
        try:
+            from opentelemetry import metrics
            from opentelemetry.sdk.metrics import Histogram, MeterProvider
            from opentelemetry.sdk.metrics.export import AggregationTemporality, PeriodicExportingMetricReader

@@ -128,7 +99,7 @@ class TencentTraceClient:
            use_http_protobuf = protocol in {"http/protobuf", "http-protobuf"}
            use_http_json = protocol in {"http/json", "http-json"}

-            # Tencent APM works best with delta aggregation temporality
+            # Set preferred temporality for histograms to DELTA
            preferred_temporality: dict[type, AggregationTemporality] = {Histogram: AggregationTemporality.DELTA}

            def _create_metric_exporter(exporter_cls, **kwargs):
@@ -203,66 +174,23 @@ class TencentTraceClient:
                )

            if metric_reader is not None:
-                # Use instance-level MeterProvider instead of global to support config changes
-                # without worker restart. Each TencentTraceClient manages its own MeterProvider.
                provider = MeterProvider(resource=self.resource, metric_readers=[metric_reader])
-                self.meter_provider = provider
-                self.meter = provider.get_meter("dify-sdk", dify_config.project.version)
-
-                # LLM operation duration histogram
+                metrics.set_meter_provider(provider)
+                self.meter = metrics.get_meter("dify-sdk", dify_config.project.version)
                self.hist_llm_duration = self.meter.create_histogram(
                    name=LLM_OPERATION_DURATION,
                    unit="s",
                    description="LLM operation duration (seconds)",
                )
-
-                # Token usage histogram with exponential buckets
-                self.hist_token_usage = self.meter.create_histogram(
-                    name=GEN_AI_TOKEN_USAGE,
-                    unit="token",
-                    description="Number of tokens used in prompt and completions",
-                )
-
-                # Time to first token histogram
-                self.hist_time_to_first_token = self.meter.create_histogram(
-                    name=GEN_AI_SERVER_TIME_TO_FIRST_TOKEN,
-                    unit="s",
-                    description="Time to first token for streaming LLM responses (seconds)",
-                )
-
-                # Time to generate histogram
-                self.hist_time_to_generate = self.meter.create_histogram(
-                    name=GEN_AI_STREAMING_TIME_TO_GENERATE,
-                    unit="s",
-                    description="Total time to generate streaming LLM responses (seconds)",
-                )
-
-                # Trace duration histogram
-                self.hist_trace_duration = self.meter.create_histogram(
-                    name=GEN_AI_TRACE_DURATION,
-                    unit="s",
-                    description="End-to-end GenAI trace duration (seconds)",
-                )
-
                self.metric_reader = metric_reader
            else:
                self.meter = None
-                self.meter_provider = None
                self.hist_llm_duration = None
-                self.hist_token_usage = None
-                self.hist_time_to_first_token = None
-                self.hist_time_to_generate = None
-                self.hist_trace_duration = None
                self.metric_reader = None
        except Exception:
            logger.exception("[Tencent APM] Metrics initialization failed; metrics disabled")
            self.meter = None
-            self.meter_provider = None
            self.hist_llm_duration = None
-            self.hist_token_usage = None
-            self.hist_time_to_first_token = None
-            self.hist_time_to_generate = None
-            self.hist_trace_duration = None
            self.metric_reader = None

    def add_span(self, span_data: SpanData) -> None:
@@ -284,158 +212,10 @@ class TencentTraceClient:
            if attributes:
                for k, v in attributes.items():
                    attrs[k] = str(v) if not isinstance(v, (str, int, float, bool)) else v  # type: ignore[assignment]
-
-            logger.info(
-                "[Tencent Metrics] Metric: %s | Value: %.4f | Attributes: %s",
-                LLM_OPERATION_DURATION,
-                latency_seconds,
-                json.dumps(attrs, ensure_ascii=False),
-            )
-
            self.hist_llm_duration.record(latency_seconds, attrs)  # type: ignore[attr-defined]
        except Exception:
            logger.debug("[Tencent APM] Failed to record LLM duration", exc_info=True)

-    def record_token_usage(
-        self,
-        token_count: int,
-        token_type: str,
-        operation_name: str,
-        request_model: str,
-        response_model: str,
-        server_address: str,
-        provider: str,
-    ) -> None:
-        """Record token usage histogram.
-
-        Args:
-            token_count: Number of tokens used
-            token_type: "input" or "output"
-            operation_name: Operation name (e.g., "chat")
-            request_model: Model used in request
-            response_model: Model used in response
-            server_address: Server address
-            provider: Model provider name
-        """
-        try:
-            if not hasattr(self, "hist_token_usage") or self.hist_token_usage is None:
-                return
-
-            attributes = {
-                "gen_ai.operation.name": operation_name,
-                "gen_ai.request.model": request_model,
-                "gen_ai.response.model": response_model,
-                "gen_ai.system": provider,
-                "gen_ai.token.type": token_type,
-                "server.address": server_address,
-            }
-
-            logger.info(
-                "[Tencent Metrics] Metric: %s | Value: %d | Attributes: %s",
-                GEN_AI_TOKEN_USAGE,
-                token_count,
-                json.dumps(attributes, ensure_ascii=False),
-            )
-
-            self.hist_token_usage.record(token_count, attributes)  # type: ignore[attr-defined]
-        except Exception:
-            logger.debug("[Tencent APM] Failed to record token usage", exc_info=True)
-
-    def record_time_to_first_token(
-        self, ttft_seconds: float, provider: str, model: str, operation_name: str = "chat"
-    ) -> None:
-        """Record time to first token histogram.
-
-        Args:
-            ttft_seconds: Time to first token in seconds
-            provider: Model provider name
-            model: Model name
-            operation_name: Operation name (default: "chat")
-        """
-        try:
-            if not hasattr(self, "hist_time_to_first_token") or self.hist_time_to_first_token is None:
-                return
-
-            attributes = {
-                "gen_ai.operation.name": operation_name,
-                "gen_ai.system": provider,
-                "gen_ai.request.model": model,
-                "gen_ai.response.model": model,
-                "stream": "true",
-            }
-
-            logger.info(
-                "[Tencent Metrics] Metric: %s | Value: %.4f | Attributes: %s",
-                GEN_AI_SERVER_TIME_TO_FIRST_TOKEN,
-                ttft_seconds,
-                json.dumps(attributes, ensure_ascii=False),
-            )
-
-            self.hist_time_to_first_token.record(ttft_seconds, attributes)  # type: ignore[attr-defined]
-        except Exception:
-            logger.debug("[Tencent APM] Failed to record time to first token", exc_info=True)
-
-    def record_time_to_generate(
-        self, ttg_seconds: float, provider: str, model: str, operation_name: str = "chat"
-    ) -> None:
-        """Record time to generate histogram.
-
-        Args:
-            ttg_seconds: Time to generate in seconds
-            provider: Model provider name
-            model: Model name
-            operation_name: Operation name (default: "chat")
-        """
-        try:
-            if not hasattr(self, "hist_time_to_generate") or self.hist_time_to_generate is None:
-                return
-
-            attributes = {
-                "gen_ai.operation.name": operation_name,
-                "gen_ai.system": provider,
-                "gen_ai.request.model": model,
-                "gen_ai.response.model": model,
-                "stream": "true",
-            }
-
-            logger.info(
-                "[Tencent Metrics] Metric: %s | Value: %.4f | Attributes: %s",
-                GEN_AI_STREAMING_TIME_TO_GENERATE,
-                ttg_seconds,
-                json.dumps(attributes, ensure_ascii=False),
-            )
-
-            self.hist_time_to_generate.record(ttg_seconds, attributes)  # type: ignore[attr-defined]
-        except Exception:
-            logger.debug("[Tencent APM] Failed to record time to generate", exc_info=True)
-
-    def record_trace_duration(self, duration_seconds: float, attributes: dict[str, str] | None = None) -> None:
-        """Record end-to-end trace duration histogram in seconds.
-
-        Args:
-            duration_seconds: Trace duration in seconds
-            attributes: Optional attributes (e.g., conversation_mode, app_id)
-        """
-        try:
-            if not hasattr(self, "hist_trace_duration") or self.hist_trace_duration is None:
-                return
-
-            attrs: dict[str, str] = {}
-            if attributes:
-                for k, v in attributes.items():
-                    attrs[k] = str(v) if not isinstance(v, (str, int, float, bool)) else v  # type: ignore[assignment]
-
-            logger.info(
-                "[Tencent Metrics] Metric: %s | Value: %.4f | Attributes: %s",
-                GEN_AI_TRACE_DURATION,
-                duration_seconds,
-                json.dumps(attrs, ensure_ascii=False),
-            )
-
-            self.hist_trace_duration.record(duration_seconds, attrs)  # type: ignore[attr-defined]
-        except Exception:
-            logger.debug("[Tencent APM] Failed to record trace duration", exc_info=True)
-
    def _create_and_export_span(self, span_data: SpanData) -> None:
        """Create span using OpenTelemetry Tracer API"""
        try:
@@ -516,19 +296,11 @@ class TencentTraceClient:

            if self.tracer_provider:
                self.tracer_provider.shutdown()
-
-            # Shutdown instance-level meter provider
-            if self.meter_provider is not None:
-                try:
-                    self.meter_provider.shutdown()  # type: ignore[attr-defined]
-                except Exception:
-                    logger.debug("[Tencent APM] Error shutting down meter provider", exc_info=True)
-
            if self.metric_reader is not None:
                try:
                    self.metric_reader.shutdown()  # type: ignore[attr-defined]
                except Exception:
-                    logger.debug("[Tencent APM] Error shutting down metric reader", exc_info=True)
+                    pass

        except Exception:
            logger.exception("[Tencent APM] Error during client shutdown")
@@ -47,9 +47,6 @@ GEN_AI_COMPLETION = "gen_ai.completion"

 GEN_AI_RESPONSE_FINISH_REASON = "gen_ai.response.finish_reason"

-# Streaming Span Attributes
-GEN_AI_IS_STREAMING_REQUEST = "llm.is_streaming"  # Same as OpenLLMetry semconv
-
 # Tool
 TOOL_NAME = "tool.name"

@@ -65,19 +62,6 @@ INSTRUMENTATION_LANGUAGE = "python"

 # Metrics
 LLM_OPERATION_DURATION = "gen_ai.client.operation.duration"
-GEN_AI_TOKEN_USAGE = "gen_ai.client.token.usage"
-GEN_AI_SERVER_TIME_TO_FIRST_TOKEN = "gen_ai.server.time_to_first_token"
-GEN_AI_STREAMING_TIME_TO_GENERATE = "gen_ai.streaming.time_to_generate"
-# The LLM trace duration which is exclusive to tencent apm
-GEN_AI_TRACE_DURATION = "gen_ai.trace.duration"
-
-# Token Usage Attributes
-GEN_AI_OPERATION_NAME = "gen_ai.operation.name"
-GEN_AI_REQUEST_MODEL = "gen_ai.request.model"
-GEN_AI_RESPONSE_MODEL = "gen_ai.response.model"
-GEN_AI_SYSTEM = "gen_ai.system"
-GEN_AI_TOKEN_TYPE = "gen_ai.token.type"
-SERVER_ADDRESS = "server.address"


 class GenAISpanKind(Enum):
@@ -14,11 +14,10 @@ from core.ops.entities.trace_entity import (
    ToolTraceInfo,
    WorkflowTraceInfo,
 )
-from core.ops.tencent_trace.entities.semconv import (
+from core.ops.tencent_trace.entities.tencent_semconv import (
    GEN_AI_COMPLETION,
    GEN_AI_FRAMEWORK,
    GEN_AI_IS_ENTRY,
-    GEN_AI_IS_STREAMING_REQUEST,
    GEN_AI_MODEL_NAME,
    GEN_AI_PROMPT,
    GEN_AI_PROVIDER,
@@ -157,25 +156,6 @@ class TencentSpanBuilder:
        outputs = node_execution.outputs or {}
        usage_data = process_data.get("usage", {}) if "usage" in process_data else outputs.get("usage", {})

-        attributes = {
-            GEN_AI_SESSION_ID: trace_info.metadata.get("conversation_id", ""),
-            GEN_AI_SPAN_KIND: GenAISpanKind.GENERATION.value,
-            GEN_AI_FRAMEWORK: "dify",
-            GEN_AI_MODEL_NAME: process_data.get("model_name", ""),
-            GEN_AI_PROVIDER: process_data.get("model_provider", ""),
-            GEN_AI_USAGE_INPUT_TOKENS: str(usage_data.get("prompt_tokens", 0)),
-            GEN_AI_USAGE_OUTPUT_TOKENS: str(usage_data.get("completion_tokens", 0)),
-            GEN_AI_USAGE_TOTAL_TOKENS: str(usage_data.get("total_tokens", 0)),
-            GEN_AI_PROMPT: json.dumps(process_data.get("prompts", []), ensure_ascii=False),
-            GEN_AI_COMPLETION: str(outputs.get("text", "")),
-            GEN_AI_RESPONSE_FINISH_REASON: outputs.get("finish_reason", ""),
-            INPUT_VALUE: json.dumps(process_data.get("prompts", []), ensure_ascii=False),
-            OUTPUT_VALUE: str(outputs.get("text", "")),
-        }
-
-        if usage_data.get("time_to_first_token") is not None:
-            attributes[GEN_AI_IS_STREAMING_REQUEST] = "true"
-
        return SpanData(
            trace_id=trace_id,
            parent_span_id=workflow_span_id,
@@ -183,7 +163,21 @@ class TencentSpanBuilder:
            name="GENERATION",
            start_time=TencentSpanBuilder._get_time_nanoseconds(node_execution.created_at),
            end_time=TencentSpanBuilder._get_time_nanoseconds(node_execution.finished_at),
-            attributes=attributes,
+            attributes={
+                GEN_AI_SESSION_ID: trace_info.metadata.get("conversation_id", ""),
+                GEN_AI_SPAN_KIND: GenAISpanKind.GENERATION.value,
+                GEN_AI_FRAMEWORK: "dify",
+                GEN_AI_MODEL_NAME: process_data.get("model_name", ""),
+                GEN_AI_PROVIDER: process_data.get("model_provider", ""),
+                GEN_AI_USAGE_INPUT_TOKENS: str(usage_data.get("prompt_tokens", 0)),
+                GEN_AI_USAGE_OUTPUT_TOKENS: str(usage_data.get("completion_tokens", 0)),
+                GEN_AI_USAGE_TOTAL_TOKENS: str(usage_data.get("total_tokens", 0)),
+                GEN_AI_PROMPT: json.dumps(process_data.get("prompts", []), ensure_ascii=False),
+                GEN_AI_COMPLETION: str(outputs.get("text", "")),
+                GEN_AI_RESPONSE_FINISH_REASON: outputs.get("finish_reason", ""),
+                INPUT_VALUE: json.dumps(process_data.get("prompts", []), ensure_ascii=False),
+                OUTPUT_VALUE: str(outputs.get("text", "")),
+            },
            status=TencentSpanBuilder._get_workflow_node_status(node_execution),
        )

@@ -197,19 +191,6 @@ class TencentSpanBuilder:
        if trace_info.error:
            status = Status(StatusCode.ERROR, trace_info.error)

-        attributes = {
-            GEN_AI_SESSION_ID: trace_info.metadata.get("conversation_id", ""),
-            GEN_AI_USER_ID: str(user_id),
-            GEN_AI_SPAN_KIND: GenAISpanKind.WORKFLOW.value,
-            GEN_AI_FRAMEWORK: "dify",
-            GEN_AI_IS_ENTRY: "true",
-            INPUT_VALUE: str(trace_info.inputs or ""),
-            OUTPUT_VALUE: str(trace_info.outputs or ""),
-        }
-
-        if trace_info.is_streaming_request:
-            attributes[GEN_AI_IS_STREAMING_REQUEST] = "true"
-
        return SpanData(
            trace_id=trace_id,
            parent_span_id=None,
@@ -217,7 +198,15 @@ class TencentSpanBuilder:
            name="message",
            start_time=TencentSpanBuilder._get_time_nanoseconds(trace_info.start_time),
            end_time=TencentSpanBuilder._get_time_nanoseconds(trace_info.end_time),
-            attributes=attributes,
+            attributes={
+                GEN_AI_SESSION_ID: trace_info.metadata.get("conversation_id", ""),
+                GEN_AI_USER_ID: str(user_id),
+                GEN_AI_SPAN_KIND: GenAISpanKind.WORKFLOW.value,
+                GEN_AI_FRAMEWORK: "dify",
+                GEN_AI_IS_ENTRY: "true",
+                INPUT_VALUE: str(trace_info.inputs or ""),
+                OUTPUT_VALUE: str(trace_info.outputs or ""),
+            },
            status=status,
            links=links,
        )
@@ -90,9 +90,6 @@ class TencentDataTrace(BaseTraceInstance):

            self._process_workflow_nodes(trace_info, trace_id)

-            # Record trace duration for entry span
-            self._record_workflow_trace_duration(trace_info)
-
        except Exception:
            logger.exception("[Tencent APM] Failed to process workflow trace")

@@ -110,11 +107,6 @@ class TencentDataTrace(BaseTraceInstance):

            self.trace_client.add_span(message_span)

-            self._record_message_llm_metrics(trace_info)
-
-            # Record trace duration for entry span
-            self._record_message_trace_duration(trace_info)
-
        except Exception:
            logger.exception("[Tencent APM] Failed to process message trace")

@@ -298,219 +290,24 @@ class TencentDataTrace(BaseTraceInstance):
    def _record_llm_metrics(self, node_execution: WorkflowNodeExecution) -> None:
        """Record LLM performance metrics"""
        try:
+            if not hasattr(self.trace_client, "record_llm_duration"):
+                return
+
            process_data = node_execution.process_data or {}
-            outputs = node_execution.outputs or {}
-            usage = process_data.get("usage", {}) if "usage" in process_data else outputs.get("usage", {})
+            usage = process_data.get("usage", {})
+            latency_s = float(usage.get("latency", 0.0))

-            model_provider = process_data.get("model_provider", "unknown")
-            model_name = process_data.get("model_name", "unknown")
-            model_mode = process_data.get("model_mode", "chat")
-
-            # Record LLM duration
-            if hasattr(self.trace_client, "record_llm_duration"):
-                latency_s = float(usage.get("latency", 0.0))
-
-                if latency_s > 0:
-                    # Determine if streaming from usage metrics
-                    is_streaming = usage.get("time_to_first_token") is not None
-
-                    attributes = {
-                        "gen_ai.system": model_provider,
-                        "gen_ai.response.model": model_name,
-                        "gen_ai.operation.name": model_mode,
-                        "stream": "true" if is_streaming else "false",
-                    }
-                    self.trace_client.record_llm_duration(latency_s, attributes)
-
-            # Record streaming metrics from usage
-            time_to_first_token = usage.get("time_to_first_token")
-            if time_to_first_token is not None and hasattr(self.trace_client, "record_time_to_first_token"):
-                ttft_seconds = float(time_to_first_token)
-                if ttft_seconds > 0:
-                    self.trace_client.record_time_to_first_token(
-                        ttft_seconds=ttft_seconds, provider=model_provider, model=model_name, operation_name=model_mode
-                    )
-
-            time_to_generate = usage.get("time_to_generate")
-            if time_to_generate is not None and hasattr(self.trace_client, "record_time_to_generate"):
-                ttg_seconds = float(time_to_generate)
-                if ttg_seconds > 0:
-                    self.trace_client.record_time_to_generate(
-                        ttg_seconds=ttg_seconds, provider=model_provider, model=model_name, operation_name=model_mode
-                    )
-
-            # Record token usage
-            if hasattr(self.trace_client, "record_token_usage"):
-                # Extract token counts
-                input_tokens = int(usage.get("prompt_tokens", 0))
-                output_tokens = int(usage.get("completion_tokens", 0))
-
-                if input_tokens > 0 or output_tokens > 0:
-                    server_address = f"{model_provider}"
-
-                    # Record input tokens
-                    if input_tokens > 0:
-                        self.trace_client.record_token_usage(
-                            token_count=input_tokens,
-                            token_type="input",
-                            operation_name=model_mode,
-                            request_model=model_name,
-                            response_model=model_name,
-                            server_address=server_address,
-                            provider=model_provider,
-                        )
-
-                    # Record output tokens
-                    if output_tokens > 0:
-                        self.trace_client.record_token_usage(
-                            token_count=output_tokens,
-                            token_type="output",
-                            operation_name=model_mode,
-                            request_model=model_name,
-                            response_model=model_name,
-                            server_address=server_address,
-                            provider=model_provider,
-                        )
+            if latency_s > 0:
+                attributes = {
+                    "provider": process_data.get("model_provider", ""),
+                    "model": process_data.get("model_name", ""),
+                    "span_kind": "GENERATION",
+                }
+                self.trace_client.record_llm_duration(latency_s, attributes)

        except Exception:
            logger.debug("[Tencent APM] Failed to record LLM metrics")

-    def _record_message_llm_metrics(self, trace_info: MessageTraceInfo) -> None:
-        """Record LLM metrics for message traces"""
-        try:
-            trace_metadata = trace_info.metadata or {}
-            message_data = trace_info.message_data or {}
-            provider_latency = 0.0
-            if isinstance(message_data, dict):
-                provider_latency = float(message_data.get("provider_response_latency", 0.0) or 0.0)
-            else:
-                provider_latency = float(getattr(message_data, "provider_response_latency", 0.0) or 0.0)
-
-            model_provider = trace_metadata.get("ls_provider") or (
-                message_data.get("model_provider", "") if isinstance(message_data, dict) else ""
-            )
-            model_name = trace_metadata.get("ls_model_name") or (
-                message_data.get("model_id", "") if isinstance(message_data, dict) else ""
-            )
-
-            # Record LLM duration
-            if provider_latency > 0 and hasattr(self.trace_client, "record_llm_duration"):
-                is_streaming = trace_info.is_streaming_request
-
-                duration_attributes = {
-                    "gen_ai.system": model_provider,
-                    "gen_ai.response.model": model_name,
-                    "gen_ai.operation.name": "chat",  # Message traces are always chat
-                    "stream": "true" if is_streaming else "false",
-                }
-                self.trace_client.record_llm_duration(provider_latency, duration_attributes)
-
-            # Record streaming metrics for message traces
-            if trace_info.is_streaming_request:
-                # Record time to first token
-                if trace_info.gen_ai_server_time_to_first_token is not None and hasattr(
-                    self.trace_client, "record_time_to_first_token"
-                ):
-                    ttft_seconds = float(trace_info.gen_ai_server_time_to_first_token)
-                    if ttft_seconds > 0:
-                        self.trace_client.record_time_to_first_token(
-                            ttft_seconds=ttft_seconds, provider=str(model_provider or ""), model=str(model_name or "")
-                        )
-
-                # Record time to generate
-                if trace_info.llm_streaming_time_to_generate is not None and hasattr(
-                    self.trace_client, "record_time_to_generate"
-                ):
-                    ttg_seconds = float(trace_info.llm_streaming_time_to_generate)
-                    if ttg_seconds > 0:
-                        self.trace_client.record_time_to_generate(
-                            ttg_seconds=ttg_seconds, provider=str(model_provider or ""), model=str(model_name or "")
-                        )
-
-            # Record token usage
-            if hasattr(self.trace_client, "record_token_usage"):
-                input_tokens = int(trace_info.message_tokens or 0)
-                output_tokens = int(trace_info.answer_tokens or 0)
-
-                if input_tokens > 0:
-                    self.trace_client.record_token_usage(
-                        token_count=input_tokens,
-                        token_type="input",
-                        operation_name="chat",
-                        request_model=str(model_name or ""),
-                        response_model=str(model_name or ""),
-                        server_address=str(model_provider or ""),
-                        provider=str(model_provider or ""),
-                    )
-
-                if output_tokens > 0:
-                    self.trace_client.record_token_usage(
-                        token_count=output_tokens,
-                        token_type="output",
-                        operation_name="chat",
-                        request_model=str(model_name or ""),
-                        response_model=str(model_name or ""),
-                        server_address=str(model_provider or ""),
-                        provider=str(model_provider or ""),
-                    )
-
-        except Exception:
-            logger.debug("[Tencent APM] Failed to record message LLM metrics")
-
-    def _record_workflow_trace_duration(self, trace_info: WorkflowTraceInfo) -> None:
-        """Record end-to-end workflow trace duration."""
-        try:
-            if not hasattr(self.trace_client, "record_trace_duration"):
-                return
-
-            # Calculate duration from start_time and end_time to match span duration
-            if trace_info.start_time and trace_info.end_time:
-                duration_s = (trace_info.end_time - trace_info.start_time).total_seconds()
-            else:
-                # Fallback to workflow_run_elapsed_time if timestamps not available
-                duration_s = float(trace_info.workflow_run_elapsed_time)
-
-            if duration_s > 0:
-                attributes = {
-                    "conversation_mode": "workflow",
-                    "workflow_status": trace_info.workflow_run_status,
-                }
-
-                # Add conversation_id if available
-                if trace_info.conversation_id:
-                    attributes["has_conversation"] = "true"
-                else:
-                    attributes["has_conversation"] = "false"
-
-                self.trace_client.record_trace_duration(duration_s, attributes)
-
-        except Exception:
-            logger.debug("[Tencent APM] Failed to record workflow trace duration")
-
-    def _record_message_trace_duration(self, trace_info: MessageTraceInfo) -> None:
-        """Record end-to-end message trace duration."""
-        try:
-            if not hasattr(self.trace_client, "record_trace_duration"):
-                return
-
-            # Calculate duration from start_time and end_time
-            if trace_info.start_time and trace_info.end_time:
-                duration = (trace_info.end_time - trace_info.start_time).total_seconds()
-
-                if duration > 0:
-                    attributes = {
-                        "conversation_mode": trace_info.conversation_mode,
-                    }
-
-                    # Add streaming flag if available
-                    if hasattr(trace_info, "is_streaming_request"):
-                        attributes["stream"] = "true" if trace_info.is_streaming_request else "false"
-
-                    self.trace_client.record_trace_duration(duration, attributes)
-
-        except Exception:
-            logger.debug("[Tencent APM] Failed to record message trace duration")
-
    def __del__(self):
        """Ensure proper cleanup on garbage collection."""
        try:
@@ -45,12 +45,6 @@ class MemoryConfig(BaseModel):
        enabled: bool
        size: int | None = None

-    mode: Literal["linear", "block"] | None = "linear"
-    block_id: list[str] | None = None
    role_prefix: RolePrefix | None = None
    window: WindowConfig
    query_prompt_template: str | None = None
-
-    @property
-    def is_block_mode(self) -> bool:
-        return self.mode == "block" and bool(self.block_id)
@@ -618,18 +618,18 @@ class ProviderManager:
                )

            for quota in configuration.quotas:
-                if quota.quota_type in (ProviderQuotaType.TRIAL, ProviderQuotaType.PAID):
+                if quota.quota_type == ProviderQuotaType.TRIAL:
                    # Init trial provider records if not exists
-                    if quota.quota_type not in provider_quota_to_provider_record_dict:
+                    if ProviderQuotaType.TRIAL not in provider_quota_to_provider_record_dict:
                        try:
                            # FIXME ignore the type error, only TrialHostingQuota has limit need to change the logic
                            new_provider_record = Provider(
                                tenant_id=tenant_id,
                                # TODO: Use provider name with prefix after the data migration.
                                provider_name=ModelProviderID(provider_name).provider_name,
-                                provider_type=ProviderType.SYSTEM.value,
-                                quota_type=quota.quota_type,
-                                quota_limit=0,  # type: ignore
+                                provider_type=ProviderType.SYSTEM,
+                                quota_type=ProviderQuotaType.TRIAL,
+                                quota_limit=quota.quota_limit,  # type: ignore
                                quota_used=0,
                                is_valid=True,
                            )
@@ -641,8 +641,8 @@ class ProviderManager:
                            stmt = select(Provider).where(
                                Provider.tenant_id == tenant_id,
                                Provider.provider_name == ModelProviderID(provider_name).provider_name,
-                                Provider.provider_type == ProviderType.SYSTEM.value,
-                                Provider.quota_type == quota.quota_type,
+                                Provider.provider_type == ProviderType.SYSTEM,
+                                Provider.quota_type == ProviderQuotaType.TRIAL,
                            )
                            existed_provider_record = db.session.scalar(stmt)
                            if not existed_provider_record:
@@ -652,7 +652,7 @@ class ProviderManager:
                                existed_provider_record.is_valid = True
                                db.session.commit()

-                            provider_name_to_provider_records_dict[provider_name].append(existed_provider_record)                        
+                            provider_name_to_provider_records_dict[provider_name].append(existed_provider_record)

        return provider_name_to_provider_records_dict

@@ -912,22 +912,6 @@ class ProviderManager:
                provider_record
            )
        quota_configurations = []
-
-        if dify_config.EDITION == "CLOUD":
-            from services.credit_pool_service import CreditPoolService
-
-            trail_pool = CreditPoolService.get_pool(
-                tenant_id=tenant_id,
-                pool_type=ProviderQuotaType.TRIAL.value,
-            )
-            paid_pool = CreditPoolService.get_pool(
-                tenant_id=tenant_id,
-                pool_type=ProviderQuotaType.PAID.value,
-            )
-        else:
-            trail_pool = None
-            paid_pool = None
-
        for provider_quota in provider_hosting_configuration.quotas:
            if provider_quota.quota_type not in quota_type_to_provider_records_dict:
                if provider_quota.quota_type == ProviderQuotaType.FREE:
@@ -948,36 +932,16 @@ class ProviderManager:
                    raise ValueError("quota_used is None")
                if provider_record.quota_limit is None:
                    raise ValueError("quota_limit is None")
-                if provider_quota.quota_type == ProviderQuotaType.TRIAL and trail_pool is not None:
-                    quota_configuration = QuotaConfiguration(
-                        quota_type=provider_quota.quota_type,
-                        quota_unit=provider_hosting_configuration.quota_unit or QuotaUnit.TOKENS,
-                        quota_used=trail_pool.quota_used,
-                        quota_limit=trail_pool.quota_limit,
-                        is_valid=trail_pool.quota_limit > trail_pool.quota_used or trail_pool.quota_limit == -1,
-                        restrict_models=provider_quota.restrict_models,
-                    )

-                elif provider_quota.quota_type == ProviderQuotaType.PAID and paid_pool is not None:
-                    quota_configuration = QuotaConfiguration(
-                        quota_type=provider_quota.quota_type,
-                        quota_unit=provider_hosting_configuration.quota_unit or QuotaUnit.TOKENS,
-                        quota_used=paid_pool.quota_used,
-                        quota_limit=paid_pool.quota_limit,
-                        is_valid=paid_pool.quota_limit > paid_pool.quota_used or paid_pool.quota_limit == -1,
-                        restrict_models=provider_quota.restrict_models,
-                    )
-
-                else:
-                    quota_configuration = QuotaConfiguration(
-                        quota_type=provider_quota.quota_type,
-                        quota_unit=provider_hosting_configuration.quota_unit or QuotaUnit.TOKENS,
-                        quota_used=provider_record.quota_used,
-                        quota_limit=provider_record.quota_limit,
-                        is_valid=provider_record.quota_limit > provider_record.quota_used
-                        or provider_record.quota_limit == -1,
-                        restrict_models=provider_quota.restrict_models,
-                    )
+                quota_configuration = QuotaConfiguration(
+                    quota_type=provider_quota.quota_type,
+                    quota_unit=provider_hosting_configuration.quota_unit or QuotaUnit.TOKENS,
+                    quota_used=provider_record.quota_used,
+                    quota_limit=provider_record.quota_limit,
+                    is_valid=provider_record.quota_limit > provider_record.quota_used
+                    or provider_record.quota_limit == -1,
+                    restrict_models=provider_quota.restrict_models,
+                )

            quota_configurations.append(quota_configuration)

@@ -289,8 +289,7 @@ class OracleVector(BaseVector):
                words = pseg.cut(query)
                current_entity = ""
                for word, pos in words:
-                    # nr: person name, ns: place name, nt: organization name
-                    if pos in {"nr", "Ng", "eng", "nz", "n", "ORG", "v"}:
+                    if pos in {"nr", "Ng", "eng", "nz", "n", "ORG", "v"}:  # nr: 人名，ns: 地名，nt: 机构名
                        current_entity += word
                    else:
                        if current_entity:
@@ -213,7 +213,7 @@ class VastbaseVector(BaseVector):

            with self._get_cursor() as cur:
                cur.execute(SQL_CREATE_TABLE.format(table_name=self.table_name, dimension=dimension))
-                # Vastbase supports vector dimensions in range [1, 16000]
+                # Vastbase 支持的向量维度取值范围为 [1,16000]
                if dimension <= 16000:
                    cur.execute(SQL_CREATE_INDEX.format(table_name=self.table_name))
            redis_client.set(collection_exist_cache_key, 1, ex=3600)
@@ -39,13 +39,11 @@ class WeaviateConfig(BaseModel):

    Attributes:
        endpoint: Weaviate server endpoint URL
-        grpc_endpoint: Optional Weaviate gRPC server endpoint URL
        api_key: Optional API key for authentication
        batch_size: Number of objects to batch per insert operation
    """

    endpoint: str
-    grpc_endpoint: str | None = None
    api_key: str | None = None
    batch_size: int = 100

@@ -90,22 +88,9 @@ class WeaviateVector(BaseVector):
        http_secure = p.scheme == "https"
        http_port = p.port or (443 if http_secure else 80)

-        # Parse gRPC configuration
-        if config.grpc_endpoint:
-            # Urls without scheme won't be parsed correctly in some python verions,
-            # see https://bugs.python.org/issue27657
-            grpc_endpoint_with_scheme = (
-                config.grpc_endpoint if "://" in config.grpc_endpoint else f"grpc://{config.grpc_endpoint}"
-            )
-            grpc_p = urlparse(grpc_endpoint_with_scheme)
-            grpc_host = grpc_p.hostname or "localhost"
-            grpc_port = grpc_p.port or (443 if grpc_p.scheme == "grpcs" else 50051)
-            grpc_secure = grpc_p.scheme == "grpcs"
-        else:
-            # Infer from HTTP endpoint as fallback
-            grpc_host = host
-            grpc_secure = http_secure
-            grpc_port = 443 if grpc_secure else 50051
+        grpc_host = host
+        grpc_secure = http_secure
+        grpc_port = 443 if grpc_secure else 50051

        client = weaviate.connect_to_custom(
            http_host=host,
@@ -115,7 +100,6 @@ class WeaviateVector(BaseVector):
            grpc_port=grpc_port,
            grpc_secure=grpc_secure,
            auth_credentials=Auth.api_key(config.api_key) if config.api_key else None,
-            skip_init_checks=True,  # Skip PyPI version check to avoid unnecessary HTTP requests
        )

        if not client.is_ready():
@@ -447,7 +431,6 @@ class WeaviateVectorFactory(AbstractVectorFactory):
            collection_name=collection_name,
            config=WeaviateConfig(
                endpoint=dify_config.WEAVIATE_ENDPOINT or "",
-                grpc_endpoint=dify_config.WEAVIATE_GRPC_ENDPOINT or "",
                api_key=dify_config.WEAVIATE_API_KEY,
                batch_size=dify_config.WEAVIATE_BATCH_SIZE,
            ),
@@ -1,89 +0,0 @@
-import json
-from dataclasses import dataclass
-from typing import Any
-
-from extensions.ext_redis import redis_client
-
-TASK_WRAPPER_PREFIX = "__WRAPPER__:"
-
-
-@dataclass
-class TaskWrapper:
-    data: Any
-    
-    def serialize(self) -> str:
-        return json.dumps(self.data, ensure_ascii=False)
-    
-    @classmethod
-    def deserialize(cls, serialized_data: str) -> 'TaskWrapper':
-        data = json.loads(serialized_data)
-        return cls(data)
-
-
-class TenantSelfTaskQueue:
-    """
-    Simple queue for tenant self tasks, used for tenant self task isolation.
-    It uses Redis list to store tasks, and Redis key to store task waiting flag.
-    Support tasks that can be serialized by json.
-    """
-    DEFAULT_TASK_TTL = 60 * 60
-
-    def __init__(self, tenant_id: str, unique_key: str):
-        self.tenant_id = tenant_id
-        self.unique_key = unique_key
-        self.queue = f"tenant_self_{unique_key}_task_queue:{tenant_id}"
-        self.task_key = f"tenant_{unique_key}_task:{tenant_id}"
-
-    def get_task_key(self):
-        return redis_client.get(self.task_key)
-
-    def set_task_waiting_time(self, ttl: int | None = None):
-        ttl = ttl or self.DEFAULT_TASK_TTL
-        redis_client.setex(self.task_key, ttl, 1)
-
-    def delete_task_key(self):
-        redis_client.delete(self.task_key)
-
-    def push_tasks(self, tasks: list):
-        serialized_tasks = []
-        for task in tasks:
-            # Store str list directly, maintaining full compatibility for pipeline scenarios
-            if isinstance(task, str):
-                serialized_tasks.append(task)
-            else:
-                # Use TaskWrapper to do JSON serialization, add prefix for identification
-                wrapper = TaskWrapper(task)
-                serialized_data = wrapper.serialize()
-                serialized_tasks.append(f"{TASK_WRAPPER_PREFIX}{serialized_data}")
-        
-        redis_client.lpush(self.queue, *serialized_tasks)
-    
-    def pull_tasks(self, count: int = 1) -> list:
-        if count <= 0:
-            return []
-        
-        tasks = []
-        for _ in range(count):
-            serialized_task = redis_client.rpop(self.queue)
-            if not serialized_task:
-                break
-            
-            if isinstance(serialized_task, bytes):
-                serialized_task = serialized_task.decode('utf-8')
-            
-            # Check if use TaskWrapper or not
-            if serialized_task.startswith(TASK_WRAPPER_PREFIX):
-                try:
-                    wrapper_data = serialized_task[len(TASK_WRAPPER_PREFIX):]
-                    wrapper = TaskWrapper.deserialize(wrapper_data)
-                    tasks.append(wrapper.data)
-                except (json.JSONDecodeError, TypeError, ValueError):
-                    tasks.append(serialized_task)
-            else:
-                tasks.append(serialized_task)
-        
-        return tasks
-
-    def get_next_task(self):
-        tasks = self.pull_tasks(1)
-        return tasks[0] if tasks else None
@@ -210,13 +210,12 @@ class Tool(ABC):
            meta=meta,
        )

-    def create_json_message(self, object: dict, suppress_output: bool = False) -> ToolInvokeMessage:
+    def create_json_message(self, object: dict) -> ToolInvokeMessage:
        """
        create a json message
        """
        return ToolInvokeMessage(
-            type=ToolInvokeMessage.MessageType.JSON,
-            message=ToolInvokeMessage.JsonMessage(json_object=object, suppress_output=suppress_output),
+            type=ToolInvokeMessage.MessageType.JSON, message=ToolInvokeMessage.JsonMessage(json_object=object)
        )

    def create_variable_message(
@@ -129,7 +129,6 @@ class ToolInvokeMessage(BaseModel):

    class JsonMessage(BaseModel):
        json_object: dict
-        suppress_output: bool = Field(default=False, description="Whether to suppress JSON output in result string")

    class BlobMessage(BaseModel):
        blob: bytes
@@ -228,41 +228,29 @@ class ToolEngine:
        """
        Handle tool response
        """
-        parts: list[str] = []
-        json_parts: list[str] = []
-
+        result = ""
        for response in tool_response:
            if response.type == ToolInvokeMessage.MessageType.TEXT:
-                parts.append(cast(ToolInvokeMessage.TextMessage, response.message).text)
+                result += cast(ToolInvokeMessage.TextMessage, response.message).text
            elif response.type == ToolInvokeMessage.MessageType.LINK:
-                parts.append(
+                result += (
                    f"result link: {cast(ToolInvokeMessage.TextMessage, response.message).text}."
                    + " please tell user to check it."
                )
            elif response.type in {ToolInvokeMessage.MessageType.IMAGE_LINK, ToolInvokeMessage.MessageType.IMAGE}:
-                parts.append(
+                result += (
                    "image has been created and sent to user already, "
                    + "you do not need to create it, just tell the user to check it now."
                )
            elif response.type == ToolInvokeMessage.MessageType.JSON:
-                json_message = cast(ToolInvokeMessage.JsonMessage, response.message)
-                if json_message.suppress_output:
-                    continue
-                json_parts.append(
-                    json.dumps(
-                        safe_json_value(cast(ToolInvokeMessage.JsonMessage, response.message).json_object),
-                        ensure_ascii=False,
-                    )
+                result += json.dumps(
+                    safe_json_value(cast(ToolInvokeMessage.JsonMessage, response.message).json_object),
+                    ensure_ascii=False,
                )
            else:
-                parts.append(str(response.message))
+                result += str(response.message)

-        # Add JSON parts, avoiding duplicates from text parts.
-        if json_parts:
-            existing_parts = set(parts)
-            parts.extend(p for p in json_parts if p not in existing_parts)
-
-        return "".join(parts)
+        return result

    @staticmethod
    def _extract_tool_response_binary_and_text(
@@ -117,7 +117,7 @@ class WorkflowTool(Tool):
        self._latest_usage = self._derive_usage_from_result(data)

        yield self.create_text_message(json.dumps(outputs, ensure_ascii=False))
-        yield self.create_json_message(outputs, suppress_output=True)
+        yield self.create_json_message(outputs)

    @property
    def latest_usage(self) -> LLMUsage:
@@ -203,48 +203,6 @@ class ArrayFileSegment(ArraySegment):
        return ""


-class VersionedMemoryValue(BaseModel):
-    current_value: str = None  # type: ignore
-    versions: Mapping[str, str] = {}
-
-    model_config = ConfigDict(frozen=True)
-
-    def add_version(
-        self,
-        new_value: str,
-        version_name: str | None = None
-    ) -> "VersionedMemoryValue":
-        if version_name is None:
-            version_name = str(len(self.versions) + 1)
-        if version_name in self.versions:
-            raise ValueError(f"Version '{version_name}' already exists.")
-        self.current_value = new_value
-        return VersionedMemoryValue(
-            current_value=new_value,
-            versions={
-                version_name: new_value,
-                **self.versions,
-            }
-        )
-
-
-class VersionedMemorySegment(Segment):
-    value_type: SegmentType = SegmentType.VERSIONED_MEMORY
-    value: VersionedMemoryValue = None  # type: ignore
-
-    @property
-    def text(self) -> str:
-        return self.value.current_value
-
-    @property
-    def log(self) -> str:
-        return self.value.current_value
-
-    @property
-    def markdown(self) -> str:
-        return self.value.current_value
-
-
 class ArrayBooleanSegment(ArraySegment):
    value_type: SegmentType = SegmentType.ARRAY_BOOLEAN
    value: Sequence[bool]
@@ -290,7 +248,6 @@ SegmentUnion: TypeAlias = Annotated[
        | Annotated[ArrayObjectSegment, Tag(SegmentType.ARRAY_OBJECT)]
        | Annotated[ArrayFileSegment, Tag(SegmentType.ARRAY_FILE)]
        | Annotated[ArrayBooleanSegment, Tag(SegmentType.ARRAY_BOOLEAN)]
-        | Annotated[VersionedMemorySegment, Tag(SegmentType.VERSIONED_MEMORY)]
    ),
    Discriminator(get_segment_discriminator),
 ]
@@ -41,8 +41,6 @@ class SegmentType(StrEnum):
    ARRAY_FILE = "array[file]"
    ARRAY_BOOLEAN = "array[boolean]"

-    VERSIONED_MEMORY = "versioned_memory"
-
    NONE = "none"

    GROUP = "group"
@@ -22,7 +22,6 @@ from .segments import (
    ObjectSegment,
    Segment,
    StringSegment,
-    VersionedMemorySegment,
    get_segment_discriminator,
 )
 from .types import SegmentType
@@ -107,10 +106,6 @@ class ArrayFileVariable(ArrayFileSegment, ArrayVariable):
    pass


-class VersionedMemoryVariable(VersionedMemorySegment, Variable):
-    pass
-
-
 class ArrayBooleanVariable(ArrayBooleanSegment, ArrayVariable):
    pass

@@ -166,7 +161,6 @@ VariableUnion: TypeAlias = Annotated[
        | Annotated[ArrayFileVariable, Tag(SegmentType.ARRAY_FILE)]
        | Annotated[ArrayBooleanVariable, Tag(SegmentType.ARRAY_BOOLEAN)]
        | Annotated[SecretVariable, Tag(SegmentType.SECRET)]
-        | Annotated[VersionedMemoryVariable, Tag(SegmentType.VERSIONED_MEMORY)]
    ),
    Discriminator(get_segment_discriminator),
 ]
@@ -1,5 +1,4 @@
 SYSTEM_VARIABLE_NODE_ID = "sys"
 ENVIRONMENT_VARIABLE_NODE_ID = "env"
 CONVERSATION_VARIABLE_NODE_ID = "conversation"
-MEMORY_BLOCK_VARIABLE_NODE_ID = "memory_block"
 RAG_PIPELINE_VARIABLE_NODE_ID = "rag"
@@ -4,7 +4,6 @@ from .agent import AgentNodeStrategyInit
 from .graph_init_params import GraphInitParams
 from .workflow_execution import WorkflowExecution
 from .workflow_node_execution import WorkflowNodeExecution
-from .workflow_pause import WorkflowPauseEntity

 __all__ = [
    "AgentNodeStrategyInit",
@@ -13,5 +12,4 @@ __all__ = [
    "VariablePool",
    "WorkflowExecution",
    "WorkflowNodeExecution",
-    "WorkflowPauseEntity",
 ]
@@ -1,49 +0,0 @@
-from enum import StrEnum, auto
-from typing import Annotated, Any, ClassVar, TypeAlias
-
-from pydantic import BaseModel, Discriminator, Tag
-
-
-class _PauseReasonType(StrEnum):
-    HUMAN_INPUT_REQUIRED = auto()
-    SCHEDULED_PAUSE = auto()
-
-
-class _PauseReasonBase(BaseModel):
-    TYPE: ClassVar[_PauseReasonType]
-
-
-class HumanInputRequired(_PauseReasonBase):
-    TYPE = _PauseReasonType.HUMAN_INPUT_REQUIRED
-
-
-class SchedulingPause(_PauseReasonBase):
-    TYPE = _PauseReasonType.SCHEDULED_PAUSE
-
-    message: str
-
-
-def _get_pause_reason_discriminator(v: Any) -> _PauseReasonType | None:
-    if isinstance(v, _PauseReasonBase):
-        return v.TYPE
-    elif isinstance(v, dict):
-        reason_type_str = v.get("TYPE")
-        if reason_type_str is None:
-            return None
-        try:
-            reason_type = _PauseReasonType(reason_type_str)
-        except ValueError:
-            return None
-        return reason_type
-    else:
-        # return None if the discriminator value isn't found
-        return None
-
-
-PauseReason: TypeAlias = Annotated[
-    (
-        Annotated[HumanInputRequired, Tag(_PauseReasonType.HUMAN_INPUT_REQUIRED)]
-        | Annotated[SchedulingPause, Tag(_PauseReasonType.SCHEDULED_PAUSE)]
-    ),
-    Discriminator(_get_pause_reason_discriminator),
-]
@@ -1,61 +0,0 @@
-"""
-Domain entities for workflow pause management.
-
-This module contains the domain model for workflow pause, which is used
-by the core workflow module. These models are independent of the storage mechanism
-and don't contain implementation details like tenant_id, app_id, etc.
-"""
-
-from abc import ABC, abstractmethod
-from datetime import datetime
-
-
-class WorkflowPauseEntity(ABC):
-    """
-    Abstract base class for workflow pause entities.
-
-    This domain model represents a paused workflow execution state,
-    without implementation details like tenant_id, app_id, etc.
-    It provides the interface for managing workflow pause/resume operations
-    and state persistence through file storage.
-
-    The `WorkflowPauseEntity` is never reused. If a workflow execution pauses multiple times,
-    it will generate multiple `WorkflowPauseEntity` records.
-    """
-
-    @property
-    @abstractmethod
-    def id(self) -> str:
-        """The identifier of current WorkflowPauseEntity"""
-        pass
-
-    @property
-    @abstractmethod
-    def workflow_execution_id(self) -> str:
-        """The identifier of the workflow execution record the pause associated with.
-        Correspond to `WorkflowExecution.id`.
-        """
-
-    @abstractmethod
-    def get_state(self) -> bytes:
-        """
-        Retrieve the serialized workflow state from storage.
-
-        This method should load and return the workflow execution state
-        that was saved when the workflow was paused. The state contains
-        all necessary information to resume the workflow execution.
-
-        Returns:
-            bytes: The serialized workflow state containing
-            execution context, variable values, node states, etc.
-
-        """
-        ...
-
-    @property
-    @abstractmethod
-    def resumed_at(self) -> datetime | None:
-        """`resumed_at` return the resumption time of the current pause, or `None` if
-        the pause is not resumed yet.
-        """
-        pass
@@ -92,111 +92,13 @@ class WorkflowType(StrEnum):


 class WorkflowExecutionStatus(StrEnum):
-    # State diagram for the workflw status:
-    # (@) means start, (*) means end
-    #
-    #       ┌------------------>------------------------->------------------->--------------┐
-    #       |                                                                               |
-    #       |                       ┌-----------------------<--------------------┐          |
-    #       ^                       |                                            |          |
-    #       |                       |                                            ^          |
-    #       |                       V                                            |          |
-    # ┌-----------┐        ┌-----------------------┐                       ┌-----------┐    V
-    # | Scheduled |------->|        Running        |---------------------->|   paused  |    |
-    # └-----------┘        └-----------------------┘                       └-----------┘    |
-    #       |                |       |       |    |                              |          |
-    #       |                |       |       |    |                              |          |
-    #       ^                |       |       |    V                              V          |
-    #       |                |       |       |    |                         ┌---------┐     |
-    #      (@)               |       |       |    └------------------------>| Stopped |<----┘
-    #                        |       |       |                              └---------┘
-    #                        |       |       |                                   |
-    #                        |       |       V                                   V
-    #                        |       |  ┌-----------┐                            |
-    #                        |       |  | Succeeded |------------->--------------┤
-    #                        |       |  └-----------┘                            |
-    #                        |       V                                           V
-    #                        |  +--------┐                                       |
-    #                        |  | Failed |---------------------->----------------┤
-    #                        |  └--------┘                                       |
-    #                        V                                                   V
-    #             ┌---------------------┐                                        |
-    #             | Partially Succeeded |---------------------->-----------------┘--------> (*)
-    #             └---------------------┘
-    #
-    # Mermaid diagram:
-    #
-    #     ---
-    #     title: State diagram for Workflow run state
-    #     ---
-    #     stateDiagram-v2
-    #         scheduled: Scheduled
-    #         running: Running
-    #         succeeded: Succeeded
-    #         failed: Failed
-    #         partial_succeeded: Partial Succeeded
-    #         paused: Paused
-    #         stopped: Stopped
-    #
-    #         [*] --> scheduled:
-    #         scheduled --> running: Start Execution
-    #         running --> paused: Human input required
-    #         paused --> running: human input added
-    #         paused --> stopped: User stops execution
-    #         running --> succeeded: Execution finishes without any error
-    #         running --> failed: Execution finishes with errors
-    #         running --> stopped: User stops execution
-    #         running --> partial_succeeded: some execution occurred and handled during execution
-    #
-    #         scheduled --> stopped: User stops execution
-    #
-    #         succeeded --> [*]
-    #         failed --> [*]
-    #         partial_succeeded --> [*]
-    #         stopped --> [*]
-
-    # `SCHEDULED` means that the workflow is scheduled to run, but has not
-    # started running yet. (maybe due to possible worker saturation.)
-    #
-    # This enum value is currently unused.
-    SCHEDULED = "scheduled"
-
-    # `RUNNING` means the workflow is exeuting.
    RUNNING = "running"
-
-    # `SUCCEEDED` means the execution of workflow succeed without any error.
    SUCCEEDED = "succeeded"
-
-    # `FAILED` means the execution of workflow failed without some errors.
    FAILED = "failed"
-
-    # `STOPPED` means the execution of workflow was stopped, either manually
-    # by the user, or automatically by the Dify application (E.G. the moderation
-    # mechanism.)
    STOPPED = "stopped"
-
-    # `PARTIAL_SUCCEEDED` indicates that some errors occurred during the workflow
-    # execution, but they were successfully handled (e.g., by using an error
-    # strategy such as "fail branch" or "default value").
    PARTIAL_SUCCEEDED = "partial-succeeded"
-
-    # `PAUSED` indicates that the workflow execution is temporarily paused
-    # (e.g., awaiting human input) and is expected to resume later.
    PAUSED = "paused"

-    def is_ended(self) -> bool:
-        return self in _END_STATE
-
-
-_END_STATE = frozenset(
-    [
-        WorkflowExecutionStatus.SUCCEEDED,
-        WorkflowExecutionStatus.FAILED,
-        WorkflowExecutionStatus.PARTIAL_SUCCEEDED,
-        WorkflowExecutionStatus.STOPPED,
-    ]
-)
-

 class WorkflowNodeExecutionMetadataKey(StrEnum):
    """
@@ -3,8 +3,6 @@ from typing import final

 from typing_extensions import override

-from core.workflow.entities.pause_reason import SchedulingPause
-
 from ..domain.graph_execution import GraphExecution
 from ..entities.commands import AbortCommand, GraphEngineCommand, PauseCommand
 from .command_processor import CommandHandler
@@ -27,7 +25,4 @@ class PauseCommandHandler(CommandHandler):
    def handle(self, command: GraphEngineCommand, execution: GraphExecution) -> None:
        assert isinstance(command, PauseCommand)
        logger.debug("Pausing workflow %s: %s", execution.workflow_id, command.reason)
-        # Convert string reason to PauseReason if needed
-        reason = command.reason
-        pause_reason = SchedulingPause(message=reason)
-        execution.pause(pause_reason)
+        execution.pause(command.reason)
@@ -8,7 +8,6 @@ from typing import Literal

 from pydantic import BaseModel, Field

-from core.workflow.entities.pause_reason import PauseReason
 from core.workflow.enums import NodeState

 from .node_execution import NodeExecution
@@ -42,7 +41,7 @@ class GraphExecutionState(BaseModel):
    completed: bool = Field(default=False)
    aborted: bool = Field(default=False)
    paused: bool = Field(default=False)
-    pause_reason: PauseReason | None = Field(default=None)
+    pause_reason: str | None = Field(default=None)
    error: GraphExecutionErrorState | None = Field(default=None)
    exceptions_count: int = Field(default=0)
    node_executions: list[NodeExecutionState] = Field(default_factory=list[NodeExecutionState])
@@ -107,7 +106,7 @@ class GraphExecution:
    completed: bool = False
    aborted: bool = False
    paused: bool = False
-    pause_reason: PauseReason | None = None
+    pause_reason: str | None = None
    error: Exception | None = None
    node_executions: dict[str, NodeExecution] = field(default_factory=dict[str, NodeExecution])
    exceptions_count: int = 0
@@ -131,7 +130,7 @@ class GraphExecution:
        self.aborted = True
        self.error = RuntimeError(f"Aborted: {reason}")

-    def pause(self, reason: PauseReason) -> None:
+    def pause(self, reason: str | None = None) -> None:
        """Pause the graph execution without marking it complete."""
        if self.completed:
            raise RuntimeError("Cannot pause execution that has completed")
@@ -36,4 +36,4 @@ class PauseCommand(GraphEngineCommand):
    """Command to pause a running workflow execution."""

    command_type: CommandType = Field(default=CommandType.PAUSE, description="Type of command")
-    reason: str = Field(default="unknown reason", description="reason for pause")
+    reason: str | None = Field(default=None, description="Optional reason for pause")
@@ -210,7 +210,7 @@ class EventHandler:
    def _(self, event: NodeRunPauseRequestedEvent) -> None:
        """Handle pause requests emitted by nodes."""

-        pause_reason = event.reason
+        pause_reason = event.reason or "Awaiting human input"
        self._graph_execution.pause(pause_reason)
        self._state_manager.finish_execution(event.node_id)
        if event.node_id in self._graph.nodes:
@@ -247,11 +247,8 @@ class GraphEngine:

            # Handle completion
            if self._graph_execution.is_paused:
-                pause_reason = self._graph_execution.pause_reason
-                assert pause_reason is not None, "pause_reason should not be None when execution is paused."
-                # Ensure we have a valid PauseReason for the event
                paused_event = GraphRunPausedEvent(
-                    reason=pause_reason,
+                    reason=self._graph_execution.pause_reason,
                    outputs=self._graph_runtime_state.outputs,
                )
                self._event_manager.notify_layers(paused_event)
@@ -216,6 +216,7 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
    def _handle_graph_run_paused(self, event: GraphRunPausedEvent) -> None:
        execution = self._get_workflow_execution()
        execution.status = WorkflowExecutionStatus.PAUSED
+        execution.error_message = event.reason or "Workflow execution paused"
        execution.outputs = event.outputs
        self._populate_completion_statistics(execution, update_finished=False)

@@ -295,7 +296,7 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
            domain_execution,
            event.node_run_result,
            WorkflowNodeExecutionStatus.PAUSED,
-            error="",
+            error=event.reason,
            update_outputs=False,
        )

--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
Novice	3d02f425d0	chore: fix test cases	2025-10-29 14:51:19 +08:00
NoviceandGitHub	5f65466551	Merge branch 'main' into mcp-conditional-capability	2025-10-29 14:15:32 +08:00
VivecandGitHub	baa985df4d	Merge branch 'main' into mcp-conditional-capability	2025-09-29 12:26:26 +08:00
autofix-ci[bot]andGitHub	09d7d5670e	[autofix.ci] apply automated fixes	2025-09-25 09:14:44 +00:00
Vivecccccc	6e3442bd12	set conditional capabilities for SamplingCapability and RootsCapability upon MCP client session initialization to prevent servers from attempting callbacks when the client does not actually support them	2025-09-25 16:46:00 +08:00