Merge branch 'feat/memory-orchestration-be' into feat/memory-orchestration-be-dev-env

# Conflicts:
#	api/controllers/console/app/workflow.py
#	api/services/workflow_service.py

.devcontainer/post_create_command.sh

@@ -1,7 +1,6 @@
 #!/bin/bash
 WORKSPACE_ROOT=$(pwd)
 
-npm add -g pnpm@10.15.0
 corepack enable
 cd web && pnpm install
 pipx install uv

.github/workflows/autofix.yml

@@ -2,8 +2,6 @@ name: autofix.ci
 on:
   pull_request:
     branches: ["main"]
-  push:
-    branches: ["main"]
 permissions:
   contents: read
 
@@ -32,8 +30,6 @@ jobs:
         run: |
           uvx --from ast-grep-cli sg --pattern 'db.session.query($WHATEVER).filter($HERE)' --rewrite 'db.session.query($WHATEVER).where($HERE)' -l py --update-all
           uvx --from ast-grep-cli sg --pattern 'session.query($WHATEVER).filter($HERE)' --rewrite 'session.query($WHATEVER).where($HERE)' -l py --update-all
-          uvx --from ast-grep-cli sg -p '$A = db.Column($$$B)' -r '$A = mapped_column($$$B)' -l py --update-all
-          uvx --from ast-grep-cli sg -p '$A : $T = db.Column($$$B)' -r '$A : $T = mapped_column($$$B)' -l py --update-all
           # Convert Optional[T] to T | None (ignoring quoted types)
           cat > /tmp/optional-rule.yml << 'EOF'
           id: convert-optional-to-union

.github/workflows/build-push.yml

@@ -4,7 +4,8 @@ on:
   push:
     branches:
       - "main"
-      - "deploy/**"
+      - "deploy/dev"
+      - "deploy/enterprise"
       - "build/**"
       - "release/e-*"
       - "hotfix/**"

.github/workflows/deploy-rag-dev.yml

@@ -1,4 +1,4 @@
-name: Deploy Trigger Dev
+name: Deploy RAG Dev
 
 permissions:
   contents: read
@@ -7,7 +7,7 @@ on:
   workflow_run:
     workflows: ["Build and Push API & Web"]
     branches:
-      - "deploy/trigger-dev"
+      - "deploy/rag-dev"
     types:
       - completed
 
@@ -16,12 +16,12 @@ jobs:
     runs-on: ubuntu-latest
     if: |
       github.event.workflow_run.conclusion == 'success' &&
-      github.event.workflow_run.head_branch == 'deploy/trigger-dev'
+      github.event.workflow_run.head_branch == 'deploy/rag-dev'
     steps:
       - name: Deploy to server
         uses: appleboy/ssh-action@v0.1.8
         with:
-          host: ${{ secrets.TRIGGER_SSH_HOST }}
+          host: ${{ secrets.RAG_SSH_HOST }}
           username: ${{ secrets.SSH_USER }}
           key: ${{ secrets.SSH_PRIVATE_KEY }}
           script: |

.gitignore

@@ -234,7 +234,4 @@ scripts/stress-test/reports/
 
 # mcp
 .playwright-mcp/
-.serena/
-
-# settings
-*.local.json
+.serena/

api/.env.example

@@ -343,15 +343,6 @@ OCEANBASE_VECTOR_DATABASE=test
 OCEANBASE_MEMORY_LIMIT=6G
 OCEANBASE_ENABLE_HYBRID_SEARCH=false
 
-# AlibabaCloud MySQL Vector configuration
-ALIBABACLOUD_MYSQL_HOST=127.0.0.1
-ALIBABACLOUD_MYSQL_PORT=3306
-ALIBABACLOUD_MYSQL_USER=root
-ALIBABACLOUD_MYSQL_PASSWORD=root
-ALIBABACLOUD_MYSQL_DATABASE=dify
-ALIBABACLOUD_MYSQL_MAX_CONNECTION=5
-ALIBABACLOUD_MYSQL_HNSW_M=6
-
 # openGauss configuration
 OPENGAUSS_HOST=127.0.0.1
 OPENGAUSS_PORT=6600
@@ -454,9 +445,6 @@ HTTP_REQUEST_NODE_MAX_BINARY_SIZE=10485760
 HTTP_REQUEST_NODE_MAX_TEXT_SIZE=1048576
 HTTP_REQUEST_NODE_SSL_VERIFY=True
 
-# Webhook request configuration
-WEBHOOK_REQUEST_BODY_MAX_SIZE=10485760
-
 # Respect X-* headers to redirect clients
 RESPECT_XFORWARD_HEADERS_ENABLED=false
 
@@ -534,12 +522,6 @@ ENABLE_CLEAN_MESSAGES=false
 ENABLE_MAIL_CLEAN_DOCUMENT_NOTIFY_TASK=false
 ENABLE_DATASETS_QUEUE_MONITOR=false
 ENABLE_CHECK_UPGRADABLE_PLUGIN_TASK=true
-ENABLE_WORKFLOW_SCHEDULE_POLLER_TASK=true
-# Interval time in minutes for polling scheduled workflows(default: 1 min)
-WORKFLOW_SCHEDULE_POLLER_INTERVAL=1
-WORKFLOW_SCHEDULE_POLLER_BATCH_SIZE=100
-# Maximum number of scheduled workflows to dispatch per tick (0 for unlimited)
-WORKFLOW_SCHEDULE_MAX_DISPATCH_PER_TICK=0
 
 # Position configuration
 POSITION_TOOL_PINS=

api/.vscode/launch.json.example

@@ -54,7 +54,7 @@
                 "--loglevel",
                 "DEBUG",
                 "-Q",
-                "dataset,generation,mail,ops_trace,app_deletion,workflow"
+                "dataset,generation,mail,ops_trace,app_deletion"
             ]
         }
     ]

api/app.py

@@ -1,3 +1,4 @@
+import os
 import sys
 
 
@@ -8,10 +9,16 @@ 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:
     # It seems that JetBrains Python debugger does not work well with gevent,
     # so we need to disable gevent in debug mode.
@@ -33,8 +40,15 @@ else:
 
     from app_factory import create_app
 
-    app = create_app()
-    celery = app.extensions["celery"]
+    socketio_app, flask_app = create_app()
+    app = flask_app
+    celery = flask_app.extensions["celery"]
 
 if __name__ == "__main__":
-    app.run(host="0.0.0.0", port=5001)
+    from gevent import pywsgi
+    from geventwebsocket.handler import WebSocketHandler
+
+    host = os.environ.get("HOST", "0.0.0.0")
+    port = int(os.environ.get("PORT", 5001))
+    server = pywsgi.WSGIServer((host, port), socketio_app, handler_class=WebSocketHandler)
+    server.serve_forever()

api/app_factory.py

@@ -31,14 +31,22 @@ def create_flask_app_with_configs() -> DifyApp:
     return dify_app
 
 
-def create_app() -> DifyApp:
+def create_app() -> tuple[any, 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 app
+    return socketio_app, app
 
 
 def initialize_extensions(app: DifyApp):

api/commands.py

@@ -15,12 +15,12 @@ from sqlalchemy.orm import sessionmaker
 from configs import dify_config
 from constants.languages import languages
 from core.helper import encrypter
-from core.plugin.entities.plugin_daemon import CredentialType
 from core.plugin.impl.plugin import PluginInstaller
 from core.rag.datasource.vdb.vector_factory import Vector
 from core.rag.datasource.vdb.vector_type import VectorType
 from core.rag.index_processor.constant.built_in_field import BuiltInField
 from core.rag.models.document import Document
+from core.tools.entities.tool_entities import CredentialType
 from core.tools.utils.system_oauth_encryption import encrypt_system_oauth_params
 from events.app_event import app_was_created
 from extensions.ext_database import db
@@ -1227,55 +1227,6 @@ def setup_system_tool_oauth_client(provider, client_params):
     click.echo(click.style(f"OAuth client params setup successfully. id: {oauth_client.id}", fg="green"))
 
 
-@click.command("setup-system-trigger-oauth-client", help="Setup system trigger oauth client.")
-@click.option("--provider", prompt=True, help="Provider name")
-@click.option("--client-params", prompt=True, help="Client Params")
-def setup_system_trigger_oauth_client(provider, client_params):
-    """
-    Setup system trigger oauth client
-    """
-    from models.provider_ids import TriggerProviderID
-    from models.trigger import TriggerOAuthSystemClient
-
-    provider_id = TriggerProviderID(provider)
-    provider_name = provider_id.provider_name
-    plugin_id = provider_id.plugin_id
-
-    try:
-        # json validate
-        click.echo(click.style(f"Validating client params: {client_params}", fg="yellow"))
-        client_params_dict = TypeAdapter(dict[str, Any]).validate_json(client_params)
-        click.echo(click.style("Client params validated successfully.", fg="green"))
-
-        click.echo(click.style(f"Encrypting client params: {client_params}", fg="yellow"))
-        click.echo(click.style(f"Using SECRET_KEY: `{dify_config.SECRET_KEY}`", fg="yellow"))
-        oauth_client_params = encrypt_system_oauth_params(client_params_dict)
-        click.echo(click.style("Client params encrypted successfully.", fg="green"))
-    except Exception as e:
-        click.echo(click.style(f"Error parsing client params: {str(e)}", fg="red"))
-        return
-
-    deleted_count = (
-        db.session.query(TriggerOAuthSystemClient)
-        .filter_by(
-            provider=provider_name,
-            plugin_id=plugin_id,
-        )
-        .delete()
-    )
-    if deleted_count > 0:
-        click.echo(click.style(f"Deleted {deleted_count} existing oauth client params.", fg="yellow"))
-
-    oauth_client = TriggerOAuthSystemClient(
-        provider=provider_name,
-        plugin_id=plugin_id,
-        encrypted_oauth_params=oauth_client_params,
-    )
-    db.session.add(oauth_client)
-    db.session.commit()
-    click.echo(click.style(f"OAuth client params setup successfully. id: {oauth_client.id}", fg="green"))
-
-
 def _find_orphaned_draft_variables(batch_size: int = 1000) -> list[str]:
     """
     Find draft variables that reference non-existent apps.
@@ -1570,14 +1521,6 @@ def transform_datasource_credentials():
                     auth_count = 0
                     for firecrawl_tenant_credential in firecrawl_tenant_credentials:
                         auth_count += 1
-                        if not firecrawl_tenant_credential.credentials:
-                            click.echo(
-                                click.style(
-                                    f"Skipping firecrawl credential for tenant {tenant_id} due to missing credentials.",
-                                    fg="yellow",
-                                )
-                            )
-                            continue
                         # get credential api key
                         credentials_json = json.loads(firecrawl_tenant_credential.credentials)
                         api_key = credentials_json.get("config", {}).get("api_key")
@@ -1633,14 +1576,6 @@ def transform_datasource_credentials():
                     auth_count = 0
                     for jina_tenant_credential in jina_tenant_credentials:
                         auth_count += 1
-                        if not jina_tenant_credential.credentials:
-                            click.echo(
-                                click.style(
-                                    f"Skipping jina credential for tenant {tenant_id} due to missing credentials.",
-                                    fg="yellow",
-                                )
-                            )
-                            continue
                         # get credential api key
                         credentials_json = json.loads(jina_tenant_credential.credentials)
                         api_key = credentials_json.get("config", {}).get("api_key")

api/configs/feature/__init__.py

@@ -174,17 +174,6 @@ class CodeExecutionSandboxConfig(BaseSettings):
     )
 
 
-class TriggerConfig(BaseSettings):
-    """
-    Configuration for trigger
-    """
-
-    WEBHOOK_REQUEST_BODY_MAX_SIZE: PositiveInt = Field(
-        description="Maximum allowed size for webhook request bodies in bytes",
-        default=10485760,
-    )
-
-
 class PluginConfig(BaseSettings):
     """
     Plugin configs
@@ -847,6 +836,16 @@ 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):
     """
@@ -996,22 +995,6 @@ class CeleryScheduleTasksConfig(BaseSettings):
         description="Enable check upgradable plugin task",
         default=True,
     )
-    ENABLE_WORKFLOW_SCHEDULE_POLLER_TASK: bool = Field(
-        description="Enable workflow schedule poller task",
-        default=True,
-    )
-    WORKFLOW_SCHEDULE_POLLER_INTERVAL: int = Field(
-        description="Workflow schedule poller interval in minutes",
-        default=1,
-    )
-    WORKFLOW_SCHEDULE_POLLER_BATCH_SIZE: int = Field(
-        description="Maximum number of schedules to process in each poll batch",
-        default=100,
-    )
-    WORKFLOW_SCHEDULE_MAX_DISPATCH_PER_TICK: int = Field(
-        description="Maximum schedules to dispatch per tick (0=unlimited, circuit breaker)",
-        default=0,
-    )
 
 
 class PositionConfig(BaseSettings):
@@ -1135,7 +1118,6 @@ class FeatureConfig(
     AuthConfig,  # Changed from OAuthConfig to AuthConfig
     BillingConfig,
     CodeExecutionSandboxConfig,
-    TriggerConfig,
     PluginConfig,
     MarketplaceConfig,
     DataSetConfig,

api/configs/feature/hosted_service/__init__.py

@@ -8,6 +8,11 @@ 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.
@@ -70,11 +75,6 @@ class HostedOpenAiConfig(BaseSettings):
         "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(
         description="Enable paid access to hosted OpenAI service",
         default=False,
@@ -98,6 +98,129 @@ class HostedOpenAiConfig(BaseSettings):
     )
 
 
+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="grok-3,grok-3-mini,grok-3-mini-fast",
+    )
+
+
 class HostedAzureOpenAiConfig(BaseSettings):
     """
     Configuration for hosted Azure OpenAI service
@@ -144,16 +267,32 @@ 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-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-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):
     """
@@ -250,5 +389,8 @@ class HostedServiceConfig(
     HostedModerationConfig,
     # credit config
     HostedCreditConfig,
+    HostedGeminiConfig,
+    HostedXAIConfig,
+    HostedDeepseekConfig,
 ):
     pass

api/configs/middleware/__init__.py

@@ -18,7 +18,6 @@ from .storage.opendal_storage_config import OpenDALStorageConfig
 from .storage.supabase_storage_config import SupabaseStorageConfig
 from .storage.tencent_cos_storage_config import TencentCloudCOSStorageConfig
 from .storage.volcengine_tos_storage_config import VolcengineTOSStorageConfig
-from .vdb.alibabacloud_mysql_config import AlibabaCloudMySQLConfig
 from .vdb.analyticdb_config import AnalyticdbConfig
 from .vdb.baidu_vector_config import BaiduVectorDBConfig
 from .vdb.chroma_config import ChromaConfig
@@ -331,7 +330,6 @@ class MiddlewareConfig(
     ClickzettaConfig,
     HuaweiCloudConfig,
     MilvusConfig,
-    AlibabaCloudMySQLConfig,
     MyScaleConfig,
     OpenSearchConfig,
     OracleConfig,

api/configs/middleware/vdb/alibabacloud_mysql_config.py

@@ -1,54 +0,0 @@
-from pydantic import Field, PositiveInt
-from pydantic_settings import BaseSettings
-
-
-class AlibabaCloudMySQLConfig(BaseSettings):
-    """
-    Configuration settings for AlibabaCloud MySQL vector database
-    """
-
-    ALIBABACLOUD_MYSQL_HOST: str = Field(
-        description="Hostname or IP address of the AlibabaCloud MySQL server (e.g., 'localhost' or 'mysql.aliyun.com')",
-        default="localhost",
-    )
-
-    ALIBABACLOUD_MYSQL_PORT: PositiveInt = Field(
-        description="Port number on which the AlibabaCloud MySQL server is listening (default is 3306)",
-        default=3306,
-    )
-
-    ALIBABACLOUD_MYSQL_USER: str = Field(
-        description="Username for authenticating with AlibabaCloud MySQL (default is 'root')",
-        default="root",
-    )
-
-    ALIBABACLOUD_MYSQL_PASSWORD: str = Field(
-        description="Password for authenticating with AlibabaCloud MySQL (default is an empty string)",
-        default="",
-    )
-
-    ALIBABACLOUD_MYSQL_DATABASE: str = Field(
-        description="Name of the AlibabaCloud MySQL database to connect to (default is 'dify')",
-        default="dify",
-    )
-
-    ALIBABACLOUD_MYSQL_MAX_CONNECTION: PositiveInt = Field(
-        description="Maximum number of connections in the connection pool",
-        default=5,
-    )
-
-    ALIBABACLOUD_MYSQL_CHARSET: str = Field(
-        description="Character set for AlibabaCloud MySQL connection (default is 'utf8mb4')",
-        default="utf8mb4",
-    )
-
-    ALIBABACLOUD_MYSQL_DISTANCE_FUNCTION: str = Field(
-        description="Distance function used for vector similarity search in AlibabaCloud MySQL "
-        "(e.g., 'cosine', 'euclidean')",
-        default="cosine",
-    )
-
-    ALIBABACLOUD_MYSQL_HNSW_M: PositiveInt = Field(
-        description="Maximum number of connections per layer for HNSW vector index (default is 6, range: 3-200)",
-        default=6,
-    )

api/configs/middleware/vdb/opensearch_config.py

@@ -1,24 +1,23 @@
-from enum import StrEnum
+from enum import Enum
 from typing import Literal
 
 from pydantic import Field, PositiveInt
 from pydantic_settings import BaseSettings
 
 
-class AuthMethod(StrEnum):
-    """
-    Authentication method for OpenSearch
-    """
-
-    BASIC = "basic"
-    AWS_MANAGED_IAM = "aws_managed_iam"
-
-
 class OpenSearchConfig(BaseSettings):
     """
     Configuration settings for OpenSearch
     """
 
+    class AuthMethod(Enum):
+        """
+        Authentication method for OpenSearch
+        """
+
+        BASIC = "basic"
+        AWS_MANAGED_IAM = "aws_managed_iam"
+
     OPENSEARCH_HOST: str | None = Field(
         description="Hostname or IP address of the OpenSearch server (e.g., 'localhost' or 'opensearch.example.com')",
         default=None,

api/contexts/__init__.py

@@ -9,7 +9,6 @@ if TYPE_CHECKING:
     from core.model_runtime.entities.model_entities import AIModelEntity
     from core.plugin.entities.plugin_daemon import PluginModelProviderEntity
     from core.tools.plugin_tool.provider import PluginToolProviderController
-    from core.trigger.provider import PluginTriggerProviderController
 
 
 """
@@ -42,11 +41,3 @@ datasource_plugin_providers: RecyclableContextVar[dict[str, "DatasourcePluginPro
 datasource_plugin_providers_lock: RecyclableContextVar[Lock] = RecyclableContextVar(
     ContextVar("datasource_plugin_providers_lock")
 )
-
-plugin_trigger_providers: RecyclableContextVar[dict[str, "PluginTriggerProviderController"]] = RecyclableContextVar(
-    ContextVar("plugin_trigger_providers")
-)
-
-plugin_trigger_providers_lock: RecyclableContextVar[Lock] = RecyclableContextVar(
-    ContextVar("plugin_trigger_providers_lock")
-)

api/controllers/console/__init__.py

@@ -58,15 +58,16 @@ 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,
-    workflow_trigger,
 )
 
 # Import auth controllers
@@ -107,10 +108,12 @@ from .datasets.rag_pipeline import (
 
 # Import explore controllers
 from .explore import (
+    banner,
     installed_app,
     parameter,
     recommended_app,
     saved_message,
+    trial,
 )
 
 # Import tag controllers
@@ -127,7 +130,6 @@ from .workspace import (
     models,
     plugin,
     tool_providers,
-    trigger_providers,
     workspace,
 )
 
@@ -145,6 +147,7 @@ __all__ = [
     "apikey",
     "app",
     "audio",
+    "banner",
     "billing",
     "bp",
     "completion",
@@ -198,7 +201,7 @@ __all__ = [
     "statistic",
     "tags",
     "tool_providers",
-    "trigger_providers",
+    "trial",
     "version",
     "website",
     "workflow",
@@ -206,6 +209,5 @@ __all__ = [
     "workflow_draft_variable",
     "workflow_run",
     "workflow_statistic",
-    "workflow_trigger",
     "workspace",
 ]

api/controllers/console/admin.py

@@ -15,7 +15,7 @@ from constants.languages import supported_language
 from controllers.console import api, console_ns
 from controllers.console.wraps import only_edition_cloud
 from extensions.ext_database import db
-from models.model import App, InstalledApp, RecommendedApp
+from models.model import App, ExporleBanner, InstalledApp, RecommendedApp, TrialApp
 
 
 def admin_required(view: Callable[P, R]):
@@ -61,6 +61,8 @@ 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"),
             },
         )
     )
@@ -79,6 +81,8 @@ class InsertExploreAppListApi(Resource):
         parser.add_argument("language", type=supported_language, required=True, nullable=False, location="json")
         parser.add_argument("category", type=str, required=True, nullable=False, location="json")
         parser.add_argument("position", type=int, required=True, nullable=False, location="json")
+        parser.add_argument("can_trial", type=bool, required=True, nullable=False, location="json")
+        parser.add_argument("trial_limit", type=int, required=True, nullable=False, location="json")
         args = parser.parse_args()
 
         app = db.session.execute(select(App).where(App.id == args["app_id"])).scalar_one_or_none()
@@ -115,6 +119,20 @@ 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()
@@ -129,6 +147,20 @@ 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()
@@ -174,7 +206,67 @@ 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("content", 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()
+
+        banner = ExporleBanner(
+            content=args["content"],
+            link=args["link"],
+            sort=args["sort"],
+        )
+        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

api/controllers/console/apikey.py

@@ -1,4 +1,5 @@
 import flask_restx
+from flask_login import current_user
 from flask_restx import Resource, fields, marshal_with
 from flask_restx._http import HTTPStatus
 from sqlalchemy import select
@@ -7,8 +8,7 @@ from werkzeug.exceptions import Forbidden
 
 from extensions.ext_database import db
 from libs.helper import TimestampField
-from libs.login import current_user, login_required
-from models.account import Account
+from libs.login import login_required
 from models.dataset import Dataset
 from models.model import ApiToken, App
 
@@ -57,8 +57,6 @@ class BaseApiKeyListResource(Resource):
     def get(self, resource_id):
         assert self.resource_id_field is not None, "resource_id_field must be set"
         resource_id = str(resource_id)
-        assert isinstance(current_user, Account)
-        assert current_user.current_tenant_id is not None
         _get_resource(resource_id, current_user.current_tenant_id, self.resource_model)
         keys = db.session.scalars(
             select(ApiToken).where(
@@ -71,10 +69,8 @@ class BaseApiKeyListResource(Resource):
     def post(self, resource_id):
         assert self.resource_id_field is not None, "resource_id_field must be set"
         resource_id = str(resource_id)
-        assert isinstance(current_user, Account)
-        assert current_user.current_tenant_id is not None
         _get_resource(resource_id, current_user.current_tenant_id, self.resource_model)
-        if not current_user.has_edit_permission:
+        if not current_user.is_editor:
             raise Forbidden()
 
         current_key_count = (
@@ -112,8 +108,6 @@ class BaseApiKeyResource(Resource):
         assert self.resource_id_field is not None, "resource_id_field must be set"
         resource_id = str(resource_id)
         api_key_id = str(api_key_id)
-        assert isinstance(current_user, Account)
-        assert current_user.current_tenant_id is not None
         _get_resource(resource_id, current_user.current_tenant_id, self.resource_model)
 
         # The role of the current user in the ta table must be admin or owner

api/controllers/console/app/app.py

@@ -304,7 +304,7 @@ class AppCopyApi(Resource):
             account = cast(Account, current_user)
             result = import_service.import_app(
                 account=account,
-                import_mode=ImportMode.YAML_CONTENT,
+                import_mode=ImportMode.YAML_CONTENT.value,
                 yaml_content=yaml_content,
                 name=args.get("name"),
                 description=args.get("description"),

api/controllers/console/app/app_import.py

@@ -70,9 +70,9 @@ class AppImportApi(Resource):
             EnterpriseService.WebAppAuth.update_app_access_mode(result.app_id, "private")
         # Return appropriate status code based on result
         status = result.status
-        if status == ImportStatus.FAILED:
+        if status == ImportStatus.FAILED.value:
             return result.model_dump(mode="json"), 400
-        elif status == ImportStatus.PENDING:
+        elif status == ImportStatus.PENDING.value:
             return result.model_dump(mode="json"), 202
         return result.model_dump(mode="json"), 200
 
@@ -97,7 +97,7 @@ class AppImportConfirmApi(Resource):
             session.commit()
 
         # Return appropriate status code based on result
-        if result.status == ImportStatus.FAILED:
+        if result.status == ImportStatus.FAILED.value:
             return result.model_dump(mode="json"), 400
         return result.model_dump(mode="json"), 200
 

api/controllers/console/app/conversation.py

@@ -309,7 +309,7 @@ class ChatConversationApi(Resource):
             )
 
         if app_model.mode == AppMode.ADVANCED_CHAT:
-            query = query.where(Conversation.invoke_from != InvokeFrom.DEBUGGER)
+            query = query.where(Conversation.invoke_from != InvokeFrom.DEBUGGER.value)
 
         match args["sort_by"]:
             case "created_at":

api/controllers/console/app/generator.py

@@ -12,7 +12,6 @@ from controllers.console.app.error import (
 )
 from controllers.console.wraps import account_initialization_required, setup_required
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
-from core.helper.code_executor.code_node_provider import CodeNodeProvider
 from core.helper.code_executor.javascript.javascript_code_provider import JavascriptCodeProvider
 from core.helper.code_executor.python3.python3_code_provider import Python3CodeProvider
 from core.llm_generator.llm_generator import LLMGenerator
@@ -199,11 +198,13 @@ class InstructionGenerateApi(Resource):
         parser.add_argument("model_config", type=dict, required=True, nullable=False, location="json")
         parser.add_argument("ideal_output", type=str, required=False, default="", location="json")
         args = parser.parse_args()
-        providers: list[type[CodeNodeProvider]] = [Python3CodeProvider, JavascriptCodeProvider]
-        code_provider: type[CodeNodeProvider] | None = next(
-            (p for p in providers if p.is_accept_language(args["language"])), None
+        code_template = (
+            Python3CodeProvider.get_default_code()
+            if args["language"] == "python"
+            else (JavascriptCodeProvider.get_default_code())
+            if args["language"] == "javascript"
+            else ""
         )
-        code_template = code_provider.get_default_code() if code_provider else ""
         try:
             # Generate from nothing for a workflow node
             if (args["current"] == code_template or args["current"] == "") and args["node_id"] != "":

api/controllers/console/app/model_config.py

@@ -14,7 +14,6 @@ from core.tools.tool_manager import ToolManager
 from core.tools.utils.configuration import ToolParameterConfigurationManager
 from events.app_event import app_model_config_was_updated
 from extensions.ext_database import db
-from libs.datetime_utils import naive_utc_now
 from libs.login import login_required
 from models.account import Account
 from models.model import AppMode, AppModelConfig
@@ -173,8 +172,6 @@ class ModelConfigResource(Resource):
         db.session.flush()
 
         app_model.app_model_config_id = new_app_model_config.id
-        app_model.updated_by = current_user.id
-        app_model.updated_at = naive_utc_now()
         db.session.commit()
 
         app_model_config_was_updated.send(app_model, app_model_config=new_app_model_config)

api/controllers/console/app/online_user.py

@@ -0,0 +1,291 @@
+import json
+import time
+
+from extensions.ext_redis import redis_client
+from extensions.ext_socketio import sio
+from libs.passport import PassportService
+from services.account_service import AccountService
+
+
+@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:
+        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
+    }
+
+    # Store session info with sid as key
+    redis_client.hset(f"workflow_online_users:{workflow_id}", sid, json.dumps(session_info))
+    redis_client.set(f"ws_sid_map:{sid}", json.dumps({"workflow_id": workflow_id, "user_id": user_id}))
+
+    # 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(f"ws_sid_map:{sid}")
+    if mapping:
+        data = json.loads(mapping)
+        workflow_id = data["workflow_id"]
+
+        # Remove this specific session
+        redis_client.hdel(f"workflow_online_users:{workflow_id}", sid)
+        redis_client.delete(f"ws_sid_map:{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(f"workflow_online_users:{workflow_id}", sid)
+    redis_client.delete(f"ws_sid_map:{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(f"workflow_online_users:{workflow_id}", sid):
+        return False
+
+    if not redis_client.exists(f"ws_sid_map:{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).
+    """
+    leader_key = f"workflow_leader:{workflow_id}"
+    raw_leader = redis_client.get(leader_key)
+    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)
+        current_leader = None
+        leader_replaced = True
+
+    if not current_leader:
+        redis_client.set(leader_key, sid, ex=3600)  # 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.
+    """
+    leader_key = f"workflow_leader:{workflow_id}"
+    current_leader = redis_client.get(leader_key)
+
+    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(f"workflow_online_users:{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, new_leader_sid, ex=3600)
+
+                # 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)
+
+
+def broadcast_leader_change(workflow_id, new_leader_sid):
+    """
+    Broadcast leader change to all sessions in the workflow.
+    """
+    sessions_json = redis_client.hgetall(f"workflow_online_users:{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_key = f"workflow_leader:{workflow_id}"
+    leader = redis_client.get(leader_key)
+    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(f"workflow_online_users:{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. mouseMove
+    2. varsAndFeaturesUpdate
+    3. syncRequest(ask leader to update graph)
+    4. appStateUpdate
+    5. mcpServerUpdate
+
+    """
+    mapping = redis_client.get(f"ws_sid_map:{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"]
+
+    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(f"ws_sid_map:{sid}")
+
+    if not mapping:
+        return {"msg": "unauthorized"}, 401
+
+    mapping_data = json.loads(mapping)
+    workflow_id = mapping_data["workflow_id"]
+
+    sio.emit("graph_update", data, room=workflow_id, skip_sid=sid)
+
+    return {"msg": "graph_update_broadcasted"}

api/controllers/console/app/statistic.py

@@ -52,7 +52,7 @@ FROM
 WHERE
     app_id = :app_id
     AND invoke_from != :invoke_from"""
-        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
+        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER.value}
 
         timezone = pytz.timezone(account.timezone)
         utc_timezone = pytz.utc
@@ -127,7 +127,7 @@ class DailyConversationStatistic(Resource):
                 sa.func.count(sa.distinct(Message.conversation_id)).label("conversation_count"),
             )
             .select_from(Message)
-            .where(Message.app_id == app_model.id, Message.invoke_from != InvokeFrom.DEBUGGER)
+            .where(Message.app_id == app_model.id, Message.invoke_from != InvokeFrom.DEBUGGER.value)
         )
 
         if args["start"]:
@@ -190,7 +190,7 @@ FROM
 WHERE
     app_id = :app_id
     AND invoke_from != :invoke_from"""
-        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
+        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER.value}
 
         timezone = pytz.timezone(account.timezone)
         utc_timezone = pytz.utc
@@ -263,7 +263,7 @@ FROM
 WHERE
     app_id = :app_id
     AND invoke_from != :invoke_from"""
-        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
+        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER.value}
 
         timezone = pytz.timezone(account.timezone)
         utc_timezone = pytz.utc
@@ -345,7 +345,7 @@ FROM
         WHERE
             c.app_id = :app_id
             AND m.invoke_from != :invoke_from"""
-        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
+        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER.value}
 
         timezone = pytz.timezone(account.timezone)
         utc_timezone = pytz.utc
@@ -432,7 +432,7 @@ LEFT JOIN
 WHERE
     m.app_id = :app_id
     AND m.invoke_from != :invoke_from"""
-        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
+        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER.value}
 
         timezone = pytz.timezone(account.timezone)
         utc_timezone = pytz.utc
@@ -509,7 +509,7 @@ FROM
 WHERE
     app_id = :app_id
     AND invoke_from != :invoke_from"""
-        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
+        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER.value}
 
         timezone = pytz.timezone(account.timezone)
         utc_timezone = pytz.utc
@@ -584,7 +584,7 @@ FROM
 WHERE
     app_id = :app_id
     AND invoke_from != :invoke_from"""
-        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
+        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER.value}
 
         timezone = pytz.timezone(account.timezone)
         utc_timezone = pytz.utc

api/controllers/console/app/workflow.py

@@ -5,6 +5,7 @@ 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
 
@@ -19,30 +20,23 @@ from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.file.models import File
 from core.helper.trace_id_helper import get_external_trace_id
-from core.model_runtime.utils.encoders import jsonable_encoder
 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
-from libs.datetime_utils import naive_utc_now
 from libs.helper import TimestampField, uuid_value
 from libs.login import current_user, login_required
 from models import App
 from models.account import Account
 from models.model import AppMode
-from models.provider_ids import TriggerProviderID
-from models.workflow import NodeType, Workflow
+from models.workflow import Workflow
 from services.app_generate_service import AppGenerateService
 from services.errors.app import WorkflowHashNotEqualError
 from services.errors.llm import InvokeRateLimitError
-from services.trigger.trigger_debug_service import (
-    PluginTriggerDebugEvent,
-    TriggerDebugService,
-    WebhookDebugEvent,
-)
-from services.trigger.webhook_service import WebhookService
 from services.workflow_service import DraftWorkflowDeletionError, WorkflowInUseError, WorkflowService
 
 logger = logging.getLogger(__name__)
@@ -112,6 +106,7 @@ 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"),
             },
         )
     )
@@ -136,6 +131,8 @@ class DraftWorkflowApi(Resource):
             parser.add_argument("hash", type=str, required=False, location="json")
             parser.add_argument("environment_variables", type=list, required=True, location="json")
             parser.add_argument("conversation_variables", type=list, required=False, location="json")
+            parser.add_argument("force_upload", type=bool, required=False, default=False, location="json")
+            parser.add_argument("memory_blocks", type=list, required=False, location="json")
             args = parser.parse_args()
         elif "text/plain" in content_type:
             try:
@@ -152,6 +149,8 @@ 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
@@ -172,6 +171,11 @@ 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"],
@@ -180,9 +184,13 @@ 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",
@@ -683,12 +691,8 @@ class PublishedWorkflowApi(Resource):
                 marked_comment=args.marked_comment or "",
             )
 
-            # Update app_model within the same session to ensure atomicity
-            app_model_in_session = session.get(App, app_model.id)
-            if app_model_in_session:
-                app_model_in_session.workflow_id = workflow.id
-                app_model_in_session.updated_by = current_user.id
-                app_model_in_session.updated_at = naive_utc_now()
+            app_model.workflow_id = workflow.id
+            db.session.commit()  # NOTE: this is necessary for update app_model.workflow_id
 
             workflow_created_at = TimestampField().format(workflow.created_at)
 
@@ -809,6 +813,45 @@ 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")
@@ -1000,415 +1043,103 @@ class DraftWorkflowNodeLastRunApi(Resource):
         return node_exec
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/trigger")
-class DraftWorkflowTriggerNodeApi(Resource):
-    """
-    Single node debug - Polling API for trigger events
-    Path: /apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/trigger
-    """
-
-    @api.doc("poll_draft_workflow_trigger_node")
-    @api.doc(description="Poll for trigger events and execute single node when event arrives")
-    @api.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
-    @api.expect(
-        api.model(
-            "DraftWorkflowTriggerNodeRequest",
-            {
-                "event_name": fields.String(required=True, description="Event name"),
-                "subscription_id": fields.String(required=True, description="Subscription ID"),
-                "provider_id": fields.String(required=True, description="Provider ID"),
-            },
-        )
-    )
-    @api.response(200, "Trigger event received and node executed successfully")
-    @api.response(403, "Permission denied")
-    @api.response(500, "Internal server error")
+class WorkflowOnlineUsersApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @get_app_model(mode=[AppMode.WORKFLOW])
-    def post(self, app_model: App, node_id: str):
-        """
-        Poll for trigger events and execute single node when event arrives
-        """
-        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
-            raise Forbidden()
-
+    @marshal_with(online_user_list_fields)
+    def get(self):
         parser = reqparse.RequestParser()
-        parser.add_argument("event_name", type=str, required=True, location="json", nullable=False)
-        parser.add_argument("subscription_id", type=str, required=True, location="json", nullable=False)
-        parser.add_argument("provider_id", type=str, required=True, location="json", nullable=False)
+        parser.add_argument("workflow_ids", type=str, required=True, location="args")
         args = parser.parse_args()
-        event_name = args["event_name"]
-        subscription_id = args["subscription_id"]
-        provider_id = args["provider_id"]
 
-        pool_key = PluginTriggerDebugEvent.build_pool_key(
-            tenant_id=app_model.tenant_id,
-            provider_id=provider_id,
-            subscription_id=subscription_id,
-            event_name=event_name,
-        )
-        event: PluginTriggerDebugEvent | None = TriggerDebugService.poll(
-            event_type=PluginTriggerDebugEvent,
-            pool_key=pool_key,
-            tenant_id=app_model.tenant_id,
-            user_id=current_user.id,
-            app_id=app_model.id,
-            node_id=node_id,
-        )
-        if not event:
-            return jsonable_encoder({"status": "waiting"})
+        workflow_ids = [id.strip() for id in args["workflow_ids"].split(",")]
 
-        try:
-            workflow_service = WorkflowService()
-            draft_workflow = workflow_service.get_draft_workflow(app_model)
-            if not draft_workflow:
-                raise ValueError("Workflow not found")
+        results = []
+        for workflow_id in workflow_ids:
+            users_json = redis_client.hgetall(f"workflow_online_users:{workflow_id}")
 
-            user_inputs = event.model_dump()
-            node_execution = workflow_service.run_draft_workflow_node(
-                app_model=app_model,
-                draft_workflow=draft_workflow,
-                node_id=node_id,
-                user_inputs=user_inputs,
-                account=current_user,
-                query="",
-                files=[],
-            )
-            return jsonable_encoder(node_execution)
-        except Exception:
-            logger.exception("Error running draft workflow trigger node")
-            return jsonable_encoder(
-                {
-                    "status": "error",
-                }
-            ), 500
+            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}
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/trigger/plugin/run")
-class DraftWorkflowTriggerRunApi(Resource):
-    """
-    Full workflow debug - Polling API for trigger events
-    Path: /apps/<uuid:app_id>/workflows/draft/trigger/run
-    """
-
-    @api.doc("poll_draft_workflow_trigger_run")
-    @api.doc(description="Poll for trigger events and execute full workflow when event arrives")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.expect(
-        api.model(
-            "DraftWorkflowTriggerRunRequest",
-            {
-                "node_id": fields.String(required=True, description="Node ID"),
-            },
-        )
-    )
-    @api.response(200, "Trigger event received and workflow executed successfully")
-    @api.response(403, "Permission denied")
-    @api.response(500, "Internal server error")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.WORKFLOW])
-    def post(self, app_model: App):
-        """
-        Poll for trigger events and execute full workflow when event arrives
-        """
-        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
-            raise Forbidden()
-
-        parser = reqparse.RequestParser()
-        parser.add_argument("node_id", type=str, required=True, location="json", nullable=False)
-        args = parser.parse_args()
-        node_id = args["node_id"]
-        workflow_service = WorkflowService()
-        workflow: Workflow | None = workflow_service.get_draft_workflow(
-            app_model=app_model,
-            workflow_id=None,
-        )
-
-        if not workflow:
-            return jsonable_encoder({"status": "error", "message": "Workflow not found"}), 404
-
-        node_data = workflow.get_node_config_by_id(node_id=node_id).get("data")
-        if not node_data:
-            return jsonable_encoder({"status": "error", "message": "Node config not found"}), 404
-
-        event_name = node_data.get("event_name")
-        subscription_id = node_data.get("subscription_id")
-        if not subscription_id:
-            return jsonable_encoder({"status": "error", "message": "Subscription ID not found"}), 404
-
-        provider_id = TriggerProviderID(node_data.get("provider_id"))
-        pool_key: str = PluginTriggerDebugEvent.build_pool_key(
-            tenant_id=app_model.tenant_id,
-            provider_id=provider_id,
-            subscription_id=subscription_id,
-            event_name=event_name,
-        )
-        event: PluginTriggerDebugEvent | None = TriggerDebugService.poll(
-            event_type=PluginTriggerDebugEvent,
-            pool_key=pool_key,
-            tenant_id=app_model.tenant_id,
-            user_id=current_user.id,
-            app_id=app_model.id,
-            node_id=node_id,
-        )
-        if not event:
-            return jsonable_encoder({"status": "waiting", "retry_in": 2000})
-
-        workflow_args = {
-            "inputs": event.model_dump(),
-            "query": "",
-            "files": [],
-        }
-        external_trace_id = get_external_trace_id(request)
-        if external_trace_id:
-            workflow_args["external_trace_id"] = external_trace_id
-
-        try:
-            response = AppGenerateService.generate(
-                app_model=app_model,
-                user=current_user,
-                args=workflow_args,
-                invoke_from=InvokeFrom.DEBUGGER,
-                streaming=True,
-                root_node_id=node_id,
-            )
-            return helper.compact_generate_response(response)
-        except InvokeRateLimitError as ex:
-            raise InvokeRateLimitHttpError(ex.description)
-        except Exception:
-            logger.exception("Error running draft workflow trigger run")
-            return jsonable_encoder(
-                {
-                    "status": "error",
-                }
-            ), 500
-
-
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/trigger/webhook/run")
-class DraftWorkflowTriggerWebhookRunApi(Resource):
-    """
-    Full workflow debug when the start node is a webhook trigger
-    Path: /apps/<uuid:app_id>/workflows/draft/trigger/webhook/run
-    """
-
-    @api.doc("draft_workflow_trigger_webhook_run")
-    @api.doc(description="Full workflow debug when the start node is a webhook trigger")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.expect(
-        api.model(
-            "DraftWorkflowTriggerWebhookRunRequest",
-            {
-                "node_id": fields.String(required=True, description="Node ID"),
-            },
-        )
-    )
-    @api.response(200, "Workflow executed successfully")
-    @api.response(403, "Permission denied")
-    @api.response(500, "Internal server error")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.WORKFLOW])
-    def post(self, app_model: App):
-        """
-        Full workflow debug when the start node is a webhook trigger
-        """
-        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
-            raise Forbidden()
-
-        parser = reqparse.RequestParser()
-        parser.add_argument("node_id", type=str, required=True, location="json", nullable=False)
-        args = parser.parse_args()
-        node_id = args["node_id"]
-
-        pool_key = WebhookDebugEvent.build_pool_key(
-            tenant_id=app_model.tenant_id,
-            app_id=app_model.id,
-            node_id=node_id,
-        )
-        event: WebhookDebugEvent | None = TriggerDebugService.poll(
-            event_type=WebhookDebugEvent,
-            pool_key=pool_key,
-            tenant_id=app_model.tenant_id,
-            user_id=current_user.id,
-            app_id=app_model.id,
-            node_id=node_id,
-        )
-
-        if not event:
-            return jsonable_encoder({"status": "waiting", "retry_in": 2000})
-
-        payload = event.payload or {}
-        workflow_inputs = payload.get("inputs")
-        if workflow_inputs is None:
-            webhook_data = payload.get("webhook_data", {})
-            workflow_inputs = WebhookService.build_workflow_inputs(webhook_data)
-
-        workflow_args = {
-            "inputs": workflow_inputs or {},
-            "query": "",
-            "files": [],
-        }
-
-        external_trace_id = get_external_trace_id(request)
-        if external_trace_id:
-            workflow_args["external_trace_id"] = external_trace_id
-
-        try:
-            response = AppGenerateService.generate(
-                app_model=app_model,
-                user=current_user,
-                args=workflow_args,
-                invoke_from=InvokeFrom.DEBUGGER,
-                streaming=True,
-                root_node_id=node_id,
-            )
-            return helper.compact_generate_response(response)
-        except InvokeRateLimitError as ex:
-            raise InvokeRateLimitHttpError(ex.description)
-        except Exception:
-            logger.exception("Error running draft workflow trigger webhook run")
-            return jsonable_encoder(
-                {
-                    "status": "error",
-                }
-            ), 500
-
-
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/debug/webhook/run")
-class DraftWorkflowNodeWebhookDebugRunApi(Resource):
-    """Single node debug when the node is a webhook trigger."""
-
-    @api.doc("draft_workflow_node_webhook_debug_run")
-    @api.doc(description="Poll for webhook debug payload and execute single node when event arrives")
-    @api.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
-    @api.response(200, "Node executed successfully")
-    @api.response(403, "Permission denied")
-    @api.response(400, "Invalid node type")
-    @api.response(500, "Internal server error")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.WORKFLOW])
-    def post(self, app_model: App, node_id: str):
-        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
-            raise Forbidden()
-
-        pool_key = WebhookDebugEvent.build_pool_key(
-            tenant_id=app_model.tenant_id,
-            app_id=app_model.id,
-            node_id=node_id,
-        )
-        event: WebhookDebugEvent | None = TriggerDebugService.poll(
-            event_type=WebhookDebugEvent,
-            pool_key=pool_key,
-            tenant_id=app_model.tenant_id,
-            user_id=current_user.id,
-            app_id=app_model.id,
-            node_id=node_id,
-        )
-
-        if not event:
-            return jsonable_encoder({"status": "waiting", "retry_in": 2000})
-
-        workflow_service = WorkflowService()
-        draft_workflow = workflow_service.get_draft_workflow(app_model=app_model)
-
-        if not draft_workflow:
-            raise DraftWorkflowNotExist()
-
-        node_config = draft_workflow.get_node_config_by_id(node_id)
-        node_type = Workflow.get_node_type_from_node_config(node_config)
-        if node_type != NodeType.TRIGGER_WEBHOOK:
-            return jsonable_encoder(
-                {
-                    "status": "error",
-                    "message": "node is not webhook trigger",
-                }
-            ), 400
-
-        payload = event.payload or {}
-        workflow_inputs = payload.get("inputs")
-        if workflow_inputs is None:
-            webhook_data = payload.get("webhook_data", {})
-            workflow_inputs = WebhookService.build_workflow_inputs(webhook_data)
-
-        workflow_node_execution = workflow_service.run_draft_workflow_node(
-            app_model=app_model,
-            draft_workflow=draft_workflow,
-            node_id=node_id,
-            user_inputs=workflow_inputs or {},
-            account=current_user,
-            query="",
-            files=[],
-        )
-
-        return jsonable_encoder(workflow_node_execution)
-
-
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/trigger/schedule/run")
-class DraftWorkflowTriggerScheduleRunApi(Resource):
-    """
-    Full workflow debug when the start node is a schedule trigger
-    Path: /apps/<uuid:app_id>/workflows/draft/trigger/schedule/run
-    """
-
-    @api.doc("draft_workflow_trigger_schedule_run")
-    @api.doc(description="Full workflow debug when the start node is a schedule trigger")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.expect(
-        api.model(
-            "DraftWorkflowTriggerScheduleRunRequest",
-            {
-                "node_id": fields.String(required=True, description="Node ID"),
-            },
-        )
-    )
-    @api.response(200, "Workflow executed successfully")
-    @api.response(403, "Permission denied")
-    @api.response(500, "Internal server error")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.WORKFLOW])
-    def post(self, app_model: App):
-        """
-        Full workflow debug when the start node is a schedule trigger
-        """
-        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
-            raise Forbidden()
-
-        parser = reqparse.RequestParser()
-        parser.add_argument("node_id", type=str, required=True, location="json", nullable=False)
-        args = parser.parse_args()
-        node_id = args["node_id"]
-
-        workflow_args = {
-            "inputs": {},
-            "query": "",
-            "files": [],
-        }
-
-        try:
-            response = AppGenerateService.generate(
-                app_model=app_model,
-                user=current_user,
-                args=workflow_args,
-                invoke_from=InvokeFrom.DEBUGGER,
-                streaming=True,
-                root_node_id=node_id,
-            )
-            return helper.compact_generate_response(response)
-        except InvokeRateLimitError as ex:
-            raise InvokeRateLimitHttpError(ex.description)
-        except Exception:
-            logger.exception("Error running draft workflow trigger schedule run")
-            return jsonable_encoder(
-                {
-                    "status": "error",
-                }
-            ), 500
+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")

api/controllers/console/app/workflow_comment.py

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

api/controllers/console/app/workflow_draft_variable.py

@@ -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 App, AppMode
 from models.account import Account
@@ -353,7 +353,7 @@ class VariableApi(Resource):
                 if len(raw_value) > 0 and not isinstance(raw_value[0], dict):
                     raise InvalidArgumentError(description=f"expected dict for files[0], got {type(raw_value)}")
                 raw_value = build_from_mappings(mappings=raw_value, tenant_id=app_model.tenant_id)
-            new_value = build_segment_with_type(variable.value_type, raw_value)
+            new_value = variable_factory.build_segment_with_type(variable.value_type, raw_value)
         draft_var_srv.update_variable(variable, name=new_name, value=new_value)
         db.session.commit()
         return variable
@@ -446,8 +446,35 @@ 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")
@@ -497,3 +524,44 @@ 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")

api/controllers/console/app/workflow_statistic.py

@@ -47,7 +47,7 @@ WHERE
         arg_dict = {
             "tz": account.timezone,
             "app_id": app_model.id,
-            "triggered_from": WorkflowRunTriggeredFrom.APP_RUN,
+            "triggered_from": WorkflowRunTriggeredFrom.APP_RUN.value,
         }
 
         timezone = pytz.timezone(account.timezone)
@@ -115,7 +115,7 @@ WHERE
         arg_dict = {
             "tz": account.timezone,
             "app_id": app_model.id,
-            "triggered_from": WorkflowRunTriggeredFrom.APP_RUN,
+            "triggered_from": WorkflowRunTriggeredFrom.APP_RUN.value,
         }
 
         timezone = pytz.timezone(account.timezone)
@@ -183,7 +183,7 @@ WHERE
         arg_dict = {
             "tz": account.timezone,
             "app_id": app_model.id,
-            "triggered_from": WorkflowRunTriggeredFrom.APP_RUN,
+            "triggered_from": WorkflowRunTriggeredFrom.APP_RUN.value,
         }
 
         timezone = pytz.timezone(account.timezone)
@@ -269,7 +269,7 @@ GROUP BY
         arg_dict = {
             "tz": account.timezone,
             "app_id": app_model.id,
-            "triggered_from": WorkflowRunTriggeredFrom.APP_RUN,
+            "triggered_from": WorkflowRunTriggeredFrom.APP_RUN.value,
         }
 
         timezone = pytz.timezone(account.timezone)

api/controllers/console/app/workflow_trigger.py

@@ -1,149 +0,0 @@
-import logging
-
-from flask_restx import Resource, marshal_with, reqparse
-from sqlalchemy import select
-from sqlalchemy.orm import Session
-from werkzeug.exceptions import Forbidden, NotFound
-
-from configs import dify_config
-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 extensions.ext_database import db
-from fields.workflow_trigger_fields import trigger_fields, triggers_list_fields, webhook_trigger_fields
-from libs.login import current_user, login_required
-from models.model import Account, AppMode
-from models.workflow import AppTrigger, AppTriggerStatus, WorkflowWebhookTrigger
-
-logger = logging.getLogger(__name__)
-
-
-class WebhookTriggerApi(Resource):
-    """Webhook Trigger API"""
-
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=AppMode.WORKFLOW)
-    @marshal_with(webhook_trigger_fields)
-    def get(self, app_model):
-        """Get webhook trigger for a node"""
-        parser = reqparse.RequestParser()
-        parser.add_argument("node_id", type=str, required=True, help="Node ID is required")
-        args = parser.parse_args()
-
-        node_id = args["node_id"]
-
-        with Session(db.engine) as session:
-            # Get webhook trigger for this app and node
-            webhook_trigger = (
-                session.query(WorkflowWebhookTrigger)
-                .filter(
-                    WorkflowWebhookTrigger.app_id == app_model.id,
-                    WorkflowWebhookTrigger.node_id == node_id,
-                )
-                .first()
-            )
-
-            if not webhook_trigger:
-                raise NotFound("Webhook trigger not found for this node")
-
-            # Add computed fields for marshal_with
-            base_url = dify_config.SERVICE_API_URL
-            webhook_trigger.webhook_url = f"{base_url}/triggers/webhook/{webhook_trigger.webhook_id}"  # type: ignore
-            webhook_trigger.webhook_debug_url = f"{base_url}/triggers/webhook-debug/{webhook_trigger.webhook_id}"  # type: ignore
-
-            return webhook_trigger
-
-
-class AppTriggersApi(Resource):
-    """App Triggers list API"""
-
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=AppMode.WORKFLOW)
-    @marshal_with(triggers_list_fields)
-    def get(self, app_model):
-        """Get app triggers list"""
-        assert isinstance(current_user, Account)
-        assert current_user.current_tenant_id is not None
-
-        with Session(db.engine) as session:
-            # Get all triggers for this app using select API
-            triggers = (
-                session.execute(
-                    select(AppTrigger)
-                    .where(
-                        AppTrigger.tenant_id == current_user.current_tenant_id,
-                        AppTrigger.app_id == app_model.id,
-                    )
-                    .order_by(AppTrigger.created_at.desc(), AppTrigger.id.desc())
-                )
-                .scalars()
-                .all()
-            )
-
-        # Add computed icon field for each trigger
-        url_prefix = dify_config.CONSOLE_API_URL + "/console/api/workspaces/current/tool-provider/builtin/"
-        for trigger in triggers:
-            if trigger.trigger_type == "trigger-plugin":
-                trigger.icon = url_prefix + trigger.provider_name + "/icon"  # type: ignore
-            else:
-                trigger.icon = ""  # type: ignore
-
-        return {"data": triggers}
-
-
-class AppTriggerEnableApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=AppMode.WORKFLOW)
-    @marshal_with(trigger_fields)
-    def post(self, app_model):
-        """Update app trigger (enable/disable)"""
-        parser = reqparse.RequestParser()
-        parser.add_argument("trigger_id", type=str, required=True, nullable=False, location="json")
-        parser.add_argument("enable_trigger", type=bool, required=True, nullable=False, location="json")
-        args = parser.parse_args()
-
-        assert isinstance(current_user, Account)
-        assert current_user.current_tenant_id is not None
-        if not current_user.has_edit_permission:
-            raise Forbidden()
-
-        trigger_id = args["trigger_id"]
-
-        with Session(db.engine) as session:
-            # Find the trigger using select
-            trigger = session.execute(
-                select(AppTrigger).where(
-                    AppTrigger.id == trigger_id,
-                    AppTrigger.tenant_id == current_user.current_tenant_id,
-                    AppTrigger.app_id == app_model.id,
-                )
-            ).scalar_one_or_none()
-
-            if not trigger:
-                raise NotFound("Trigger not found")
-
-            # Update status based on enable_trigger boolean
-            trigger.status = AppTriggerStatus.ENABLED if args["enable_trigger"] else AppTriggerStatus.DISABLED
-
-            session.commit()
-            session.refresh(trigger)
-
-        # Add computed icon field
-        url_prefix = dify_config.CONSOLE_API_URL + "/console/api/workspaces/current/tool-provider/builtin/"
-        if trigger.trigger_type == "trigger-plugin":
-            trigger.icon = url_prefix + trigger.provider_name + "/icon"  # type: ignore
-        else:
-            trigger.icon = ""  # type: ignore
-
-        return trigger
-
-
-api.add_resource(WebhookTriggerApi, "/apps/<uuid:app_id>/workflows/triggers/webhook")
-api.add_resource(AppTriggersApi, "/apps/<uuid:app_id>/triggers")
-api.add_resource(AppTriggerEnableApi, "/apps/<uuid:app_id>/trigger-enable")

api/controllers/console/auth/activate.py

@@ -103,7 +103,7 @@ class ActivateApi(Resource):
         account.interface_language = args["interface_language"]
         account.timezone = args["timezone"]
         account.interface_theme = "light"
-        account.status = AccountStatus.ACTIVE
+        account.status = AccountStatus.ACTIVE.value
         account.initialized_at = naive_utc_now()
         db.session.commit()
 

api/controllers/console/auth/oauth.py

@@ -130,11 +130,11 @@ class OAuthCallback(Resource):
             return redirect(f"{dify_config.CONSOLE_WEB_URL}/signin?message={e.description}")
 
         # Check account status
-        if account.status == AccountStatus.BANNED:
+        if account.status == AccountStatus.BANNED.value:
             return redirect(f"{dify_config.CONSOLE_WEB_URL}/signin?message=Account is banned.")
 
-        if account.status == AccountStatus.PENDING:
-            account.status = AccountStatus.ACTIVE
+        if account.status == AccountStatus.PENDING.value:
+            account.status = AccountStatus.ACTIVE.value
             account.initialized_at = naive_utc_now()
             db.session.commit()
 

api/controllers/console/billing/compliance.py

@@ -1,9 +1,9 @@
 from flask import request
+from flask_login import current_user
 from flask_restx import Resource, reqparse
 
 from libs.helper import extract_remote_ip
-from libs.login import current_user, login_required
-from models.account import Account
+from libs.login import login_required
 from services.billing_service import BillingService
 
 from .. import console_ns
@@ -17,8 +17,6 @@ class ComplianceApi(Resource):
     @account_initialization_required
     @only_edition_cloud
     def get(self):
-        assert isinstance(current_user, Account)
-        assert current_user.current_tenant_id is not None
         parser = reqparse.RequestParser()
         parser.add_argument("doc_name", type=str, required=True, location="args")
         args = parser.parse_args()

api/controllers/console/datasets/data_source.py

@@ -256,7 +256,7 @@ class DataSourceNotionApi(Resource):
             credential_id = notion_info.get("credential_id")
             for page in notion_info["pages"]:
                 extract_setting = ExtractSetting(
-                    datasource_type=DatasourceType.NOTION,
+                    datasource_type=DatasourceType.NOTION.value,
                     notion_info=NotionInfo.model_validate(
                         {
                             "credential_id": credential_id,

api/controllers/console/datasets/datasets.py

@@ -45,79 +45,6 @@ def _validate_name(name: str) -> str:
     return name
 
 
-def _get_retrieval_methods_by_vector_type(vector_type: str | None, is_mock: bool = False) -> dict[str, list[str]]:
-    """
-    Get supported retrieval methods based on vector database type.
-
-    Args:
-        vector_type: Vector database type, can be None
-        is_mock: Whether this is a Mock API, affects MILVUS handling
-
-    Returns:
-        Dictionary containing supported retrieval methods
-
-    Raises:
-        ValueError: If vector_type is None or unsupported
-    """
-    if vector_type is None:
-        raise ValueError("Vector store type is not configured.")
-
-    # Define vector database types that only support semantic search
-    semantic_only_types = {
-        VectorType.RELYT,
-        VectorType.TIDB_VECTOR,
-        VectorType.CHROMA,
-        VectorType.PGVECTO_RS,
-        VectorType.VIKINGDB,
-        VectorType.UPSTASH,
-    }
-
-    # Define vector database types that support all retrieval methods
-    full_search_types = {
-        VectorType.QDRANT,
-        VectorType.WEAVIATE,
-        VectorType.OPENSEARCH,
-        VectorType.ANALYTICDB,
-        VectorType.MYSCALE,
-        VectorType.ORACLE,
-        VectorType.ELASTICSEARCH,
-        VectorType.ELASTICSEARCH_JA,
-        VectorType.PGVECTOR,
-        VectorType.VASTBASE,
-        VectorType.TIDB_ON_QDRANT,
-        VectorType.LINDORM,
-        VectorType.COUCHBASE,
-        VectorType.OPENGAUSS,
-        VectorType.OCEANBASE,
-        VectorType.TABLESTORE,
-        VectorType.HUAWEI_CLOUD,
-        VectorType.TENCENT,
-        VectorType.MATRIXONE,
-        VectorType.CLICKZETTA,
-        VectorType.BAIDU,
-        VectorType.ALIBABACLOUD_MYSQL,
-    }
-
-    semantic_methods = {"retrieval_method": [RetrievalMethod.SEMANTIC_SEARCH.value]}
-    full_methods = {
-        "retrieval_method": [
-            RetrievalMethod.SEMANTIC_SEARCH.value,
-            RetrievalMethod.FULL_TEXT_SEARCH.value,
-            RetrievalMethod.HYBRID_SEARCH.value,
-        ]
-    }
-
-    if vector_type == VectorType.MILVUS:
-        return semantic_methods if is_mock else full_methods
-
-    if vector_type in semantic_only_types:
-        return semantic_methods
-    elif vector_type in full_search_types:
-        return full_methods
-    else:
-        raise ValueError(f"Unsupported vector db type {vector_type}.")
-
-
 @console_ns.route("/datasets")
 class DatasetListApi(Resource):
     @api.doc("get_datasets")
@@ -573,7 +500,7 @@ class DatasetIndexingEstimateApi(Resource):
             if file_details:
                 for file_detail in file_details:
                     extract_setting = ExtractSetting(
-                        datasource_type=DatasourceType.FILE,
+                        datasource_type=DatasourceType.FILE.value,
                         upload_file=file_detail,
                         document_model=args["doc_form"],
                     )
@@ -585,7 +512,7 @@ class DatasetIndexingEstimateApi(Resource):
                 credential_id = notion_info.get("credential_id")
                 for page in notion_info["pages"]:
                     extract_setting = ExtractSetting(
-                        datasource_type=DatasourceType.NOTION,
+                        datasource_type=DatasourceType.NOTION.value,
                         notion_info=NotionInfo.model_validate(
                             {
                                 "credential_id": credential_id,
@@ -602,7 +529,7 @@ class DatasetIndexingEstimateApi(Resource):
             website_info_list = args["info_list"]["website_info_list"]
             for url in website_info_list["urls"]:
                 extract_setting = ExtractSetting(
-                    datasource_type=DatasourceType.WEBSITE,
+                    datasource_type=DatasourceType.WEBSITE.value,
                     website_info=WebsiteInfo.model_validate(
                         {
                             "provider": website_info_list["provider"],
@@ -850,7 +777,49 @@ class DatasetRetrievalSettingApi(Resource):
     @account_initialization_required
     def get(self):
         vector_type = dify_config.VECTOR_STORE
-        return _get_retrieval_methods_by_vector_type(vector_type, is_mock=False)
+        match vector_type:
+            case (
+                VectorType.RELYT
+                | VectorType.TIDB_VECTOR
+                | VectorType.CHROMA
+                | VectorType.PGVECTO_RS
+                | VectorType.VIKINGDB
+                | VectorType.UPSTASH
+            ):
+                return {"retrieval_method": [RetrievalMethod.SEMANTIC_SEARCH.value]}
+            case (
+                VectorType.QDRANT
+                | VectorType.WEAVIATE
+                | VectorType.OPENSEARCH
+                | VectorType.ANALYTICDB
+                | VectorType.MYSCALE
+                | VectorType.ORACLE
+                | VectorType.ELASTICSEARCH
+                | VectorType.ELASTICSEARCH_JA
+                | VectorType.PGVECTOR
+                | VectorType.VASTBASE
+                | VectorType.TIDB_ON_QDRANT
+                | VectorType.LINDORM
+                | VectorType.COUCHBASE
+                | VectorType.MILVUS
+                | VectorType.OPENGAUSS
+                | VectorType.OCEANBASE
+                | VectorType.TABLESTORE
+                | VectorType.HUAWEI_CLOUD
+                | VectorType.TENCENT
+                | VectorType.MATRIXONE
+                | VectorType.CLICKZETTA
+                | VectorType.BAIDU
+            ):
+                return {
+                    "retrieval_method": [
+                        RetrievalMethod.SEMANTIC_SEARCH.value,
+                        RetrievalMethod.FULL_TEXT_SEARCH.value,
+                        RetrievalMethod.HYBRID_SEARCH.value,
+                    ]
+                }
+            case _:
+                raise ValueError(f"Unsupported vector db type {vector_type}.")
 
 
 @console_ns.route("/datasets/retrieval-setting/<string:vector_type>")
@@ -863,7 +832,48 @@ class DatasetRetrievalSettingMockApi(Resource):
     @login_required
     @account_initialization_required
     def get(self, vector_type):
-        return _get_retrieval_methods_by_vector_type(vector_type, is_mock=True)
+        match vector_type:
+            case (
+                VectorType.MILVUS
+                | VectorType.RELYT
+                | VectorType.TIDB_VECTOR
+                | VectorType.CHROMA
+                | VectorType.PGVECTO_RS
+                | VectorType.VIKINGDB
+                | VectorType.UPSTASH
+            ):
+                return {"retrieval_method": [RetrievalMethod.SEMANTIC_SEARCH.value]}
+            case (
+                VectorType.QDRANT
+                | VectorType.WEAVIATE
+                | VectorType.OPENSEARCH
+                | VectorType.ANALYTICDB
+                | VectorType.MYSCALE
+                | VectorType.ORACLE
+                | VectorType.ELASTICSEARCH
+                | VectorType.ELASTICSEARCH_JA
+                | VectorType.COUCHBASE
+                | VectorType.PGVECTOR
+                | VectorType.VASTBASE
+                | VectorType.LINDORM
+                | VectorType.OPENGAUSS
+                | VectorType.OCEANBASE
+                | VectorType.TABLESTORE
+                | VectorType.TENCENT
+                | VectorType.HUAWEI_CLOUD
+                | VectorType.MATRIXONE
+                | VectorType.CLICKZETTA
+                | VectorType.BAIDU
+            ):
+                return {
+                    "retrieval_method": [
+                        RetrievalMethod.SEMANTIC_SEARCH.value,
+                        RetrievalMethod.FULL_TEXT_SEARCH.value,
+                        RetrievalMethod.HYBRID_SEARCH.value,
+                    ]
+                }
+            case _:
+                raise ValueError(f"Unsupported vector db type {vector_type}.")
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/error-docs")

api/controllers/console/datasets/datasets_document.py

@@ -475,7 +475,7 @@ class DocumentIndexingEstimateApi(DocumentResource):
                     raise NotFound("File not found.")
 
                 extract_setting = ExtractSetting(
-                    datasource_type=DatasourceType.FILE, upload_file=file, document_model=document.doc_form
+                    datasource_type=DatasourceType.FILE.value, upload_file=file, document_model=document.doc_form
                 )
 
                 indexing_runner = IndexingRunner()
@@ -538,7 +538,7 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
                     raise NotFound("File not found.")
 
                 extract_setting = ExtractSetting(
-                    datasource_type=DatasourceType.FILE, upload_file=file_detail, document_model=document.doc_form
+                    datasource_type=DatasourceType.FILE.value, upload_file=file_detail, document_model=document.doc_form
                 )
                 extract_settings.append(extract_setting)
 
@@ -546,7 +546,7 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
                 if not data_source_info:
                     continue
                 extract_setting = ExtractSetting(
-                    datasource_type=DatasourceType.NOTION,
+                    datasource_type=DatasourceType.NOTION.value,
                     notion_info=NotionInfo.model_validate(
                         {
                             "credential_id": data_source_info["credential_id"],
@@ -563,7 +563,7 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
                 if not data_source_info:
                     continue
                 extract_setting = ExtractSetting(
-                    datasource_type=DatasourceType.WEBSITE,
+                    datasource_type=DatasourceType.WEBSITE.value,
                     website_info=WebsiteInfo.model_validate(
                         {
                             "provider": data_source_info["provider"],

api/controllers/console/datasets/hit_testing_base.py

@@ -1,5 +1,7 @@
 import logging
+from typing import cast
 
+from flask_login import current_user
 from flask_restx import marshal, reqparse
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
 
@@ -19,7 +21,6 @@ from core.errors.error import (
 )
 from core.model_runtime.errors.invoke import InvokeError
 from fields.hit_testing_fields import hit_testing_record_fields
-from libs.login import current_user
 from models.account import Account
 from services.dataset_service import DatasetService
 from services.hit_testing_service import HitTestingService
@@ -30,7 +31,6 @@ logger = logging.getLogger(__name__)
 class DatasetsHitTestingBase:
     @staticmethod
     def get_and_validate_dataset(dataset_id: str):
-        assert isinstance(current_user, Account)
         dataset = DatasetService.get_dataset(dataset_id)
         if dataset is None:
             raise NotFound("Dataset not found.")
@@ -57,12 +57,11 @@ class DatasetsHitTestingBase:
 
     @staticmethod
     def perform_hit_testing(dataset, args):
-        assert isinstance(current_user, Account)
         try:
             response = HitTestingService.retrieve(
                 dataset=dataset,
                 query=args["query"],
-                account=current_user,
+                account=cast(Account, current_user),
                 retrieval_model=args["retrieval_model"],
                 external_retrieval_model=args["external_retrieval_model"],
                 limit=10,

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

@@ -60,9 +60,9 @@ class RagPipelineImportApi(Resource):
 
         # Return appropriate status code based on result
         status = result.status
-        if status == ImportStatus.FAILED:
+        if status == ImportStatus.FAILED.value:
             return result.model_dump(mode="json"), 400
-        elif status == ImportStatus.PENDING:
+        elif status == ImportStatus.PENDING.value:
             return result.model_dump(mode="json"), 202
         return result.model_dump(mode="json"), 200
 
@@ -87,7 +87,7 @@ class RagPipelineImportConfirmApi(Resource):
             session.commit()
 
         # Return appropriate status code based on result
-        if result.status == ImportStatus.FAILED:
+        if result.status == ImportStatus.FAILED.value:
             return result.model_dump(mode="json"), 400
         return result.model_dump(mode="json"), 200
 

api/controllers/console/explore/banner.py

@@ -0,0 +1,34 @@
+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."""
+        banners = (
+            db.session.query(ExporleBanner).where(ExporleBanner.status == "enabled").order_by(ExporleBanner.sort).all()
+        )
+
+        # Convert banners to serializable format
+        result = []
+        for banner in banners:
+            banner_data = {
+                "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")

api/controllers/console/explore/error.py

@@ -29,3 +29,25 @@ 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."

api/controllers/console/explore/recommended_app.py

@@ -27,6 +27,7 @@ recommended_app_fields = {
     "category": fields.String,
     "position": fields.Integer,
     "is_listed": fields.Boolean,
+    "can_trial": fields.Boolean,
 }
 
 recommended_app_list_fields = {

api/controllers/console/explore/trial.py

@@ -0,0 +1,375 @@
+import logging
+
+from flask import request
+from flask_restx import Resource, 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,
+    NoAudioUploadedError,
+    ProviderModelCurrentlyNotSupportError,
+    ProviderNotInitializeError,
+    ProviderNotSupportSpeechToTextError,
+    ProviderQuotaExceededError,
+    UnsupportedAudioTypeError,
+)
+from controllers.console.app.wraps import get_app_model
+from controllers.console.explore.error import (
+    AppSuggestedQuestionsAfterAnswerDisabledError,
+    NotChatAppError,
+    NotCompletionAppError,
+)
+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.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.app_fields import app_detail_fields_with_site
+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 services.app_generate_service import AppGenerateService
+from services.app_service import AppService
+from services.audio_service import AudioService
+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 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")
+            response = AppGenerateService.generate(
+                app_model=app_model, user=current_user, args=args, invoke_from=InvokeFrom.EXPLORE, streaming=True
+            )
+            RecommendedAppService.add_trial_app_record(app_model.id, current_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")
+            response = AudioService.transcript_asr(app_model=app_model, file=file, end_user=None)
+            RecommendedAppService.add_trial_app_record(app_model.id, current_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")
+            response = AudioService.transcript_tts(app_model=app_model, text=text, voice=voice, message_id=message_id)
+            RecommendedAppService.add_trial_app_record(app_model.id, current_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")
+            response = AppGenerateService.generate(
+                app_model=app_model, user=current_user, args=args, invoke_from=InvokeFrom.EXPLORE, streaming=streaming
+            )
+
+            RecommendedAppService.add_trial_app_record(app_model.id, current_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
+    @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
+    @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
+    @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
+
+
+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/controllers/console/explore/wraps.py

@@ -2,15 +2,16 @@ 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
+from controllers.console.explore.error import AppAccessDeniedError, TrialAppLimitExceeded, TrialAppNotAllowed
 from controllers.console.wraps import account_initialization_required
 from extensions.ext_database import db
-from libs.login import current_user, login_required
-from models import InstalledApp
-from models.account import Account
+from libs.login import login_required
+from models import AccountTrialAppRecord, App, InstalledApp, TrialApp
 from services.app_service import AppService
 from services.enterprise.enterprise_service import EnterpriseService
 from services.feature_service import FeatureService
@@ -24,8 +25,6 @@ def installed_app_required(view: Callable[Concatenate[InstalledApp, P], R] | Non
     def decorator(view: Callable[Concatenate[InstalledApp, P], R]):
         @wraps(view)
         def decorated(installed_app_id: str, *args: P.args, **kwargs: P.kwargs):
-            assert isinstance(current_user, Account)
-            assert current_user.current_tenant_id is not None
             installed_app = (
                 db.session.query(InstalledApp)
                 .where(
@@ -58,7 +57,6 @@ def user_allowed_to_access_app(view: Callable[Concatenate[InstalledApp, P], R] |
         def decorated(installed_app: InstalledApp, *args: P.args, **kwargs: P.kwargs):
             feature = FeatureService.get_system_features()
             if feature.webapp_auth.enabled:
-                assert isinstance(current_user, Account)
                 app_id = installed_app.app_id
                 app_code = AppService.get_app_code_by_id(app_id)
                 res = EnterpriseService.WebAppAuth.is_user_allowed_to_access_webapp(
@@ -77,6 +75,59 @@ 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
 
@@ -86,3 +137,13 @@ 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,
+    ]

api/controllers/console/extension.py

@@ -1,11 +1,11 @@
+from flask_login import current_user
 from flask_restx import Resource, fields, marshal_with, reqparse
 
 from constants import HIDDEN_VALUE
 from controllers.console import api, console_ns
 from controllers.console.wraps import account_initialization_required, setup_required
 from fields.api_based_extension_fields import api_based_extension_fields
-from libs.login import current_user, login_required
-from models.account import Account
+from libs.login import login_required
 from models.api_based_extension import APIBasedExtension
 from services.api_based_extension_service import APIBasedExtensionService
 from services.code_based_extension_service import CodeBasedExtensionService
@@ -47,8 +47,6 @@ class APIBasedExtensionAPI(Resource):
     @account_initialization_required
     @marshal_with(api_based_extension_fields)
     def get(self):
-        assert isinstance(current_user, Account)
-        assert current_user.current_tenant_id is not None
         tenant_id = current_user.current_tenant_id
         return APIBasedExtensionService.get_all_by_tenant_id(tenant_id)
 
@@ -70,8 +68,6 @@ class APIBasedExtensionAPI(Resource):
     @account_initialization_required
     @marshal_with(api_based_extension_fields)
     def post(self):
-        assert isinstance(current_user, Account)
-        assert current_user.current_tenant_id is not None
         parser = reqparse.RequestParser()
         parser.add_argument("name", type=str, required=True, location="json")
         parser.add_argument("api_endpoint", type=str, required=True, location="json")
@@ -99,8 +95,6 @@ class APIBasedExtensionDetailAPI(Resource):
     @account_initialization_required
     @marshal_with(api_based_extension_fields)
     def get(self, id):
-        assert isinstance(current_user, Account)
-        assert current_user.current_tenant_id is not None
         api_based_extension_id = str(id)
         tenant_id = current_user.current_tenant_id
 
@@ -125,8 +119,6 @@ class APIBasedExtensionDetailAPI(Resource):
     @account_initialization_required
     @marshal_with(api_based_extension_fields)
     def post(self, id):
-        assert isinstance(current_user, Account)
-        assert current_user.current_tenant_id is not None
         api_based_extension_id = str(id)
         tenant_id = current_user.current_tenant_id
 
@@ -154,8 +146,6 @@ class APIBasedExtensionDetailAPI(Resource):
     @login_required
     @account_initialization_required
     def delete(self, id):
-        assert isinstance(current_user, Account)
-        assert current_user.current_tenant_id is not None
         api_based_extension_id = str(id)
         tenant_id = current_user.current_tenant_id
 

api/controllers/console/feature.py

@@ -1,7 +1,7 @@
+from flask_login import current_user
 from flask_restx import Resource, fields
 
-from libs.login import current_user, login_required
-from models.account import Account
+from libs.login import login_required
 from services.feature_service import FeatureService
 
 from . import api, console_ns
@@ -23,8 +23,6 @@ class FeatureApi(Resource):
     @cloud_utm_record
     def get(self):
         """Get feature configuration for current tenant"""
-        assert isinstance(current_user, Account)
-        assert current_user.current_tenant_id is not None
         return FeatureService.get_features(current_user.current_tenant_id).model_dump()
 
 

api/controllers/console/remote_files.py

@@ -1,6 +1,8 @@
 import urllib.parse
+from typing import cast
 
 import httpx
+from flask_login import current_user
 from flask_restx import Resource, marshal_with, reqparse
 
 import services
@@ -14,7 +16,6 @@ from core.file import helpers as file_helpers
 from core.helper import ssrf_proxy
 from extensions.ext_database import db
 from fields.file_fields import file_fields_with_signed_url, remote_file_info_fields
-from libs.login import current_user
 from models.account import Account
 from services.file_service import FileService
 
@@ -64,8 +65,7 @@ class RemoteFileUploadApi(Resource):
         content = resp.content if resp.request.method == "GET" else ssrf_proxy.get(url).content
 
         try:
-            assert isinstance(current_user, Account)
-            user = current_user
+            user = cast(Account, current_user)
             upload_file = FileService(db.engine).upload_file(
                 filename=file_info.filename,
                 content=content,

api/controllers/console/tag/tags.py

@@ -1,12 +1,12 @@
 from flask import request
+from flask_login import current_user
 from flask_restx import Resource, marshal_with, reqparse
 from werkzeug.exceptions import Forbidden
 
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, setup_required
 from fields.tag_fields import dataset_tag_fields
-from libs.login import current_user, login_required
-from models.account import Account
+from libs.login import login_required
 from models.model import Tag
 from services.tag_service import TagService
 
@@ -24,8 +24,6 @@ class TagListApi(Resource):
     @account_initialization_required
     @marshal_with(dataset_tag_fields)
     def get(self):
-        assert isinstance(current_user, Account)
-        assert current_user.current_tenant_id is not None
         tag_type = request.args.get("type", type=str, default="")
         keyword = request.args.get("keyword", default=None, type=str)
         tags = TagService.get_tags(tag_type, current_user.current_tenant_id, keyword)
@@ -36,10 +34,8 @@ class TagListApi(Resource):
     @login_required
     @account_initialization_required
     def post(self):
-        assert isinstance(current_user, Account)
-        assert current_user.current_tenant_id is not None
         # The role of the current user in the ta table must be admin, owner, or editor
-        if not (current_user.has_edit_permission or current_user.is_dataset_editor):
+        if not (current_user.is_editor or current_user.is_dataset_editor):
             raise Forbidden()
 
         parser = reqparse.RequestParser()
@@ -63,11 +59,9 @@ class TagUpdateDeleteApi(Resource):
     @login_required
     @account_initialization_required
     def patch(self, tag_id):
-        assert isinstance(current_user, Account)
-        assert current_user.current_tenant_id is not None
         tag_id = str(tag_id)
         # The role of the current user in the ta table must be admin, owner, or editor
-        if not (current_user.has_edit_permission or current_user.is_dataset_editor):
+        if not (current_user.is_editor or current_user.is_dataset_editor):
             raise Forbidden()
 
         parser = reqparse.RequestParser()
@@ -87,11 +81,9 @@ class TagUpdateDeleteApi(Resource):
     @login_required
     @account_initialization_required
     def delete(self, tag_id):
-        assert isinstance(current_user, Account)
-        assert current_user.current_tenant_id is not None
         tag_id = str(tag_id)
         # The role of the current user in the ta table must be admin, owner, or editor
-        if not current_user.has_edit_permission:
+        if not current_user.is_editor:
             raise Forbidden()
 
         TagService.delete_tag(tag_id)
@@ -105,10 +97,8 @@ class TagBindingCreateApi(Resource):
     @login_required
     @account_initialization_required
     def post(self):
-        assert isinstance(current_user, Account)
-        assert current_user.current_tenant_id is not None
         # The role of the current user in the ta table must be admin, owner, editor, or dataset_operator
-        if not (current_user.has_edit_permission or current_user.is_dataset_editor):
+        if not (current_user.is_editor or current_user.is_dataset_editor):
             raise Forbidden()
 
         parser = reqparse.RequestParser()
@@ -133,10 +123,8 @@ class TagBindingDeleteApi(Resource):
     @login_required
     @account_initialization_required
     def post(self):
-        assert isinstance(current_user, Account)
-        assert current_user.current_tenant_id is not None
         # The role of the current user in the ta table must be admin, owner, editor, or dataset_operator
-        if not (current_user.has_edit_permission or current_user.is_dataset_editor):
+        if not (current_user.is_editor or current_user.is_dataset_editor):
             raise Forbidden()
 
         parser = reqparse.RequestParser()

api/controllers/console/workspace/account.py

@@ -33,6 +33,7 @@ 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
@@ -135,6 +136,17 @@ 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

api/controllers/console/workspace/agent_providers.py

@@ -1,10 +1,10 @@
+from flask_login import current_user
 from flask_restx import Resource, fields
 
 from controllers.console import api, console_ns
 from controllers.console.wraps import account_initialization_required, setup_required
 from core.model_runtime.utils.encoders import jsonable_encoder
-from libs.login import current_user, login_required
-from models.account import Account
+from libs.login import login_required
 from services.agent_service import AgentService
 
 
@@ -21,9 +21,7 @@ class AgentProviderListApi(Resource):
     @login_required
     @account_initialization_required
     def get(self):
-        assert isinstance(current_user, Account)
         user = current_user
-        assert user.current_tenant_id is not None
 
         user_id = user.id
         tenant_id = user.current_tenant_id
@@ -45,9 +43,7 @@ class AgentProviderApi(Resource):
     @login_required
     @account_initialization_required
     def get(self, provider_name: str):
-        assert isinstance(current_user, Account)
         user = current_user
-        assert user.current_tenant_id is not None
         user_id = user.id
         tenant_id = user.current_tenant_id
         return jsonable_encoder(AgentService.get_agent_provider(user_id, tenant_id, provider_name))

api/controllers/console/workspace/endpoint.py

@@ -1,3 +1,4 @@
+from flask_login import current_user
 from flask_restx import Resource, fields, reqparse
 from werkzeug.exceptions import Forbidden
 
@@ -5,18 +6,10 @@ from controllers.console import api, console_ns
 from controllers.console.wraps import account_initialization_required, setup_required
 from core.model_runtime.utils.encoders import jsonable_encoder
 from core.plugin.impl.exc import PluginPermissionDeniedError
-from libs.login import current_user, login_required
-from models.account import Account
+from libs.login import login_required
 from services.plugin.endpoint_service import EndpointService
 
 
-def _current_account_with_tenant() -> tuple[Account, str]:
-    assert isinstance(current_user, Account)
-    tenant_id = current_user.current_tenant_id
-    assert tenant_id is not None
-    return current_user, tenant_id
-
-
 @console_ns.route("/workspaces/current/endpoints/create")
 class EndpointCreateApi(Resource):
     @api.doc("create_endpoint")
@@ -41,7 +34,7 @@ class EndpointCreateApi(Resource):
     @login_required
     @account_initialization_required
     def post(self):
-        user, tenant_id = _current_account_with_tenant()
+        user = current_user
         if not user.is_admin_or_owner:
             raise Forbidden()
 
@@ -58,7 +51,7 @@ class EndpointCreateApi(Resource):
         try:
             return {
                 "success": EndpointService.create_endpoint(
-                    tenant_id=tenant_id,
+                    tenant_id=user.current_tenant_id,
                     user_id=user.id,
                     plugin_unique_identifier=plugin_unique_identifier,
                     name=name,
@@ -87,7 +80,7 @@ class EndpointListApi(Resource):
     @login_required
     @account_initialization_required
     def get(self):
-        user, tenant_id = _current_account_with_tenant()
+        user = current_user
 
         parser = reqparse.RequestParser()
         parser.add_argument("page", type=int, required=True, location="args")
@@ -100,7 +93,7 @@ class EndpointListApi(Resource):
         return jsonable_encoder(
             {
                 "endpoints": EndpointService.list_endpoints(
-                    tenant_id=tenant_id,
+                    tenant_id=user.current_tenant_id,
                     user_id=user.id,
                     page=page,
                     page_size=page_size,
@@ -130,7 +123,7 @@ class EndpointListForSinglePluginApi(Resource):
     @login_required
     @account_initialization_required
     def get(self):
-        user, tenant_id = _current_account_with_tenant()
+        user = current_user
 
         parser = reqparse.RequestParser()
         parser.add_argument("page", type=int, required=True, location="args")
@@ -145,7 +138,7 @@ class EndpointListForSinglePluginApi(Resource):
         return jsonable_encoder(
             {
                 "endpoints": EndpointService.list_endpoints_for_single_plugin(
-                    tenant_id=tenant_id,
+                    tenant_id=user.current_tenant_id,
                     user_id=user.id,
                     plugin_id=plugin_id,
                     page=page,
@@ -172,7 +165,7 @@ class EndpointDeleteApi(Resource):
     @login_required
     @account_initialization_required
     def post(self):
-        user, tenant_id = _current_account_with_tenant()
+        user = current_user
 
         parser = reqparse.RequestParser()
         parser.add_argument("endpoint_id", type=str, required=True)
@@ -184,7 +177,9 @@ class EndpointDeleteApi(Resource):
         endpoint_id = args["endpoint_id"]
 
         return {
-            "success": EndpointService.delete_endpoint(tenant_id=tenant_id, user_id=user.id, endpoint_id=endpoint_id)
+            "success": EndpointService.delete_endpoint(
+                tenant_id=user.current_tenant_id, user_id=user.id, endpoint_id=endpoint_id
+            )
         }
 
 
@@ -212,7 +207,7 @@ class EndpointUpdateApi(Resource):
     @login_required
     @account_initialization_required
     def post(self):
-        user, tenant_id = _current_account_with_tenant()
+        user = current_user
 
         parser = reqparse.RequestParser()
         parser.add_argument("endpoint_id", type=str, required=True)
@@ -229,7 +224,7 @@ class EndpointUpdateApi(Resource):
 
         return {
             "success": EndpointService.update_endpoint(
-                tenant_id=tenant_id,
+                tenant_id=user.current_tenant_id,
                 user_id=user.id,
                 endpoint_id=endpoint_id,
                 name=name,
@@ -255,7 +250,7 @@ class EndpointEnableApi(Resource):
     @login_required
     @account_initialization_required
     def post(self):
-        user, tenant_id = _current_account_with_tenant()
+        user = current_user
 
         parser = reqparse.RequestParser()
         parser.add_argument("endpoint_id", type=str, required=True)
@@ -267,7 +262,9 @@ class EndpointEnableApi(Resource):
             raise Forbidden()
 
         return {
-            "success": EndpointService.enable_endpoint(tenant_id=tenant_id, user_id=user.id, endpoint_id=endpoint_id)
+            "success": EndpointService.enable_endpoint(
+                tenant_id=user.current_tenant_id, user_id=user.id, endpoint_id=endpoint_id
+            )
         }
 
 
@@ -288,7 +285,7 @@ class EndpointDisableApi(Resource):
     @login_required
     @account_initialization_required
     def post(self):
-        user, tenant_id = _current_account_with_tenant()
+        user = current_user
 
         parser = reqparse.RequestParser()
         parser.add_argument("endpoint_id", type=str, required=True)
@@ -300,5 +297,7 @@ class EndpointDisableApi(Resource):
             raise Forbidden()
 
         return {
-            "success": EndpointService.disable_endpoint(tenant_id=tenant_id, user_id=user.id, endpoint_id=endpoint_id)
+            "success": EndpointService.disable_endpoint(
+                tenant_id=user.current_tenant_id, user_id=user.id, endpoint_id=endpoint_id
+            )
         }

api/controllers/console/workspace/members.py

@@ -1,6 +1,7 @@
 from urllib import parse
 
 from flask import abort, request
+from flask_login import current_user
 from flask_restx import Resource, marshal_with, reqparse
 
 import services
@@ -25,7 +26,7 @@ from controllers.console.wraps import (
 from extensions.ext_database import db
 from fields.member_fields import account_with_role_list_fields
 from libs.helper import extract_remote_ip
-from libs.login import current_user, login_required
+from libs.login import login_required
 from models.account import Account, TenantAccountRole
 from services.account_service import AccountService, RegisterService, TenantService
 from services.errors.account import AccountAlreadyInTenantError

api/controllers/console/workspace/plugin.py

@@ -540,20 +540,18 @@ class PluginFetchDynamicSelectOptionsApi(Resource):
         parser.add_argument("provider", type=str, required=True, location="args")
         parser.add_argument("action", type=str, required=True, location="args")
         parser.add_argument("parameter", type=str, required=True, location="args")
-        parser.add_argument("credential_id", type=str, required=False, location="args")
         parser.add_argument("provider_type", type=str, required=True, location="args")
         args = parser.parse_args()
 
         try:
             options = PluginParameterService.get_dynamic_select_options(
-                tenant_id=tenant_id,
-                user_id=user_id,
-                plugin_id=args["plugin_id"],
-                provider=args["provider"],
-                action=args["action"],
-                parameter=args["parameter"],
-                credential_id=args["credential_id"],
-                provider_type=args["provider_type"],
+                tenant_id,
+                user_id,
+                args["plugin_id"],
+                args["provider"],
+                args["action"],
+                args["parameter"],
+                args["provider_type"],
             )
         except PluginDaemonClientSideError as e:
             raise ValueError(e)

api/controllers/console/workspace/tool_providers.py

@@ -21,13 +21,11 @@ from core.mcp.auth.auth_provider import OAuthClientProvider
 from core.mcp.error import MCPAuthError, MCPError
 from core.mcp.mcp_client import MCPClient
 from core.model_runtime.utils.encoders import jsonable_encoder
-from core.plugin.entities.plugin_daemon import CredentialType
 from core.plugin.impl.oauth import OAuthHandler
+from core.tools.entities.tool_entities import CredentialType
 from libs.helper import StrLen, alphanumeric, uuid_value
 from libs.login import login_required
 from models.provider_ids import ToolProviderID
-
-# from models.provider_ids import ToolProviderID
 from services.plugin.oauth_service import OAuthProxyService
 from services.tools.api_tools_manage_service import ApiToolManageService
 from services.tools.builtin_tools_manage_service import BuiltinToolManageService

api/controllers/console/workspace/trigger_providers.py

@@ -1,590 +0,0 @@
-import logging
-
-from flask import make_response, redirect, request
-from flask_restx import Resource, reqparse
-from sqlalchemy.orm import Session
-from werkzeug.exceptions import BadRequest, Forbidden
-
-from configs import dify_config
-from controllers.console import api
-from controllers.console.wraps import account_initialization_required, setup_required
-from core.model_runtime.utils.encoders import jsonable_encoder
-from core.plugin.entities.plugin_daemon import CredentialType
-from core.plugin.impl.oauth import OAuthHandler
-from core.trigger.entities.entities import SubscriptionBuilderUpdater
-from core.trigger.trigger_manager import TriggerManager
-from extensions.ext_database import db
-from libs.login import current_user, login_required
-from models.account import Account
-from models.provider_ids import TriggerProviderID
-from services.plugin.oauth_service import OAuthProxyService
-from services.trigger.trigger_provider_service import TriggerProviderService
-from services.trigger.trigger_subscription_builder_service import TriggerSubscriptionBuilderService
-from services.trigger.workflow_plugin_trigger_service import WorkflowPluginTriggerService
-
-logger = logging.getLogger(__name__)
-
-
-class TriggerProviderIconApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def get(self, provider):
-        user = current_user
-        assert isinstance(user, Account)
-        assert user.current_tenant_id is not None
-
-        return TriggerManager.get_trigger_plugin_icon(tenant_id=user.current_tenant_id, provider_id=provider)
-
-
-class TriggerProviderListApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def get(self):
-        """List all trigger providers for the current tenant"""
-        user = current_user
-        assert isinstance(user, Account)
-        assert user.current_tenant_id is not None
-        return jsonable_encoder(TriggerProviderService.list_trigger_providers(user.current_tenant_id))
-
-
-class TriggerProviderInfoApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def get(self, provider):
-        """Get info for a trigger provider"""
-        user = current_user
-        assert isinstance(user, Account)
-        assert user.current_tenant_id is not None
-        return jsonable_encoder(
-            TriggerProviderService.get_trigger_provider(user.current_tenant_id, TriggerProviderID(provider))
-        )
-
-
-class TriggerSubscriptionListApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def get(self, provider):
-        """List all trigger subscriptions for the current tenant's provider"""
-        user = current_user
-        assert isinstance(user, Account)
-        assert user.current_tenant_id is not None
-        if not user.is_admin_or_owner:
-            raise Forbidden()
-
-        try:
-            return jsonable_encoder(
-                TriggerProviderService.list_trigger_provider_subscriptions(
-                    tenant_id=user.current_tenant_id, provider_id=TriggerProviderID(provider)
-                )
-            )
-        except Exception as e:
-            logger.exception("Error listing trigger providers", exc_info=e)
-            raise
-
-
-class TriggerSubscriptionBuilderCreateApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def post(self, provider):
-        """Add a new subscription instance for a trigger provider"""
-        user = current_user
-        assert isinstance(user, Account)
-        assert user.current_tenant_id is not None
-        if not user.is_admin_or_owner:
-            raise Forbidden()
-
-        parser = reqparse.RequestParser()
-        parser.add_argument("credential_type", type=str, required=False, nullable=True, location="json")
-        args = parser.parse_args()
-
-        try:
-            credential_type = CredentialType.of(args.get("credential_type") or CredentialType.UNAUTHORIZED.value)
-            subscription_builder = TriggerSubscriptionBuilderService.create_trigger_subscription_builder(
-                tenant_id=user.current_tenant_id,
-                user_id=user.id,
-                provider_id=TriggerProviderID(provider),
-                credential_type=credential_type,
-            )
-            return jsonable_encoder({"subscription_builder": subscription_builder})
-        except Exception as e:
-            logger.exception("Error adding provider credential", exc_info=e)
-            raise
-
-
-class TriggerSubscriptionBuilderGetApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def get(self, provider, subscription_builder_id):
-        """Get a subscription instance for a trigger provider"""
-        return jsonable_encoder(
-            TriggerSubscriptionBuilderService.get_subscription_builder_by_id(subscription_builder_id)
-        )
-
-
-class TriggerSubscriptionBuilderVerifyApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def post(self, provider, subscription_builder_id):
-        """Verify a subscription instance for a trigger provider"""
-        user = current_user
-        assert isinstance(user, Account)
-        assert user.current_tenant_id is not None
-        if not user.is_admin_or_owner:
-            raise Forbidden()
-
-        parser = reqparse.RequestParser()
-        # The credentials of the subscription builder
-        parser.add_argument("credentials", type=dict, required=False, nullable=True, location="json")
-        args = parser.parse_args()
-
-        try:
-            # Use atomic update_and_verify to prevent race conditions
-            return TriggerSubscriptionBuilderService.update_and_verify_builder(
-                tenant_id=user.current_tenant_id,
-                user_id=user.id,
-                provider_id=TriggerProviderID(provider),
-                subscription_builder_id=subscription_builder_id,
-                subscription_builder_updater=SubscriptionBuilderUpdater(
-                    credentials=args.get("credentials", None),
-                ),
-            )
-        except Exception as e:
-            logger.exception("Error verifying provider credential", exc_info=e)
-            raise ValueError(str(e)) from e
-
-
-class TriggerSubscriptionBuilderUpdateApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def post(self, provider, subscription_builder_id):
-        """Update a subscription instance for a trigger provider"""
-        user = current_user
-        assert isinstance(user, Account)
-        assert user.current_tenant_id is not None
-
-        parser = reqparse.RequestParser()
-        # The name of the subscription builder
-        parser.add_argument("name", type=str, required=False, nullable=True, location="json")
-        # The parameters of the subscription builder
-        parser.add_argument("parameters", type=dict, required=False, nullable=True, location="json")
-        # The properties of the subscription builder
-        parser.add_argument("properties", type=dict, required=False, nullable=True, location="json")
-        # The credentials of the subscription builder
-        parser.add_argument("credentials", type=dict, required=False, nullable=True, location="json")
-        args = parser.parse_args()
-        try:
-            return jsonable_encoder(
-                TriggerSubscriptionBuilderService.update_trigger_subscription_builder(
-                    tenant_id=user.current_tenant_id,
-                    provider_id=TriggerProviderID(provider),
-                    subscription_builder_id=subscription_builder_id,
-                    subscription_builder_updater=SubscriptionBuilderUpdater(
-                        name=args.get("name", None),
-                        parameters=args.get("parameters", None),
-                        properties=args.get("properties", None),
-                        credentials=args.get("credentials", None),
-                    ),
-                )
-            )
-        except Exception as e:
-            logger.exception("Error updating provider credential", exc_info=e)
-            raise
-
-
-class TriggerSubscriptionBuilderLogsApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def get(self, provider, subscription_builder_id):
-        """Get the request logs for a subscription instance for a trigger provider"""
-        user = current_user
-        assert isinstance(user, Account)
-        assert user.current_tenant_id is not None
-
-        try:
-            logs = TriggerSubscriptionBuilderService.list_logs(subscription_builder_id)
-            return jsonable_encoder({"logs": [log.model_dump(mode="json") for log in logs]})
-        except Exception as e:
-            logger.exception("Error getting request logs for subscription builder", exc_info=e)
-            raise
-
-
-class TriggerSubscriptionBuilderBuildApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def post(self, provider, subscription_builder_id):
-        """Build a subscription instance for a trigger provider"""
-        user = current_user
-        assert isinstance(user, Account)
-        assert user.current_tenant_id is not None
-        if not user.is_admin_or_owner:
-            raise Forbidden()
-
-        parser = reqparse.RequestParser()
-        # The name of the subscription builder
-        parser.add_argument("name", type=str, required=False, nullable=True, location="json")
-        # The parameters of the subscription builder
-        parser.add_argument("parameters", type=dict, required=False, nullable=True, location="json")
-        # The properties of the subscription builder
-        parser.add_argument("properties", type=dict, required=False, nullable=True, location="json")
-        # The credentials of the subscription builder
-        parser.add_argument("credentials", type=dict, required=False, nullable=True, location="json")
-        args = parser.parse_args()
-        try:
-            # Use atomic update_and_build to prevent race conditions
-            TriggerSubscriptionBuilderService.update_and_build_builder(
-                tenant_id=user.current_tenant_id,
-                user_id=user.id,
-                provider_id=TriggerProviderID(provider),
-                subscription_builder_id=subscription_builder_id,
-                subscription_builder_updater=SubscriptionBuilderUpdater(
-                    name=args.get("name", None),
-                    parameters=args.get("parameters", None),
-                    properties=args.get("properties", None),
-                ),
-            )
-            return 200
-        except Exception as e:
-            logger.exception("Error building provider credential", exc_info=e)
-            raise ValueError(str(e)) from e
-
-
-class TriggerSubscriptionDeleteApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def post(self, subscription_id):
-        """Delete a subscription instance"""
-        user = current_user
-        assert isinstance(user, Account)
-        assert user.current_tenant_id is not None
-        if not user.is_admin_or_owner:
-            raise Forbidden()
-
-        try:
-            with Session(db.engine) as session:
-                # Delete trigger provider subscription
-                TriggerProviderService.delete_trigger_provider(
-                    session=session,
-                    tenant_id=user.current_tenant_id,
-                    subscription_id=subscription_id,
-                )
-                # Delete plugin triggers
-                WorkflowPluginTriggerService.delete_plugin_trigger_by_subscription(
-                    session=session,
-                    tenant_id=user.current_tenant_id,
-                    subscription_id=subscription_id,
-                )
-                session.commit()
-            return {"result": "success"}
-        except ValueError as e:
-            raise BadRequest(str(e))
-        except Exception as e:
-            logger.exception("Error deleting provider credential", exc_info=e)
-            raise
-
-
-class TriggerOAuthAuthorizeApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def get(self, provider):
-        """Initiate OAuth authorization flow for a trigger provider"""
-        user = current_user
-        assert isinstance(user, Account)
-        assert user.current_tenant_id is not None
-
-        try:
-            provider_id = TriggerProviderID(provider)
-            plugin_id = provider_id.plugin_id
-            provider_name = provider_id.provider_name
-            tenant_id = user.current_tenant_id
-
-            # Get OAuth client configuration
-            oauth_client_params = TriggerProviderService.get_oauth_client(
-                tenant_id=tenant_id,
-                provider_id=provider_id,
-            )
-
-            if oauth_client_params is None:
-                raise Forbidden("No OAuth client configuration found for this trigger provider")
-
-            # Create subscription builder
-            subscription_builder = TriggerSubscriptionBuilderService.create_trigger_subscription_builder(
-                tenant_id=tenant_id,
-                user_id=user.id,
-                provider_id=provider_id,
-                credential_type=CredentialType.OAUTH2,
-            )
-
-            # Create OAuth handler and proxy context
-            oauth_handler = OAuthHandler()
-            context_id = OAuthProxyService.create_proxy_context(
-                user_id=user.id,
-                tenant_id=tenant_id,
-                plugin_id=plugin_id,
-                provider=provider_name,
-                extra_data={
-                    "subscription_builder_id": subscription_builder.id,
-                },
-            )
-
-            # Build redirect URI for callback
-            redirect_uri = f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{provider}/trigger/callback"
-
-            # Get authorization URL
-            authorization_url_response = oauth_handler.get_authorization_url(
-                tenant_id=tenant_id,
-                user_id=user.id,
-                plugin_id=plugin_id,
-                provider=provider_name,
-                redirect_uri=redirect_uri,
-                system_credentials=oauth_client_params,
-            )
-
-            # Create response with cookie
-            response = make_response(
-                jsonable_encoder(
-                    {
-                        "authorization_url": authorization_url_response.authorization_url,
-                        "subscription_builder_id": subscription_builder.id,
-                        "subscription_builder": subscription_builder,
-                    }
-                )
-            )
-            response.set_cookie(
-                "context_id",
-                context_id,
-                httponly=True,
-                samesite="Lax",
-                max_age=OAuthProxyService.__MAX_AGE__,
-            )
-
-            return response
-
-        except Exception as e:
-            logger.exception("Error initiating OAuth flow", exc_info=e)
-            raise
-
-
-class TriggerOAuthCallbackApi(Resource):
-    @setup_required
-    def get(self, provider):
-        """Handle OAuth callback for trigger provider"""
-        context_id = request.cookies.get("context_id")
-        if not context_id:
-            raise Forbidden("context_id not found")
-
-        # Use and validate proxy context
-        context = OAuthProxyService.use_proxy_context(context_id)
-        if context is None:
-            raise Forbidden("Invalid context_id")
-
-        # Parse provider ID
-        provider_id = TriggerProviderID(provider)
-        plugin_id = provider_id.plugin_id
-        provider_name = provider_id.provider_name
-        user_id = context.get("user_id")
-        tenant_id = context.get("tenant_id")
-        subscription_builder_id = context.get("subscription_builder_id")
-
-        # Get OAuth client configuration
-        oauth_client_params = TriggerProviderService.get_oauth_client(
-            tenant_id=tenant_id,
-            provider_id=provider_id,
-        )
-
-        if oauth_client_params is None:
-            raise Forbidden("No OAuth client configuration found for this trigger provider")
-
-        # Get OAuth credentials from callback
-        oauth_handler = OAuthHandler()
-        redirect_uri = f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{provider}/trigger/callback"
-
-        credentials_response = oauth_handler.get_credentials(
-            tenant_id=tenant_id,
-            user_id=user_id,
-            plugin_id=plugin_id,
-            provider=provider_name,
-            redirect_uri=redirect_uri,
-            system_credentials=oauth_client_params,
-            request=request,
-        )
-
-        credentials = credentials_response.credentials
-        expires_at = credentials_response.expires_at
-
-        if not credentials:
-            raise Exception("Failed to get OAuth credentials")
-
-        # Update subscription builder
-        TriggerSubscriptionBuilderService.update_trigger_subscription_builder(
-            tenant_id=tenant_id,
-            provider_id=provider_id,
-            subscription_builder_id=subscription_builder_id,
-            subscription_builder_updater=SubscriptionBuilderUpdater(
-                credentials=credentials,
-                credential_expires_at=expires_at,
-            ),
-        )
-        # Redirect to OAuth callback page
-        return redirect(f"{dify_config.CONSOLE_WEB_URL}/oauth-callback")
-
-
-class TriggerOAuthClientManageApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def get(self, provider):
-        """Get OAuth client configuration for a provider"""
-        user = current_user
-        assert isinstance(user, Account)
-        assert user.current_tenant_id is not None
-        if not user.is_admin_or_owner:
-            raise Forbidden()
-
-        try:
-            provider_id = TriggerProviderID(provider)
-
-            # Get custom OAuth client params if exists
-            custom_params = TriggerProviderService.get_custom_oauth_client_params(
-                tenant_id=user.current_tenant_id,
-                provider_id=provider_id,
-            )
-
-            # Check if custom client is enabled
-            is_custom_enabled = TriggerProviderService.is_oauth_custom_client_enabled(
-                tenant_id=user.current_tenant_id,
-                provider_id=provider_id,
-            )
-
-            # Check if there's a system OAuth client
-            system_client = TriggerProviderService.get_oauth_client(
-                tenant_id=user.current_tenant_id,
-                provider_id=provider_id,
-            )
-            provider_controller = TriggerManager.get_trigger_provider(user.current_tenant_id, provider_id)
-            redirect_uri = f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{provider}/trigger/callback"
-            return jsonable_encoder(
-                {
-                    "configured": bool(custom_params or system_client),
-                    "oauth_client_schema": provider_controller.get_oauth_client_schema(),
-                    "custom_configured": bool(custom_params),
-                    "custom_enabled": is_custom_enabled,
-                    "redirect_uri": redirect_uri,
-                    "params": custom_params or {},
-                }
-            )
-
-        except Exception as e:
-            logger.exception("Error getting OAuth client", exc_info=e)
-            raise
-
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def post(self, provider):
-        """Configure custom OAuth client for a provider"""
-        user = current_user
-        assert isinstance(user, Account)
-        assert user.current_tenant_id is not None
-        if not user.is_admin_or_owner:
-            raise Forbidden()
-
-        parser = reqparse.RequestParser()
-        parser.add_argument("client_params", type=dict, required=False, nullable=True, location="json")
-        parser.add_argument("enabled", type=bool, required=False, nullable=True, location="json")
-        args = parser.parse_args()
-
-        try:
-            provider_id = TriggerProviderID(provider)
-            return TriggerProviderService.save_custom_oauth_client_params(
-                tenant_id=user.current_tenant_id,
-                provider_id=provider_id,
-                client_params=args.get("client_params"),
-                enabled=args.get("enabled"),
-            )
-
-        except ValueError as e:
-            raise BadRequest(str(e))
-        except Exception as e:
-            logger.exception("Error configuring OAuth client", exc_info=e)
-            raise
-
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def delete(self, provider):
-        """Remove custom OAuth client configuration"""
-        user = current_user
-        assert isinstance(user, Account)
-        assert user.current_tenant_id is not None
-        if not user.is_admin_or_owner:
-            raise Forbidden()
-
-        try:
-            provider_id = TriggerProviderID(provider)
-
-            return TriggerProviderService.delete_custom_oauth_client_params(
-                tenant_id=user.current_tenant_id,
-                provider_id=provider_id,
-            )
-        except ValueError as e:
-            raise BadRequest(str(e))
-        except Exception as e:
-            logger.exception("Error removing OAuth client", exc_info=e)
-            raise
-
-
-# Trigger Subscription
-api.add_resource(TriggerProviderIconApi, "/workspaces/current/trigger-provider/<path:provider>/icon")
-api.add_resource(TriggerProviderListApi, "/workspaces/current/triggers")
-api.add_resource(TriggerProviderInfoApi, "/workspaces/current/trigger-provider/<path:provider>/info")
-api.add_resource(TriggerSubscriptionListApi, "/workspaces/current/trigger-provider/<path:provider>/subscriptions/list")
-api.add_resource(
-    TriggerSubscriptionDeleteApi,
-    "/workspaces/current/trigger-provider/<path:subscription_id>/subscriptions/delete",
-)
-
-# Trigger Subscription Builder
-api.add_resource(
-    TriggerSubscriptionBuilderCreateApi,
-    "/workspaces/current/trigger-provider/<path:provider>/subscriptions/builder/create",
-)
-api.add_resource(
-    TriggerSubscriptionBuilderGetApi,
-    "/workspaces/current/trigger-provider/<path:provider>/subscriptions/builder/<path:subscription_builder_id>",
-)
-api.add_resource(
-    TriggerSubscriptionBuilderUpdateApi,
-    "/workspaces/current/trigger-provider/<path:provider>/subscriptions/builder/update/<path:subscription_builder_id>",
-)
-api.add_resource(
-    TriggerSubscriptionBuilderVerifyApi,
-    "/workspaces/current/trigger-provider/<path:provider>/subscriptions/builder/verify/<path:subscription_builder_id>",
-)
-api.add_resource(
-    TriggerSubscriptionBuilderBuildApi,
-    "/workspaces/current/trigger-provider/<path:provider>/subscriptions/builder/build/<path:subscription_builder_id>",
-)
-api.add_resource(
-    TriggerSubscriptionBuilderLogsApi,
-    "/workspaces/current/trigger-provider/<path:provider>/subscriptions/builder/logs/<path:subscription_builder_id>",
-)
-
-
-# OAuth
-api.add_resource(
-    TriggerOAuthAuthorizeApi, "/workspaces/current/trigger-provider/<path:provider>/subscriptions/oauth/authorize"
-)
-api.add_resource(TriggerOAuthCallbackApi, "/oauth/plugin/<path:provider>/trigger/callback")
-api.add_resource(TriggerOAuthClientManageApi, "/workspaces/current/trigger-provider/<path:provider>/oauth/client")

api/controllers/console/workspace/workspace.py

@@ -1,6 +1,7 @@
 import logging
 
 from flask import request
+from flask_login import current_user
 from flask_restx import Resource, fields, inputs, marshal, marshal_with, reqparse
 from sqlalchemy import select
 from werkzeug.exceptions import Unauthorized
@@ -23,7 +24,7 @@ from controllers.console.wraps import (
 )
 from extensions.ext_database import db
 from libs.helper import TimestampField
-from libs.login import current_user, login_required
+from libs.login import login_required
 from models.account import Account, Tenant, TenantStatus
 from services.account_service import TenantService
 from services.feature_service import FeatureService
@@ -50,6 +51,8 @@ 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,
 }
 
 tenants_fields = {

api/controllers/console/wraps.py

@@ -7,13 +7,13 @@ from functools import wraps
 from typing import ParamSpec, TypeVar
 
 from flask import abort, request
+from flask_login import current_user
 
 from configs import dify_config
 from controllers.console.workspace.error import AccountNotInitializedError
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
-from libs.login import current_user
-from models.account import Account, AccountStatus
+from models.account import AccountStatus
 from models.dataset import RateLimitLog
 from models.model import DifySetup
 from services.feature_service import FeatureService, LicenseStatus
@@ -25,16 +25,11 @@ P = ParamSpec("P")
 R = TypeVar("R")
 
 
-def _current_account() -> Account:
-    assert isinstance(current_user, Account)
-    return current_user
-
-
 def account_initialization_required(view: Callable[P, R]):
     @wraps(view)
     def decorated(*args: P.args, **kwargs: P.kwargs):
         # check account initialization
-        account = _current_account()
+        account = current_user
 
         if account.status == AccountStatus.UNINITIALIZED:
             raise AccountNotInitializedError()
@@ -80,9 +75,7 @@ def only_edition_self_hosted(view: Callable[P, R]):
 def cloud_edition_billing_enabled(view: Callable[P, R]):
     @wraps(view)
     def decorated(*args: P.args, **kwargs: P.kwargs):
-        account = _current_account()
-        assert account.current_tenant_id is not None
-        features = FeatureService.get_features(account.current_tenant_id)
+        features = FeatureService.get_features(current_user.current_tenant_id)
         if not features.billing.enabled:
             abort(403, "Billing feature is not enabled.")
         return view(*args, **kwargs)
@@ -94,10 +87,7 @@ def cloud_edition_billing_resource_check(resource: str):
     def interceptor(view: Callable[P, R]):
         @wraps(view)
         def decorated(*args: P.args, **kwargs: P.kwargs):
-            account = _current_account()
-            assert account.current_tenant_id is not None
-            tenant_id = account.current_tenant_id
-            features = FeatureService.get_features(tenant_id)
+            features = FeatureService.get_features(current_user.current_tenant_id)
             if features.billing.enabled:
                 members = features.members
                 apps = features.apps
@@ -138,9 +128,7 @@ def cloud_edition_billing_knowledge_limit_check(resource: str):
     def interceptor(view: Callable[P, R]):
         @wraps(view)
         def decorated(*args: P.args, **kwargs: P.kwargs):
-            account = _current_account()
-            assert account.current_tenant_id is not None
-            features = FeatureService.get_features(account.current_tenant_id)
+            features = FeatureService.get_features(current_user.current_tenant_id)
             if features.billing.enabled:
                 if resource == "add_segment":
                     if features.billing.subscription.plan == "sandbox":
@@ -163,13 +151,10 @@ def cloud_edition_billing_rate_limit_check(resource: str):
         @wraps(view)
         def decorated(*args: P.args, **kwargs: P.kwargs):
             if resource == "knowledge":
-                account = _current_account()
-                assert account.current_tenant_id is not None
-                tenant_id = account.current_tenant_id
-                knowledge_rate_limit = FeatureService.get_knowledge_rate_limit(tenant_id)
+                knowledge_rate_limit = FeatureService.get_knowledge_rate_limit(current_user.current_tenant_id)
                 if knowledge_rate_limit.enabled:
                     current_time = int(time.time() * 1000)
-                    key = f"rate_limit_{tenant_id}"
+                    key = f"rate_limit_{current_user.current_tenant_id}"
 
                     redis_client.zadd(key, {current_time: current_time})
 
@@ -180,7 +165,7 @@ def cloud_edition_billing_rate_limit_check(resource: str):
                     if request_count > knowledge_rate_limit.limit:
                         # add ratelimit record
                         rate_limit_log = RateLimitLog(
-                            tenant_id=tenant_id,
+                            tenant_id=current_user.current_tenant_id,
                             subscription_plan=knowledge_rate_limit.subscription_plan,
                             operation="knowledge",
                         )
@@ -200,17 +185,14 @@ def cloud_utm_record(view: Callable[P, R]):
     @wraps(view)
     def decorated(*args: P.args, **kwargs: P.kwargs):
         with contextlib.suppress(Exception):
-            account = _current_account()
-            assert account.current_tenant_id is not None
-            tenant_id = account.current_tenant_id
-            features = FeatureService.get_features(tenant_id)
+            features = FeatureService.get_features(current_user.current_tenant_id)
 
             if features.billing.enabled:
                 utm_info = request.cookies.get("utm_info")
 
                 if utm_info:
                     utm_info_dict: dict = json.loads(utm_info)
-                    OperationService.record_utm(tenant_id, utm_info_dict)
+                    OperationService.record_utm(current_user.current_tenant_id, utm_info_dict)
 
         return view(*args, **kwargs)
 
@@ -289,9 +271,7 @@ def enable_change_email(view: Callable[P, R]):
 def is_allow_transfer_owner(view: Callable[P, R]):
     @wraps(view)
     def decorated(*args: P.args, **kwargs: P.kwargs):
-        account = _current_account()
-        assert account.current_tenant_id is not None
-        features = FeatureService.get_features(account.current_tenant_id)
+        features = FeatureService.get_features(current_user.current_tenant_id)
         if features.is_allow_transfer_workspace:
             return view(*args, **kwargs)
 
@@ -304,9 +284,7 @@ def is_allow_transfer_owner(view: Callable[P, R]):
 def knowledge_pipeline_publish_enabled(view):
     @wraps(view)
     def decorated(*args, **kwargs):
-        account = _current_account()
-        assert account.current_tenant_id is not None
-        features = FeatureService.get_features(account.current_tenant_id)
+        features = FeatureService.get_features(current_user.current_tenant_id)
         if features.knowledge_pipeline.publish_enabled:
             return view(*args, **kwargs)
         abort(403)

api/controllers/inner_api/plugin/wraps.py

@@ -25,8 +25,8 @@ def get_user(tenant_id: str, user_id: str | None) -> EndUser:
     As a result, it could only be considered as an end user id.
     """
     if not user_id:
-        user_id = DefaultEndUserSessionID.DEFAULT_SESSION_ID
-    is_anonymous = user_id == DefaultEndUserSessionID.DEFAULT_SESSION_ID
+        user_id = DefaultEndUserSessionID.DEFAULT_SESSION_ID.value
+    is_anonymous = user_id == DefaultEndUserSessionID.DEFAULT_SESSION_ID.value
     try:
         with Session(db.engine) as session:
             user_model = None
@@ -85,7 +85,7 @@ def get_user_tenant(view: Callable[P, R] | None = None):
                 raise ValueError("tenant_id is required")
 
             if not user_id:
-                user_id = DefaultEndUserSessionID.DEFAULT_SESSION_ID
+                user_id = DefaultEndUserSessionID.DEFAULT_SESSION_ID.value
 
             try:
                 tenant_model = (

api/controllers/mcp/mcp.py

@@ -9,9 +9,10 @@ from controllers.console.app.mcp_server import AppMCPServerStatus
 from controllers.mcp import mcp_ns
 from core.app.app_config.entities import VariableEntity
 from core.mcp import types as mcp_types
+from core.mcp.server.streamable_http import handle_mcp_request
 from extensions.ext_database import db
 from libs import helper
-from models.model import App, AppMCPServer, AppMode
+from models.model import App, AppMCPServer, AppMode, EndUser
 
 
 class MCPRequestError(Exception):
@@ -194,6 +195,50 @@ class MCPAppApi(Resource):
             except ValidationError as e:
                 raise MCPRequestError(mcp_types.INVALID_PARAMS, f"Invalid MCP request: {str(e)}")
 
-        mcp_server_handler = MCPServerStreamableHTTPRequestHandler(app, request, converted_user_input_form)
-        response = mcp_server_handler.handle()
-        return helper.compact_generate_response(response)
+    def _retrieve_end_user(self, tenant_id: str, mcp_server_id: str, session: Session) -> EndUser | None:
+        """Get end user from existing session - optimized query"""
+        return (
+            session.query(EndUser)
+            .where(EndUser.tenant_id == tenant_id)
+            .where(EndUser.session_id == mcp_server_id)
+            .where(EndUser.type == "mcp")
+            .first()
+        )
+
+    def _create_end_user(
+        self, client_name: str, tenant_id: str, app_id: str, mcp_server_id: str, session: Session
+    ) -> EndUser:
+        """Create end user in existing session"""
+        end_user = EndUser(
+            tenant_id=tenant_id,
+            app_id=app_id,
+            type="mcp",
+            name=client_name,
+            session_id=mcp_server_id,
+        )
+        session.add(end_user)
+        session.flush()  # Use flush instead of commit to keep transaction open
+        session.refresh(end_user)
+        return end_user
+
+    def _handle_mcp_request(
+        self,
+        app: App,
+        mcp_server: AppMCPServer,
+        mcp_request: mcp_types.ClientRequest,
+        user_input_form: list[VariableEntity],
+        session: Session,
+        request_id: Union[int, str],
+    ) -> mcp_types.JSONRPCResponse | mcp_types.JSONRPCError | None:
+        """Handle MCP request and return response"""
+        end_user = self._retrieve_end_user(mcp_server.tenant_id, mcp_server.id, session)
+
+        if not end_user and isinstance(mcp_request.root, mcp_types.InitializeRequest):
+            client_info = mcp_request.root.params.clientInfo
+            client_name = f"{client_info.name}@{client_info.version}"
+            # Commit the session before creating end user to avoid transaction conflicts
+            session.commit()
+            with Session(db.engine, expire_on_commit=False) as create_session, create_session.begin():
+                end_user = self._create_end_user(client_name, app.tenant_id, app.id, mcp_server.id, create_session)
+
+        return handle_mcp_request(app, mcp_request, user_input_form, mcp_server, end_user, request_id)

api/controllers/service_api/__init__.py

@@ -19,6 +19,7 @@ from .app import (
     annotation,
     app,
     audio,
+    chatflow_memory,
     completion,
     conversation,
     file,
@@ -40,6 +41,7 @@ __all__ = [
     "annotation",
     "app",
     "audio",
+    "chatflow_memory",
     "completion",
     "conversation",
     "dataset",

api/controllers/service_api/app/chatflow_memory.py

@@ -0,0 +1,124 @@
+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.entities.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')

api/controllers/service_api/wraps.py

@@ -313,7 +313,7 @@ def create_or_update_end_user_for_user_id(app_model: App, user_id: str | None =
     Create or update session terminal based on user ID.
     """
     if not user_id:
-        user_id = DefaultEndUserSessionID.DEFAULT_SESSION_ID
+        user_id = DefaultEndUserSessionID.DEFAULT_SESSION_ID.value
 
     with Session(db.engine, expire_on_commit=False) as session:
         end_user = (
@@ -332,7 +332,7 @@ def create_or_update_end_user_for_user_id(app_model: App, user_id: str | None =
                 tenant_id=app_model.tenant_id,
                 app_id=app_model.id,
                 type="service_api",
-                is_anonymous=user_id == DefaultEndUserSessionID.DEFAULT_SESSION_ID,
+                is_anonymous=user_id == DefaultEndUserSessionID.DEFAULT_SESSION_ID.value,
                 session_id=user_id,
             )
             session.add(end_user)

api/controllers/trigger/__init__.py

@@ -1,12 +0,0 @@
-from flask import Blueprint
-
-# Create trigger blueprint
-bp = Blueprint("trigger", __name__, url_prefix="/triggers")
-
-# Import routes after blueprint creation to avoid circular imports
-from . import trigger, webhook
-
-__all__ = [
-    "trigger",
-    "webhook",
-]

api/controllers/trigger/trigger.py

@@ -1,43 +0,0 @@
-import logging
-import re
-
-from flask import jsonify, request
-from werkzeug.exceptions import NotFound
-
-from controllers.trigger import bp
-from services.trigger.trigger_service import TriggerService
-from services.trigger.trigger_subscription_builder_service import TriggerSubscriptionBuilderService
-
-logger = logging.getLogger(__name__)
-
-UUID_PATTERN = r"^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$"
-UUID_MATCHER = re.compile(UUID_PATTERN)
-
-
-@bp.route("/plugin/<string:endpoint_id>", methods=["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"])
-def trigger_endpoint(endpoint_id: str):
-    """
-    Handle endpoint trigger calls.
-    """
-    # endpoint_id must be UUID
-    if not UUID_MATCHER.match(endpoint_id):
-        raise NotFound("Invalid endpoint ID")
-    handling_chain = [
-        TriggerService.process_endpoint,
-        TriggerSubscriptionBuilderService.process_builder_validation_endpoint,
-    ]
-    response = None
-    try:
-        for handler in handling_chain:
-            response = handler(endpoint_id, request)
-            if response:
-                break
-        if not response:
-            logger.error("Endpoint not found for {endpoint_id}")
-            return jsonify({"error": "Endpoint not found"}), 404
-        return response
-    except ValueError as e:
-        raise NotFound(str(e))
-    except Exception as e:
-        logger.exception("Webhook processing failed for {endpoint_id}")
-        return jsonify({"error": "Internal server error", "message": str(e)}), 500

api/controllers/trigger/webhook.py

@@ -1,104 +0,0 @@
-import logging
-import time
-
-from flask import jsonify
-from werkzeug.exceptions import NotFound, RequestEntityTooLarge
-
-from controllers.trigger import bp
-from services.trigger.trigger_debug_service import TriggerDebugService, WebhookDebugEvent
-from services.trigger.webhook_service import WebhookService
-
-logger = logging.getLogger(__name__)
-
-
-def _prepare_webhook_execution(webhook_id: str, is_debug: bool = False):
-    """Fetch trigger context, extract request data, and validate payload using unified processing.
-
-    Args:
-        webhook_id: The webhook ID to process
-        is_debug: If True, skip status validation for debug mode
-    """
-    webhook_trigger, workflow, node_config = WebhookService.get_webhook_trigger_and_workflow(
-        webhook_id, is_debug=is_debug
-    )
-
-    try:
-        # Use new unified extraction and validation
-        webhook_data = WebhookService.extract_and_validate_webhook_data(webhook_trigger, node_config)
-        return webhook_trigger, workflow, node_config, webhook_data, None
-    except ValueError as e:
-        # Fall back to raw extraction for error reporting
-        webhook_data = WebhookService.extract_webhook_data(webhook_trigger)
-        return webhook_trigger, workflow, node_config, webhook_data, str(e)
-
-
-@bp.route("/webhook/<string:webhook_id>", methods=["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"])
-def handle_webhook(webhook_id: str):
-    """
-    Handle webhook trigger calls.
-
-    This endpoint receives webhook calls and processes them according to the
-    configured webhook trigger settings.
-    """
-    try:
-        webhook_trigger, workflow, node_config, webhook_data, error = _prepare_webhook_execution(webhook_id)
-        if error:
-            return jsonify({"error": "Bad Request", "message": error}), 400
-
-        # Process webhook call (send to Celery)
-        WebhookService.trigger_workflow_execution(webhook_trigger, webhook_data, workflow)
-
-        # Return configured response
-        response_data, status_code = WebhookService.generate_webhook_response(node_config)
-        return jsonify(response_data), status_code
-
-    except ValueError as e:
-        raise NotFound(str(e))
-    except RequestEntityTooLarge:
-        raise
-    except Exception as e:
-        logger.exception("Webhook processing failed for %s", webhook_id)
-        return jsonify({"error": "Internal server error", "message": str(e)}), 500
-
-
-@bp.route("/webhook-debug/<string:webhook_id>", methods=["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"])
-def handle_webhook_debug(webhook_id: str):
-    """Handle webhook debug calls without triggering production workflow execution."""
-    try:
-        webhook_trigger, _, node_config, webhook_data, error = _prepare_webhook_execution(webhook_id, is_debug=True)
-        if error:
-            return jsonify({"error": "Bad Request", "message": error}), 400
-
-        workflow_inputs = WebhookService.build_workflow_inputs(webhook_data)
-
-        # Generate pool key and dispatch debug event
-        pool_key: str = WebhookDebugEvent.build_pool_key(
-            tenant_id=webhook_trigger.tenant_id,
-            app_id=webhook_trigger.app_id,
-            node_id=webhook_trigger.node_id,
-        )
-        event = WebhookDebugEvent(
-            request_id=f"webhook_debug_{webhook_trigger.webhook_id}_{int(time.time() * 1000)}",
-            timestamp=int(time.time()),
-            node_id=webhook_trigger.node_id,
-            payload={
-                "inputs": workflow_inputs,
-                "webhook_data": webhook_data,
-                "method": webhook_data.get("method"),
-            },
-        )
-        TriggerDebugService.dispatch(
-            tenant_id=webhook_trigger.tenant_id,
-            event=event,
-            pool_key=pool_key,
-        )
-        response_data, status_code = WebhookService.generate_webhook_response(node_config)
-        return jsonify(response_data), status_code
-
-    except ValueError as e:
-        raise NotFound(str(e))
-    except RequestEntityTooLarge:
-        raise
-    except Exception as e:
-        logger.exception("Webhook debug processing failed for %s", webhook_id)
-        return jsonify({"error": "Internal server error", "message": str(e)}), 500

api/controllers/web/__init__.py

@@ -18,6 +18,7 @@ web_ns = Namespace("web", description="Web application API operations", path="/"
 from . import (
     app,
     audio,
+    chatflow_memory,
     completion,
     conversation,
     feature,
@@ -39,6 +40,7 @@ __all__ = [
     "app",
     "audio",
     "bp",
+    "chatflow_memory",
     "completion",
     "conversation",
     "feature",

api/controllers/web/chatflow_memory.py

@@ -0,0 +1,123 @@
+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.entities.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')

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

@@ -197,12 +197,12 @@ class DatasetConfigManager:
 
         # strategy
         if "strategy" not in config["agent_mode"] or not config["agent_mode"].get("strategy"):
-            config["agent_mode"]["strategy"] = PlanningStrategy.ROUTER
+            config["agent_mode"]["strategy"] = PlanningStrategy.ROUTER.value
 
         has_datasets = False
         if config.get("agent_mode", {}).get("strategy") in {
-            PlanningStrategy.ROUTER,
-            PlanningStrategy.REACT_ROUTER,
+            PlanningStrategy.ROUTER.value,
+            PlanningStrategy.REACT_ROUTER.value,
         }:
             for tool in config.get("agent_mode", {}).get("tools", []):
                 key = list(tool.keys())[0]

api/core/app/app_config/easy_ui_based_app/model_config/converter.py

@@ -68,13 +68,9 @@ class ModelConfigConverter:
         # get model mode
         model_mode = model_config.mode
         if not model_mode:
-            model_mode = LLMMode.CHAT
+            model_mode = LLMMode.CHAT.value
             if model_schema and model_schema.model_properties.get(ModelPropertyKey.MODE):
-                try:
-                    model_mode = LLMMode(model_schema.model_properties[ModelPropertyKey.MODE])
-                except ValueError:
-                    # Fall back to CHAT mode if the stored value is invalid
-                    model_mode = LLMMode.CHAT
+                model_mode = LLMMode(model_schema.model_properties[ModelPropertyKey.MODE]).value
 
         if not model_schema:
             raise ValueError(f"Model {model_name} not exist.")

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

@@ -100,7 +100,7 @@ class PromptTemplateConfigManager:
             if config["model"]["mode"] not in model_mode_vals:
                 raise ValueError(f"model.mode must be in {model_mode_vals} when prompt_type is advanced")
 
-            if app_mode == AppMode.CHAT and config["model"]["mode"] == ModelMode.COMPLETION:
+            if app_mode == AppMode.CHAT and config["model"]["mode"] == ModelMode.COMPLETION.value:
                 user_prefix = config["completion_prompt_config"]["conversation_histories_role"]["user_prefix"]
                 assistant_prefix = config["completion_prompt_config"]["conversation_histories_role"]["assistant_prefix"]
 
@@ -110,7 +110,7 @@ class PromptTemplateConfigManager:
                 if not assistant_prefix:
                     config["completion_prompt_config"]["conversation_histories_role"]["assistant_prefix"] = "Assistant"
 
-            if config["model"]["mode"] == ModelMode.CHAT:
+            if config["model"]["mode"] == ModelMode.CHAT.value:
                 prompt_list = config["chat_prompt_config"]["prompt"]
 
                 if len(prompt_list) > 10:

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

@@ -1,10 +1,11 @@
 import logging
 import time
-from collections.abc import Mapping
+from collections.abc import Mapping, MutableMapping
 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
@@ -20,11 +21,14 @@ 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.entities import GraphRuntimeState, VariablePool
 from core.workflow.graph_engine.command_channels.redis_channel import RedisChannel
+from core.workflow.graph_events import GraphRunSucceededEvent
 from core.workflow.system_variable import SystemVariable
 from core.workflow.variable_loader import VariableLoader
 from core.workflow.workflow_entry import WorkflowEntry
@@ -34,6 +38,8 @@ 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__)
 
@@ -70,6 +76,11 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
         self._app = app
 
     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)
 
@@ -133,6 +144,7 @@ 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
@@ -177,6 +189,31 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
         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,
@@ -374,3 +411,67 @@ 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)

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

@@ -186,7 +186,7 @@ class AgentChatAppConfigManager(BaseAppConfigManager):
             raise ValueError("enabled in agent_mode must be of boolean type")
 
         if not agent_mode.get("strategy"):
-            agent_mode["strategy"] = PlanningStrategy.ROUTER
+            agent_mode["strategy"] = PlanningStrategy.ROUTER.value
 
         if agent_mode["strategy"] not in [member.value for member in list(PlanningStrategy.__members__.values())]:
             raise ValueError("strategy in agent_mode must be in the specified strategy list")

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

@@ -198,9 +198,9 @@ class AgentChatAppRunner(AppRunner):
         # start agent runner
         if agent_entity.strategy == AgentEntity.Strategy.CHAIN_OF_THOUGHT:
             # check LLM mode
-            if model_schema.model_properties.get(ModelPropertyKey.MODE) == LLMMode.CHAT:
+            if model_schema.model_properties.get(ModelPropertyKey.MODE) == LLMMode.CHAT.value:
                 runner_cls = CotChatAgentRunner
-            elif model_schema.model_properties.get(ModelPropertyKey.MODE) == LLMMode.COMPLETION:
+            elif model_schema.model_properties.get(ModelPropertyKey.MODE) == LLMMode.COMPLETION.value:
                 runner_cls = CotCompletionAgentRunner
             else:
                 raise ValueError(f"Invalid LLM mode: {model_schema.model_properties.get(ModelPropertyKey.MODE)}")

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

@@ -38,7 +38,6 @@ from core.file import FILE_MODEL_IDENTITY, File
 from core.plugin.impl.datasource import PluginDatasourceManager
 from core.tools.entities.tool_entities import ToolProviderType
 from core.tools.tool_manager import ToolManager
-from core.trigger.trigger_manager import TriggerManager
 from core.variables.segments import ArrayFileSegment, FileSegment, Segment
 from core.workflow.entities import WorkflowExecution, WorkflowNodeExecution
 from core.workflow.enums import NodeType, WorkflowNodeExecutionStatus
@@ -182,11 +181,6 @@ class WorkflowResponseConverter:
             response.data.extras["icon"] = provider_entity.declaration.identity.generate_datasource_icon_url(
                 self._application_generate_entity.app_config.tenant_id
             )
-        elif event.node_type == NodeType.TRIGGER_PLUGIN:
-            response.data.extras["icon"] = TriggerManager.get_trigger_plugin_icon(
-                self._application_generate_entity.app_config.tenant_id,
-                event.provider_id,
-            )
 
         return response
 

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

@@ -229,8 +229,8 @@ class PipelineRunner(WorkflowBasedAppRunner):
             workflow_id=workflow.id,
             graph_config=graph_config,
             user_id=self.application_generate_entity.user_id,
-            user_from=UserFrom.ACCOUNT,
-            invoke_from=InvokeFrom.SERVICE_API,
+            user_from=UserFrom.ACCOUNT.value,
+            invoke_from=InvokeFrom.SERVICE_API.value,
             call_depth=0,
         )
 

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

@@ -3,7 +3,7 @@ import logging
 import threading
 import uuid
 from collections.abc import Generator, Mapping, Sequence
-from typing import Any, Literal, Optional, Union, overload
+from typing import Any, Literal, Union, overload
 
 from flask import Flask, current_app
 from pydantic import ValidationError
@@ -53,8 +53,6 @@ class WorkflowAppGenerator(BaseAppGenerator):
         invoke_from: InvokeFrom,
         streaming: Literal[True],
         call_depth: int,
-        triggered_from: Optional[WorkflowRunTriggeredFrom] = None,
-        root_node_id: Optional[str] = None,
     ) -> Generator[Mapping | str, None, None]: ...
 
     @overload
@@ -68,8 +66,6 @@ class WorkflowAppGenerator(BaseAppGenerator):
         invoke_from: InvokeFrom,
         streaming: Literal[False],
         call_depth: int,
-        triggered_from: Optional[WorkflowRunTriggeredFrom] = None,
-        root_node_id: Optional[str] = None,
     ) -> Mapping[str, Any]: ...
 
     @overload
@@ -83,8 +79,6 @@ class WorkflowAppGenerator(BaseAppGenerator):
         invoke_from: InvokeFrom,
         streaming: bool,
         call_depth: int,
-        triggered_from: Optional[WorkflowRunTriggeredFrom] = None,
-        root_node_id: Optional[str] = None,
     ) -> Union[Mapping[str, Any], Generator[Mapping | str, None, None]]: ...
 
     def generate(
@@ -97,8 +91,6 @@ class WorkflowAppGenerator(BaseAppGenerator):
         invoke_from: InvokeFrom,
         streaming: bool = True,
         call_depth: int = 0,
-        triggered_from: Optional[WorkflowRunTriggeredFrom] = None,
-        root_node_id: Optional[str] = None,
     ) -> Union[Mapping[str, Any], Generator[Mapping | str, None, None]]:
         files: Sequence[Mapping[str, Any]] = args.get("files") or []
 
@@ -134,20 +126,17 @@ class WorkflowAppGenerator(BaseAppGenerator):
             **extract_external_trace_id_from_args(args),
         }
         workflow_run_id = str(uuid.uuid4())
-        if triggered_from in (WorkflowRunTriggeredFrom.DEBUGGING, WorkflowRunTriggeredFrom.APP_RUN):
-            # start node get inputs
-            inputs = self._prepare_user_inputs(
-                user_inputs=inputs,
-                variables=app_config.variables,
-                tenant_id=app_model.tenant_id,
-                strict_type_validation=True if invoke_from == InvokeFrom.SERVICE_API else False,
-            )
         # init application generate entity
         application_generate_entity = WorkflowAppGenerateEntity(
             task_id=str(uuid.uuid4()),
             app_config=app_config,
             file_upload_config=file_extra_config,
-            inputs=inputs,
+            inputs=self._prepare_user_inputs(
+                user_inputs=inputs,
+                variables=app_config.variables,
+                tenant_id=app_model.tenant_id,
+                strict_type_validation=True if invoke_from == InvokeFrom.SERVICE_API else False,
+            ),
             files=list(system_files),
             user_id=user.id,
             stream=streaming,
@@ -166,10 +155,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
         # Create session factory
         session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
         # Create workflow execution(aka workflow run) repository
-        if triggered_from is not None:
-            # Use explicitly provided triggered_from (for async triggers)
-            workflow_triggered_from = triggered_from
-        elif invoke_from == InvokeFrom.DEBUGGER:
+        if invoke_from == InvokeFrom.DEBUGGER:
             workflow_triggered_from = WorkflowRunTriggeredFrom.DEBUGGING
         else:
             workflow_triggered_from = WorkflowRunTriggeredFrom.APP_RUN
@@ -196,7 +182,6 @@ class WorkflowAppGenerator(BaseAppGenerator):
             workflow_execution_repository=workflow_execution_repository,
             workflow_node_execution_repository=workflow_node_execution_repository,
             streaming=streaming,
-            root_node_id=root_node_id,
         )
 
     def _generate(
@@ -211,7 +196,6 @@ class WorkflowAppGenerator(BaseAppGenerator):
         workflow_node_execution_repository: WorkflowNodeExecutionRepository,
         streaming: bool = True,
         variable_loader: VariableLoader = DUMMY_VARIABLE_LOADER,
-        root_node_id: Optional[str] = None,
     ) -> Union[Mapping[str, Any], Generator[str | Mapping[str, Any], None, None]]:
         """
         Generate App response.
@@ -247,7 +231,6 @@ class WorkflowAppGenerator(BaseAppGenerator):
                 "queue_manager": queue_manager,
                 "context": context,
                 "variable_loader": variable_loader,
-                "root_node_id": root_node_id,
             },
         )
 
@@ -441,7 +424,6 @@ class WorkflowAppGenerator(BaseAppGenerator):
         queue_manager: AppQueueManager,
         context: contextvars.Context,
         variable_loader: VariableLoader,
-        root_node_id: Optional[str] = None,
     ) -> None:
         """
         Generate worker in a new thread.
@@ -483,7 +465,6 @@ class WorkflowAppGenerator(BaseAppGenerator):
                 variable_loader=variable_loader,
                 workflow=workflow,
                 system_user_id=system_user_id,
-                root_node_id=root_node_id,
             )
 
             try:

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

@@ -1,6 +1,6 @@
 import logging
 import time
-from typing import Optional, cast
+from typing import cast
 
 from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.apps.workflow.app_config_manager import WorkflowAppConfig
@@ -34,7 +34,6 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
         variable_loader: VariableLoader,
         workflow: Workflow,
         system_user_id: str,
-        root_node_id: Optional[str] = None,
     ):
         super().__init__(
             queue_manager=queue_manager,
@@ -44,7 +43,6 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
         self.application_generate_entity = application_generate_entity
         self._workflow = workflow
         self._sys_user_id = system_user_id
-        self._root_node_id = root_node_id
 
     def run(self):
         """
@@ -90,7 +88,6 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
                 workflow_id=self._workflow.id,
                 tenant_id=self._workflow.tenant_id,
                 user_id=self.application_generate_entity.user_id,
-                root_node_id=self._root_node_id,
             )
 
         # RUN WORKFLOW

api/core/app/apps/workflow_app_runner.py

@@ -1,6 +1,6 @@
 import time
 from collections.abc import Mapping
-from typing import Any, Optional, cast
+from typing import Any, cast
 
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
 from core.app.entities.app_invoke_entities import InvokeFrom
@@ -80,7 +80,6 @@ class WorkflowBasedAppRunner:
         workflow_id: str = "",
         tenant_id: str = "",
         user_id: str = "",
-        root_node_id: Optional[str] = None,
     ) -> Graph:
         """
         Init graph
@@ -101,8 +100,8 @@ class WorkflowBasedAppRunner:
             workflow_id=workflow_id,
             graph_config=graph_config,
             user_id=user_id,
-            user_from=UserFrom.ACCOUNT,
-            invoke_from=InvokeFrom.SERVICE_API,
+            user_from=UserFrom.ACCOUNT.value,
+            invoke_from=InvokeFrom.SERVICE_API.value,
             call_depth=0,
         )
 
@@ -114,7 +113,7 @@ class WorkflowBasedAppRunner:
         )
 
         # init graph
-        graph = Graph.init(graph_config=graph_config, node_factory=node_factory, root_node_id=root_node_id)
+        graph = Graph.init(graph_config=graph_config, node_factory=node_factory)
 
         if not graph:
             raise ValueError("graph not found in workflow")
@@ -245,8 +244,8 @@ class WorkflowBasedAppRunner:
             workflow_id=workflow.id,
             graph_config=graph_config,
             user_id="",
-            user_from=UserFrom.ACCOUNT,
-            invoke_from=InvokeFrom.SERVICE_API,
+            user_from=UserFrom.ACCOUNT.value,
+            invoke_from=InvokeFrom.SERVICE_API.value,
             call_depth=0,
         )
 

api/core/datasource/entities/api_entities.py

@@ -49,7 +49,7 @@ class DatasourceProviderApiEntity(BaseModel):
         for datasource in datasources:
             if datasource.get("parameters"):
                 for parameter in datasource.get("parameters"):
-                    if parameter.get("type") == DatasourceParameter.DatasourceParameterType.SYSTEM_FILES:
+                    if parameter.get("type") == DatasourceParameter.DatasourceParameterType.SYSTEM_FILES.value:
                         parameter["type"] = "files"
         # -------------
 

api/core/datasource/entities/datasource_entities.py

@@ -1,5 +1,5 @@
 import enum
-from enum import StrEnum
+from enum import Enum
 from typing import Any
 
 from pydantic import BaseModel, Field, ValidationInfo, field_validator
@@ -54,16 +54,16 @@ class DatasourceParameter(PluginParameter):
         removes TOOLS_SELECTOR from PluginParameterType
         """
 
-        STRING = PluginParameterType.STRING
-        NUMBER = PluginParameterType.NUMBER
-        BOOLEAN = PluginParameterType.BOOLEAN
-        SELECT = PluginParameterType.SELECT
-        SECRET_INPUT = PluginParameterType.SECRET_INPUT
-        FILE = PluginParameterType.FILE
-        FILES = PluginParameterType.FILES
+        STRING = PluginParameterType.STRING.value
+        NUMBER = PluginParameterType.NUMBER.value
+        BOOLEAN = PluginParameterType.BOOLEAN.value
+        SELECT = PluginParameterType.SELECT.value
+        SECRET_INPUT = PluginParameterType.SECRET_INPUT.value
+        FILE = PluginParameterType.FILE.value
+        FILES = PluginParameterType.FILES.value
 
         # deprecated, should not use.
-        SYSTEM_FILES = PluginParameterType.SYSTEM_FILES
+        SYSTEM_FILES = PluginParameterType.SYSTEM_FILES.value
 
         def as_normal_type(self):
             return as_normal_type(self)
@@ -218,7 +218,7 @@ class DatasourceLabel(BaseModel):
     icon: str = Field(..., description="The icon of the tool")
 
 
-class DatasourceInvokeFrom(StrEnum):
+class DatasourceInvokeFrom(Enum):
     """
     Enum class for datasource invoke
     """

api/core/entities/parameter_entities.py

@@ -14,7 +14,6 @@ class CommonParameterType(StrEnum):
     APP_SELECTOR = "app-selector"
     MODEL_SELECTOR = "model-selector"
     TOOLS_SELECTOR = "array[tools]"
-    CHECKBOX = "checkbox"
     ANY = auto()
 
     # Dynamic select parameter

api/core/entities/provider_configuration.py

@@ -207,7 +207,7 @@ class ProviderConfiguration(BaseModel):
         """
         stmt = select(Provider).where(
             Provider.tenant_id == self.tenant_id,
-            Provider.provider_type == ProviderType.CUSTOM,
+            Provider.provider_type == ProviderType.CUSTOM.value,
             Provider.provider_name.in_(self._get_provider_names()),
         )
 
@@ -458,7 +458,7 @@ class ProviderConfiguration(BaseModel):
                     provider_record = Provider(
                         tenant_id=self.tenant_id,
                         provider_name=self.provider.provider,
-                        provider_type=ProviderType.CUSTOM,
+                        provider_type=ProviderType.CUSTOM.value,
                         is_valid=True,
                         credential_id=new_record.id,
                     )
@@ -1414,7 +1414,7 @@ class ProviderConfiguration(BaseModel):
         """
         secret_input_form_variables = []
         for credential_form_schema in credential_form_schemas:
-            if credential_form_schema.type == FormType.SECRET_INPUT:
+            if credential_form_schema.type.value == FormType.SECRET_INPUT.value:
                 secret_input_form_variables.append(credential_form_schema.variable)
 
         return secret_input_form_variables

api/core/entities/provider_entities.py

@@ -1,5 +1,5 @@
 from enum import StrEnum, auto
-from typing import Optional, Union
+from typing import Union
 
 from pydantic import BaseModel, ConfigDict, Field
 
@@ -84,9 +84,9 @@ class SystemConfiguration(BaseModel):
     """
 
     enabled: bool
-    current_quota_type: Optional[ProviderQuotaType] = None
+    current_quota_type: ProviderQuotaType | None = None
     quota_configurations: list[QuotaConfiguration] = []
-    credentials: Optional[dict] = None
+    credentials: dict | None = None
 
 
 class CustomProviderConfiguration(BaseModel):
@@ -95,8 +95,8 @@ class CustomProviderConfiguration(BaseModel):
     """
 
     credentials: dict
-    current_credential_id: Optional[str] = None
-    current_credential_name: Optional[str] = None
+    current_credential_id: str | None = None
+    current_credential_name: str | None = None
     available_credentials: list[CredentialConfiguration] = []
 
 
@@ -107,9 +107,9 @@ class CustomModelConfiguration(BaseModel):
 
     model: str
     model_type: ModelType
-    credentials: dict | None
-    current_credential_id: Optional[str] = None
-    current_credential_name: Optional[str] = None
+    credentials: dict | None = None
+    current_credential_id: str | None = None
+    current_credential_name: str | None = None
     available_model_credentials: list[CredentialConfiguration] = []
     unadded_to_model_list: bool | None = False
 
@@ -131,7 +131,7 @@ class CustomConfiguration(BaseModel):
     Model class for provider custom configuration.
     """
 
-    provider: Optional[CustomProviderConfiguration] = None
+    provider: CustomProviderConfiguration | None = None
     models: list[CustomModelConfiguration] = []
     can_added_models: list[UnaddedModelConfiguration] = []
 
@@ -207,7 +207,6 @@ class ProviderConfig(BasicProviderConfig):
     required: bool = False
     default: Union[int, str, float, bool] | None = None
     options: list[Option] | None = None
-    multiple: bool | None = False
     label: I18nObject | None = None
     help: I18nObject | None = None
     url: str | None = None

api/core/extension/api_based_extension_requestor.py

@@ -1,13 +1,13 @@
 from typing import cast
 
-import httpx
+import requests
 
 from configs import dify_config
 from models.api_based_extension import APIBasedExtensionPoint
 
 
 class APIBasedExtensionRequestor:
-    timeout: httpx.Timeout = httpx.Timeout(60.0, connect=5.0)
+    timeout: tuple[int, int] = (5, 60)
     """timeout for request connect and read"""
 
     def __init__(self, api_endpoint: str, api_key: str):
@@ -27,23 +27,25 @@ class APIBasedExtensionRequestor:
         url = self.api_endpoint
 
         try:
-            mounts: dict[str, httpx.BaseTransport] | None = None
+            # proxy support for security
+            proxies = None
             if dify_config.SSRF_PROXY_HTTP_URL and dify_config.SSRF_PROXY_HTTPS_URL:
-                mounts = {
-                    "http://": httpx.HTTPTransport(proxy=dify_config.SSRF_PROXY_HTTP_URL),
-                    "https://": httpx.HTTPTransport(proxy=dify_config.SSRF_PROXY_HTTPS_URL),
+                proxies = {
+                    "http": dify_config.SSRF_PROXY_HTTP_URL,
+                    "https": dify_config.SSRF_PROXY_HTTPS_URL,
                 }
 
-            with httpx.Client(mounts=mounts, timeout=self.timeout) as client:
-                response = client.request(
-                    method="POST",
-                    url=url,
-                    json={"point": point.value, "params": params},
-                    headers=headers,
-                )
-        except httpx.TimeoutException:
+            response = requests.request(
+                method="POST",
+                url=url,
+                json={"point": point.value, "params": params},
+                headers=headers,
+                timeout=self.timeout,
+                proxies=proxies,
+            )
+        except requests.Timeout:
             raise ValueError("request timeout")
-        except httpx.RequestError:
+        except requests.ConnectionError:
             raise ValueError("request connection error")
 
         if response.status_code != 200:

api/core/helper/name_generator.py

@@ -3,7 +3,7 @@ import re
 from collections.abc import Sequence
 from typing import Any
 
-from core.plugin.entities.plugin_daemon import CredentialType
+from core.tools.entities.tool_entities import CredentialType
 
 logger = logging.getLogger(__name__)
 

api/core/helper/provider_encryption.py

@@ -1,128 +0,0 @@
-import contextlib
-from copy import deepcopy
-from typing import Any, Optional, Protocol
-
-from core.entities.provider_entities import BasicProviderConfig
-from core.helper import encrypter
-
-
-class ProviderConfigCache(Protocol):
-    """
-    Interface for provider configuration cache operations
-    """
-
-    def get(self) -> Optional[dict]:
-        """Get cached provider configuration"""
-        ...
-
-    def set(self, config: dict[str, Any]) -> None:
-        """Cache provider configuration"""
-        ...
-
-    def delete(self) -> None:
-        """Delete cached provider configuration"""
-        ...
-
-
-class ProviderConfigEncrypter:
-    tenant_id: str
-    config: list[BasicProviderConfig]
-    provider_config_cache: ProviderConfigCache
-
-    def __init__(
-        self,
-        tenant_id: str,
-        config: list[BasicProviderConfig],
-        provider_config_cache: ProviderConfigCache,
-    ):
-        self.tenant_id = tenant_id
-        self.config = config
-        self.provider_config_cache = provider_config_cache
-
-    def _deep_copy(self, data: dict[str, str]) -> dict[str, str]:
-        """
-        deep copy data
-        """
-        return deepcopy(data)
-
-    def encrypt(self, data: dict[str, str]) -> dict[str, str]:
-        """
-        encrypt tool credentials with tenant id
-
-        return a deep copy of credentials with encrypted values
-        """
-        data = self._deep_copy(data)
-
-        # get fields need to be decrypted
-        fields = dict[str, BasicProviderConfig]()
-        for credential in self.config:
-            fields[credential.name] = credential
-
-        for field_name, field in fields.items():
-            if field.type == BasicProviderConfig.Type.SECRET_INPUT:
-                if field_name in data:
-                    encrypted = encrypter.encrypt_token(self.tenant_id, data[field_name] or "")
-                    data[field_name] = encrypted
-
-        return data
-
-    def mask_credentials(self, data: dict[str, Any]) -> dict[str, Any]:
-        """
-        mask credentials
-
-        return a deep copy of credentials with masked values
-        """
-        data = self._deep_copy(data)
-
-        # get fields need to be decrypted
-        fields = dict[str, BasicProviderConfig]()
-        for credential in self.config:
-            fields[credential.name] = credential
-
-        for field_name, field in fields.items():
-            if field.type == BasicProviderConfig.Type.SECRET_INPUT:
-                if field_name in data:
-                    if len(data[field_name]) > 6:
-                        data[field_name] = (
-                            data[field_name][:2] + "*" * (len(data[field_name]) - 4) + data[field_name][-2:]
-                        )
-                    else:
-                        data[field_name] = "*" * len(data[field_name])
-
-        return data
-
-    def mask_tool_credentials(self, data: dict[str, Any]) -> dict[str, Any]:
-        return self.mask_credentials(data)
-
-    def decrypt(self, data: dict[str, str]) -> dict[str, Any]:
-        """
-        decrypt tool credentials with tenant id
-
-        return a deep copy of credentials with decrypted values
-        """
-        cached_credentials = self.provider_config_cache.get()
-        if cached_credentials:
-            return cached_credentials
-
-        data = self._deep_copy(data)
-        # get fields need to be decrypted
-        fields = dict[str, BasicProviderConfig]()
-        for credential in self.config:
-            fields[credential.name] = credential
-
-        for field_name, field in fields.items():
-            if field.type == BasicProviderConfig.Type.SECRET_INPUT:
-                if field_name in data:
-                    with contextlib.suppress(Exception):
-                        # if the value is None or empty string, skip decrypt
-                        if not data[field_name]:
-                            continue
-
-                        data[field_name] = encrypter.decrypt_token(self.tenant_id, data[field_name])
-
-        self.provider_config_cache.set(data)
-        return data
-
-
-def create_provider_encrypter(tenant_id: str, config: list[BasicProviderConfig], cache: ProviderConfigCache):
-    return ProviderConfigEncrypter(tenant_id=tenant_id, config=config, provider_config_cache=cache), cache

api/core/hosting_configuration.py

@@ -56,6 +56,9 @@ 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()
 
@@ -128,7 +131,7 @@ class HostingConfiguration:
         quotas: list[HostingQuota] = []
 
         if dify_config.HOSTED_OPENAI_TRIAL_ENABLED:
-            hosted_quota_limit = dify_config.HOSTED_OPENAI_QUOTA_LIMIT
+            hosted_quota_limit = 0
             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)
@@ -156,18 +159,49 @@ class HostingConfiguration:
             quota_unit=quota_unit,
         )
 
-    @staticmethod
-    def init_anthropic() -> HostingProvider:
-        quota_unit = QuotaUnit.TOKENS
+    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
         quotas: list[HostingQuota] = []
 
         if dify_config.HOSTED_ANTHROPIC_TRIAL_ENABLED:
-            hosted_quota_limit = dify_config.HOSTED_ANTHROPIC_QUOTA_LIMIT
-            trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit)
+            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)
             quotas.append(trial_quota)
 
         if dify_config.HOSTED_ANTHROPIC_PAID_ENABLED:
-            paid_quota = PaidHostingQuota()
+            paid_models = self.parse_restrict_models_from_env("HOSTED_ANTHROPIC_PAID_MODELS")
+            paid_quota = PaidHostingQuota(restrict_models=paid_models)
             quotas.append(paid_quota)
 
         if len(quotas) > 0:
@@ -185,6 +219,66 @@ 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

api/core/indexing_runner.py

@@ -343,7 +343,7 @@ class IndexingRunner:
 
             if file_detail:
                 extract_setting = ExtractSetting(
-                    datasource_type=DatasourceType.FILE,
+                    datasource_type=DatasourceType.FILE.value,
                     upload_file=file_detail,
                     document_model=dataset_document.doc_form,
                 )
@@ -356,7 +356,7 @@ class IndexingRunner:
             ):
                 raise ValueError("no notion import info found")
             extract_setting = ExtractSetting(
-                datasource_type=DatasourceType.NOTION,
+                datasource_type=DatasourceType.NOTION.value,
                 notion_info=NotionInfo.model_validate(
                     {
                         "credential_id": data_source_info["credential_id"],
@@ -379,7 +379,7 @@ class IndexingRunner:
             ):
                 raise ValueError("no website import info found")
             extract_setting = ExtractSetting(
-                datasource_type=DatasourceType.WEBSITE,
+                datasource_type=DatasourceType.WEBSITE.value,
                 website_info=WebsiteInfo.model_validate(
                     {
                         "provider": data_source_info["provider"],

api/core/llm_generator/llm_generator.py

@@ -14,10 +14,12 @@ 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
@@ -27,6 +29,7 @@ from core.ops.entities.trace_entity import TraceTaskName
 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.variable_pool import VariablePool
 from core.workflow.entities.workflow_node_execution import WorkflowNodeExecutionMetadataKey
 from extensions.ext_database import db
 from extensions.ext_storage import storage
@@ -560,3 +563,35 @@ 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()

api/core/llm_generator/output_parser/structured_output.py

@@ -224,8 +224,8 @@ def _handle_native_json_schema(
 
     # Set appropriate response format if required by the model
     for rule in rules:
-        if rule.name == "response_format" and ResponseFormat.JSON_SCHEMA in rule.options:
-            model_parameters["response_format"] = ResponseFormat.JSON_SCHEMA
+        if rule.name == "response_format" and ResponseFormat.JSON_SCHEMA.value in rule.options:
+            model_parameters["response_format"] = ResponseFormat.JSON_SCHEMA.value
 
     return model_parameters
 
@@ -239,10 +239,10 @@ def _set_response_format(model_parameters: dict, rules: list):
     """
     for rule in rules:
         if rule.name == "response_format":
-            if ResponseFormat.JSON in rule.options:
-                model_parameters["response_format"] = ResponseFormat.JSON
-            elif ResponseFormat.JSON_OBJECT in rule.options:
-                model_parameters["response_format"] = ResponseFormat.JSON_OBJECT
+            if ResponseFormat.JSON.value in rule.options:
+                model_parameters["response_format"] = ResponseFormat.JSON.value
+            elif ResponseFormat.JSON_OBJECT.value in rule.options:
+                model_parameters["response_format"] = ResponseFormat.JSON_OBJECT.value
 
 
 def _handle_prompt_based_schema(

api/core/llm_generator/prompts.py

@@ -422,3 +422,18 @@ 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:
+"""

api/core/memory/entities.py

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

api/core/memory/errors.py

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

api/core/model_runtime/entities/provider_entities.py

@@ -1,5 +1,5 @@
 from collections.abc import Sequence
-from enum import StrEnum, auto
+from enum import Enum, StrEnum, auto
 
 from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
@@ -7,7 +7,7 @@ from core.model_runtime.entities.common_entities import I18nObject
 from core.model_runtime.entities.model_entities import AIModelEntity, ModelType
 
 
-class ConfigurateMethod(StrEnum):
+class ConfigurateMethod(Enum):
     """
     Enum class for configurate method of provider model.
     """

api/core/ops/arize_phoenix_trace/arize_phoenix_trace.py

@@ -213,9 +213,9 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
                     node_metadata.update(json.loads(node_execution.execution_metadata))
 
                 # Determine the correct span kind based on node type
-                span_kind = OpenInferenceSpanKindValues.CHAIN
+                span_kind = OpenInferenceSpanKindValues.CHAIN.value
                 if node_execution.node_type == "llm":
-                    span_kind = OpenInferenceSpanKindValues.LLM
+                    span_kind = OpenInferenceSpanKindValues.LLM.value
                     provider = process_data.get("model_provider")
                     model = process_data.get("model_name")
                     if provider:
@@ -230,18 +230,18 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
                         node_metadata["prompt_tokens"] = usage_data.get("prompt_tokens", 0)
                         node_metadata["completion_tokens"] = usage_data.get("completion_tokens", 0)
                 elif node_execution.node_type == "dataset_retrieval":
-                    span_kind = OpenInferenceSpanKindValues.RETRIEVER
+                    span_kind = OpenInferenceSpanKindValues.RETRIEVER.value
                 elif node_execution.node_type == "tool":
-                    span_kind = OpenInferenceSpanKindValues.TOOL
+                    span_kind = OpenInferenceSpanKindValues.TOOL.value
                 else:
-                    span_kind = OpenInferenceSpanKindValues.CHAIN
+                    span_kind = OpenInferenceSpanKindValues.CHAIN.value
 
                 node_span = self.tracer.start_span(
                     name=node_execution.node_type,
                     attributes={
                         SpanAttributes.INPUT_VALUE: node_execution.inputs or "{}",
                         SpanAttributes.OUTPUT_VALUE: node_execution.outputs or "{}",
-                        SpanAttributes.OPENINFERENCE_SPAN_KIND: span_kind.value,
+                        SpanAttributes.OPENINFERENCE_SPAN_KIND: span_kind,
                         SpanAttributes.METADATA: json.dumps(node_metadata, ensure_ascii=False),
                         SpanAttributes.SESSION_ID: trace_info.conversation_id or "",
                     },

api/core/ops/langfuse_trace/langfuse_trace.py

@@ -73,7 +73,7 @@ class LangFuseDataTrace(BaseTraceInstance):
 
         if trace_info.message_id:
             trace_id = trace_info.trace_id or trace_info.message_id
-            name = TraceTaskName.MESSAGE_TRACE
+            name = TraceTaskName.MESSAGE_TRACE.value
             trace_data = LangfuseTrace(
                 id=trace_id,
                 user_id=user_id,
@@ -88,7 +88,7 @@ class LangFuseDataTrace(BaseTraceInstance):
             self.add_trace(langfuse_trace_data=trace_data)
             workflow_span_data = LangfuseSpan(
                 id=trace_info.workflow_run_id,
-                name=TraceTaskName.WORKFLOW_TRACE,
+                name=TraceTaskName.WORKFLOW_TRACE.value,
                 input=dict(trace_info.workflow_run_inputs),
                 output=dict(trace_info.workflow_run_outputs),
                 trace_id=trace_id,
@@ -103,7 +103,7 @@ class LangFuseDataTrace(BaseTraceInstance):
             trace_data = LangfuseTrace(
                 id=trace_id,
                 user_id=user_id,
-                name=TraceTaskName.WORKFLOW_TRACE,
+                name=TraceTaskName.WORKFLOW_TRACE.value,
                 input=dict(trace_info.workflow_run_inputs),
                 output=dict(trace_info.workflow_run_outputs),
                 metadata=metadata,
@@ -253,7 +253,7 @@ class LangFuseDataTrace(BaseTraceInstance):
         trace_data = LangfuseTrace(
             id=trace_id,
             user_id=user_id,
-            name=TraceTaskName.MESSAGE_TRACE,
+            name=TraceTaskName.MESSAGE_TRACE.value,
             input={
                 "message": trace_info.inputs,
                 "files": file_list,
@@ -303,7 +303,7 @@ class LangFuseDataTrace(BaseTraceInstance):
         if trace_info.message_data is None:
             return
         span_data = LangfuseSpan(
-            name=TraceTaskName.MODERATION_TRACE,
+            name=TraceTaskName.MODERATION_TRACE.value,
             input=trace_info.inputs,
             output={
                 "action": trace_info.action,
@@ -331,7 +331,7 @@ class LangFuseDataTrace(BaseTraceInstance):
         )
 
         generation_data = LangfuseGeneration(
-            name=TraceTaskName.SUGGESTED_QUESTION_TRACE,
+            name=TraceTaskName.SUGGESTED_QUESTION_TRACE.value,
             input=trace_info.inputs,
             output=str(trace_info.suggested_question),
             trace_id=trace_info.trace_id or trace_info.message_id,
@@ -349,7 +349,7 @@ class LangFuseDataTrace(BaseTraceInstance):
         if trace_info.message_data is None:
             return
         dataset_retrieval_span_data = LangfuseSpan(
-            name=TraceTaskName.DATASET_RETRIEVAL_TRACE,
+            name=TraceTaskName.DATASET_RETRIEVAL_TRACE.value,
             input=trace_info.inputs,
             output={"documents": trace_info.documents},
             trace_id=trace_info.trace_id or trace_info.message_id,
@@ -377,7 +377,7 @@ class LangFuseDataTrace(BaseTraceInstance):
 
     def generate_name_trace(self, trace_info: GenerateNameTraceInfo):
         name_generation_trace_data = LangfuseTrace(
-            name=TraceTaskName.GENERATE_NAME_TRACE,
+            name=TraceTaskName.GENERATE_NAME_TRACE.value,
             input=trace_info.inputs,
             output=trace_info.outputs,
             user_id=trace_info.tenant_id,
@@ -388,7 +388,7 @@ class LangFuseDataTrace(BaseTraceInstance):
         self.add_trace(langfuse_trace_data=name_generation_trace_data)
 
         name_generation_span_data = LangfuseSpan(
-            name=TraceTaskName.GENERATE_NAME_TRACE,
+            name=TraceTaskName.GENERATE_NAME_TRACE.value,
             input=trace_info.inputs,
             output=trace_info.outputs,
             trace_id=trace_info.conversation_id,

api/core/ops/langsmith_trace/langsmith_trace.py

@@ -81,7 +81,7 @@ class LangSmithDataTrace(BaseTraceInstance):
         if trace_info.message_id:
             message_run = LangSmithRunModel(
                 id=trace_info.message_id,
-                name=TraceTaskName.MESSAGE_TRACE,
+                name=TraceTaskName.MESSAGE_TRACE.value,
                 inputs=dict(trace_info.workflow_run_inputs),
                 outputs=dict(trace_info.workflow_run_outputs),
                 run_type=LangSmithRunType.chain,
@@ -110,7 +110,7 @@ class LangSmithDataTrace(BaseTraceInstance):
             file_list=trace_info.file_list,
             total_tokens=trace_info.total_tokens,
             id=trace_info.workflow_run_id,
-            name=TraceTaskName.WORKFLOW_TRACE,
+            name=TraceTaskName.WORKFLOW_TRACE.value,
             inputs=dict(trace_info.workflow_run_inputs),
             run_type=LangSmithRunType.tool,
             start_time=trace_info.workflow_data.created_at,
@@ -271,7 +271,7 @@ class LangSmithDataTrace(BaseTraceInstance):
             output_tokens=trace_info.answer_tokens,
             total_tokens=trace_info.total_tokens,
             id=message_id,
-            name=TraceTaskName.MESSAGE_TRACE,
+            name=TraceTaskName.MESSAGE_TRACE.value,
             inputs=trace_info.inputs,
             run_type=LangSmithRunType.chain,
             start_time=trace_info.start_time,
@@ -327,7 +327,7 @@ class LangSmithDataTrace(BaseTraceInstance):
         if trace_info.message_data is None:
             return
         langsmith_run = LangSmithRunModel(
-            name=TraceTaskName.MODERATION_TRACE,
+            name=TraceTaskName.MODERATION_TRACE.value,
             inputs=trace_info.inputs,
             outputs={
                 "action": trace_info.action,
@@ -362,7 +362,7 @@ class LangSmithDataTrace(BaseTraceInstance):
         if message_data is None:
             return
         suggested_question_run = LangSmithRunModel(
-            name=TraceTaskName.SUGGESTED_QUESTION_TRACE,
+            name=TraceTaskName.SUGGESTED_QUESTION_TRACE.value,
             inputs=trace_info.inputs,
             outputs=trace_info.suggested_question,
             run_type=LangSmithRunType.tool,
@@ -391,7 +391,7 @@ class LangSmithDataTrace(BaseTraceInstance):
         if trace_info.message_data is None:
             return
         dataset_retrieval_run = LangSmithRunModel(
-            name=TraceTaskName.DATASET_RETRIEVAL_TRACE,
+            name=TraceTaskName.DATASET_RETRIEVAL_TRACE.value,
             inputs=trace_info.inputs,
             outputs={"documents": trace_info.documents},
             run_type=LangSmithRunType.retriever,
@@ -447,7 +447,7 @@ class LangSmithDataTrace(BaseTraceInstance):
 
     def generate_name_trace(self, trace_info: GenerateNameTraceInfo):
         name_run = LangSmithRunModel(
-            name=TraceTaskName.GENERATE_NAME_TRACE,
+            name=TraceTaskName.GENERATE_NAME_TRACE.value,
             inputs=trace_info.inputs,
             outputs=trace_info.outputs,
             run_type=LangSmithRunType.tool,

api/core/ops/opik_trace/opik_trace.py

@@ -108,7 +108,7 @@ class OpikDataTrace(BaseTraceInstance):
 
             trace_data = {
                 "id": opik_trace_id,
-                "name": TraceTaskName.MESSAGE_TRACE,
+                "name": TraceTaskName.MESSAGE_TRACE.value,
                 "start_time": trace_info.start_time,
                 "end_time": trace_info.end_time,
                 "metadata": workflow_metadata,
@@ -125,7 +125,7 @@ class OpikDataTrace(BaseTraceInstance):
                 "id": root_span_id,
                 "parent_span_id": None,
                 "trace_id": opik_trace_id,
-                "name": TraceTaskName.WORKFLOW_TRACE,
+                "name": TraceTaskName.WORKFLOW_TRACE.value,
                 "input": wrap_dict("input", trace_info.workflow_run_inputs),
                 "output": wrap_dict("output", trace_info.workflow_run_outputs),
                 "start_time": trace_info.start_time,
@@ -138,7 +138,7 @@ class OpikDataTrace(BaseTraceInstance):
         else:
             trace_data = {
                 "id": opik_trace_id,
-                "name": TraceTaskName.MESSAGE_TRACE,
+                "name": TraceTaskName.MESSAGE_TRACE.value,
                 "start_time": trace_info.start_time,
                 "end_time": trace_info.end_time,
                 "metadata": workflow_metadata,
@@ -290,7 +290,7 @@ class OpikDataTrace(BaseTraceInstance):
 
         trace_data = {
             "id": prepare_opik_uuid(trace_info.start_time, dify_trace_id),
-            "name": TraceTaskName.MESSAGE_TRACE,
+            "name": TraceTaskName.MESSAGE_TRACE.value,
             "start_time": trace_info.start_time,
             "end_time": trace_info.end_time,
             "metadata": wrap_metadata(metadata),
@@ -329,7 +329,7 @@ class OpikDataTrace(BaseTraceInstance):
 
         span_data = {
             "trace_id": prepare_opik_uuid(start_time, trace_info.trace_id or trace_info.message_id),
-            "name": TraceTaskName.MODERATION_TRACE,
+            "name": TraceTaskName.MODERATION_TRACE.value,
             "type": "tool",
             "start_time": start_time,
             "end_time": trace_info.end_time or trace_info.message_data.updated_at,
@@ -355,7 +355,7 @@ class OpikDataTrace(BaseTraceInstance):
 
         span_data = {
             "trace_id": prepare_opik_uuid(start_time, trace_info.trace_id or trace_info.message_id),
-            "name": TraceTaskName.SUGGESTED_QUESTION_TRACE,
+            "name": TraceTaskName.SUGGESTED_QUESTION_TRACE.value,
             "type": "tool",
             "start_time": start_time,
             "end_time": trace_info.end_time or message_data.updated_at,
@@ -375,7 +375,7 @@ class OpikDataTrace(BaseTraceInstance):
 
         span_data = {
             "trace_id": prepare_opik_uuid(start_time, trace_info.trace_id or trace_info.message_id),
-            "name": TraceTaskName.DATASET_RETRIEVAL_TRACE,
+            "name": TraceTaskName.DATASET_RETRIEVAL_TRACE.value,
             "type": "tool",
             "start_time": start_time,
             "end_time": trace_info.end_time or trace_info.message_data.updated_at,
@@ -405,7 +405,7 @@ class OpikDataTrace(BaseTraceInstance):
     def generate_name_trace(self, trace_info: GenerateNameTraceInfo):
         trace_data = {
             "id": prepare_opik_uuid(trace_info.start_time, trace_info.trace_id or trace_info.message_id),
-            "name": TraceTaskName.GENERATE_NAME_TRACE,
+            "name": TraceTaskName.GENERATE_NAME_TRACE.value,
             "start_time": trace_info.start_time,
             "end_time": trace_info.end_time,
             "metadata": wrap_metadata(trace_info.metadata),
@@ -420,7 +420,7 @@ class OpikDataTrace(BaseTraceInstance):
 
         span_data = {
             "trace_id": trace.id,
-            "name": TraceTaskName.GENERATE_NAME_TRACE,
+            "name": TraceTaskName.GENERATE_NAME_TRACE.value,
             "start_time": trace_info.start_time,
             "end_time": trace_info.end_time,
             "metadata": wrap_metadata(trace_info.metadata),

api/core/ops/weave_trace/weave_trace.py

@@ -104,7 +104,7 @@ class WeaveDataTrace(BaseTraceInstance):
 
             message_run = WeaveTraceModel(
                 id=trace_info.message_id,
-                op=str(TraceTaskName.MESSAGE_TRACE),
+                op=str(TraceTaskName.MESSAGE_TRACE.value),
                 inputs=dict(trace_info.workflow_run_inputs),
                 outputs=dict(trace_info.workflow_run_outputs),
                 total_tokens=trace_info.total_tokens,
@@ -126,7 +126,7 @@ class WeaveDataTrace(BaseTraceInstance):
             file_list=trace_info.file_list,
             total_tokens=trace_info.total_tokens,
             id=trace_info.workflow_run_id,
-            op=str(TraceTaskName.WORKFLOW_TRACE),
+            op=str(TraceTaskName.WORKFLOW_TRACE.value),
             inputs=dict(trace_info.workflow_run_inputs),
             outputs=dict(trace_info.workflow_run_outputs),
             attributes=workflow_attributes,
@@ -253,7 +253,7 @@ class WeaveDataTrace(BaseTraceInstance):
 
         message_run = WeaveTraceModel(
             id=trace_id,
-            op=str(TraceTaskName.MESSAGE_TRACE),
+            op=str(TraceTaskName.MESSAGE_TRACE.value),
             input_tokens=trace_info.message_tokens,
             output_tokens=trace_info.answer_tokens,
             total_tokens=trace_info.total_tokens,
@@ -300,7 +300,7 @@ class WeaveDataTrace(BaseTraceInstance):
 
         moderation_run = WeaveTraceModel(
             id=str(uuid.uuid4()),
-            op=str(TraceTaskName.MODERATION_TRACE),
+            op=str(TraceTaskName.MODERATION_TRACE.value),
             inputs=trace_info.inputs,
             outputs={
                 "action": trace_info.action,
@@ -330,7 +330,7 @@ class WeaveDataTrace(BaseTraceInstance):
 
         suggested_question_run = WeaveTraceModel(
             id=str(uuid.uuid4()),
-            op=str(TraceTaskName.SUGGESTED_QUESTION_TRACE),
+            op=str(TraceTaskName.SUGGESTED_QUESTION_TRACE.value),
             inputs=trace_info.inputs,
             outputs=trace_info.suggested_question,
             attributes=attributes,
@@ -355,7 +355,7 @@ class WeaveDataTrace(BaseTraceInstance):
 
         dataset_retrieval_run = WeaveTraceModel(
             id=str(uuid.uuid4()),
-            op=str(TraceTaskName.DATASET_RETRIEVAL_TRACE),
+            op=str(TraceTaskName.DATASET_RETRIEVAL_TRACE.value),
             inputs=trace_info.inputs,
             outputs={"documents": trace_info.documents},
             attributes=attributes,
@@ -397,7 +397,7 @@ class WeaveDataTrace(BaseTraceInstance):
 
         name_run = WeaveTraceModel(
             id=str(uuid.uuid4()),
-            op=str(TraceTaskName.GENERATE_NAME_TRACE),
+            op=str(TraceTaskName.GENERATE_NAME_TRACE.value),
             inputs=trace_info.inputs,
             outputs=trace_info.outputs,
             attributes=attributes,

api/core/plugin/backwards_invocation/node.py

@@ -52,7 +52,7 @@ class PluginNodeBackwardsInvocation(BaseBackwardsInvocation):
             instruction=instruction,  # instruct with variables are not supported
         )
         node_data_dict = node_data.model_dump()
-        node_data_dict["type"] = NodeType.PARAMETER_EXTRACTOR
+        node_data_dict["type"] = NodeType.PARAMETER_EXTRACTOR.value
         execution = workflow_service.run_free_workflow_node(
             node_data_dict,
             tenant_id=tenant_id,