if session unauthorized, rejoin

- Introduced AmplitudeProvider component for initializing Amplitude analytics.
- Integrated user tracking in AppContextProvider to report user and workspace information.
- Added event tracking for workflow log filter selections.
- Updated package.json and pnpm-lock.yaml to include @amplitude/unified dependency.

api/.env.example

@@ -30,6 +30,9 @@ INTERNAL_FILES_URL=http://127.0.0.1:5001
 # The time in seconds after the signature is rejected
 FILES_ACCESS_TIMEOUT=300
 
+# Collaboration mode toggle
+ENABLE_COLLABORATION_MODE=false
+
 # Access token expiration time in minutes
 ACCESS_TOKEN_EXPIRE_MINUTES=60
 
@@ -371,6 +374,12 @@ UPLOAD_IMAGE_FILE_SIZE_LIMIT=10
 UPLOAD_VIDEO_FILE_SIZE_LIMIT=100
 UPLOAD_AUDIO_FILE_SIZE_LIMIT=50
 
+# Comma-separated list of file extensions blocked from upload for security reasons.
+# Extensions should be lowercase without dots (e.g., exe,bat,sh,dll).
+# Empty by default to allow all file types.
+# Recommended: exe,bat,cmd,com,scr,vbs,ps1,msi,dll
+UPLOAD_FILE_EXTENSION_BLACKLIST=
+
 # Model configuration
 MULTIMODAL_SEND_FORMAT=base64
 PROMPT_GENERATION_MAX_TOKENS=512
@@ -608,3 +617,6 @@ SWAGGER_UI_PATH=/swagger-ui.html
 # Whether to encrypt dataset IDs when exporting DSL files (default: true)
 # Set to false to export dataset IDs as plain text for easier cross-environment import
 DSL_EXPORT_ENCRYPT_DATASET_ID=true
+
+# Maximum number of segments for dataset segments API (0 for unlimited)
+DATASET_MAX_SEGMENTS_PER_REQUEST=0

api/Dockerfile

@@ -15,7 +15,11 @@ FROM base AS packages
 # RUN sed -i 's@deb.debian.org@mirrors.aliyun.com@g' /etc/apt/sources.list.d/debian.sources
 
 RUN apt-get update \
-    && apt-get install -y --no-install-recommends gcc g++ libc-dev libffi-dev libgmp-dev libmpfr-dev libmpc-dev
+    && apt-get install -y --no-install-recommends \
+          # basic environment
+          g++ \
+          # for building gmpy2
+          libmpfr-dev libmpc-dev
 
 # Install Python dependencies
 COPY pyproject.toml uv.lock ./
@@ -49,7 +53,9 @@ RUN \
     # Install dependencies
     && apt-get install -y --no-install-recommends \
         # basic environment
-        curl nodejs libgmp-dev libmpfr-dev libmpc-dev \
+        curl nodejs \
+        # for gmpy2 \
+        libgmp-dev libmpfr-dev libmpc-dev \
         # For Security
         expat libldap-2.5-0 perl libsqlite3-0 zlib1g \
         # install fonts to support the use of tools like pypdfium2

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:
     # Gunicorn and Celery handle monkey patching automatically in production by
     # specifying the `gevent` worker class. Manual monkey patching is not required here.
@@ -22,8 +29,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

@@ -1601,7 +1601,7 @@ def transform_datasource_credentials():
                             "integration_secret": api_key,
                         }
                         datasource_provider = DatasourceProvider(
-                            provider="jina",
+                            provider="jinareader",
                             tenant_id=tenant_id,
                             plugin_id=jina_plugin_id,
                             auth_type=api_key_credential_type.value,

api/configs/feature/__init__.py

@@ -331,6 +331,31 @@ class FileUploadConfig(BaseSettings):
         default=10,
     )
 
+    inner_UPLOAD_FILE_EXTENSION_BLACKLIST: str = Field(
+        description=(
+            "Comma-separated list of file extensions that are blocked from upload. "
+            "Extensions should be lowercase without dots (e.g., 'exe,bat,sh,dll'). "
+            "Empty by default to allow all file types."
+        ),
+        validation_alias=AliasChoices("UPLOAD_FILE_EXTENSION_BLACKLIST"),
+        default="",
+    )
+
+    @computed_field  # type: ignore[misc]
+    @property
+    def UPLOAD_FILE_EXTENSION_BLACKLIST(self) -> set[str]:
+        """
+        Parse and return the blacklist as a set of lowercase extensions.
+        Returns an empty set if no blacklist is configured.
+        """
+        if not self.inner_UPLOAD_FILE_EXTENSION_BLACKLIST:
+            return set()
+        return {
+            ext.strip().lower().strip(".")
+            for ext in self.inner_UPLOAD_FILE_EXTENSION_BLACKLIST.split(",")
+            if ext.strip()
+        }
+
 
 class HttpConfig(BaseSettings):
     """
@@ -846,6 +871,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):
     """
@@ -920,6 +955,11 @@ class DataSetConfig(BaseSettings):
         default=True,
     )
 
+    DATASET_MAX_SEGMENTS_PER_REQUEST: NonNegativeInt = Field(
+        description="Maximum number of segments for dataset segments API (0 for unlimited)",
+        default=0,
+    )
+
 
 class WorkspaceConfig(BaseSettings):
     """
@@ -1053,6 +1093,13 @@ class PositionConfig(BaseSettings):
         return {item.strip() for item in self.POSITION_TOOL_EXCLUDES.split(",") if item.strip() != ""}
 
 
+class CollaborationConfig(BaseSettings):
+    ENABLE_COLLABORATION_MODE: bool = Field(
+        description="Whether to enable collaboration mode features across the workspace",
+        default=False,
+    )
+
+
 class LoginConfig(BaseSettings):
     ENABLE_EMAIL_CODE_LOGIN: bool = Field(
         description="whether to enable email code login",
@@ -1112,6 +1159,13 @@ class SwaggerUIConfig(BaseSettings):
     )
 
 
+class TenantSelfTaskQueueConfig(BaseSettings):
+    TENANT_SELF_TASK_QUEUE_PULL_SIZE: int = Field(
+        description="Default batch size for tenant self task queue pull operations",
+        default=1,
+    )
+
+
 class FeatureConfig(
     # place the configs in alphabet order
     AppExecutionConfig,
@@ -1136,11 +1190,13 @@ class FeatureConfig(
     RagEtlConfig,
     RepositoryConfig,
     SecurityConfig,
+    TenantSelfTaskQueueConfig,
     ToolConfig,
     UpdateConfig,
     WorkflowConfig,
     WorkflowNodeExecutionConfig,
     WorkspaceConfig,
+    CollaborationConfig,
     LoginConfig,
     AccountConfig,
     SwaggerUIConfig,

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.
@@ -60,19 +65,46 @@ class HostedOpenAiConfig(BaseSettings):
 
     HOSTED_OPENAI_TRIAL_MODELS: str = Field(
         description="Comma-separated list of available models for trial access",
-        default="gpt-3.5-turbo,"
-        "gpt-3.5-turbo-1106,"
-        "gpt-3.5-turbo-instruct,"
+        default="gpt-4,"
+        "gpt-4-turbo-preview,"
+        "gpt-4-turbo-2024-04-09,"
+        "gpt-4-1106-preview,"
+        "gpt-4-0125-preview,"
+        "gpt-4-turbo,"
+        "gpt-4.1,"
+        "gpt-4.1-2025-04-14,"
+        "gpt-4.1-mini,"
+        "gpt-4.1-mini-2025-04-14,"
+        "gpt-4.1-nano,"
+        "gpt-4.1-nano-2025-04-14,"
+        "gpt-3.5-turbo,"
         "gpt-3.5-turbo-16k,"
         "gpt-3.5-turbo-16k-0613,"
+        "gpt-3.5-turbo-1106,"
         "gpt-3.5-turbo-0613,"
         "gpt-3.5-turbo-0125,"
-        "text-davinci-003",
-    )
-
-    HOSTED_OPENAI_QUOTA_LIMIT: NonNegativeInt = Field(
-        description="Quota limit for hosted OpenAI service usage",
-        default=200,
+        "gpt-3.5-turbo-instruct,"
+        "text-davinci-003,"
+        "chatgpt-4o-latest,"
+        "gpt-4o,"
+        "gpt-4o-2024-05-13,"
+        "gpt-4o-2024-08-06,"
+        "gpt-4o-2024-11-20,"
+        "gpt-4o-audio-preview,"
+        "gpt-4o-audio-preview-2025-06-03,"
+        "gpt-4o-mini,"
+        "gpt-4o-mini-2024-07-18,"
+        "o3-mini,"
+        "o3-mini-2025-01-31,"
+        "gpt-5-mini-2025-08-07,"
+        "gpt-5-mini,"
+        "o4-mini,"
+        "o4-mini-2025-04-16,"
+        "gpt-5-chat-latest,"
+        "gpt-5,"
+        "gpt-5-2025-08-07,"
+        "gpt-5-nano,"
+        "gpt-5-nano-2025-08-07",
     )
 
     HOSTED_OPENAI_PAID_ENABLED: bool = Field(
@@ -87,6 +119,13 @@ class HostedOpenAiConfig(BaseSettings):
         "gpt-4-turbo-2024-04-09,"
         "gpt-4-1106-preview,"
         "gpt-4-0125-preview,"
+        "gpt-4-turbo,"
+        "gpt-4.1,"
+        "gpt-4.1-2025-04-14,"
+        "gpt-4.1-mini,"
+        "gpt-4.1-mini-2025-04-14,"
+        "gpt-4.1-nano,"
+        "gpt-4.1-nano-2025-04-14,"
         "gpt-3.5-turbo,"
         "gpt-3.5-turbo-16k,"
         "gpt-3.5-turbo-16k-0613,"
@@ -94,7 +133,150 @@ class HostedOpenAiConfig(BaseSettings):
         "gpt-3.5-turbo-0613,"
         "gpt-3.5-turbo-0125,"
         "gpt-3.5-turbo-instruct,"
-        "text-davinci-003",
+        "text-davinci-003,"
+        "chatgpt-4o-latest,"
+        "gpt-4o,"
+        "gpt-4o-2024-05-13,"
+        "gpt-4o-2024-08-06,"
+        "gpt-4o-2024-11-20,"
+        "gpt-4o-audio-preview,"
+        "gpt-4o-audio-preview-2025-06-03,"
+        "gpt-4o-mini,"
+        "gpt-4o-mini-2024-07-18,"
+        "o3-mini,"
+        "o3-mini-2025-01-31,"
+        "gpt-5-mini-2025-08-07,"
+        "gpt-5-mini,"
+        "o4-mini,"
+        "o4-mini-2025-04-16,"
+        "gpt-5-chat-latest,"
+        "gpt-5,"
+        "gpt-5-2025-08-07,"
+        "gpt-5-nano,"
+        "gpt-5-nano-2025-08-07",
+    )
+
+
+class HostedGeminiConfig(BaseSettings):
+    """
+    Configuration for fetching Gemini service
+    """
+
+    HOSTED_GEMINI_API_KEY: str | None = Field(
+        description="API key for hosted Gemini service",
+        default=None,
+    )
+
+    HOSTED_GEMINI_API_BASE: str | None = Field(
+        description="Base URL for hosted Gemini API",
+        default=None,
+    )
+
+    HOSTED_GEMINI_API_ORGANIZATION: str | None = Field(
+        description="Organization ID for hosted Gemini service",
+        default=None,
+    )
+
+    HOSTED_GEMINI_TRIAL_ENABLED: bool = Field(
+        description="Enable trial access to hosted Gemini service",
+        default=False,
+    )
+
+    HOSTED_GEMINI_TRIAL_MODELS: str = Field(
+        description="Comma-separated list of available models for trial access",
+        default="gemini-2.5-flash,gemini-2.0-flash,gemini-2.0-flash-lite,",
+    )
+
+    HOSTED_GEMINI_PAID_ENABLED: bool = Field(
+        description="Enable paid access to hosted gemini service",
+        default=False,
+    )
+
+    HOSTED_GEMINI_PAID_MODELS: str = Field(
+        description="Comma-separated list of available models for paid access",
+        default="gemini-2.5-flash,gemini-2.0-flash,gemini-2.0-flash-lite,",
+    )
+
+
+class HostedXAIConfig(BaseSettings):
+    """
+    Configuration for fetching XAI service
+    """
+
+    HOSTED_XAI_API_KEY: str | None = Field(
+        description="API key for hosted XAI service",
+        default=None,
+    )
+
+    HOSTED_XAI_API_BASE: str | None = Field(
+        description="Base URL for hosted XAI API",
+        default=None,
+    )
+
+    HOSTED_XAI_API_ORGANIZATION: str | None = Field(
+        description="Organization ID for hosted XAI service",
+        default=None,
+    )
+
+    HOSTED_XAI_TRIAL_ENABLED: bool = Field(
+        description="Enable trial access to hosted XAI service",
+        default=False,
+    )
+
+    HOSTED_XAI_TRIAL_MODELS: str = Field(
+        description="Comma-separated list of available models for trial access",
+        default="grok-3,grok-3-mini,grok-3-mini-fast",
+    )
+
+    HOSTED_XAI_PAID_ENABLED: bool = Field(
+        description="Enable paid access to hosted XAI service",
+        default=False,
+    )
+
+    HOSTED_XAI_PAID_MODELS: str = Field(
+        description="Comma-separated list of available models for paid access",
+        default="grok-3,grok-3-mini,grok-3-mini-fast",
+    )
+
+
+class HostedDeepseekConfig(BaseSettings):
+    """
+    Configuration for fetching Deepseek service
+    """
+
+    HOSTED_DEEPSEEK_API_KEY: str | None = Field(
+        description="API key for hosted Deepseek service",
+        default=None,
+    )
+
+    HOSTED_DEEPSEEK_API_BASE: str | None = Field(
+        description="Base URL for hosted Deepseek API",
+        default=None,
+    )
+
+    HOSTED_DEEPSEEK_API_ORGANIZATION: str | None = Field(
+        description="Organization ID for hosted Deepseek service",
+        default=None,
+    )
+
+    HOSTED_DEEPSEEK_TRIAL_ENABLED: bool = Field(
+        description="Enable trial access to hosted Deepseek service",
+        default=False,
+    )
+
+    HOSTED_DEEPSEEK_TRIAL_MODELS: str = Field(
+        description="Comma-separated list of available models for trial access",
+        default="deepseek-chat,deepseek-reasoner",
+    )
+
+    HOSTED_DEEPSEEK_PAID_ENABLED: bool = Field(
+        description="Enable paid access to hosted XAI service",
+        default=False,
+    )
+
+    HOSTED_DEEPSEEK_PAID_MODELS: str = Field(
+        description="Comma-separated list of available models for paid access",
+        default="deepseek-chat,deepseek-reasoner",
     )
 
 
@@ -144,16 +326,30 @@ class HostedAnthropicConfig(BaseSettings):
         default=False,
     )
 
-    HOSTED_ANTHROPIC_QUOTA_LIMIT: NonNegativeInt = Field(
-        description="Quota limit for hosted Anthropic service usage",
-        default=600000,
-    )
-
     HOSTED_ANTHROPIC_PAID_ENABLED: bool = Field(
         description="Enable paid access to hosted Anthropic service",
         default=False,
     )
 
+    HOSTED_ANTHROPIC_TRIAL_MODELS: str = Field(
+        description="Comma-separated list of available models for paid access",
+        default="claude-opus-4-20250514,"
+        "claude-sonnet-4-20250514,"
+        "claude-3-5-haiku-20241022,"
+        "claude-3-opus-20240229,"
+        "claude-3-7-sonnet-20250219,"
+        "claude-3-haiku-20240307",
+    )
+    HOSTED_ANTHROPIC_PAID_MODELS: str = Field(
+        description="Comma-separated list of available models for paid access",
+        default="claude-opus-4-20250514,"
+        "claude-sonnet-4-20250514,"
+        "claude-3-5-haiku-20241022,"
+        "claude-3-opus-20240229,"
+        "claude-3-7-sonnet-20250219,"
+        "claude-3-haiku-20240307",
+    )
+
 
 class HostedMinmaxConfig(BaseSettings):
     """
@@ -250,5 +446,8 @@ class HostedServiceConfig(
     HostedModerationConfig,
     # credit config
     HostedCreditConfig,
+    HostedGeminiConfig,
+    HostedXAIConfig,
+    HostedDeepseekConfig,
 ):
     pass

api/configs/middleware/vdb/weaviate_config.py

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

api/controllers/common/errors.py

@@ -25,6 +25,12 @@ class UnsupportedFileTypeError(BaseHTTPException):
     code = 415
 
 
+class BlockedFileExtensionError(BaseHTTPException):
+    error_code = "file_extension_blocked"
+    description = "The file extension is blocked for security reasons."
+    code = 400
+
+
 class TooManyFilesError(BaseHTTPException):
     error_code = "too_many_files"
     description = "Only one file is allowed."

api/controllers/console/__init__.py

@@ -58,11 +58,13 @@ 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,
@@ -106,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
@@ -143,6 +147,7 @@ __all__ = [
     "apikey",
     "app",
     "audio",
+    "banner",
     "billing",
     "bp",
     "completion",
@@ -196,6 +201,7 @@ __all__ = [
     "statistic",
     "tags",
     "tool_providers",
+    "trial",
     "version",
     "website",
     "workflow",

api/controllers/console/admin.py

@@ -16,7 +16,7 @@ from controllers.console import api, console_ns
 from controllers.console.wraps import only_edition_cloud
 from extensions.ext_database import db
 from libs.token import extract_access_token
-from models.model import App, InstalledApp, RecommendedApp
+from models.model import App, ExporleBanner, InstalledApp, RecommendedApp, TrialApp
 
 
 def admin_required(view: Callable[P, R]):
@@ -52,6 +52,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"),
             },
         )
     )
@@ -71,6 +73,8 @@ class InsertExploreAppListApi(Resource):
             .add_argument("language", type=supported_language, required=True, nullable=False, location="json")
             .add_argument("category", type=str, required=True, nullable=False, location="json")
             .add_argument("position", type=int, required=True, nullable=False, location="json")
+            .add_argument("can_trial", type=bool, required=True, nullable=False, location="json")
+            .add_argument("trial_limit", type=int, required=True, nullable=False, location="json")
         )
         args = parser.parse_args()
 
@@ -108,6 +112,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()
@@ -122,6 +140,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()
@@ -167,7 +199,83 @@ class InsertExploreAppApi(Resource):
             for installed_app in installed_apps:
                 session.delete(installed_app)
 
+            trial_app = session.execute(
+                select(TrialApp).where(TrialApp.app_id == recommended_app.app_id)
+            ).scalar_one_or_none()
+            if trial_app:
+                session.delete(trial_app)
+
         db.session.delete(recommended_app)
         db.session.commit()
 
         return {"result": "success"}, 204
+
+
+@console_ns.route("/admin/insert-explore-banner")
+class InsertExploreBanner(Resource):
+    @api.doc("insert_explore_banner")
+    @api.doc(description="Insert an explore banner")
+    @api.expect(
+        api.model(
+            "InsertExploreBannerRequest",
+            {
+                "content": fields.String(required=True, description="Banner content"),
+                "link": fields.String(required=True, description="Banner link"),
+                "sort": fields.Integer(required=True, description="Banner sort"),
+            },
+        )
+    )
+    @api.response(200, "Banner inserted successfully")
+    @admin_required
+    @only_edition_cloud
+    def post(self):
+        parser = reqparse.RequestParser()
+        parser.add_argument("category", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("title", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("description", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("img-src", type=str, required=True, nullable=False, location="json")
+
+        parser.add_argument("language", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("link", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("sort", type=int, required=True, nullable=False, location="json")
+
+        args = parser.parse_args()
+
+        content = {
+            "category": args["category"],
+            "title": args["title"],
+            "description": args["description"],
+            "img-src": args["img-src"],
+        }
+
+        if not args["language"]:
+            args["language"] = "en-US"
+
+        banner = ExporleBanner(
+            content=content,
+            link=args["link"],
+            sort=args["sort"],
+            language=args["language"],
+        )
+        db.session.add(banner)
+        db.session.commit()
+
+        return {"result": "success"}, 200
+
+
+@console_ns.route("/admin/delete-explore-banner/<uuid:banner_id>")
+class DeleteExploreBanner(Resource):
+    @api.doc("delete_explore_banner")
+    @api.doc(description="Delete an explore banner")
+    @api.response(204, "Banner deleted successfully")
+    @admin_required
+    @only_edition_cloud
+    def delete(self, banner_id):
+        banner = db.session.execute(select(ExporleBanner).where(ExporleBanner.id == banner_id)).scalar_one_or_none()
+        if not banner:
+            raise NotFound(f"Banner '{banner_id}' is not found")
+
+        db.session.delete(banner)
+        db.session.commit()
+
+        return {"result": "success"}, 204

api/controllers/console/app/annotation.py

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

api/controllers/console/app/conversation.py

@@ -1,7 +1,5 @@
-from datetime import datetime
-
-import pytz
 import sqlalchemy as sa
+from flask import abort
 from flask_restx import Resource, marshal_with, reqparse
 from flask_restx.inputs import int_range
 from sqlalchemy import func, or_
@@ -19,7 +17,7 @@ from fields.conversation_fields import (
     conversation_pagination_fields,
     conversation_with_summary_pagination_fields,
 )
-from libs.datetime_utils import naive_utc_now
+from libs.datetime_utils import naive_utc_now, parse_time_range
 from libs.helper import DatetimeString
 from libs.login import current_account_with_tenant, login_required
 from models import Conversation, EndUser, Message, MessageAnnotation
@@ -90,25 +88,17 @@ class CompletionConversationApi(Resource):
 
         account = current_user
         assert account.timezone is not None
-        timezone = pytz.timezone(account.timezone)
-        utc_timezone = pytz.utc
 
-        if args["start"]:
-            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
-            start_datetime = start_datetime.replace(second=0)
-
-            start_datetime_timezone = timezone.localize(start_datetime)
-            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
+        try:
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
+        except ValueError as e:
+            abort(400, description=str(e))
 
+        if start_datetime_utc:
             query = query.where(Conversation.created_at >= start_datetime_utc)
 
-        if args["end"]:
-            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
-            end_datetime = end_datetime.replace(second=59)
-
-            end_datetime_timezone = timezone.localize(end_datetime)
-            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
-
+        if end_datetime_utc:
+            end_datetime_utc = end_datetime_utc.replace(second=59)
             query = query.where(Conversation.created_at < end_datetime_utc)
 
         # FIXME, the type ignore in this file
@@ -270,29 +260,21 @@ class ChatConversationApi(Resource):
 
         account = current_user
         assert account.timezone is not None
-        timezone = pytz.timezone(account.timezone)
-        utc_timezone = pytz.utc
 
-        if args["start"]:
-            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
-            start_datetime = start_datetime.replace(second=0)
-
-            start_datetime_timezone = timezone.localize(start_datetime)
-            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
+        try:
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
+        except ValueError as e:
+            abort(400, description=str(e))
 
+        if start_datetime_utc:
             match args["sort_by"]:
                 case "updated_at" | "-updated_at":
                     query = query.where(Conversation.updated_at >= start_datetime_utc)
                 case "created_at" | "-created_at" | _:
                     query = query.where(Conversation.created_at >= start_datetime_utc)
 
-        if args["end"]:
-            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
-            end_datetime = end_datetime.replace(second=59)
-
-            end_datetime_timezone = timezone.localize(end_datetime)
-            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
-
+        if end_datetime_utc:
+            end_datetime_utc = end_datetime_utc.replace(second=59)
             match args["sort_by"]:
                 case "updated_at" | "-updated_at":
                     query = query.where(Conversation.updated_at <= end_datetime_utc)

api/controllers/console/app/error.py

@@ -115,3 +115,9 @@ class InvokeRateLimitError(BaseHTTPException):
     error_code = "rate_limit_error"
     description = "Rate Limit Error"
     code = 429
+
+
+class NeedAddIdsError(BaseHTTPException):
+    error_code = "need_add_ids"
+    description = "Need to add ids."
+    code = 400

api/controllers/console/app/message.py

@@ -16,7 +16,6 @@ from controllers.console.app.wraps import get_app_model
 from controllers.console.explore.error import AppSuggestedQuestionsAfterAnswerDisabledError
 from controllers.console.wraps import (
     account_initialization_required,
-    cloud_edition_billing_resource_check,
     edit_permission_required,
     setup_required,
 )
@@ -24,12 +23,11 @@ from core.app.entities.app_invoke_entities import InvokeFrom
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
 from core.model_runtime.errors.invoke import InvokeError
 from extensions.ext_database import db
-from fields.conversation_fields import annotation_fields, message_detail_fields
+from fields.conversation_fields import message_detail_fields
 from libs.helper import uuid_value
 from libs.infinite_scroll_pagination import InfiniteScrollPagination
 from libs.login import current_account_with_tenant, login_required
 from models.model import AppMode, Conversation, Message, MessageAnnotation, MessageFeedback
-from services.annotation_service import AppAnnotationService
 from services.errors.conversation import ConversationNotExistsError
 from services.errors.message import MessageNotExistsError, SuggestedQuestionsAfterAnswerDisabledError
 from services.message_service import MessageService
@@ -194,45 +192,6 @@ class MessageFeedbackApi(Resource):
         return {"result": "success"}
 
 
-@console_ns.route("/apps/<uuid:app_id>/annotations")
-class MessageAnnotationApi(Resource):
-    @api.doc("create_message_annotation")
-    @api.doc(description="Create message annotation")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.expect(
-        api.model(
-            "MessageAnnotationRequest",
-            {
-                "message_id": fields.String(description="Message ID"),
-                "question": fields.String(required=True, description="Question text"),
-                "answer": fields.String(required=True, description="Answer text"),
-                "annotation_reply": fields.Raw(description="Annotation reply"),
-            },
-        )
-    )
-    @api.response(200, "Annotation created successfully", annotation_fields)
-    @api.response(403, "Insufficient permissions")
-    @marshal_with(annotation_fields)
-    @get_app_model
-    @setup_required
-    @login_required
-    @cloud_edition_billing_resource_check("annotation")
-    @account_initialization_required
-    @edit_permission_required
-    def post(self, app_model):
-        parser = (
-            reqparse.RequestParser()
-            .add_argument("message_id", required=False, type=uuid_value, location="json")
-            .add_argument("question", required=True, type=str, location="json")
-            .add_argument("answer", required=True, type=str, location="json")
-            .add_argument("annotation_reply", required=False, type=dict, location="json")
-        )
-        args = parser.parse_args()
-        annotation = AppAnnotationService.up_insert_app_annotation_from_message(args, app_model.id)
-
-        return annotation
-
-
 @console_ns.route("/apps/<uuid:app_id>/annotations/count")
 class MessageAnnotationCountApi(Resource):
     @api.doc("get_annotation_count")

api/controllers/console/app/online_user.py

@@ -0,0 +1,339 @@
+import json
+import time
+
+from werkzeug.wrappers import Request as WerkzeugRequest
+
+from extensions.ext_redis import redis_client
+from extensions.ext_socketio import sio
+from libs.passport import PassportService
+from libs.token import extract_access_token
+from services.account_service import AccountService
+
+SESSION_STATE_TTL_SECONDS = 3600
+WORKFLOW_ONLINE_USERS_PREFIX = "workflow_online_users:"
+WORKFLOW_LEADER_PREFIX = "workflow_leader:"
+WS_SID_MAP_PREFIX = "ws_sid_map:"
+
+
+def _workflow_key(workflow_id: str) -> str:
+    return f"{WORKFLOW_ONLINE_USERS_PREFIX}{workflow_id}"
+
+
+def _leader_key(workflow_id: str) -> str:
+    return f"{WORKFLOW_LEADER_PREFIX}{workflow_id}"
+
+
+def _sid_key(sid: str) -> str:
+    return f"{WS_SID_MAP_PREFIX}{sid}"
+
+
+def _refresh_session_state(workflow_id: str, sid: str) -> None:
+    """
+    Refresh TTLs for workflow + session keys so healthy sessions do not linger forever after crashes.
+    """
+    workflow_key = _workflow_key(workflow_id)
+    sid_key = _sid_key(sid)
+    if redis_client.exists(workflow_key):
+        redis_client.expire(workflow_key, SESSION_STATE_TTL_SECONDS)
+    if redis_client.exists(sid_key):
+        redis_client.expire(sid_key, SESSION_STATE_TTL_SECONDS)
+
+
+@sio.on("connect")
+def socket_connect(sid, environ, auth):
+    """
+    WebSocket connect event, do authentication here.
+    """
+    token = None
+    if auth and isinstance(auth, dict):
+        token = auth.get("token")
+
+    if not token:
+        try:
+            request_environ = WerkzeugRequest(environ)
+            token = extract_access_token(request_environ)
+        except Exception:
+            token = None
+
+    if not token:
+        return False
+
+    try:
+        decoded = PassportService().verify(token)
+        user_id = decoded.get("user_id")
+        if not user_id:
+            return False
+
+        with sio.app.app_context():
+            user = AccountService.load_logged_in_account(account_id=user_id)
+            if not user:
+                return False
+
+            sio.save_session(sid, {"user_id": user.id, "username": user.name, "avatar": user.avatar})
+
+            return True
+
+    except Exception:
+        return False
+
+
+@sio.on("user_connect")
+def handle_user_connect(sid, data):
+    """
+    Handle user connect event. Each session (tab) is treated as an independent collaborator.
+    """
+
+    workflow_id = data.get("workflow_id")
+    if not workflow_id:
+        return {"msg": "workflow_id is required"}, 400
+
+    session = sio.get_session(sid)
+    user_id = session.get("user_id")
+
+    if not user_id:
+        return {"msg": "unauthorized"}, 401
+
+    # Each session is stored independently with sid as key
+    session_info = {
+        "user_id": user_id,
+        "username": session.get("username", "Unknown"),
+        "avatar": session.get("avatar", None),
+        "sid": sid,
+        "connected_at": int(time.time()),  # Add timestamp to differentiate tabs
+    }
+
+    workflow_key = _workflow_key(workflow_id)
+    # Store session info with sid as key
+    redis_client.hset(workflow_key, sid, json.dumps(session_info))
+    redis_client.set(
+        _sid_key(sid),
+        json.dumps({"workflow_id": workflow_id, "user_id": user_id}),
+        ex=SESSION_STATE_TTL_SECONDS,
+    )
+    _refresh_session_state(workflow_id, sid)
+
+    # Leader election: first session becomes the leader
+    leader_sid = get_or_set_leader(workflow_id, sid)
+    is_leader = leader_sid == sid
+
+    sio.enter_room(sid, workflow_id)
+    broadcast_online_users(workflow_id)
+
+    # Notify this session of their leader status
+    sio.emit("status", {"isLeader": is_leader}, room=sid)
+
+    return {"msg": "connected", "user_id": user_id, "sid": sid, "isLeader": is_leader}
+
+
+@sio.on("disconnect")
+def handle_disconnect(sid):
+    """
+    Handle session disconnect event. Remove the specific session from online users.
+    """
+    mapping = redis_client.get(_sid_key(sid))
+    if mapping:
+        data = json.loads(mapping)
+        workflow_id = data["workflow_id"]
+
+        # Remove this specific session
+        redis_client.hdel(_workflow_key(workflow_id), sid)
+        redis_client.delete(_sid_key(sid))
+
+        # Handle leader re-election if the leader session disconnected
+        handle_leader_disconnect(workflow_id, sid)
+
+        broadcast_online_users(workflow_id)
+
+
+def _clear_session_state(workflow_id: str, sid: str) -> None:
+    redis_client.hdel(_workflow_key(workflow_id), sid)
+    redis_client.delete(_sid_key(sid))
+
+
+def _is_session_active(workflow_id: str, sid: str) -> bool:
+    if not sid:
+        return False
+
+    try:
+        if not sio.manager.is_connected(sid, "/"):
+            return False
+    except AttributeError:
+        return False
+
+    if not redis_client.hexists(_workflow_key(workflow_id), sid):
+        return False
+
+    if not redis_client.exists(_sid_key(sid)):
+        return False
+
+    return True
+
+
+def get_or_set_leader(workflow_id: str, sid: str) -> str:
+    """
+    Get current leader session or set this session as leader if no valid leader exists.
+    Returns the leader session id (sid).
+    """
+    raw_leader = redis_client.get(_leader_key(workflow_id))
+    current_leader = raw_leader.decode("utf-8") if isinstance(raw_leader, bytes) else raw_leader
+    leader_replaced = False
+
+    if current_leader and not _is_session_active(workflow_id, current_leader):
+        _clear_session_state(workflow_id, current_leader)
+        redis_client.delete(_leader_key(workflow_id))
+        current_leader = None
+        leader_replaced = True
+
+    if not current_leader:
+        redis_client.set(_leader_key(workflow_id), sid, ex=SESSION_STATE_TTL_SECONDS)  # Expire in 1 hour
+        if leader_replaced:
+            broadcast_leader_change(workflow_id, sid)
+        return sid
+
+    return current_leader
+
+
+def handle_leader_disconnect(workflow_id, disconnected_sid):
+    """
+    Handle leader re-election when a session disconnects.
+    If the disconnected session was the leader, elect a new leader from remaining sessions.
+    """
+    current_leader = redis_client.get(_leader_key(workflow_id))
+
+    if current_leader:
+        current_leader = current_leader.decode("utf-8") if isinstance(current_leader, bytes) else current_leader
+
+        if current_leader == disconnected_sid:
+            # Leader session disconnected, elect a new leader
+            sessions_json = redis_client.hgetall(_workflow_key(workflow_id))
+
+            if sessions_json:
+                # Get the first remaining session as new leader
+                new_leader_sid = list(sessions_json.keys())[0]
+                if isinstance(new_leader_sid, bytes):
+                    new_leader_sid = new_leader_sid.decode("utf-8")
+
+                redis_client.set(_leader_key(workflow_id), new_leader_sid, ex=SESSION_STATE_TTL_SECONDS)
+
+                # Notify all sessions about the new leader
+                broadcast_leader_change(workflow_id, new_leader_sid)
+            else:
+                # No sessions left, remove leader
+                redis_client.delete(_leader_key(workflow_id))
+
+
+def broadcast_leader_change(workflow_id, new_leader_sid):
+    """
+    Broadcast leader change to all sessions in the workflow.
+    """
+    sessions_json = redis_client.hgetall(_workflow_key(workflow_id))
+
+    for sid, session_info_json in sessions_json.items():
+        try:
+            sid_str = sid.decode("utf-8") if isinstance(sid, bytes) else sid
+            is_leader = sid_str == new_leader_sid
+            # Emit to each session whether they are the new leader
+            sio.emit("status", {"isLeader": is_leader}, room=sid_str)
+        except Exception:
+            continue
+
+
+def get_current_leader(workflow_id):
+    """
+    Get the current leader for a workflow.
+    """
+    leader = redis_client.get(_leader_key(workflow_id))
+    return leader.decode("utf-8") if leader and isinstance(leader, bytes) else leader
+
+
+def broadcast_online_users(workflow_id):
+    """
+    Broadcast online users to the workflow room.
+    Each session is shown as a separate user (even if same person has multiple tabs).
+    """
+    sessions_json = redis_client.hgetall(_workflow_key(workflow_id))
+    users = []
+
+    for sid, session_info_json in sessions_json.items():
+        try:
+            session_info = json.loads(session_info_json)
+            # Each session appears as a separate "user" in the UI
+            users.append(
+                {
+                    "user_id": session_info["user_id"],
+                    "username": session_info["username"],
+                    "avatar": session_info.get("avatar"),
+                    "sid": session_info["sid"],
+                    "connected_at": session_info.get("connected_at"),
+                }
+            )
+        except Exception:
+            continue
+
+    # Sort by connection time to maintain consistent order
+    users.sort(key=lambda x: x.get("connected_at") or 0)
+
+    # Get current leader session
+    leader_sid = get_current_leader(workflow_id)
+
+    sio.emit("online_users", {"workflow_id": workflow_id, "users": users, "leader": leader_sid}, room=workflow_id)
+
+
+@sio.on("collaboration_event")
+def handle_collaboration_event(sid, data):
+    """
+    Handle general collaboration events, include:
+    1. mouse_move
+    2. vars_and_features_update
+    3. sync_request (ask leader to update graph)
+    4. app_state_update
+    5. mcp_server_update
+    6. workflow_update
+    7. comments_update
+    8. node_panel_presence
+
+    """
+    mapping = redis_client.get(_sid_key(sid))
+
+    if not mapping:
+        return {"msg": "unauthorized"}, 401
+
+    mapping_data = json.loads(mapping)
+    workflow_id = mapping_data["workflow_id"]
+    user_id = mapping_data["user_id"]
+    _refresh_session_state(workflow_id, sid)
+
+    event_type = data.get("type")
+    event_data = data.get("data")
+    timestamp = data.get("timestamp", int(time.time()))
+
+    if not event_type:
+        return {"msg": "invalid event type"}, 400
+
+    sio.emit(
+        "collaboration_update",
+        {"type": event_type, "userId": user_id, "data": event_data, "timestamp": timestamp},
+        room=workflow_id,
+        skip_sid=sid,
+    )
+
+    return {"msg": "event_broadcasted"}
+
+
+@sio.on("graph_event")
+def handle_graph_event(sid, data):
+    """
+    Handle graph events - simple broadcast relay.
+    """
+    mapping = redis_client.get(_sid_key(sid))
+
+    if not mapping:
+        return {"msg": "unauthorized"}, 401
+
+    mapping_data = json.loads(mapping)
+    workflow_id = mapping_data["workflow_id"]
+    _refresh_session_state(workflow_id, sid)
+
+    sio.emit("graph_update", data, room=workflow_id, skip_sid=sid)
+
+    return {"msg": "graph_update_broadcasted"}

api/controllers/console/app/statistic.py

@@ -1,9 +1,7 @@
-from datetime import datetime
 from decimal import Decimal
 
-import pytz
 import sqlalchemy as sa
-from flask import jsonify
+from flask import abort, jsonify
 from flask_restx import Resource, fields, reqparse
 
 from controllers.console import api, console_ns
@@ -11,6 +9,7 @@ from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
 from core.app.entities.app_invoke_entities import InvokeFrom
 from extensions.ext_database import db
+from libs.datetime_utils import parse_time_range
 from libs.helper import DatetimeString
 from libs.login import current_account_with_tenant, login_required
 from models import AppMode, Message
@@ -56,26 +55,16 @@ WHERE
         arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
         assert account.timezone is not None
 
-        timezone = pytz.timezone(account.timezone)
-        utc_timezone = pytz.utc
-
-        if args["start"]:
-            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
-            start_datetime = start_datetime.replace(second=0)
-
-            start_datetime_timezone = timezone.localize(start_datetime)
-            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
+        try:
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
+        except ValueError as e:
+            abort(400, description=str(e))
 
+        if start_datetime_utc:
             sql_query += " AND created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
-        if args["end"]:
-            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
-            end_datetime = end_datetime.replace(second=0)
-
-            end_datetime_timezone = timezone.localize(end_datetime)
-            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
-
+        if end_datetime_utc:
             sql_query += " AND created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
@@ -120,8 +109,11 @@ class DailyConversationStatistic(Resource):
         )
         args = parser.parse_args()
         assert account.timezone is not None
-        timezone = pytz.timezone(account.timezone)
-        utc_timezone = pytz.utc
+
+        try:
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
+        except ValueError as e:
+            abort(400, description=str(e))
 
         stmt = (
             sa.select(
@@ -134,18 +126,10 @@ class DailyConversationStatistic(Resource):
             .where(Message.app_id == app_model.id, Message.invoke_from != InvokeFrom.DEBUGGER)
         )
 
-        if args["start"]:
-            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
-            start_datetime = start_datetime.replace(second=0)
-            start_datetime_timezone = timezone.localize(start_datetime)
-            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
+        if start_datetime_utc:
             stmt = stmt.where(Message.created_at >= start_datetime_utc)
 
-        if args["end"]:
-            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
-            end_datetime = end_datetime.replace(second=0)
-            end_datetime_timezone = timezone.localize(end_datetime)
-            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
+        if end_datetime_utc:
             stmt = stmt.where(Message.created_at < end_datetime_utc)
 
         stmt = stmt.group_by("date").order_by("date")
@@ -198,26 +182,17 @@ WHERE
     AND invoke_from != :invoke_from"""
         arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
         assert account.timezone is not None
-        timezone = pytz.timezone(account.timezone)
-        utc_timezone = pytz.utc
 
-        if args["start"]:
-            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
-            start_datetime = start_datetime.replace(second=0)
-
-            start_datetime_timezone = timezone.localize(start_datetime)
-            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
+        try:
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
+        except ValueError as e:
+            abort(400, description=str(e))
 
+        if start_datetime_utc:
             sql_query += " AND created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
-        if args["end"]:
-            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
-            end_datetime = end_datetime.replace(second=0)
-
-            end_datetime_timezone = timezone.localize(end_datetime)
-            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
-
+        if end_datetime_utc:
             sql_query += " AND created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
@@ -273,26 +248,17 @@ WHERE
     AND invoke_from != :invoke_from"""
         arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
         assert account.timezone is not None
-        timezone = pytz.timezone(account.timezone)
-        utc_timezone = pytz.utc
 
-        if args["start"]:
-            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
-            start_datetime = start_datetime.replace(second=0)
-
-            start_datetime_timezone = timezone.localize(start_datetime)
-            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
+        try:
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
+        except ValueError as e:
+            abort(400, description=str(e))
 
+        if start_datetime_utc:
             sql_query += " AND created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
-        if args["end"]:
-            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
-            end_datetime = end_datetime.replace(second=0)
-
-            end_datetime_timezone = timezone.localize(end_datetime)
-            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
-
+        if end_datetime_utc:
             sql_query += " AND created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
@@ -357,26 +323,17 @@ FROM
             AND m.invoke_from != :invoke_from"""
         arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
         assert account.timezone is not None
-        timezone = pytz.timezone(account.timezone)
-        utc_timezone = pytz.utc
 
-        if args["start"]:
-            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
-            start_datetime = start_datetime.replace(second=0)
-
-            start_datetime_timezone = timezone.localize(start_datetime)
-            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
+        try:
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
+        except ValueError as e:
+            abort(400, description=str(e))
 
+        if start_datetime_utc:
             sql_query += " AND c.created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
-        if args["end"]:
-            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
-            end_datetime = end_datetime.replace(second=0)
-
-            end_datetime_timezone = timezone.localize(end_datetime)
-            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
-
+        if end_datetime_utc:
             sql_query += " AND c.created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
@@ -446,26 +403,17 @@ WHERE
     AND m.invoke_from != :invoke_from"""
         arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
         assert account.timezone is not None
-        timezone = pytz.timezone(account.timezone)
-        utc_timezone = pytz.utc
 
-        if args["start"]:
-            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
-            start_datetime = start_datetime.replace(second=0)
-
-            start_datetime_timezone = timezone.localize(start_datetime)
-            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
+        try:
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
+        except ValueError as e:
+            abort(400, description=str(e))
 
+        if start_datetime_utc:
             sql_query += " AND m.created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
-        if args["end"]:
-            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
-            end_datetime = end_datetime.replace(second=0)
-
-            end_datetime_timezone = timezone.localize(end_datetime)
-            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
-
+        if end_datetime_utc:
             sql_query += " AND m.created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
@@ -525,26 +473,17 @@ WHERE
     AND invoke_from != :invoke_from"""
         arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
         assert account.timezone is not None
-        timezone = pytz.timezone(account.timezone)
-        utc_timezone = pytz.utc
 
-        if args["start"]:
-            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
-            start_datetime = start_datetime.replace(second=0)
-
-            start_datetime_timezone = timezone.localize(start_datetime)
-            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
+        try:
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
+        except ValueError as e:
+            abort(400, description=str(e))
 
+        if start_datetime_utc:
             sql_query += " AND created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
-        if args["end"]:
-            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
-            end_datetime = end_datetime.replace(second=0)
-
-            end_datetime_timezone = timezone.localize(end_datetime)
-            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
-
+        if end_datetime_utc:
             sql_query += " AND created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
@@ -602,26 +541,17 @@ WHERE
     AND invoke_from != :invoke_from"""
         arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
         assert account.timezone is not None
-        timezone = pytz.timezone(account.timezone)
-        utc_timezone = pytz.utc
 
-        if args["start"]:
-            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
-            start_datetime = start_datetime.replace(second=0)
-
-            start_datetime_timezone = timezone.localize(start_datetime)
-            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
+        try:
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
+        except ValueError as e:
+            abort(400, description=str(e))
 
+        if start_datetime_utc:
             sql_query += " AND created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
-        if args["end"]:
-            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
-            end_datetime = end_datetime.replace(second=0)
-
-            end_datetime_timezone = timezone.localize(end_datetime)
-            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
-
+        if end_datetime_utc:
             sql_query += " AND created_at < :end"
             arg_dict["end"] = end_datetime_utc
 

api/controllers/console/app/workflow.py

@@ -5,10 +5,12 @@ from typing import cast
 
 from flask import abort, request
 from flask_restx import Resource, fields, inputs, marshal_with, reqparse
+from pydantic_core import ValidationError
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
 
 import services
+from configs import dify_config
 from controllers.console import api, console_ns
 from controllers.console.app.error import ConversationCompletedError, DraftWorkflowNotExist, DraftWorkflowNotSync
 from controllers.console.app.wraps import get_app_model
@@ -21,7 +23,9 @@ from core.file.models import File
 from core.helper.trace_id_helper import get_external_trace_id
 from core.workflow.graph_engine.manager import GraphEngineManager
 from extensions.ext_database import db
+from extensions.ext_redis import redis_client
 from factories import file_factory, variable_factory
+from fields.online_user_fields import online_user_list_fields
 from fields.workflow_fields import workflow_fields, workflow_pagination_fields
 from fields.workflow_run_fields import workflow_run_node_execution_fields
 from libs import helper
@@ -99,10 +103,22 @@ class DraftWorkflowApi(Resource):
                 "hash": fields.String(description="Workflow hash for validation"),
                 "environment_variables": fields.List(fields.Raw, required=True, description="Environment variables"),
                 "conversation_variables": fields.List(fields.Raw, description="Conversation variables"),
+                "memory_blocks": fields.List(fields.Raw, description="Memory blocks"),
             },
         )
     )
-    @api.response(200, "Draft workflow synced successfully", workflow_fields)
+    @api.response(
+        200,
+        "Draft workflow synced successfully",
+        api.model(
+            "SyncDraftWorkflowResponse",
+            {
+                "result": fields.String,
+                "hash": fields.String,
+                "updated_at": fields.String,
+            },
+        ),
+    )
     @api.response(400, "Invalid workflow configuration")
     @api.response(403, "Permission denied")
     @edit_permission_required
@@ -122,6 +138,8 @@ class DraftWorkflowApi(Resource):
                 .add_argument("hash", type=str, required=False, location="json")
                 .add_argument("environment_variables", type=list, required=True, location="json")
                 .add_argument("conversation_variables", type=list, required=False, location="json")
+                .add_argument("force_upload", type=bool, required=False, default=False, location="json")
+                .add_argument("memory_blocks", type=list, required=False, location="json")
             )
             args = parser.parse_args()
         elif "text/plain" in content_type:
@@ -139,6 +157,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
@@ -155,6 +175,10 @@ 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"],
@@ -163,9 +187,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",
@@ -734,6 +762,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")
@@ -915,3 +982,105 @@ class DraftWorkflowNodeLastRunApi(Resource):
         if node_exec is None:
             raise NotFound("last run not found")
         return node_exec
+
+
+class WorkflowOnlineUsersApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @marshal_with(online_user_list_fields)
+    def get(self):
+        parser = reqparse.RequestParser()
+        parser.add_argument("workflow_ids", type=str, required=True, location="args")
+        args = parser.parse_args()
+
+        workflow_ids = [id.strip() for id in args["workflow_ids"].split(",")]
+
+        results = []
+        for workflow_id in workflow_ids:
+            users_json = redis_client.hgetall(f"workflow_online_users:{workflow_id}")
+
+            users = []
+            for _, user_info_json in users_json.items():
+                try:
+                    users.append(json.loads(user_info_json))
+                except Exception:
+                    continue
+            results.append({"workflow_id": workflow_id, "users": users})
+
+        return {"data": results}
+
+
+api.add_resource(
+    DraftWorkflowApi,
+    "/apps/<uuid:app_id>/workflows/draft",
+)
+api.add_resource(
+    WorkflowConfigApi,
+    "/apps/<uuid:app_id>/workflows/draft/config",
+)
+api.add_resource(
+    WorkflowFeaturesApi,
+    "/apps/<uuid:app_id>/workflows/draft/features",
+)
+api.add_resource(
+    AdvancedChatDraftWorkflowRunApi,
+    "/apps/<uuid:app_id>/advanced-chat/workflows/draft/run",
+)
+api.add_resource(
+    DraftWorkflowRunApi,
+    "/apps/<uuid:app_id>/workflows/draft/run",
+)
+api.add_resource(
+    WorkflowTaskStopApi,
+    "/apps/<uuid:app_id>/workflow-runs/tasks/<string:task_id>/stop",
+)
+api.add_resource(
+    DraftWorkflowNodeRunApi,
+    "/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/run",
+)
+api.add_resource(
+    AdvancedChatDraftRunIterationNodeApi,
+    "/apps/<uuid:app_id>/advanced-chat/workflows/draft/iteration/nodes/<string:node_id>/run",
+)
+api.add_resource(
+    WorkflowDraftRunIterationNodeApi,
+    "/apps/<uuid:app_id>/workflows/draft/iteration/nodes/<string:node_id>/run",
+)
+api.add_resource(
+    AdvancedChatDraftRunLoopNodeApi,
+    "/apps/<uuid:app_id>/advanced-chat/workflows/draft/loop/nodes/<string:node_id>/run",
+)
+api.add_resource(
+    WorkflowDraftRunLoopNodeApi,
+    "/apps/<uuid:app_id>/workflows/draft/loop/nodes/<string:node_id>/run",
+)
+api.add_resource(
+    PublishedWorkflowApi,
+    "/apps/<uuid:app_id>/workflows/publish",
+)
+api.add_resource(
+    PublishedAllWorkflowApi,
+    "/apps/<uuid:app_id>/workflows",
+)
+api.add_resource(
+    DefaultBlockConfigsApi,
+    "/apps/<uuid:app_id>/workflows/default-workflow-block-configs",
+)
+api.add_resource(
+    DefaultBlockConfigApi,
+    "/apps/<uuid:app_id>/workflows/default-workflow-block-configs/<string:block_type>",
+)
+api.add_resource(
+    ConvertToWorkflowApi,
+    "/apps/<uuid:app_id>/convert-to-workflow",
+)
+api.add_resource(
+    WorkflowByIdApi,
+    "/apps/<uuid:app_id>/workflows/<string:workflow_id>",
+)
+api.add_resource(
+    DraftWorkflowNodeLastRunApi,
+    "/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/last-run",
+)
+api.add_resource(WorkflowOnlineUsersApi, "/apps/workflows/online-users")

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 Account, App, AppMode
 from models.workflow import WorkflowDraftVariable
@@ -355,7 +355,7 @@ class VariableApi(Resource):
                 if len(raw_value) > 0 and not isinstance(raw_value[0], dict):
                     raise InvalidArgumentError(description=f"expected dict for files[0], got {type(raw_value)}")
                 raw_value = build_from_mappings(mappings=raw_value, tenant_id=app_model.tenant_id)
-            new_value = 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
@@ -448,8 +448,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")
@@ -499,3 +526,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

@@ -1,7 +1,4 @@
-from datetime import datetime
-
-import pytz
-from flask import jsonify
+from flask import abort, jsonify
 from flask_restx import Resource, reqparse
 from sqlalchemy.orm import sessionmaker
 
@@ -9,6 +6,7 @@ from controllers.console import api, console_ns
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
 from extensions.ext_database import db
+from libs.datetime_utils import parse_time_range
 from libs.helper import DatetimeString
 from libs.login import current_account_with_tenant, login_required
 from models.enums import WorkflowRunTriggeredFrom
@@ -43,23 +41,11 @@ class WorkflowDailyRunsStatistic(Resource):
         args = parser.parse_args()
 
         assert account.timezone is not None
-        timezone = pytz.timezone(account.timezone)
-        utc_timezone = pytz.utc
 
-        start_date = None
-        end_date = None
-
-        if args["start"]:
-            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
-            start_datetime = start_datetime.replace(second=0)
-            start_datetime_timezone = timezone.localize(start_datetime)
-            start_date = start_datetime_timezone.astimezone(utc_timezone)
-
-        if args["end"]:
-            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
-            end_datetime = end_datetime.replace(second=0)
-            end_datetime_timezone = timezone.localize(end_datetime)
-            end_date = end_datetime_timezone.astimezone(utc_timezone)
+        try:
+            start_date, end_date = parse_time_range(args["start"], args["end"], account.timezone)
+        except ValueError as e:
+            abort(400, description=str(e))
 
         response_data = self._workflow_run_repo.get_daily_runs_statistics(
             tenant_id=app_model.tenant_id,
@@ -100,23 +86,11 @@ class WorkflowDailyTerminalsStatistic(Resource):
         args = parser.parse_args()
 
         assert account.timezone is not None
-        timezone = pytz.timezone(account.timezone)
-        utc_timezone = pytz.utc
 
-        start_date = None
-        end_date = None
-
-        if args["start"]:
-            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
-            start_datetime = start_datetime.replace(second=0)
-            start_datetime_timezone = timezone.localize(start_datetime)
-            start_date = start_datetime_timezone.astimezone(utc_timezone)
-
-        if args["end"]:
-            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
-            end_datetime = end_datetime.replace(second=0)
-            end_datetime_timezone = timezone.localize(end_datetime)
-            end_date = end_datetime_timezone.astimezone(utc_timezone)
+        try:
+            start_date, end_date = parse_time_range(args["start"], args["end"], account.timezone)
+        except ValueError as e:
+            abort(400, description=str(e))
 
         response_data = self._workflow_run_repo.get_daily_terminals_statistics(
             tenant_id=app_model.tenant_id,
@@ -157,23 +131,11 @@ class WorkflowDailyTokenCostStatistic(Resource):
         args = parser.parse_args()
 
         assert account.timezone is not None
-        timezone = pytz.timezone(account.timezone)
-        utc_timezone = pytz.utc
 
-        start_date = None
-        end_date = None
-
-        if args["start"]:
-            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
-            start_datetime = start_datetime.replace(second=0)
-            start_datetime_timezone = timezone.localize(start_datetime)
-            start_date = start_datetime_timezone.astimezone(utc_timezone)
-
-        if args["end"]:
-            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
-            end_datetime = end_datetime.replace(second=0)
-            end_datetime_timezone = timezone.localize(end_datetime)
-            end_date = end_datetime_timezone.astimezone(utc_timezone)
+        try:
+            start_date, end_date = parse_time_range(args["start"], args["end"], account.timezone)
+        except ValueError as e:
+            abort(400, description=str(e))
 
         response_data = self._workflow_run_repo.get_daily_token_cost_statistics(
             tenant_id=app_model.tenant_id,
@@ -214,23 +176,11 @@ class WorkflowAverageAppInteractionStatistic(Resource):
         args = parser.parse_args()
 
         assert account.timezone is not None
-        timezone = pytz.timezone(account.timezone)
-        utc_timezone = pytz.utc
 
-        start_date = None
-        end_date = None
-
-        if args["start"]:
-            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
-            start_datetime = start_datetime.replace(second=0)
-            start_datetime_timezone = timezone.localize(start_datetime)
-            start_date = start_datetime_timezone.astimezone(utc_timezone)
-
-        if args["end"]:
-            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
-            end_datetime = end_datetime.replace(second=0)
-            end_datetime_timezone = timezone.localize(end_datetime)
-            end_date = end_datetime_timezone.astimezone(utc_timezone)
+        try:
+            start_date, end_date = parse_time_range(args["start"], args["end"], account.timezone)
+        except ValueError as e:
+            abort(400, description=str(e))
 
         response_data = self._workflow_run_repo.get_average_app_interaction_statistics(
             tenant_id=app_model.tenant_id,

api/controllers/console/app/wraps.py

@@ -23,6 +23,15 @@ def _load_app_model(app_id: str) -> App | None:
     return app_model
 
 
+def _load_app_model_with_trial(app_id: str) -> App | None:
+    app_model = (
+        db.session.query(App)
+        .where(App.id == app_id, App.status == "normal")
+        .first()
+    )
+    return app_model
+
+
 def get_app_model(view: Callable[P, R] | None = None, *, mode: Union[AppMode, list[AppMode], None] = None):
     def decorator(view_func: Callable[P1, R1]):
         @wraps(view_func)
@@ -62,3 +71,44 @@ def get_app_model(view: Callable[P, R] | None = None, *, mode: Union[AppMode, li
         return decorator
     else:
         return decorator(view)
+
+
+def get_app_model_with_trial(view: Callable[P, R] | None = None, *, mode: Union[AppMode, list[AppMode], None] = None):
+    def decorator(view_func: Callable[P, R]):
+        @wraps(view_func)
+        def decorated_view(*args: P.args, **kwargs: P.kwargs):
+            if not kwargs.get("app_id"):
+                raise ValueError("missing app_id in path parameters")
+
+            app_id = kwargs.get("app_id")
+            app_id = str(app_id)
+
+            del kwargs["app_id"]
+
+            app_model = _load_app_model_with_trial(app_id)
+
+            if not app_model:
+                raise AppNotFoundError()
+
+            app_mode = AppMode.value_of(app_model.mode)
+
+            if mode is not None:
+                if isinstance(mode, list):
+                    modes = mode
+                else:
+                    modes = [mode]
+
+                if app_mode not in modes:
+                    mode_values = {m.value for m in modes}
+                    raise AppNotFoundError(f"App mode is not in the supported list: {mode_values}")
+
+            kwargs["app_model"] = app_model
+
+            return view_func(*args, **kwargs)
+
+        return decorated_view
+
+    if view is None:
+        return decorator
+    else:
+        return decorator(view)

api/controllers/console/auth/login.py

@@ -25,10 +25,13 @@ from controllers.console.wraps import email_password_login_enabled, setup_requir
 from events.tenant_event import tenant_was_created
 from libs.helper import email, extract_remote_ip
 from libs.login import current_account_with_tenant
+from libs.passport import PassportService
 from libs.token import (
+    check_csrf_token,
     clear_access_token_from_cookie,
     clear_csrf_token_from_cookie,
     clear_refresh_token_from_cookie,
+    extract_access_token,
     extract_refresh_token,
     set_access_token_to_cookie,
     set_csrf_token_to_cookie,
@@ -289,3 +292,18 @@ class RefreshTokenApi(Resource):
             return response
         except Exception as e:
             return {"result": "fail", "message": str(e)}, 401
+
+
+# this api helps frontend to check whether user is authenticated
+# TODO: remove in the future. frontend should redirect to login page by catching 401 status
+@console_ns.route("/login/status")
+class LoginStatus(Resource):
+    def get(self):
+        token = extract_access_token(request)
+        res = True
+        try:
+            validated = PassportService().verify(token=token)
+            check_csrf_token(request=request, user_id=validated.get("user_id"))
+        except:
+            res = False
+        return {"logged_in": res}

api/controllers/console/billing/billing.py

@@ -2,6 +2,7 @@ from flask_restx import Resource, reqparse
 
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, only_edition_cloud, setup_required
+from enums.cloud_plan import CloudPlan
 from libs.login import current_account_with_tenant, login_required
 from services.billing_service import BillingService
 
@@ -16,7 +17,13 @@ class Subscription(Resource):
         current_user, current_tenant_id = current_account_with_tenant()
         parser = (
             reqparse.RequestParser()
-            .add_argument("plan", type=str, required=True, location="args", choices=["professional", "team"])
+            .add_argument(
+                "plan",
+                type=str,
+                required=True,
+                location="args",
+                choices=[CloudPlan.PROFESSIONAL, CloudPlan.TEAM],
+            )
             .add_argument("interval", type=str, required=True, location="args", choices=["month", "year"])
         )
         args = parser.parse_args()

api/controllers/console/datasets/datasets_document.py

@@ -746,7 +746,7 @@ class DocumentApi(DocumentResource):
                 "name": document.name,
                 "created_from": document.created_from,
                 "created_by": document.created_by,
-                "created_at": document.created_at.timestamp(),
+                "created_at": int(document.created_at.timestamp()),
                 "tokens": document.tokens,
                 "indexing_status": document.indexing_status,
                 "completed_at": int(document.completed_at.timestamp()) if document.completed_at else None,
@@ -779,7 +779,7 @@ class DocumentApi(DocumentResource):
                 "name": document.name,
                 "created_from": document.created_from,
                 "created_by": document.created_by,
-                "created_at": document.created_at.timestamp(),
+                "created_at": int(document.created_at.timestamp()),
                 "tokens": document.tokens,
                 "indexing_status": document.indexing_status,
                 "completed_at": int(document.completed_at.timestamp()) if document.completed_at else None,

api/controllers/console/explore/banner.py

@@ -0,0 +1,43 @@
+from flask import request
+from flask_restx import Resource
+
+from controllers.console import api
+from controllers.console.explore.wraps import explore_banner_enabled
+from extensions.ext_database import db
+from models.model import ExporleBanner
+
+
+class BannerApi(Resource):
+    """Resource for banner list."""
+
+    @explore_banner_enabled
+    def get(self):
+        """Get banner list."""
+        language = request.args.get("language", "en-US")
+
+        # Build base query for enabled banners
+        base_query = db.session.query(ExporleBanner).where(ExporleBanner.status == "enabled")
+        
+        # Try to get banners in the requested language
+        banners = base_query.where(ExporleBanner.language == language).order_by(ExporleBanner.sort).all()
+        
+        # Fallback to en-US if no banners found and language is not en-US
+        if not banners and language != "en-US":
+            banners = base_query.where(ExporleBanner.language == "en-US").order_by(ExporleBanner.sort).all()
+        # Convert banners to serializable format
+        result = []
+        for banner in banners:
+            banner_data = {
+                "id": banner.id,
+                "content": banner.content,  # Already parsed as JSON by SQLAlchemy
+                "link": banner.link,
+                "sort": banner.sort,
+                "status": banner.status,
+                "created_at": banner.created_at.isoformat() if banner.created_at else None,
+            }
+            result.append(banner_data)
+
+        return result
+
+
+api.add_resource(BannerApi, "/explore/banners")

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

api/controllers/console/explore/wraps.py

@@ -2,14 +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_account_with_tenant, login_required
-from models import InstalledApp
+from models import AccountTrialAppRecord, App, InstalledApp, TrialApp
 from services.enterprise.enterprise_service import EnterpriseService
 from services.feature_service import FeatureService
 
@@ -71,6 +73,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
 
@@ -80,3 +135,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/files.py

@@ -8,6 +8,7 @@ import services
 from configs import dify_config
 from constants import DOCUMENT_EXTENSIONS
 from controllers.common.errors import (
+    BlockedFileExtensionError,
     FilenameNotExistsError,
     FileTooLargeError,
     NoFileUploadedError,
@@ -83,6 +84,8 @@ class FileApi(Resource):
             raise FileTooLargeError(file_too_large_error.description)
         except services.errors.file.UnsupportedFileTypeError:
             raise UnsupportedFileTypeError()
+        except services.errors.file.BlockedFileExtensionError as blocked_extension_error:
+            raise BlockedFileExtensionError(blocked_extension_error.description)
 
         return upload_file, 201
 

api/controllers/console/workspace/account.py

@@ -32,6 +32,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
@@ -128,6 +129,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/workspace.py

@@ -21,6 +21,7 @@ from controllers.console.wraps import (
     cloud_edition_billing_resource_check,
     setup_required,
 )
+from enums.cloud_plan import CloudPlan
 from extensions.ext_database import db
 from libs.helper import TimestampField
 from libs.login import current_account_with_tenant, login_required
@@ -50,6 +51,9 @@ tenant_fields = {
     "in_trial": fields.Boolean,
     "trial_end_reason": fields.String,
     "custom_config": fields.Raw(attribute="custom_config"),
+    "trial_credits": fields.Integer,
+    "trial_credits_used": fields.Integer,
+    "next_credit_reset_date": fields.Integer,
 }
 
 tenants_fields = {
@@ -83,7 +87,7 @@ class TenantListApi(Resource):
                 "name": tenant.name,
                 "status": tenant.status,
                 "created_at": tenant.created_at,
-                "plan": features.billing.subscription.plan if features.billing.enabled else "sandbox",
+                "plan": features.billing.subscription.plan if features.billing.enabled else CloudPlan.SANDBOX,
                 "current": tenant.id == current_tenant_id if current_tenant_id else False,
             }
 

api/controllers/console/wraps.py

@@ -10,6 +10,7 @@ from flask import abort, request
 
 from configs import dify_config
 from controllers.console.workspace.error import AccountNotInitializedError
+from enums.cloud_plan import CloudPlan
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from libs.login import current_account_with_tenant
@@ -133,7 +134,7 @@ def cloud_edition_billing_knowledge_limit_check(resource: str):
             features = FeatureService.get_features(current_tenant_id)
             if features.billing.enabled:
                 if resource == "add_segment":
-                    if features.billing.subscription.plan == "sandbox":
+                    if features.billing.subscription.plan == CloudPlan.SANDBOX:
                         abort(
                             403,
                             "To unlock this feature and elevate your Dify experience, please upgrade to a paid plan.",

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,109 @@
+from flask_restx import Resource, reqparse
+
+from controllers.service_api import api
+from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate_app_token
+from core.memory.entities import MemoryBlock, MemoryCreatedBy
+from core.workflow.runtime.variable_pool import VariablePool
+from models import App, EndUser
+from services.chatflow_memory_service import ChatflowMemoryService
+from services.workflow_service import WorkflowService
+
+
+class MemoryListApi(Resource):
+    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))
+    def get(self, app_model: App, end_user: EndUser):
+        parser = reqparse.RequestParser()
+        parser.add_argument("conversation_id", required=False, type=str | None, default=None)
+        parser.add_argument("memory_id", required=False, type=str | None, default=None)
+        parser.add_argument("version", required=False, type=int | None, default=None)
+        args = parser.parse_args()
+        conversation_id: str | None = args.get("conversation_id")
+        memory_id = args.get("memory_id")
+        version = args.get("version")
+
+        if conversation_id:
+            result = ChatflowMemoryService.get_persistent_memories_with_conversation(
+                app_model, MemoryCreatedBy(end_user_id=end_user.id), conversation_id, version
+            )
+            session_memories = ChatflowMemoryService.get_session_memories_with_conversation(
+                app_model, MemoryCreatedBy(end_user_id=end_user.id), conversation_id, version
+            )
+            result = [*result, *session_memories]
+        else:
+            result = ChatflowMemoryService.get_persistent_memories(
+                app_model, MemoryCreatedBy(end_user_id=end_user.id), version
+            )
+
+        if memory_id:
+            result = [it for it in result if it.spec.id == memory_id]
+        return [it for it in result if it.spec.end_user_visible]
+
+
+class MemoryEditApi(Resource):
+    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))
+    def put(self, app_model: App, end_user: EndUser):
+        parser = reqparse.RequestParser()
+        parser.add_argument("id", type=str, required=True)
+        parser.add_argument("conversation_id", type=str | None, required=False, default=None)
+        parser.add_argument("node_id", type=str | None, required=False, default=None)
+        parser.add_argument("update", type=str, required=True)
+        args = parser.parse_args()
+        workflow = WorkflowService().get_published_workflow(app_model)
+        update = args.get("update")
+        conversation_id = args.get("conversation_id")
+        node_id = args.get("node_id")
+        if not isinstance(update, str):
+            return {"error": "Invalid update"}, 400
+        if not workflow:
+            return {"error": "Workflow not found"}, 404
+        memory_spec = next((it for it in workflow.memory_blocks if it.id == args["id"]), None)
+        if not memory_spec:
+            return {"error": "Memory not found"}, 404
+
+        # First get existing memory
+        existing_memory = ChatflowMemoryService.get_memory_by_spec(
+            spec=memory_spec,
+            tenant_id=app_model.tenant_id,
+            app_id=app_model.id,
+            created_by=MemoryCreatedBy(end_user_id=end_user.id),
+            conversation_id=conversation_id,
+            node_id=node_id,
+            is_draft=False,
+        )
+
+        # Create updated memory instance with incremented version
+        updated_memory = MemoryBlock(
+            spec=existing_memory.spec,
+            tenant_id=existing_memory.tenant_id,
+            app_id=existing_memory.app_id,
+            conversation_id=existing_memory.conversation_id,
+            node_id=existing_memory.node_id,
+            value=update,  # New value
+            version=existing_memory.version + 1,  # Increment version for update
+            edited_by_user=True,
+            created_by=existing_memory.created_by,
+        )
+
+        ChatflowMemoryService.save_memory(updated_memory, VariablePool(), False)
+        return "", 204
+
+
+class MemoryDeleteApi(Resource):
+    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))
+    def delete(self, app_model: App, end_user: EndUser):
+        parser = reqparse.RequestParser()
+        parser.add_argument("id", type=str, required=False, default=None)
+        args = parser.parse_args()
+        memory_id = args.get("id")
+
+        if memory_id:
+            ChatflowMemoryService.delete_memory(app_model, memory_id, MemoryCreatedBy(end_user_id=end_user.id))
+            return "", 204
+        else:
+            ChatflowMemoryService.delete_all_user_memories(app_model, MemoryCreatedBy(end_user_id=end_user.id))
+            return "", 200
+
+
+api.add_resource(MemoryListApi, "/memories")
+api.add_resource(MemoryEditApi, "/memory-edit")
+api.add_resource(MemoryDeleteApi, "/memories")

api/controllers/service_api/dataset/document.py

@@ -592,7 +592,7 @@ class DocumentApi(DatasetApiResource):
                 "name": document.name,
                 "created_from": document.created_from,
                 "created_by": document.created_by,
-                "created_at": document.created_at.timestamp(),
+                "created_at": int(document.created_at.timestamp()),
                 "tokens": document.tokens,
                 "indexing_status": document.indexing_status,
                 "completed_at": int(document.completed_at.timestamp()) if document.completed_at else None,
@@ -625,7 +625,7 @@ class DocumentApi(DatasetApiResource):
                 "name": document.name,
                 "created_from": document.created_from,
                 "created_by": document.created_by,
-                "created_at": document.created_at.timestamp(),
+                "created_at": int(document.created_at.timestamp()),
                 "tokens": document.tokens,
                 "indexing_status": document.indexing_status,
                 "completed_at": int(document.completed_at.timestamp()) if document.completed_at else None,

api/controllers/service_api/dataset/segment.py

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

api/controllers/service_api/wraps.py

@@ -13,6 +13,7 @@ from sqlalchemy import select, update
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden, NotFound, Unauthorized
 
+from enums.cloud_plan import CloudPlan
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from libs.datetime_utils import naive_utc_now
@@ -66,6 +67,7 @@ def validate_app_token(view: Callable[P, R] | None = None, *, fetch_user_arg: Fe
 
             kwargs["app_model"] = app_model
 
+            # If caller needs end-user context, attach EndUser to current_user
             if fetch_user_arg:
                 if fetch_user_arg.fetch_from == WhereisUserArg.QUERY:
                     user_id = request.args.get("user")
@@ -74,7 +76,6 @@ def validate_app_token(view: Callable[P, R] | None = None, *, fetch_user_arg: Fe
                 elif fetch_user_arg.fetch_from == WhereisUserArg.FORM:
                     user_id = request.form.get("user")
                 else:
-                    # use default-user
                     user_id = None
 
                 if not user_id and fetch_user_arg.required:
@@ -89,6 +90,28 @@ def validate_app_token(view: Callable[P, R] | None = None, *, fetch_user_arg: Fe
                 # Set EndUser as current logged-in user for flask_login.current_user
                 current_app.login_manager._update_request_context_with_user(end_user)  # type: ignore
                 user_logged_in.send(current_app._get_current_object(), user=end_user)  # type: ignore
+            else:
+                # For service API without end-user context, ensure an Account is logged in
+                # so services relying on current_account_with_tenant() work correctly.
+                tenant_owner_info = (
+                    db.session.query(Tenant, Account)
+                    .join(TenantAccountJoin, Tenant.id == TenantAccountJoin.tenant_id)
+                    .join(Account, TenantAccountJoin.account_id == Account.id)
+                    .where(
+                        Tenant.id == app_model.tenant_id,
+                        TenantAccountJoin.role == "owner",
+                        Tenant.status == TenantStatus.NORMAL,
+                    )
+                    .one_or_none()
+                )
+
+                if tenant_owner_info:
+                    tenant_model, account = tenant_owner_info
+                    account.current_tenant = tenant_model
+                    current_app.login_manager._update_request_context_with_user(account)  # type: ignore
+                    user_logged_in.send(current_app._get_current_object(), user=current_user)  # type: ignore
+                else:
+                    raise Unauthorized("Tenant owner account not found or tenant is not active.")
 
             return view_func(*args, **kwargs)
 
@@ -138,7 +161,7 @@ def cloud_edition_billing_knowledge_limit_check(resource: str, api_token_type: s
             features = FeatureService.get_features(api_token.tenant_id)
             if features.billing.enabled:
                 if resource == "add_segment":
-                    if features.billing.subscription.plan == "sandbox":
+                    if features.billing.subscription.plan == CloudPlan.SANDBOX:
                         raise Forbidden(
                             "To unlock this feature and elevate your Dify experience, please upgrade to a paid plan."
                         )

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,108 @@
+from flask_restx import reqparse
+
+from controllers.web import api
+from controllers.web.wraps import WebApiResource
+from core.memory.entities import MemoryBlock, MemoryCreatedBy
+from core.workflow.runtime.variable_pool import VariablePool
+from models import App, EndUser
+from services.chatflow_memory_service import ChatflowMemoryService
+from services.workflow_service import WorkflowService
+
+
+class MemoryListApi(WebApiResource):
+    def get(self, app_model: App, end_user: EndUser):
+        parser = reqparse.RequestParser()
+        parser.add_argument("conversation_id", required=False, type=str | None, default=None)
+        parser.add_argument("memory_id", required=False, type=str | None, default=None)
+        parser.add_argument("version", required=False, type=int | None, default=None)
+        args = parser.parse_args()
+        conversation_id: str | None = args.get("conversation_id")
+        memory_id = args.get("memory_id")
+        version = args.get("version")
+
+        if conversation_id:
+            result = ChatflowMemoryService.get_persistent_memories_with_conversation(
+                app_model, MemoryCreatedBy(end_user_id=end_user.id), conversation_id, version
+            )
+            session_memories = ChatflowMemoryService.get_session_memories_with_conversation(
+                app_model, MemoryCreatedBy(end_user_id=end_user.id), conversation_id, version
+            )
+            result = [*result, *session_memories]
+        else:
+            result = ChatflowMemoryService.get_persistent_memories(
+                app_model, MemoryCreatedBy(end_user_id=end_user.id), version
+            )
+
+        if memory_id:
+            result = [it for it in result if it.spec.id == memory_id]
+        return [it for it in result if it.spec.end_user_visible]
+
+
+class MemoryEditApi(WebApiResource):
+    def put(self, app_model: App, end_user: EndUser):
+        parser = reqparse.RequestParser()
+        parser.add_argument("id", type=str, required=True)
+        parser.add_argument("conversation_id", type=str | None, required=False, default=None)
+        parser.add_argument("node_id", type=str | None, required=False, default=None)
+        parser.add_argument("update", type=str, required=True)
+        args = parser.parse_args()
+        workflow = WorkflowService().get_published_workflow(app_model)
+        update = args.get("update")
+        conversation_id = args.get("conversation_id")
+        node_id = args.get("node_id")
+        if not isinstance(update, str):
+            return {"error": "Update must be a string"}, 400
+        if not workflow:
+            return {"error": "Workflow not found"}, 404
+        memory_spec = next((it for it in workflow.memory_blocks if it.id == args["id"]), None)
+        if not memory_spec:
+            return {"error": "Memory not found"}, 404
+        if not memory_spec.end_user_editable:
+            return {"error": "Memory not editable"}, 403
+
+        # First get existing memory
+        existing_memory = ChatflowMemoryService.get_memory_by_spec(
+            spec=memory_spec,
+            tenant_id=app_model.tenant_id,
+            app_id=app_model.id,
+            created_by=MemoryCreatedBy(end_user_id=end_user.id),
+            conversation_id=conversation_id,
+            node_id=node_id,
+            is_draft=False,
+        )
+
+        # Create updated memory instance with incremented version
+        updated_memory = MemoryBlock(
+            spec=existing_memory.spec,
+            tenant_id=existing_memory.tenant_id,
+            app_id=existing_memory.app_id,
+            conversation_id=existing_memory.conversation_id,
+            node_id=existing_memory.node_id,
+            value=update,  # New value
+            version=existing_memory.version + 1,  # Increment version for update
+            edited_by_user=True,
+            created_by=existing_memory.created_by,
+        )
+
+        ChatflowMemoryService.save_memory(updated_memory, VariablePool(), False)
+        return "", 204
+
+
+class MemoryDeleteApi(WebApiResource):
+    def delete(self, app_model: App, end_user: EndUser):
+        parser = reqparse.RequestParser()
+        parser.add_argument("id", type=str, required=False, default=None)
+        args = parser.parse_args()
+        memory_id = args.get("id")
+
+        if memory_id:
+            ChatflowMemoryService.delete_memory(app_model, memory_id, MemoryCreatedBy(end_user_id=end_user.id))
+            return "", 204
+        else:
+            ChatflowMemoryService.delete_all_user_memories(app_model, MemoryCreatedBy(end_user_id=end_user.id))
+            return "", 200
+
+
+api.add_resource(MemoryListApi, "/memories")
+api.add_resource(MemoryEditApi, "/memory-edit")
+api.add_resource(MemoryDeleteApi, "/memories")

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

@@ -1,10 +1,11 @@
 import logging
 import time
-from collections.abc import Mapping, Sequence
+from collections.abc import Mapping, MutableMapping, Sequence
 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,6 +21,8 @@ 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
@@ -27,6 +30,7 @@ from core.workflow.enums import WorkflowType
 from core.workflow.graph_engine.command_channels.redis_channel import RedisChannel
 from core.workflow.graph_engine.layers.base import GraphEngineLayer
 from core.workflow.graph_engine.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
+from core.workflow.graph_events import GraphRunSucceededEvent
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 from core.workflow.runtime import GraphRuntimeState, VariablePool
@@ -39,6 +43,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__)
 
@@ -81,6 +87,10 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
         self._workflow_node_execution_repository = workflow_node_execution_repository
 
     def run(self):
+        ChatflowMemoryService.wait_for_sync_memory_completion(
+            workflow=self._workflow, conversation_id=self.conversation.id
+        )
+
         app_config = self.application_generate_entity.app_config
         app_config = cast(AdvancedChatAppConfig, app_config)
 
@@ -143,6 +153,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
@@ -206,6 +217,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,
@@ -403,3 +439,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_runner.py

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

api/core/app/apps/base_app_runner.py

@@ -79,7 +79,7 @@ class AppRunner:
         prompt_template_entity: PromptTemplateEntity,
         inputs: Mapping[str, str],
         files: Sequence["File"],
-        query: str | None = None,
+        query: str = "",
         context: str | None = None,
         memory: TokenBufferMemory | None = None,
         image_detail_config: ImagePromptMessageContent.DETAIL | None = None,
@@ -105,7 +105,7 @@ class AppRunner:
                 app_mode=AppMode.value_of(app_record.mode),
                 prompt_template_entity=prompt_template_entity,
                 inputs=inputs,
-                query=query or "",
+                query=query,
                 files=files,
                 context=context,
                 memory=memory,

api/core/app/apps/message_based_app_generator.py

@@ -190,7 +190,7 @@ class MessageBasedAppGenerator(BaseAppGenerator):
             override_model_configs=json.dumps(override_model_configs) if override_model_configs else None,
             conversation_id=conversation.id,
             inputs=application_generate_entity.inputs,
-            query=application_generate_entity.query or "",
+            query=application_generate_entity.query,
             message="",
             message_tokens=0,
             message_unit_price=0,

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

@@ -40,19 +40,15 @@ from core.workflow.repositories.draft_variable_repository import DraftVariableSa
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 from core.workflow.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader
+from enums.cloud_plan import CloudPlan
 from extensions.ext_database import db
-from extensions.ext_redis import redis_client
 from libs.flask_utils import preserve_flask_contexts
 from models import Account, EndUser, Workflow, WorkflowNodeExecutionTriggeredFrom
 from models.dataset import Document, DocumentPipelineExecutionLog, Pipeline
 from models.enums import WorkflowRunTriggeredFrom
 from models.model import AppMode
 from services.datasource_provider_service import DatasourceProviderService
-from services.feature_service import FeatureService
-from services.file_service import FileService
 from services.workflow_draft_variable_service import DraftVarLoader, WorkflowDraftVariableService
-from tasks.rag_pipeline.priority_rag_pipeline_run_task import priority_rag_pipeline_run_task
-from tasks.rag_pipeline.rag_pipeline_run_task import rag_pipeline_run_task
 
 logger = logging.getLogger(__name__)
 
@@ -255,7 +251,7 @@ class PipelineGenerator(BaseAppGenerator):
             json_text = json.dumps(text)
             upload_file = FileService(db.engine).upload_text(json_text, name, user.id, dataset.tenant_id)
             features = FeatureService.get_features(dataset.tenant_id)
-            if features.billing.enabled and features.billing.subscription.plan == "sandbox":
+            if features.billing.enabled and features.billing.subscription.plan == CloudPlan.SANDBOX:
                 tenant_pipeline_task_key = f"tenant_pipeline_task:{dataset.tenant_id}"
                 tenant_self_pipeline_task_queue = f"tenant_self_pipeline_task_queue:{dataset.tenant_id}"
 

api/core/app/entities/app_invoke_entities.py

@@ -129,7 +129,7 @@ class EasyUIBasedAppGenerateEntity(AppGenerateEntity):
     app_config: EasyUIBasedAppConfig = None  # type: ignore
     model_conf: ModelConfigWithCredentialsEntity
 
-    query: str | None = None
+    query: str = ""
 
     # pydantic configs
     model_config = ConfigDict(protected_namespaces=())

api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py

@@ -121,7 +121,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
         if self._application_generate_entity.app_config.app_mode != AppMode.COMPLETION:
             # start generate conversation name thread
             self._conversation_name_generate_thread = self._message_cycle_manager.generate_conversation_name(
-                conversation_id=self._conversation_id, query=self._application_generate_entity.query or ""
+                conversation_id=self._conversation_id, query=self._application_generate_entity.query
             )
 
         generator = self._wrapper_process_stream_response(trace_manager=self._application_generate_entity.trace_manager)

api/core/app/task_pipeline/message_cycle_manager.py

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

api/core/entities/document_task.py

@@ -0,0 +1,13 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class DocumentTask:
+    """Document task entity for document indexing operations.
+    
+    This class represents a document indexing task that can be queued
+    and processed by the document indexing system.
+    """
+    tenant_id: str
+    dataset_id: str
+    document_ids: list[str]

api/core/entities/mcp_provider.py

@@ -14,6 +14,7 @@ from core.helper.provider_cache import NoOpProviderCredentialCache
 from core.mcp.types import OAuthClientInformation, OAuthClientMetadata, OAuthTokens
 from core.tools.entities.common_entities import I18nObject
 from core.tools.entities.tool_entities import ToolProviderType
+from core.tools.utils.encryption import create_provider_encrypter
 
 if TYPE_CHECKING:
     from models.tools import MCPToolProvider
@@ -271,7 +272,6 @@ class MCPProviderEntity(BaseModel):
 
     def _decrypt_dict(self, data: dict[str, Any]) -> dict[str, Any]:
         """Generic method to decrypt dictionary fields"""
-        from core.tools.utils.encryption import create_provider_encrypter
 
         if not data:
             return {}

api/core/file/models.py

@@ -74,6 +74,10 @@ class File(BaseModel):
         storage_key: str | None = None,
         dify_model_identity: str | None = FILE_MODEL_IDENTITY,
         url: str | None = None,
+        # Legacy compatibility fields - explicitly handle known extra fields
+        tool_file_id: str | None = None,
+        upload_file_id: str | None = None,
+        datasource_file_id: str | None = None,
     ):
         super().__init__(
             id=id,

api/core/helper/code_executor/javascript/javascript_transformer.py

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

api/core/helper/code_executor/python3/python3_transformer.py

@@ -6,9 +6,7 @@ from core.helper.code_executor.template_transformer import TemplateTransformer
 class Python3TemplateTransformer(TemplateTransformer):
     @classmethod
     def get_runner_script(cls) -> str:
-        runner_script = dedent(f"""
-            # declare main function
-            {cls._code_placeholder}
+        runner_script = dedent(f"""            {cls._code_placeholder}
 
             import json
             from base64 import b64decode

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/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
@@ -28,6 +30,7 @@ from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
 from core.ops.utils import measure_time
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
 from core.workflow.entities.workflow_node_execution import WorkflowNodeExecutionMetadataKey
+from core.workflow.runtime.variable_pool import VariablePool
 from extensions.ext_database import db
 from extensions.ext_storage import storage
 from models import App, Message, WorkflowNodeExecutionModel
@@ -562,3 +565,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/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/ops/tencent_trace/client.py

@@ -5,6 +5,7 @@ Tencent APM Trace Client - handles network operations, metrics, and API communic
 from __future__ import annotations
 
 import importlib
+import json
 import logging
 import os
 import socket
@@ -110,6 +111,7 @@ class TencentTraceClient:
         self.span_contexts: dict[int, trace_api.SpanContext] = {}
 
         self.meter: Meter | None = None
+        self.meter_provider: MeterProvider | None = None
         self.hist_llm_duration: Histogram | None = None
         self.hist_token_usage: Histogram | None = None
         self.hist_time_to_first_token: Histogram | None = None
@@ -119,7 +121,6 @@ class TencentTraceClient:
 
         # Metrics exporter and instruments
         try:
-            from opentelemetry import metrics
             from opentelemetry.sdk.metrics import Histogram, MeterProvider
             from opentelemetry.sdk.metrics.export import AggregationTemporality, PeriodicExportingMetricReader
 
@@ -202,9 +203,11 @@ class TencentTraceClient:
                 )
 
             if metric_reader is not None:
+                # Use instance-level MeterProvider instead of global to support config changes
+                # without worker restart. Each TencentTraceClient manages its own MeterProvider.
                 provider = MeterProvider(resource=self.resource, metric_readers=[metric_reader])
-                metrics.set_meter_provider(provider)
-                self.meter = metrics.get_meter("dify-sdk", dify_config.project.version)
+                self.meter_provider = provider
+                self.meter = provider.get_meter("dify-sdk", dify_config.project.version)
 
                 # LLM operation duration histogram
                 self.hist_llm_duration = self.meter.create_histogram(
@@ -244,6 +247,7 @@ class TencentTraceClient:
                 self.metric_reader = metric_reader
             else:
                 self.meter = None
+                self.meter_provider = None
                 self.hist_llm_duration = None
                 self.hist_token_usage = None
                 self.hist_time_to_first_token = None
@@ -253,6 +257,7 @@ class TencentTraceClient:
         except Exception:
             logger.exception("[Tencent APM] Metrics initialization failed; metrics disabled")
             self.meter = None
+            self.meter_provider = None
             self.hist_llm_duration = None
             self.hist_token_usage = None
             self.hist_time_to_first_token = None
@@ -279,6 +284,14 @@ class TencentTraceClient:
             if attributes:
                 for k, v in attributes.items():
                     attrs[k] = str(v) if not isinstance(v, (str, int, float, bool)) else v  # type: ignore[assignment]
+
+            logger.info(
+                "[Tencent Metrics] Metric: %s | Value: %.4f | Attributes: %s",
+                LLM_OPERATION_DURATION,
+                latency_seconds,
+                json.dumps(attrs, ensure_ascii=False),
+            )
+
             self.hist_llm_duration.record(latency_seconds, attrs)  # type: ignore[attr-defined]
         except Exception:
             logger.debug("[Tencent APM] Failed to record LLM duration", exc_info=True)
@@ -317,6 +330,13 @@ class TencentTraceClient:
                 "server.address": server_address,
             }
 
+            logger.info(
+                "[Tencent Metrics] Metric: %s | Value: %d | Attributes: %s",
+                GEN_AI_TOKEN_USAGE,
+                token_count,
+                json.dumps(attributes, ensure_ascii=False),
+            )
+
             self.hist_token_usage.record(token_count, attributes)  # type: ignore[attr-defined]
         except Exception:
             logger.debug("[Tencent APM] Failed to record token usage", exc_info=True)
@@ -344,6 +364,13 @@ class TencentTraceClient:
                 "stream": "true",
             }
 
+            logger.info(
+                "[Tencent Metrics] Metric: %s | Value: %.4f | Attributes: %s",
+                GEN_AI_SERVER_TIME_TO_FIRST_TOKEN,
+                ttft_seconds,
+                json.dumps(attributes, ensure_ascii=False),
+            )
+
             self.hist_time_to_first_token.record(ttft_seconds, attributes)  # type: ignore[attr-defined]
         except Exception:
             logger.debug("[Tencent APM] Failed to record time to first token", exc_info=True)
@@ -371,6 +398,13 @@ class TencentTraceClient:
                 "stream": "true",
             }
 
+            logger.info(
+                "[Tencent Metrics] Metric: %s | Value: %.4f | Attributes: %s",
+                GEN_AI_STREAMING_TIME_TO_GENERATE,
+                ttg_seconds,
+                json.dumps(attributes, ensure_ascii=False),
+            )
+
             self.hist_time_to_generate.record(ttg_seconds, attributes)  # type: ignore[attr-defined]
         except Exception:
             logger.debug("[Tencent APM] Failed to record time to generate", exc_info=True)
@@ -390,6 +424,14 @@ class TencentTraceClient:
             if attributes:
                 for k, v in attributes.items():
                     attrs[k] = str(v) if not isinstance(v, (str, int, float, bool)) else v  # type: ignore[assignment]
+
+            logger.info(
+                "[Tencent Metrics] Metric: %s | Value: %.4f | Attributes: %s",
+                GEN_AI_TRACE_DURATION,
+                duration_seconds,
+                json.dumps(attrs, ensure_ascii=False),
+            )
+
             self.hist_trace_duration.record(duration_seconds, attrs)  # type: ignore[attr-defined]
         except Exception:
             logger.debug("[Tencent APM] Failed to record trace duration", exc_info=True)
@@ -474,11 +516,19 @@ class TencentTraceClient:
 
             if self.tracer_provider:
                 self.tracer_provider.shutdown()
+
+            # Shutdown instance-level meter provider
+            if self.meter_provider is not None:
+                try:
+                    self.meter_provider.shutdown()  # type: ignore[attr-defined]
+                except Exception:
+                    logger.debug("[Tencent APM] Error shutting down meter provider", exc_info=True)
+
             if self.metric_reader is not None:
                 try:
                     self.metric_reader.shutdown()  # type: ignore[attr-defined]
                 except Exception:
-                    pass
+                    logger.debug("[Tencent APM] Error shutting down metric reader", exc_info=True)
 
         except Exception:
             logger.exception("[Tencent APM] Error during client shutdown")

api/core/prompt/entities/advanced_prompt_entities.py

@@ -45,6 +45,12 @@ class MemoryConfig(BaseModel):
         enabled: bool
         size: int | None = None
 
+    mode: Literal["linear", "block"] | None = "linear"
+    block_id: list[str] | None = None
     role_prefix: RolePrefix | None = None
     window: WindowConfig
     query_prompt_template: str | None = None
+
+    @property
+    def is_block_mode(self) -> bool:
+        return self.mode == "block" and bool(self.block_id)

api/core/provider_manager.py

@@ -618,18 +618,18 @@ class ProviderManager:
                 )
 
             for quota in configuration.quotas:
-                if quota.quota_type == ProviderQuotaType.TRIAL:
+                if quota.quota_type in (ProviderQuotaType.TRIAL, ProviderQuotaType.PAID):
                     # Init trial provider records if not exists
-                    if ProviderQuotaType.TRIAL not in provider_quota_to_provider_record_dict:
+                    if quota.quota_type not in provider_quota_to_provider_record_dict:
                         try:
                             # FIXME ignore the type error, only TrialHostingQuota has limit need to change the logic
                             new_provider_record = Provider(
                                 tenant_id=tenant_id,
                                 # TODO: Use provider name with prefix after the data migration.
                                 provider_name=ModelProviderID(provider_name).provider_name,
-                                provider_type=ProviderType.SYSTEM,
-                                quota_type=ProviderQuotaType.TRIAL,
-                                quota_limit=quota.quota_limit,  # type: ignore
+                                provider_type=ProviderType.SYSTEM.value,
+                                quota_type=quota.quota_type,
+                                quota_limit=0,  # type: ignore
                                 quota_used=0,
                                 is_valid=True,
                             )
@@ -641,8 +641,8 @@ class ProviderManager:
                             stmt = select(Provider).where(
                                 Provider.tenant_id == tenant_id,
                                 Provider.provider_name == ModelProviderID(provider_name).provider_name,
-                                Provider.provider_type == ProviderType.SYSTEM,
-                                Provider.quota_type == ProviderQuotaType.TRIAL,
+                                Provider.provider_type == ProviderType.SYSTEM.value,
+                                Provider.quota_type == quota.quota_type,
                             )
                             existed_provider_record = db.session.scalar(stmt)
                             if not existed_provider_record:
@@ -652,7 +652,7 @@ class ProviderManager:
                                 existed_provider_record.is_valid = True
                                 db.session.commit()
 
-                            provider_name_to_provider_records_dict[provider_name].append(existed_provider_record)
+                            provider_name_to_provider_records_dict[provider_name].append(existed_provider_record)                        
 
         return provider_name_to_provider_records_dict
 
@@ -912,6 +912,22 @@ class ProviderManager:
                 provider_record
             )
         quota_configurations = []
+
+        if dify_config.EDITION == "CLOUD":
+            from services.credit_pool_service import CreditPoolService
+
+            trail_pool = CreditPoolService.get_pool(
+                tenant_id=tenant_id,
+                pool_type=ProviderQuotaType.TRIAL.value,
+            )
+            paid_pool = CreditPoolService.get_pool(
+                tenant_id=tenant_id,
+                pool_type=ProviderQuotaType.PAID.value,
+            )
+        else:
+            trail_pool = None
+            paid_pool = None
+
         for provider_quota in provider_hosting_configuration.quotas:
             if provider_quota.quota_type not in quota_type_to_provider_records_dict:
                 if provider_quota.quota_type == ProviderQuotaType.FREE:
@@ -932,16 +948,36 @@ class ProviderManager:
                     raise ValueError("quota_used is None")
                 if provider_record.quota_limit is None:
                     raise ValueError("quota_limit is None")
+                if provider_quota.quota_type == ProviderQuotaType.TRIAL and trail_pool is not None:
+                    quota_configuration = QuotaConfiguration(
+                        quota_type=provider_quota.quota_type,
+                        quota_unit=provider_hosting_configuration.quota_unit or QuotaUnit.TOKENS,
+                        quota_used=trail_pool.quota_used,
+                        quota_limit=trail_pool.quota_limit,
+                        is_valid=trail_pool.quota_limit > trail_pool.quota_used or trail_pool.quota_limit == -1,
+                        restrict_models=provider_quota.restrict_models,
+                    )
 
-                quota_configuration = QuotaConfiguration(
-                    quota_type=provider_quota.quota_type,
-                    quota_unit=provider_hosting_configuration.quota_unit or QuotaUnit.TOKENS,
-                    quota_used=provider_record.quota_used,
-                    quota_limit=provider_record.quota_limit,
-                    is_valid=provider_record.quota_limit > provider_record.quota_used
-                    or provider_record.quota_limit == -1,
-                    restrict_models=provider_quota.restrict_models,
-                )
+                elif provider_quota.quota_type == ProviderQuotaType.PAID and paid_pool is not None:
+                    quota_configuration = QuotaConfiguration(
+                        quota_type=provider_quota.quota_type,
+                        quota_unit=provider_hosting_configuration.quota_unit or QuotaUnit.TOKENS,
+                        quota_used=paid_pool.quota_used,
+                        quota_limit=paid_pool.quota_limit,
+                        is_valid=paid_pool.quota_limit > paid_pool.quota_used or paid_pool.quota_limit == -1,
+                        restrict_models=provider_quota.restrict_models,
+                    )
+
+                else:
+                    quota_configuration = QuotaConfiguration(
+                        quota_type=provider_quota.quota_type,
+                        quota_unit=provider_hosting_configuration.quota_unit or QuotaUnit.TOKENS,
+                        quota_used=provider_record.quota_used,
+                        quota_limit=provider_record.quota_limit,
+                        is_valid=provider_record.quota_limit > provider_record.quota_used
+                        or provider_record.quota_limit == -1,
+                        restrict_models=provider_quota.restrict_models,
+                    )
 
             quota_configurations.append(quota_configuration)
 

api/core/rag/datasource/vdb/oracle/oraclevector.py

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

api/core/rag/datasource/vdb/pyvastbase/vastbase_vector.py

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

api/core/rag/datasource/vdb/weaviate/weaviate_vector.py

@@ -39,11 +39,13 @@ class WeaviateConfig(BaseModel):
 
     Attributes:
         endpoint: Weaviate server endpoint URL
+        grpc_endpoint: Optional Weaviate gRPC server endpoint URL
         api_key: Optional API key for authentication
         batch_size: Number of objects to batch per insert operation
     """
 
     endpoint: str
+    grpc_endpoint: str | None = None
     api_key: str | None = None
     batch_size: int = 100
 
@@ -88,9 +90,22 @@ class WeaviateVector(BaseVector):
         http_secure = p.scheme == "https"
         http_port = p.port or (443 if http_secure else 80)
 
-        grpc_host = host
-        grpc_secure = http_secure
-        grpc_port = 443 if grpc_secure else 50051
+        # Parse gRPC configuration
+        if config.grpc_endpoint:
+            # Urls without scheme won't be parsed correctly in some python verions,
+            # see https://bugs.python.org/issue27657
+            grpc_endpoint_with_scheme = (
+                config.grpc_endpoint if "://" in config.grpc_endpoint else f"grpc://{config.grpc_endpoint}"
+            )
+            grpc_p = urlparse(grpc_endpoint_with_scheme)
+            grpc_host = grpc_p.hostname or "localhost"
+            grpc_port = grpc_p.port or (443 if grpc_p.scheme == "grpcs" else 50051)
+            grpc_secure = grpc_p.scheme == "grpcs"
+        else:
+            # Infer from HTTP endpoint as fallback
+            grpc_host = host
+            grpc_secure = http_secure
+            grpc_port = 443 if grpc_secure else 50051
 
         client = weaviate.connect_to_custom(
             http_host=host,
@@ -432,6 +447,7 @@ class WeaviateVectorFactory(AbstractVectorFactory):
             collection_name=collection_name,
             config=WeaviateConfig(
                 endpoint=dify_config.WEAVIATE_ENDPOINT or "",
+                grpc_endpoint=dify_config.WEAVIATE_GRPC_ENDPOINT or "",
                 api_key=dify_config.WEAVIATE_API_KEY,
                 batch_size=dify_config.WEAVIATE_BATCH_SIZE,
             ),

api/core/rag/pipeline/queue.py

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

api/core/tools/__base/tool.py

@@ -210,12 +210,13 @@ class Tool(ABC):
             meta=meta,
         )
 
-    def create_json_message(self, object: dict) -> ToolInvokeMessage:
+    def create_json_message(self, object: dict, suppress_output: bool = False) -> ToolInvokeMessage:
         """
         create a json message
         """
         return ToolInvokeMessage(
-            type=ToolInvokeMessage.MessageType.JSON, message=ToolInvokeMessage.JsonMessage(json_object=object)
+            type=ToolInvokeMessage.MessageType.JSON,
+            message=ToolInvokeMessage.JsonMessage(json_object=object, suppress_output=suppress_output),
         )
 
     def create_variable_message(

api/core/tools/entities/tool_entities.py

@@ -129,6 +129,7 @@ class ToolInvokeMessage(BaseModel):
 
     class JsonMessage(BaseModel):
         json_object: dict
+        suppress_output: bool = Field(default=False, description="Whether to suppress JSON output in result string")
 
     class BlobMessage(BaseModel):
         blob: bytes

api/core/tools/tool_engine.py

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

api/core/tools/workflow_as_tool/tool.py

@@ -117,7 +117,7 @@ class WorkflowTool(Tool):
         self._latest_usage = self._derive_usage_from_result(data)
 
         yield self.create_text_message(json.dumps(outputs, ensure_ascii=False))
-        yield self.create_json_message(outputs)
+        yield self.create_json_message(outputs, suppress_output=True)
 
     @property
     def latest_usage(self) -> LLMUsage:

api/core/variables/segments.py

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

api/core/variables/types.py

@@ -41,6 +41,8 @@ class SegmentType(StrEnum):
     ARRAY_FILE = "array[file]"
     ARRAY_BOOLEAN = "array[boolean]"
 
+    VERSIONED_MEMORY = "versioned_memory"
+
     NONE = "none"
 
     GROUP = "group"

api/core/variables/variables.py

@@ -22,6 +22,7 @@ from .segments import (
     ObjectSegment,
     Segment,
     StringSegment,
+    VersionedMemorySegment,
     get_segment_discriminator,
 )
 from .types import SegmentType
@@ -106,6 +107,10 @@ class ArrayFileVariable(ArrayFileSegment, ArrayVariable):
     pass
 
 
+class VersionedMemoryVariable(VersionedMemorySegment, Variable):
+    pass
+
+
 class ArrayBooleanVariable(ArrayBooleanSegment, ArrayVariable):
     pass
 
@@ -161,6 +166,7 @@ VariableUnion: TypeAlias = Annotated[
         | Annotated[ArrayFileVariable, Tag(SegmentType.ARRAY_FILE)]
         | Annotated[ArrayBooleanVariable, Tag(SegmentType.ARRAY_BOOLEAN)]
         | Annotated[SecretVariable, Tag(SegmentType.SECRET)]
+        | Annotated[VersionedMemoryVariable, Tag(SegmentType.VERSIONED_MEMORY)]
     ),
     Discriminator(get_segment_discriminator),
 ]

api/core/workflow/constants.py

@@ -1,4 +1,5 @@
 SYSTEM_VARIABLE_NODE_ID = "sys"
 ENVIRONMENT_VARIABLE_NODE_ID = "env"
 CONVERSATION_VARIABLE_NODE_ID = "conversation"
+MEMORY_BLOCK_VARIABLE_NODE_ID = "memory_block"
 RAG_PIPELINE_VARIABLE_NODE_ID = "rag"

api/core/workflow/nodes/llm/llm_utils.py

@@ -6,7 +6,7 @@ from sqlalchemy.orm import Session
 
 from configs import dify_config
 from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
-from core.entities.provider_entities import QuotaUnit
+from core.entities.provider_entities import ProviderQuotaType, QuotaUnit
 from core.file.models import File
 from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance, ModelManager
@@ -136,21 +136,36 @@ def deduct_llm_quota(tenant_id: str, model_instance: ModelInstance, usage: LLMUs
             used_quota = 1
 
     if used_quota is not None and system_configuration.current_quota_type is not None:
-        with Session(db.engine) as session:
-            stmt = (
-                update(Provider)
-                .where(
-                    Provider.tenant_id == tenant_id,
-                    # TODO: Use provider name with prefix after the data migration.
-                    Provider.provider_name == ModelProviderID(model_instance.provider).provider_name,
-                    Provider.provider_type == ProviderType.SYSTEM,
-                    Provider.quota_type == system_configuration.current_quota_type.value,
-                    Provider.quota_limit > Provider.quota_used,
-                )
-                .values(
-                    quota_used=Provider.quota_used + used_quota,
-                    last_used=naive_utc_now(),
-                )
+        if system_configuration.current_quota_type == ProviderQuotaType.TRIAL:
+            from services.credit_pool_service import CreditPoolService
+
+            CreditPoolService.check_and_deduct_credits(
+                tenant_id=tenant_id,
+                credits_required=used_quota,
             )
-            session.execute(stmt)
-            session.commit()
+        elif system_configuration.current_quota_type == ProviderQuotaType.PAID:
+            from services.credit_pool_service import CreditPoolService
+            CreditPoolService.check_and_deduct_credits(
+                tenant_id=tenant_id,
+                credits_required=used_quota,
+                pool_type="paid",
+            )
+        else:
+            with Session(db.engine) as session:
+                stmt = (
+                    update(Provider)
+                    .where(
+                        Provider.tenant_id == tenant_id,
+                        # TODO: Use provider name with prefix after the data migration.
+                        Provider.provider_name == ModelProviderID(model_instance.provider).provider_name,
+                        Provider.provider_type == ProviderType.SYSTEM.value,
+                        Provider.quota_type == system_configuration.current_quota_type.value,
+                        Provider.quota_limit > Provider.quota_used,
+                    )
+                    .values(
+                        quota_used=Provider.quota_used + used_quota,
+                        last_used=naive_utc_now(),
+                    )
+                )
+                session.execute(stmt)
+                session.commit()

api/core/workflow/nodes/llm/node.py

@@ -7,11 +7,15 @@ import time
 from collections.abc import Generator, Mapping, Sequence
 from typing import TYPE_CHECKING, Any, Literal
 
-from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from core.app.entities.app_invoke_entities import InvokeFrom, ModelConfigWithCredentialsEntity
 from core.file import FileType, file_manager
 from core.helper.code_executor import CodeExecutor, CodeLanguage
 from core.llm_generator.output_parser.errors import OutputParserError
 from core.llm_generator.output_parser.structured_output import invoke_llm_with_structured_output
+from core.memory.entities import MemoryCreatedBy, MemoryScope
 from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance, ModelManager
 from core.model_runtime.entities import (
@@ -73,6 +77,8 @@ from core.workflow.nodes.base.entities import BaseNodeData, RetryConfig, Variabl
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.base.variable_template_parser import VariableTemplateParser
 from core.workflow.runtime import VariablePool
+from models import UserFrom, Workflow
+from models.engine import db
 
 from . import llm_utils
 from .entities import (
@@ -317,6 +323,11 @@ class LLMNode(Node):
             if self._file_outputs:
                 outputs["files"] = ArrayFileSegment(value=self._file_outputs)
 
+            try:
+                self._handle_chatflow_memory(result_text, variable_pool)
+            except Exception as e:
+                logger.warning("Memory orchestration failed for node %s: %s", self.node_id, str(e))
+
             # Send final chunk event to indicate streaming is complete
             yield StreamChunkEvent(
                 selector=[self._node_id, "text"],
@@ -1228,6 +1239,77 @@ class LLMNode(Node):
     def retry(self) -> bool:
         return self._node_data.retry_config.retry_enabled
 
+    def _handle_chatflow_memory(self, llm_output: str, variable_pool: VariablePool):
+        if not self._node_data.memory or self._node_data.memory.mode != "block":
+            return
+
+        conversation_id_segment = variable_pool.get((SYSTEM_VARIABLE_NODE_ID, SystemVariableKey.CONVERSATION_ID))
+        if not conversation_id_segment:
+            raise ValueError("Conversation ID not found in variable pool.")
+        conversation_id = conversation_id_segment.text
+
+        user_query_segment = variable_pool.get((SYSTEM_VARIABLE_NODE_ID, SystemVariableKey.QUERY))
+        if not user_query_segment:
+            raise ValueError("User query not found in variable pool.")
+        user_query = user_query_segment.text
+
+        from core.model_runtime.entities.message_entities import AssistantPromptMessage, UserPromptMessage
+        from services.chatflow_history_service import ChatflowHistoryService
+
+        ChatflowHistoryService.save_node_message(
+            prompt_message=(UserPromptMessage(content=user_query)),
+            node_id=self.node_id,
+            conversation_id=conversation_id,
+            app_id=self.app_id,
+            tenant_id=self.tenant_id,
+        )
+        ChatflowHistoryService.save_node_message(
+            prompt_message=(AssistantPromptMessage(content=llm_output)),
+            node_id=self.node_id,
+            conversation_id=conversation_id,
+            app_id=self.app_id,
+            tenant_id=self.tenant_id,
+        )
+
+        memory_config = self._node_data.memory
+        if not memory_config:
+            return
+        block_ids = memory_config.block_id
+        if not block_ids:
+            return
+
+        # FIXME: This is dirty workaround and may cause incorrect resolution for workflow version
+        with Session(db.engine) as session:
+            stmt = select(Workflow).where(Workflow.tenant_id == self.tenant_id, Workflow.app_id == self.app_id)
+            workflow = session.scalars(stmt).first()
+            if not workflow:
+                raise ValueError("Workflow not found.")
+        memory_blocks = workflow.memory_blocks
+
+        for block_id in block_ids:
+            memory_block_spec = next((block for block in memory_blocks if block.id == block_id), None)
+
+            if memory_block_spec and memory_block_spec.scope == MemoryScope.NODE:
+                is_draft = self.invoke_from == InvokeFrom.DEBUGGER
+                from services.chatflow_memory_service import ChatflowMemoryService
+
+                ChatflowMemoryService.update_node_memory_if_needed(
+                    tenant_id=self.tenant_id,
+                    app_id=self.app_id,
+                    node_id=self.id,
+                    conversation_id=conversation_id,
+                    memory_block_spec=memory_block_spec,
+                    variable_pool=variable_pool,
+                    is_draft=is_draft,
+                    created_by=self._get_user_from_context(),
+                )
+
+    def _get_user_from_context(self) -> MemoryCreatedBy:
+        if self.user_from == UserFrom.ACCOUNT:
+            return MemoryCreatedBy(account_id=self.user_id)
+        else:
+            return MemoryCreatedBy(end_user_id=self.user_id)
+
 
 def _combine_message_content_with_role(
     *, contents: str | list[PromptMessageContentUnionTypes] | None = None, role: PromptMessageRole

api/core/workflow/runtime/variable_pool.py

@@ -9,11 +9,12 @@ from pydantic import BaseModel, Field
 from core.file import File, FileAttribute, file_manager
 from core.variables import Segment, SegmentGroup, Variable
 from core.variables.consts import SELECTORS_LENGTH
-from core.variables.segments import FileSegment, ObjectSegment
-from core.variables.variables import RAGPipelineVariableInput, VariableUnion
+from core.variables.segments import FileSegment, ObjectSegment, VersionedMemoryValue
+from core.variables.variables import RAGPipelineVariableInput, VariableUnion, VersionedMemoryVariable
 from core.workflow.constants import (
     CONVERSATION_VARIABLE_NODE_ID,
     ENVIRONMENT_VARIABLE_NODE_ID,
+    MEMORY_BLOCK_VARIABLE_NODE_ID,
     RAG_PIPELINE_VARIABLE_NODE_ID,
     SYSTEM_VARIABLE_NODE_ID,
 )
@@ -56,6 +57,10 @@ class VariablePool(BaseModel):
         description="RAG pipeline variables.",
         default_factory=list,
     )
+    memory_blocks: Mapping[str, str] = Field(
+        description="Memory blocks.",
+        default_factory=dict,
+    )
 
     def model_post_init(self, context: Any, /):
         # Create a mapping from field names to SystemVariableKey enum values
@@ -76,6 +81,18 @@ class VariablePool(BaseModel):
                 rag_pipeline_variables_map[node_id][key] = value
             for key, value in rag_pipeline_variables_map.items():
                 self.add((RAG_PIPELINE_VARIABLE_NODE_ID, key), value)
+        # Add memory blocks to the variable pool
+        for memory_id, memory_value in self.memory_blocks.items():
+            self.add(
+                [MEMORY_BLOCK_VARIABLE_NODE_ID, memory_id],
+                VersionedMemoryVariable(
+                    value=VersionedMemoryValue(
+                        current_value=memory_value,
+                        versions={"1": memory_value},
+                    ),
+                    name=memory_id,
+                )
+            )
 
     def add(self, selector: Sequence[str], value: Any, /):
         """

api/docker/entrypoint.sh

@@ -39,14 +39,16 @@ elif [[ "${MODE}" == "beat" ]]; then
   exec celery -A app.celery beat --loglevel ${LOG_LEVEL:-INFO}
 else
   if [[ "${DEBUG}" == "true" ]]; then
-    exec flask run --host=${DIFY_BIND_ADDRESS:-0.0.0.0} --port=${DIFY_PORT:-5001} --debug
+    export HOST=${DIFY_BIND_ADDRESS:-0.0.0.0}
+    export PORT=${DIFY_PORT:-5001}
+    exec python -m app
   else
     exec gunicorn \
       --bind "${DIFY_BIND_ADDRESS:-0.0.0.0}:${DIFY_PORT:-5001}" \
       --workers ${SERVER_WORKER_AMOUNT:-1} \
-      --worker-class ${SERVER_WORKER_CLASS:-gevent} \
+      --worker-class ${SERVER_WORKER_CLASS:-geventwebsocket.gunicorn.workers.GeventWebSocketWorker} \
       --worker-connections ${SERVER_WORKER_CONNECTIONS:-10} \
       --timeout ${GUNICORN_TIMEOUT:-200} \
-      app:app
+      app:socketio_app
   fi
 fi

api/enums/cloud_plan.py

@@ -0,0 +1,15 @@
+from enum import StrEnum, auto
+
+
+class CloudPlan(StrEnum):
+    """
+    Enum representing user plan types in the cloud platform.
+
+    SANDBOX: Free/default plan with limited features
+    PROFESSIONAL: Professional paid plan
+    TEAM: Team collaboration paid plan
+    """
+
+    SANDBOX = auto()
+    PROFESSIONAL = auto()
+    TEAM = auto()

api/events/event_handlers/update_provider_when_message_created.py

@@ -10,7 +10,7 @@ from sqlalchemy.orm import Session
 
 from configs import dify_config
 from core.app.entities.app_invoke_entities import AgentChatAppGenerateEntity, ChatAppGenerateEntity
-from core.entities.provider_entities import QuotaUnit, SystemConfiguration
+from core.entities.provider_entities import ProviderQuotaType, QuotaUnit, SystemConfiguration
 from events.message_event import message_was_created
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client, redis_fallback
@@ -134,22 +134,38 @@ def handle(sender: Message, **kwargs):
             system_configuration=system_configuration,
             model_name=model_config.model,
         )
-
         if used_quota is not None:
-            quota_update = _ProviderUpdateOperation(
-                filters=_ProviderUpdateFilters(
+            if provider_configuration.system_configuration.current_quota_type == ProviderQuotaType.TRIAL:
+                from services.credit_pool_service import CreditPoolService
+
+                CreditPoolService.check_and_deduct_credits(
                     tenant_id=tenant_id,
-                    provider_name=ModelProviderID(model_config.provider).provider_name,
-                    provider_type=ProviderType.SYSTEM,
-                    quota_type=provider_configuration.system_configuration.current_quota_type.value,
-                ),
-                values=_ProviderUpdateValues(quota_used=Provider.quota_used + used_quota, last_used=current_time),
-                additional_filters=_ProviderUpdateAdditionalFilters(
-                    quota_limit_check=True  # Provider.quota_limit > Provider.quota_used
-                ),
-                description="quota_deduction_update",
-            )
-            updates_to_perform.append(quota_update)
+                    credits_required=used_quota,
+                    pool_type="trial",
+                )
+            elif provider_configuration.system_configuration.current_quota_type == ProviderQuotaType.PAID:
+                from services.credit_pool_service import CreditPoolService
+
+                CreditPoolService.check_and_deduct_credits(
+                    tenant_id=tenant_id,
+                    credits_required=used_quota,
+                    pool_type="paid",
+                )
+            else:
+                quota_update = _ProviderUpdateOperation(
+                    filters=_ProviderUpdateFilters(
+                        tenant_id=tenant_id,
+                        provider_name=ModelProviderID(model_config.provider).provider_name,
+                        provider_type=ProviderType.SYSTEM.value,
+                        quota_type=provider_configuration.system_configuration.current_quota_type.value,
+                    ),
+                    values=_ProviderUpdateValues(quota_used=Provider.quota_used + used_quota, last_used=current_time),
+                    additional_filters=_ProviderUpdateAdditionalFilters(
+                        quota_limit_check=True  # Provider.quota_limit > Provider.quota_used
+                    ),
+                    description="quota_deduction_update",
+                )
+                updates_to_perform.append(quota_update)
 
     # Execute all updates
     start_time = time_module.perf_counter()

api/extensions/ext_blueprints.py

@@ -31,7 +31,13 @@ def init_app(app: DifyApp):
         web_bp,
         resources={r"/*": {"origins": dify_config.WEB_API_CORS_ALLOW_ORIGINS}},
         supports_credentials=True,
-        allow_headers=list(AUTHENTICATED_HEADERS),
+        allow_headers=[
+            "Content-Type",
+            "Authorization",
+            HEADER_NAME_APP_CODE,
+            HEADER_NAME_CSRF_TOKEN,
+            HEADER_NAME_PASSPORT,
+        ],
         methods=["GET", "PUT", "POST", "DELETE", "OPTIONS", "PATCH"],
         expose_headers=["X-Version", "X-Env"],
     )

api/extensions/ext_socketio.py

@@ -0,0 +1,3 @@
+import socketio
+
+sio = socketio.Server(async_mode="gevent", cors_allowed_origins="*")

api/factories/variable_factory.py

@@ -21,6 +21,8 @@ from core.variables.segments import (
     ObjectSegment,
     Segment,
     StringSegment,
+    VersionedMemorySegment,
+    VersionedMemoryValue,
 )
 from core.variables.types import SegmentType
 from core.variables.variables import (
@@ -39,6 +41,7 @@ from core.variables.variables import (
     SecretVariable,
     StringVariable,
     Variable,
+    VersionedMemoryVariable,
 )
 from core.workflow.constants import (
     CONVERSATION_VARIABLE_NODE_ID,
@@ -69,6 +72,7 @@ SEGMENT_TO_VARIABLE_MAP = {
     NoneSegment: NoneVariable,
     ObjectSegment: ObjectVariable,
     StringSegment: StringVariable,
+    VersionedMemorySegment: VersionedMemoryVariable
 }
 
 
@@ -193,6 +197,7 @@ _segment_factory: Mapping[SegmentType, type[Segment]] = {
     SegmentType.FILE: FileSegment,
     SegmentType.BOOLEAN: BooleanSegment,
     SegmentType.OBJECT: ObjectSegment,
+    SegmentType.VERSIONED_MEMORY: VersionedMemorySegment,
     # Array types
     SegmentType.ARRAY_ANY: ArrayAnySegment,
     SegmentType.ARRAY_STRING: ArrayStringSegment,
@@ -259,6 +264,12 @@ def build_segment_with_type(segment_type: SegmentType, value: Any) -> Segment:
         else:
             raise TypeMismatchError(f"Type mismatch: expected {segment_type}, but got empty list")
 
+    if segment_type == SegmentType.VERSIONED_MEMORY:
+        return VersionedMemorySegment(
+            value_type=segment_type,
+            value=VersionedMemoryValue.model_validate(value)
+        )
+
     inferred_type = SegmentType.infer_segment_type(value)
     # Type compatibility checking
     if inferred_type is None:

api/fields/online_user_fields.py

@@ -0,0 +1,17 @@
+from flask_restx import fields
+
+online_user_partial_fields = {
+    "user_id": fields.String,
+    "username": fields.String,
+    "avatar": fields.String,
+    "sid": fields.String,
+}
+
+workflow_online_users_fields = {
+    "workflow_id": fields.String,
+    "users": fields.List(fields.Nested(online_user_partial_fields)),
+}
+
+online_user_list_fields = {
+    "data": fields.List(fields.Nested(workflow_online_users_fields)),
+}

api/fields/workflow_comment_fields.py

@@ -0,0 +1,96 @@
+from flask_restx import fields
+
+from libs.helper import AvatarUrlField, TimestampField
+
+# basic account fields for comments
+account_fields = {
+    "id": fields.String,
+    "name": fields.String,
+    "email": fields.String,
+    "avatar_url": AvatarUrlField,
+}
+
+# Comment mention fields
+workflow_comment_mention_fields = {
+    "mentioned_user_id": fields.String,
+    "mentioned_user_account": fields.Nested(account_fields, allow_null=True),
+    "reply_id": fields.String,
+}
+
+# Comment reply fields
+workflow_comment_reply_fields = {
+    "id": fields.String,
+    "content": fields.String,
+    "created_by": fields.String,
+    "created_by_account": fields.Nested(account_fields, allow_null=True),
+    "created_at": TimestampField,
+}
+
+# Basic comment fields (for list views)
+workflow_comment_basic_fields = {
+    "id": fields.String,
+    "position_x": fields.Float,
+    "position_y": fields.Float,
+    "content": fields.String,
+    "created_by": fields.String,
+    "created_by_account": fields.Nested(account_fields, allow_null=True),
+    "created_at": TimestampField,
+    "updated_at": TimestampField,
+    "resolved": fields.Boolean,
+    "resolved_at": TimestampField,
+    "resolved_by": fields.String,
+    "resolved_by_account": fields.Nested(account_fields, allow_null=True),
+    "reply_count": fields.Integer,
+    "mention_count": fields.Integer,
+    "participants": fields.List(fields.Nested(account_fields)),
+}
+
+# Detailed comment fields (for single comment view)
+workflow_comment_detail_fields = {
+    "id": fields.String,
+    "position_x": fields.Float,
+    "position_y": fields.Float,
+    "content": fields.String,
+    "created_by": fields.String,
+    "created_by_account": fields.Nested(account_fields, allow_null=True),
+    "created_at": TimestampField,
+    "updated_at": TimestampField,
+    "resolved": fields.Boolean,
+    "resolved_at": TimestampField,
+    "resolved_by": fields.String,
+    "resolved_by_account": fields.Nested(account_fields, allow_null=True),
+    "replies": fields.List(fields.Nested(workflow_comment_reply_fields)),
+    "mentions": fields.List(fields.Nested(workflow_comment_mention_fields)),
+}
+
+# Comment creation response fields (simplified)
+workflow_comment_create_fields = {
+    "id": fields.String,
+    "created_at": TimestampField,
+}
+
+# Comment update response fields (simplified)
+workflow_comment_update_fields = {
+    "id": fields.String,
+    "updated_at": TimestampField,
+}
+
+# Comment resolve response fields
+workflow_comment_resolve_fields = {
+    "id": fields.String,
+    "resolved": fields.Boolean,
+    "resolved_at": TimestampField,
+    "resolved_by": fields.String,
+}
+
+# Reply creation response fields (simplified)
+workflow_comment_reply_create_fields = {
+    "id": fields.String,
+    "created_at": TimestampField,
+}
+
+# Reply update response fields
+workflow_comment_reply_update_fields = {
+    "id": fields.String,
+    "updated_at": TimestampField,
+}

api/libs/datetime_utils.py

@@ -2,6 +2,8 @@ import abc
 import datetime
 from typing import Protocol
 
+import pytz
+
 
 class _NowFunction(Protocol):
     @abc.abstractmethod
@@ -20,3 +22,51 @@ def naive_utc_now() -> datetime.datetime:
     representing current UTC time.
     """
     return _now_func(datetime.UTC).replace(tzinfo=None)
+
+
+def parse_time_range(
+    start: str | None, end: str | None, tzname: str
+) -> tuple[datetime.datetime | None, datetime.datetime | None]:
+    """
+    Parse time range strings and convert to UTC datetime objects.
+    Handles DST ambiguity and non-existent times gracefully.
+
+    Args:
+        start: Start time string (YYYY-MM-DD HH:MM)
+        end: End time string (YYYY-MM-DD HH:MM)
+        tzname: Timezone name
+
+    Returns:
+        tuple: (start_datetime_utc, end_datetime_utc)
+
+    Raises:
+        ValueError: When time range is invalid or start > end
+    """
+    tz = pytz.timezone(tzname)
+    utc = pytz.utc
+
+    def _parse(time_str: str | None, label: str) -> datetime.datetime | None:
+        if not time_str:
+            return None
+
+        try:
+            dt = datetime.datetime.strptime(time_str, "%Y-%m-%d %H:%M").replace(second=0)
+        except ValueError as e:
+            raise ValueError(f"Invalid {label} time format: {e}")
+
+        try:
+            return tz.localize(dt, is_dst=None).astimezone(utc)
+        except pytz.AmbiguousTimeError:
+            return tz.localize(dt, is_dst=False).astimezone(utc)
+        except pytz.NonExistentTimeError:
+            dt += datetime.timedelta(hours=1)
+            return tz.localize(dt, is_dst=None).astimezone(utc)
+
+    start_dt = _parse(start, "start")
+    end_dt = _parse(end, "end")
+
+    # Range validation
+    if start_dt and end_dt and start_dt > end_dt:
+        raise ValueError("start must be earlier than or equal to end")
+
+    return start_dt, end_dt

api/migrations/versions/2025_09_18_1726-227822d22895_add_workflow_comments_table.py

@@ -0,0 +1,90 @@
+"""Add workflow comments table
+
+Revision ID: 227822d22895
+Revises: 68519ad5cd18
+Create Date: 2025-08-22 17:26:15.255980
+
+"""
+from alembic import op
+import models as models
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '227822d22895'
+down_revision = '68519ad5cd18'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('workflow_comments',
+    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+    sa.Column('app_id', models.types.StringUUID(), nullable=False),
+    sa.Column('position_x', sa.Float(), nullable=False),
+    sa.Column('position_y', sa.Float(), nullable=False),
+    sa.Column('content', sa.Text(), nullable=False),
+    sa.Column('created_by', models.types.StringUUID(), nullable=False),
+    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+    sa.Column('resolved', sa.Boolean(), server_default=sa.text('false'), nullable=False),
+    sa.Column('resolved_at', sa.DateTime(), nullable=True),
+    sa.Column('resolved_by', models.types.StringUUID(), nullable=True),
+    sa.PrimaryKeyConstraint('id', name='workflow_comments_pkey')
+    )
+    with op.batch_alter_table('workflow_comments', schema=None) as batch_op:
+        batch_op.create_index('workflow_comments_app_idx', ['tenant_id', 'app_id'], unique=False)
+        batch_op.create_index('workflow_comments_created_at_idx', ['created_at'], unique=False)
+
+    op.create_table('workflow_comment_replies',
+    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+    sa.Column('comment_id', models.types.StringUUID(), nullable=False),
+    sa.Column('content', sa.Text(), nullable=False),
+    sa.Column('created_by', models.types.StringUUID(), nullable=False),
+    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+    sa.ForeignKeyConstraint(['comment_id'], ['workflow_comments.id'], name=op.f('workflow_comment_replies_comment_id_fkey'), ondelete='CASCADE'),
+    sa.PrimaryKeyConstraint('id', name='workflow_comment_replies_pkey')
+    )
+    with op.batch_alter_table('workflow_comment_replies', schema=None) as batch_op:
+        batch_op.create_index('comment_replies_comment_idx', ['comment_id'], unique=False)
+        batch_op.create_index('comment_replies_created_at_idx', ['created_at'], unique=False)
+
+    op.create_table('workflow_comment_mentions',
+    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+    sa.Column('comment_id', models.types.StringUUID(), nullable=False),
+    sa.Column('reply_id', models.types.StringUUID(), nullable=True),
+    sa.Column('mentioned_user_id', models.types.StringUUID(), nullable=False),
+    sa.ForeignKeyConstraint(['comment_id'], ['workflow_comments.id'], name=op.f('workflow_comment_mentions_comment_id_fkey'), ondelete='CASCADE'),
+    sa.ForeignKeyConstraint(['reply_id'], ['workflow_comment_replies.id'], name=op.f('workflow_comment_mentions_reply_id_fkey'), ondelete='CASCADE'),
+    sa.PrimaryKeyConstraint('id', name='workflow_comment_mentions_pkey')
+    )
+    with op.batch_alter_table('workflow_comment_mentions', schema=None) as batch_op:
+        batch_op.create_index('comment_mentions_comment_idx', ['comment_id'], unique=False)
+        batch_op.create_index('comment_mentions_reply_idx', ['reply_id'], unique=False)
+        batch_op.create_index('comment_mentions_user_idx', ['mentioned_user_id'], unique=False)
+
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('workflow_comment_mentions', schema=None) as batch_op:
+        batch_op.drop_index('comment_mentions_user_idx')
+        batch_op.drop_index('comment_mentions_reply_idx')
+        batch_op.drop_index('comment_mentions_comment_idx')
+
+    op.drop_table('workflow_comment_mentions')
+    with op.batch_alter_table('workflow_comment_replies', schema=None) as batch_op:
+        batch_op.drop_index('comment_replies_created_at_idx')
+        batch_op.drop_index('comment_replies_comment_idx')
+
+    op.drop_table('workflow_comment_replies')
+    with op.batch_alter_table('workflow_comments', schema=None) as batch_op:
+        batch_op.drop_index('workflow_comments_created_at_idx')
+        batch_op.drop_index('workflow_comments_app_idx')
+
+    op.drop_table('workflow_comments')
+    # ### end Alembic commands ###

api/migrations/versions/2025_09_25_1520-58a70d22fdbd_add_table_credit_pool.py

@@ -0,0 +1,104 @@
+"""add table credit pool
+
+Revision ID: 58a70d22fdbd
+Revises: 68519ad5cd18
+Create Date: 2025-09-25 15:20:40.367078
+
+"""
+from alembic import op
+import models as models
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '58a70d22fdbd'
+down_revision = '68519ad5cd18'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('tenant_credit_pools',
+    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+    sa.Column('pool_type', sa.String(length=40), nullable=False),
+    sa.Column('quota_limit', sa.BigInteger(), nullable=False),
+    sa.Column('quota_used', sa.BigInteger(), nullable=False),
+    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+    sa.PrimaryKeyConstraint('id', name='tenant_credit_pool_pkey')
+    )
+    with op.batch_alter_table('tenant_credit_pools', schema=None) as batch_op:
+        batch_op.create_index('tenant_credit_pool_pool_type_idx', ['pool_type'], unique=False)
+        batch_op.create_index('tenant_credit_pool_tenant_id_idx', ['tenant_id'], unique=False)
+    # Data migration: Move quota data from providers to tenant_credit_pools
+    migrate_quota_data()
+    
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('tenant_credit_pools', schema=None) as batch_op:
+        batch_op.drop_index('tenant_credit_pool_tenant_id_idx')
+        batch_op.drop_index('tenant_credit_pool_pool_type_idx')
+
+    op.drop_table('tenant_credit_pools')
+    # ### end Alembic commands ###
+
+
+def migrate_quota_data():
+    """
+    Migrate quota data from providers table to tenant_credit_pools table
+    for providers with quota_type='trial' or 'paid', provider_name='openai', provider_type='system'
+    """
+    # Create connection
+    bind = op.get_bind()
+    
+    # Define quota type mappings
+    quota_type_mappings = ['trial', 'paid']
+    
+    for quota_type in quota_type_mappings:
+        # Query providers that match the criteria
+        select_sql = sa.text("""
+            SELECT tenant_id, quota_limit, quota_used
+            FROM providers 
+            WHERE quota_type = :quota_type
+            AND provider_name = 'openai' 
+            AND provider_type = 'system'
+            AND quota_limit IS NOT NULL
+        """)
+        
+        result = bind.execute(select_sql, {"quota_type": quota_type})
+        providers_data = result.fetchall()
+        
+        # Insert data into tenant_credit_pools
+        for provider_data in providers_data:
+            tenant_id, quota_limit, quota_used = provider_data
+            
+            # Check if credit pool already exists for this tenant and pool type
+            check_sql = sa.text("""
+                SELECT COUNT(*) 
+                FROM tenant_credit_pools 
+                WHERE tenant_id = :tenant_id AND pool_type = :pool_type
+            """)
+            
+            existing_count = bind.execute(check_sql, {
+                "tenant_id": tenant_id,
+                "pool_type": quota_type
+            }).scalar()
+            
+            if existing_count == 0:
+                # Insert new credit pool record
+                insert_sql = sa.text("""
+                    INSERT INTO tenant_credit_pools (tenant_id, pool_type, quota_limit, quota_used, created_at, updated_at)
+                    VALUES (:tenant_id, :pool_type, :quota_limit, :quota_used, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
+                """)
+                
+                bind.execute(insert_sql, {
+                    "tenant_id": tenant_id,
+                    "pool_type": quota_type,
+                    "quota_limit": quota_limit or 0,
+                    "quota_used": quota_used or 0
+                })

api/migrations/versions/2025_10_11_1442-3993fd9e9c2f_add_table_explore_banner_and_trial.py

@@ -0,0 +1,92 @@
+"""add table explore banner and trial
+
+Revision ID: 3993fd9e9c2f
+Revises: 68519ad5cd18
+Create Date: 2025-10-11 14:42:01.954865
+
+"""
+from alembic import op
+import models as models
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '3993fd9e9c2f'
+down_revision = '68519ad5cd18'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('account_trial_app_records',
+    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+    sa.Column('account_id', models.types.StringUUID(), nullable=False),
+    sa.Column('app_id', models.types.StringUUID(), nullable=False),
+    sa.Column('count', sa.Integer(), nullable=False),
+    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+    sa.PrimaryKeyConstraint('id', name='user_trial_app_pkey'),
+    sa.UniqueConstraint('account_id', 'app_id', name='unique_account_trial_app_record')
+    )
+    with op.batch_alter_table('account_trial_app_records', schema=None) as batch_op:
+        batch_op.create_index('account_trial_app_record_account_id_idx', ['account_id'], unique=False)
+        batch_op.create_index('account_trial_app_record_app_id_idx', ['app_id'], unique=False)
+
+    op.create_table('exporle_banners',
+    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+    sa.Column('content', sa.JSON(), nullable=False),
+    sa.Column('link', sa.String(length=255), nullable=False),
+    sa.Column('sort', sa.Integer(), nullable=False),
+    sa.Column('status', sa.String(length=255), server_default=sa.text("'enabled'::character varying"), nullable=False),
+    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+    sa.Column('language', sa.String(length=255), server_default=sa.text("'en-US'::character varying"), nullable=False),
+    sa.PrimaryKeyConstraint('id', name='exporler_banner_pkey')
+    )
+    op.create_table('trial_apps',
+    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+    sa.Column('app_id', models.types.StringUUID(), nullable=False),
+    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+    sa.Column('trial_limit', sa.Integer(), nullable=False),
+    sa.PrimaryKeyConstraint('id', name='trial_app_pkey'),
+    sa.UniqueConstraint('app_id', name='unique_trail_app_id')
+    )
+    with op.batch_alter_table('trial_apps', schema=None) as batch_op:
+        batch_op.create_index('trial_app_app_id_idx', ['app_id'], unique=False)
+        batch_op.create_index('trial_app_tenant_id_idx', ['tenant_id'], unique=False)
+
+    with op.batch_alter_table('datasource_providers', schema=None) as batch_op:
+        batch_op.alter_column('avatar_url',
+               existing_type=sa.TEXT(),
+               type_=sa.String(length=255),
+               existing_nullable=True)
+
+    with op.batch_alter_table('providers', schema=None) as batch_op:
+        batch_op.drop_column('credential_status')
+
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('providers', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('credential_status', sa.VARCHAR(length=20), server_default=sa.text("'active'::character varying"), autoincrement=False, nullable=True))
+
+    with op.batch_alter_table('datasource_providers', schema=None) as batch_op:
+        batch_op.alter_column('avatar_url',
+               existing_type=sa.String(length=255),
+               type_=sa.TEXT(),
+               existing_nullable=True)
+
+    with op.batch_alter_table('trial_apps', schema=None) as batch_op:
+        batch_op.drop_index('trial_app_tenant_id_idx')
+        batch_op.drop_index('trial_app_app_id_idx')
+
+    op.drop_table('trial_apps')
+    op.drop_table('exporle_banners')
+    with op.batch_alter_table('account_trial_app_records', schema=None) as batch_op:
+        batch_op.drop_index('account_trial_app_record_app_id_idx')
+        batch_op.drop_index('account_trial_app_record_account_id_idx')
+
+    op.drop_table('account_trial_app_records')
+    # ### end Alembic commands ###

api/migrations/versions/2025_10_11_1529-d00b2b40ea3e_add_chatflow_memory_tables.py

@@ -0,0 +1,104 @@
+"""add_chatflow_memory_tables
+
+Revision ID: d00b2b40ea3e
+Revises: 68519ad5cd18
+Create Date: 2025-10-11 15:29:20.244675
+
+"""
+from alembic import op
+import models as models
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = 'd00b2b40ea3e'
+down_revision = '68519ad5cd18'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('chatflow_conversations',
+    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+    sa.Column('app_id', models.types.StringUUID(), nullable=False),
+    sa.Column('node_id', sa.Text(), nullable=True),
+    sa.Column('original_conversation_id', models.types.StringUUID(), nullable=True),
+    sa.Column('conversation_metadata', sa.Text(), nullable=False),
+    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+    sa.PrimaryKeyConstraint('id', name='chatflow_conversations_pkey')
+    )
+    with op.batch_alter_table('chatflow_conversations', schema=None) as batch_op:
+        batch_op.create_index('chatflow_conversations_original_conversation_id_idx', ['tenant_id', 'app_id', 'node_id', 'original_conversation_id'], unique=False)
+
+    op.create_table('chatflow_memory_variables',
+    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+    sa.Column('app_id', models.types.StringUUID(), nullable=True),
+    sa.Column('conversation_id', models.types.StringUUID(), nullable=True),
+    sa.Column('node_id', sa.Text(), nullable=True),
+    sa.Column('memory_id', sa.Text(), nullable=False),
+    sa.Column('value', sa.Text(), nullable=False),
+    sa.Column('name', sa.Text(), nullable=False),
+    sa.Column('scope', sa.String(length=10), nullable=False),
+    sa.Column('term', sa.String(length=20), nullable=False),
+    sa.Column('version', sa.Integer(), nullable=False),
+    sa.Column('created_by_role', sa.String(length=20), nullable=False),
+    sa.Column('created_by', models.types.StringUUID(), nullable=False),
+    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+    sa.PrimaryKeyConstraint('id', name='chatflow_memory_variables_pkey')
+    )
+    with op.batch_alter_table('chatflow_memory_variables', schema=None) as batch_op:
+        batch_op.create_index('chatflow_memory_variables_memory_id_idx', ['tenant_id', 'app_id', 'node_id', 'memory_id'], unique=False)
+
+    op.create_table('chatflow_messages',
+    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+    sa.Column('conversation_id', models.types.StringUUID(), nullable=False),
+    sa.Column('index', sa.Integer(), nullable=False),
+    sa.Column('version', sa.Integer(), nullable=False),
+    sa.Column('data', sa.Text(), nullable=False),
+    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+    sa.PrimaryKeyConstraint('id', name='chatflow_messages_pkey')
+    )
+    with op.batch_alter_table('chatflow_messages', schema=None) as batch_op:
+        batch_op.create_index('chatflow_messages_version_idx', ['conversation_id', 'index', 'version'], unique=False)
+
+    with op.batch_alter_table('datasource_providers', schema=None) as batch_op:
+        batch_op.alter_column('avatar_url',
+               existing_type=sa.TEXT(),
+               type_=sa.String(length=255),
+               existing_nullable=True)
+
+    with op.batch_alter_table('providers', schema=None) as batch_op:
+        batch_op.drop_column('credential_status')
+
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('providers', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('credential_status', sa.VARCHAR(length=20), server_default=sa.text("'active'::character varying"), autoincrement=False, nullable=True))
+
+    with op.batch_alter_table('datasource_providers', schema=None) as batch_op:
+        batch_op.alter_column('avatar_url',
+               existing_type=sa.String(length=255),
+               type_=sa.TEXT(),
+               existing_nullable=True)
+
+    with op.batch_alter_table('chatflow_messages', schema=None) as batch_op:
+        batch_op.drop_index('chatflow_messages_version_idx')
+
+    op.drop_table('chatflow_messages')
+    with op.batch_alter_table('chatflow_memory_variables', schema=None) as batch_op:
+        batch_op.drop_index('chatflow_memory_variables_memory_id_idx')
+
+    op.drop_table('chatflow_memory_variables')
+    with op.batch_alter_table('chatflow_conversations', schema=None) as batch_op:
+        batch_op.drop_index('chatflow_conversations_original_conversation_id_idx')
+
+    op.drop_table('chatflow_conversations')
+    # ### end Alembic commands ###

api/migrations/versions/2025_10_14_1618-d98acf217d43_add_app_mode_for_messsage.py

@@ -22,6 +22,55 @@ def upgrade():
         batch_op.add_column(sa.Column('app_mode', sa.String(length=255), nullable=True))
         batch_op.create_index('message_app_mode_idx', ['app_mode'], unique=False)
 
+    conn = op.get_bind()
+
+    # Strategy: Update in batches to minimize lock time
+    # For large tables (millions of rows), this prevents long-running transactions
+    batch_size = 10000
+
+    print("Starting backfill of app_mode from conversations...")
+
+    # Use a more efficient UPDATE with JOIN
+    # This query updates messages.app_mode from conversations.mode
+    # Using string formatting for LIMIT since it's a constant
+    update_query = f"""
+        UPDATE messages m
+        SET app_mode = c.mode
+        FROM conversations c
+        WHERE m.conversation_id = c.id
+          AND m.app_mode IS NULL
+          AND m.id IN (
+              SELECT id FROM messages
+              WHERE app_mode IS NULL
+              LIMIT {batch_size}
+          )
+    """
+
+    # Execute batched updates
+    total_updated = 0
+    iteration = 0
+    while True:
+        iteration += 1
+        result = conn.execute(sa.text(update_query))
+
+        # Check if result is None or has no rowcount
+        if result is None:
+            print("Warning: Query returned None, stopping backfill")
+            break
+
+        rows_updated = result.rowcount if hasattr(result, 'rowcount') else 0
+        total_updated += rows_updated
+
+        if rows_updated == 0:
+            break
+
+        print(f"Iteration {iteration}: Updated {rows_updated} messages (total: {total_updated})")
+
+        # For very large tables, add a small delay to reduce load
+        # Uncomment if needed: import time; time.sleep(0.1)
+
+    print(f"Backfill completed. Total messages updated: {total_updated}")
+
     # ### end Alembic commands ###
 
 

api/models/__init__.py

@@ -9,6 +9,12 @@ from .account import (
     TenantStatus,
 )
 from .api_based_extension import APIBasedExtension, APIBasedExtensionPoint
+from .chatflow_memory import ChatflowConversation, ChatflowMemoryVariable, ChatflowMessage
+from .comment import (
+    WorkflowComment,
+    WorkflowCommentMention,
+    WorkflowCommentReply,
+)
 from .dataset import (
     AppDatasetJoin,
     Dataset,
@@ -28,6 +34,7 @@ from .dataset import (
 )
 from .enums import CreatorUserRole, UserFrom, WorkflowRunTriggeredFrom
 from .model import (
+    AccountTrialAppRecord,
     ApiRequest,
     ApiToken,
     App,
@@ -40,6 +47,7 @@ from .model import (
     DatasetRetrieverResource,
     DifySetup,
     EndUser,
+    ExporleBanner,
     IconType,
     InstalledApp,
     Message,
@@ -53,7 +61,9 @@ from .model import (
     Site,
     Tag,
     TagBinding,
+    TenantCreditPool,
     TraceAppConfig,
+    TrialApp,
     UploadFile,
 )
 from .oauth import DatasourceOauthParamConfig, DatasourceProvider
@@ -99,6 +109,7 @@ __all__ = [
     "Account",
     "AccountIntegrate",
     "AccountStatus",
+    "AccountTrialAppRecord",
     "ApiRequest",
     "ApiToken",
     "ApiToolProvider",
@@ -112,6 +123,9 @@ __all__ = [
     "BuiltinToolProvider",
     "CeleryTask",
     "CeleryTaskSet",
+    "ChatflowConversation",
+    "ChatflowMemoryVariable",
+    "ChatflowMessage",
     "Conversation",
     "ConversationVariable",
     "CreatorUserRole",
@@ -132,6 +146,7 @@ __all__ = [
     "DocumentSegment",
     "Embedding",
     "EndUser",
+    "ExporleBanner",
     "ExternalKnowledgeApis",
     "ExternalKnowledgeBindings",
     "IconType",
@@ -160,6 +175,7 @@ __all__ = [
     "Tenant",
     "TenantAccountJoin",
     "TenantAccountRole",
+    "TenantCreditPool",
     "TenantDefaultModel",
     "TenantPreferredModelProvider",
     "TenantStatus",
@@ -169,12 +185,16 @@ __all__ = [
     "ToolLabelBinding",
     "ToolModelInvoke",
     "TraceAppConfig",
+    "TrialApp",
     "UploadFile",
     "UserFrom",
     "Whitelist",
     "Workflow",
     "WorkflowAppLog",
     "WorkflowAppLogCreatedFrom",
+    "WorkflowComment",
+    "WorkflowCommentMention",
+    "WorkflowCommentReply",
     "WorkflowNodeExecutionModel",
     "WorkflowNodeExecutionOffload",
     "WorkflowNodeExecutionTriggeredFrom",

api/models/chatflow_memory.py

@@ -0,0 +1,76 @@
+from datetime import datetime
+
+import sqlalchemy as sa
+from sqlalchemy import DateTime, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from .base import Base
+from .types import StringUUID
+
+
+class ChatflowMemoryVariable(Base):
+    __tablename__ = "chatflow_memory_variables"
+    __table_args__ = (
+        sa.PrimaryKeyConstraint("id", name="chatflow_memory_variables_pkey"),
+        sa.Index("chatflow_memory_variables_memory_id_idx", "tenant_id", "app_id", "node_id", "memory_id"),
+    )
+
+    id: Mapped[str] = mapped_column(StringUUID, primary_key=True, server_default=sa.text("uuid_generate_v4()"))
+    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    app_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
+    conversation_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
+    node_id: Mapped[str | None] = mapped_column(sa.Text, nullable=True)
+    memory_id: Mapped[str] = mapped_column(sa.Text, nullable=False)
+    value: Mapped[str] = mapped_column(sa.Text, nullable=False)
+    name: Mapped[str] = mapped_column(sa.Text, nullable=False)
+    scope: Mapped[str] = mapped_column(sa.String(10), nullable=False)  # 'app' or 'node'
+    term: Mapped[str] = mapped_column(sa.String(20), nullable=False)  # 'session' or 'persistent'
+    version: Mapped[int] = mapped_column(sa.Integer, nullable=False, default=1)
+    created_by_role: Mapped[str] = mapped_column(sa.String(20))  # 'end_user' or 'account`
+    created_by: Mapped[str] = mapped_column(StringUUID)
+
+    created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, server_default=func.current_timestamp())
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime, nullable=False, server_default=func.current_timestamp(), onupdate=func.current_timestamp()
+    )
+
+
+class ChatflowConversation(Base):
+    __tablename__ = "chatflow_conversations"
+    __table_args__ = (
+        sa.PrimaryKeyConstraint("id", name="chatflow_conversations_pkey"),
+        sa.Index(
+            "chatflow_conversations_original_conversation_id_idx",
+            "tenant_id", "app_id", "node_id", "original_conversation_id"
+        ),
+    )
+
+    id: Mapped[str] = mapped_column(StringUUID, primary_key=True, server_default=sa.text("uuid_generate_v4()"))
+    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    node_id: Mapped[str | None] = mapped_column(sa.Text, nullable=True)
+    original_conversation_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
+    conversation_metadata: Mapped[str] = mapped_column(sa.Text, nullable=False)  # JSON
+
+    created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, server_default=func.current_timestamp())
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime, nullable=False, server_default=func.current_timestamp(), onupdate=func.current_timestamp()
+    )
+
+
+class ChatflowMessage(Base):
+    __tablename__ = "chatflow_messages"
+    __table_args__ = (
+        sa.PrimaryKeyConstraint("id", name="chatflow_messages_pkey"),
+        sa.Index("chatflow_messages_version_idx", "conversation_id", "index", "version"),
+    )
+
+    id: Mapped[str] = mapped_column(StringUUID, primary_key=True, server_default=sa.text("uuid_generate_v4()"))
+    conversation_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    index: Mapped[int] = mapped_column(sa.Integer, nullable=False)  # This index starts from 0
+    version: Mapped[int] = mapped_column(sa.Integer, nullable=False)
+    data: Mapped[str] = mapped_column(sa.Text, nullable=False)  # Serialized PromptMessage JSON
+    created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, server_default=func.current_timestamp())
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime, nullable=False, server_default=func.current_timestamp(), onupdate=func.current_timestamp()
+    )

api/models/comment.py

@@ -0,0 +1,189 @@
+"""Workflow comment models."""
+
+from datetime import datetime
+from typing import TYPE_CHECKING, Optional
+
+from sqlalchemy import Index, func
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from .account import Account
+from .base import Base
+from .engine import db
+from .types import StringUUID
+
+if TYPE_CHECKING:
+    pass
+
+
+class WorkflowComment(Base):
+    """Workflow comment model for canvas commenting functionality.
+
+    Comments are associated with apps rather than specific workflow versions,
+    since an app has only one draft workflow at a time and comments should persist
+    across workflow version changes.
+
+    Attributes:
+        id: Comment ID
+        tenant_id: Workspace ID
+        app_id: App ID (primary association, comments belong to apps)
+        position_x: X coordinate on canvas
+        position_y: Y coordinate on canvas
+        content: Comment content
+        created_by: Creator account ID
+        created_at: Creation time
+        updated_at: Last update time
+        resolved: Whether comment is resolved
+        resolved_at: Resolution time
+        resolved_by: Resolver account ID
+    """
+
+    __tablename__ = "workflow_comments"
+    __table_args__ = (
+        db.PrimaryKeyConstraint("id", name="workflow_comments_pkey"),
+        Index("workflow_comments_app_idx", "tenant_id", "app_id"),
+        Index("workflow_comments_created_at_idx", "created_at"),
+    )
+
+    id: Mapped[str] = mapped_column(StringUUID, server_default=db.text("uuid_generate_v4()"))
+    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    position_x: Mapped[float] = mapped_column(db.Float)
+    position_y: Mapped[float] = mapped_column(db.Float)
+    content: Mapped[str] = mapped_column(db.Text, nullable=False)
+    created_by: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    created_at: Mapped[datetime] = mapped_column(db.DateTime, nullable=False, server_default=func.current_timestamp())
+    updated_at: Mapped[datetime] = mapped_column(
+        db.DateTime, nullable=False, server_default=func.current_timestamp(), onupdate=func.current_timestamp()
+    )
+    resolved: Mapped[bool] = mapped_column(db.Boolean, nullable=False, server_default=db.text("false"))
+    resolved_at: Mapped[Optional[datetime]] = mapped_column(db.DateTime)
+    resolved_by: Mapped[Optional[str]] = mapped_column(StringUUID)
+
+    # Relationships
+    replies: Mapped[list["WorkflowCommentReply"]] = relationship(
+        "WorkflowCommentReply", back_populates="comment", cascade="all, delete-orphan"
+    )
+    mentions: Mapped[list["WorkflowCommentMention"]] = relationship(
+        "WorkflowCommentMention", back_populates="comment", cascade="all, delete-orphan"
+    )
+
+    @property
+    def created_by_account(self):
+        """Get creator account."""
+        return db.session.get(Account, self.created_by)
+
+    @property
+    def resolved_by_account(self):
+        """Get resolver account."""
+        if self.resolved_by:
+            return db.session.get(Account, self.resolved_by)
+        return None
+
+    @property
+    def reply_count(self):
+        """Get reply count."""
+        return len(self.replies)
+
+    @property
+    def mention_count(self):
+        """Get mention count."""
+        return len(self.mentions)
+
+    @property
+    def participants(self):
+        """Get all participants (creator + repliers + mentioned users)."""
+        participant_ids = set()
+
+        # Add comment creator
+        participant_ids.add(self.created_by)
+
+        # Add reply creators
+        participant_ids.update(reply.created_by for reply in self.replies)
+
+        # Add mentioned users
+        participant_ids.update(mention.mentioned_user_id for mention in self.mentions)
+
+        # Get account objects
+        participants = []
+        for user_id in participant_ids:
+            account = db.session.get(Account, user_id)
+            if account:
+                participants.append(account)
+
+        return participants
+
+
+class WorkflowCommentReply(Base):
+    """Workflow comment reply model.
+
+    Attributes:
+        id: Reply ID
+        comment_id: Parent comment ID
+        content: Reply content
+        created_by: Creator account ID
+        created_at: Creation time
+    """
+
+    __tablename__ = "workflow_comment_replies"
+    __table_args__ = (
+        db.PrimaryKeyConstraint("id", name="workflow_comment_replies_pkey"),
+        Index("comment_replies_comment_idx", "comment_id"),
+        Index("comment_replies_created_at_idx", "created_at"),
+    )
+
+    id: Mapped[str] = mapped_column(StringUUID, server_default=db.text("uuid_generate_v4()"))
+    comment_id: Mapped[str] = mapped_column(
+        StringUUID, db.ForeignKey("workflow_comments.id", ondelete="CASCADE"), nullable=False
+    )
+    content: Mapped[str] = mapped_column(db.Text, nullable=False)
+    created_by: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    created_at: Mapped[datetime] = mapped_column(db.DateTime, nullable=False, server_default=func.current_timestamp())
+    updated_at: Mapped[datetime] = mapped_column(
+        db.DateTime, nullable=False, server_default=func.current_timestamp(), onupdate=func.current_timestamp()
+    )
+    # Relationships
+    comment: Mapped["WorkflowComment"] = relationship("WorkflowComment", back_populates="replies")
+
+    @property
+    def created_by_account(self):
+        """Get creator account."""
+        return db.session.get(Account, self.created_by)
+
+
+class WorkflowCommentMention(Base):
+    """Workflow comment mention model.
+
+    Mentions are only for internal accounts since end users
+    cannot access workflow canvas and commenting features.
+
+    Attributes:
+        id: Mention ID
+        comment_id: Parent comment ID
+        mentioned_user_id: Mentioned account ID
+    """
+
+    __tablename__ = "workflow_comment_mentions"
+    __table_args__ = (
+        db.PrimaryKeyConstraint("id", name="workflow_comment_mentions_pkey"),
+        Index("comment_mentions_comment_idx", "comment_id"),
+        Index("comment_mentions_reply_idx", "reply_id"),
+        Index("comment_mentions_user_idx", "mentioned_user_id"),
+    )
+
+    id: Mapped[str] = mapped_column(StringUUID, server_default=db.text("uuid_generate_v4()"))
+    comment_id: Mapped[str] = mapped_column(
+        StringUUID, db.ForeignKey("workflow_comments.id", ondelete="CASCADE"), nullable=False
+    )
+    reply_id: Mapped[Optional[str]] = mapped_column(
+        StringUUID, db.ForeignKey("workflow_comment_replies.id", ondelete="CASCADE"), nullable=True
+    )
+    mentioned_user_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+
+    # Relationships
+    comment: Mapped["WorkflowComment"] = relationship("WorkflowComment", back_populates="mentions")
+    reply: Mapped[Optional["WorkflowCommentReply"]] = relationship("WorkflowCommentReply")
+
+    @property
+    def mentioned_user_account(self):
+        """Get mentioned account."""
+        return db.session.get(Account, self.mentioned_user_id)

api/models/enums.py

@@ -23,6 +23,7 @@ class DraftVariableType(StrEnum):
     NODE = "node"
     SYS = "sys"
     CONVERSATION = "conversation"
+    MEMORY_BLOCK = "memory_block"
 
 
 class MessageStatus(StrEnum):

api/models/model.py

@@ -3,13 +3,14 @@ import re
 import uuid
 from collections.abc import Mapping
 from datetime import datetime
+from decimal import Decimal
 from enum import StrEnum, auto
 from typing import TYPE_CHECKING, Any, Literal, Optional, cast
 
 import sqlalchemy as sa
 from flask import request
-from flask_login import UserMixin
-from sqlalchemy import Float, Index, PrimaryKeyConstraint, String, exists, func, select, text
+from flask_login import UserMixin  # type: ignore[import-untyped]
+from sqlalchemy import BigInteger, Float, Index, PrimaryKeyConstraint, String, exists, func, select, text
 from sqlalchemy.orm import Mapped, Session, mapped_column
 
 from configs import dify_config
@@ -581,6 +582,64 @@ class InstalledApp(Base):
         return tenant
 
 
+class TrialApp(Base):
+    __tablename__ = "trial_apps"
+    __table_args__ = (
+        sa.PrimaryKeyConstraint("id", name="trial_app_pkey"),
+        sa.Index("trial_app_app_id_idx", "app_id"),
+        sa.Index("trial_app_tenant_id_idx", "tenant_id"),
+        sa.UniqueConstraint("app_id", name="unique_trail_app_id"),
+    )
+
+    id = mapped_column(StringUUID, server_default=sa.text("uuid_generate_v4()"))
+    app_id = mapped_column(StringUUID, nullable=False)
+    tenant_id = mapped_column(StringUUID, nullable=False)
+    created_at = mapped_column(sa.DateTime, nullable=False, server_default=func.current_timestamp())
+    trial_limit = mapped_column(sa.Integer, nullable=False, default=3)
+
+    @property
+    def app(self) -> App | None:
+        app = db.session.query(App).where(App.id == self.app_id).first()
+        return app
+
+
+class AccountTrialAppRecord(Base):
+    __tablename__ = "account_trial_app_records"
+    __table_args__ = (
+        sa.PrimaryKeyConstraint("id", name="user_trial_app_pkey"),
+        sa.Index("account_trial_app_record_account_id_idx", "account_id"),
+        sa.Index("account_trial_app_record_app_id_idx", "app_id"),
+        sa.UniqueConstraint("account_id", "app_id", name="unique_account_trial_app_record"),
+    )
+    id = mapped_column(StringUUID, server_default=sa.text("uuid_generate_v4()"))
+    account_id = mapped_column(StringUUID, nullable=False)
+    app_id = mapped_column(StringUUID, nullable=False)
+    count = mapped_column(sa.Integer, nullable=False, default=0)
+    created_at = mapped_column(sa.DateTime, nullable=False, server_default=func.current_timestamp())
+
+    @property
+    def app(self) -> App | None:
+        app = db.session.query(App).where(App.id == self.app_id).first()
+        return app
+
+    @property
+    def user(self) -> Account | None:
+        user = db.session.query(Account).where(Account.id == self.account_id).first()
+        return user
+
+
+class ExporleBanner(Base):
+    __tablename__ = "exporle_banners"
+    __table_args__ = (sa.PrimaryKeyConstraint("id", name="exporler_banner_pkey"),)
+    id = mapped_column(StringUUID, server_default=sa.text("uuid_generate_v4()"))
+    content = mapped_column(sa.JSON, nullable=False)
+    link = mapped_column(String(255), nullable=False)
+    sort = mapped_column(sa.Integer, nullable=False)
+    status = mapped_column(sa.String(255), nullable=False, server_default=sa.text("'enabled'::character varying"))
+    created_at = mapped_column(sa.DateTime, nullable=False, server_default=func.current_timestamp())
+    language = mapped_column(String(255), nullable=False, server_default=sa.text("'en-US'::character varying"))
+
+
 class OAuthProviderApp(Base):
     """
     Globally shared OAuth provider app information.
@@ -914,34 +973,40 @@ class Message(Base):
     )
 
     id: Mapped[str] = mapped_column(StringUUID, server_default=sa.text("uuid_generate_v4()"))
-    app_id = mapped_column(StringUUID, nullable=False)
-    model_provider = mapped_column(String(255), nullable=True)
-    model_id = mapped_column(String(255), nullable=True)
-    override_model_configs = mapped_column(sa.Text)
-    conversation_id = mapped_column(StringUUID, sa.ForeignKey("conversations.id"), nullable=False)
+    app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    model_provider: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    model_id: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    override_model_configs: Mapped[str | None] = mapped_column(sa.Text)
+    conversation_id: Mapped[str] = mapped_column(StringUUID, sa.ForeignKey("conversations.id"), nullable=False)
     _inputs: Mapped[dict[str, Any]] = mapped_column("inputs", sa.JSON)
     query: Mapped[str] = mapped_column(sa.Text, nullable=False)
-    message = mapped_column(sa.JSON, nullable=False)
+    message: Mapped[dict[str, Any]] = mapped_column(sa.JSON, nullable=False)
     message_tokens: Mapped[int] = mapped_column(sa.Integer, nullable=False, server_default=sa.text("0"))
-    message_unit_price = mapped_column(sa.Numeric(10, 4), nullable=False)
-    message_price_unit = mapped_column(sa.Numeric(10, 7), nullable=False, server_default=sa.text("0.001"))
+    message_unit_price: Mapped[Decimal] = mapped_column(sa.Numeric(10, 4), nullable=False)
+    message_price_unit: Mapped[Decimal] = mapped_column(
+        sa.Numeric(10, 7), nullable=False, server_default=sa.text("0.001")
+    )
     answer: Mapped[str] = mapped_column(sa.Text, nullable=False)
     answer_tokens: Mapped[int] = mapped_column(sa.Integer, nullable=False, server_default=sa.text("0"))
-    answer_unit_price = mapped_column(sa.Numeric(10, 4), nullable=False)
-    answer_price_unit = mapped_column(sa.Numeric(10, 7), nullable=False, server_default=sa.text("0.001"))
-    parent_message_id = mapped_column(StringUUID, nullable=True)
-    provider_response_latency = mapped_column(sa.Float, nullable=False, server_default=sa.text("0"))
-    total_price = mapped_column(sa.Numeric(10, 7))
+    answer_unit_price: Mapped[Decimal] = mapped_column(sa.Numeric(10, 4), nullable=False)
+    answer_price_unit: Mapped[Decimal] = mapped_column(
+        sa.Numeric(10, 7), nullable=False, server_default=sa.text("0.001")
+    )
+    parent_message_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
+    provider_response_latency: Mapped[float] = mapped_column(sa.Float, nullable=False, server_default=sa.text("0"))
+    total_price: Mapped[Decimal | None] = mapped_column(sa.Numeric(10, 7))
     currency: Mapped[str] = mapped_column(String(255), nullable=False)
-    status = mapped_column(String(255), nullable=False, server_default=sa.text("'normal'::character varying"))
-    error = mapped_column(sa.Text)
-    message_metadata = mapped_column(sa.Text)
+    status: Mapped[str] = mapped_column(
+        String(255), nullable=False, server_default=sa.text("'normal'::character varying")
+    )
+    error: Mapped[str | None] = mapped_column(sa.Text)
+    message_metadata: Mapped[str | None] = mapped_column(sa.Text)
     invoke_from: Mapped[str | None] = mapped_column(String(255), nullable=True)
     from_source: Mapped[str] = mapped_column(String(255), nullable=False)
     from_end_user_id: Mapped[str | None] = mapped_column(StringUUID)
     from_account_id: Mapped[str | None] = mapped_column(StringUUID)
     created_at: Mapped[datetime] = mapped_column(sa.DateTime, server_default=func.current_timestamp())
-    updated_at = mapped_column(sa.DateTime, nullable=False, server_default=func.current_timestamp())
+    updated_at: Mapped[datetime] = mapped_column(sa.DateTime, nullable=False, server_default=func.current_timestamp())
     agent_based: Mapped[bool] = mapped_column(sa.Boolean, nullable=False, server_default=sa.text("false"))
     workflow_run_id: Mapped[str | None] = mapped_column(StringUUID)
     app_mode: Mapped[str | None] = mapped_column(String(255), nullable=True)
@@ -1212,9 +1277,13 @@ class Message(Base):
     @property
     def workflow_run(self):
         if self.workflow_run_id:
-            from .workflow import WorkflowRun
+            from sqlalchemy.orm import sessionmaker
 
-            return db.session.query(WorkflowRun).where(WorkflowRun.id == self.workflow_run_id).first()
+            from repositories.factory import DifyAPIRepositoryFactory
+
+            session_maker = sessionmaker(bind=db.engine, expire_on_commit=False)
+            repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker)
+            return repo.get_workflow_run_by_id_without_tenant(run_id=self.workflow_run_id)
 
         return None
 
@@ -1275,20 +1344,20 @@ class MessageFeedback(Base):
         sa.Index("message_feedback_conversation_idx", "conversation_id", "from_source", "rating"),
     )
 
-    id = mapped_column(StringUUID, server_default=sa.text("uuid_generate_v4()"))
-    app_id = mapped_column(StringUUID, nullable=False)
-    conversation_id = mapped_column(StringUUID, nullable=False)
-    message_id = mapped_column(StringUUID, nullable=False)
+    id: Mapped[str] = mapped_column(StringUUID, server_default=sa.text("uuid_generate_v4()"))
+    app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    conversation_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    message_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
     rating: Mapped[str] = mapped_column(String(255), nullable=False)
-    content = mapped_column(sa.Text)
+    content: Mapped[str | None] = mapped_column(sa.Text)
     from_source: Mapped[str] = mapped_column(String(255), nullable=False)
-    from_end_user_id = mapped_column(StringUUID)
-    from_account_id = mapped_column(StringUUID)
-    created_at = mapped_column(sa.DateTime, nullable=False, server_default=func.current_timestamp())
-    updated_at = mapped_column(sa.DateTime, nullable=False, server_default=func.current_timestamp())
+    from_end_user_id: Mapped[str | None] = mapped_column(StringUUID)
+    from_account_id: Mapped[str | None] = mapped_column(StringUUID)
+    created_at: Mapped[datetime] = mapped_column(sa.DateTime, nullable=False, server_default=func.current_timestamp())
+    updated_at: Mapped[datetime] = mapped_column(sa.DateTime, nullable=False, server_default=func.current_timestamp())
 
     @property
-    def from_account(self):
+    def from_account(self) -> Account | None:
         account = db.session.query(Account).where(Account.id == self.from_account_id).first()
         return account
 
@@ -1946,3 +2015,29 @@ class TraceAppConfig(Base):
             "created_at": str(self.created_at) if self.created_at else None,
             "updated_at": str(self.updated_at) if self.updated_at else None,
         }
+
+
+class TenantCreditPool(Base):
+    __tablename__ = "tenant_credit_pools"
+    __table_args__ = (
+        sa.PrimaryKeyConstraint("id", name="tenant_credit_pool_pkey"),
+        sa.Index("tenant_credit_pool_tenant_id_idx", "tenant_id"),
+        sa.Index("tenant_credit_pool_pool_type_idx", "pool_type"),
+    )
+
+    id = mapped_column(StringUUID, primary_key=True, server_default=text("uuid_generate_v4()"))
+    tenant_id = mapped_column(StringUUID, nullable=False)
+    pool_type = mapped_column(String(40), nullable=False, default="trial", server_default="trial")
+    quota_limit = mapped_column(BigInteger, nullable=False, default=0)
+    quota_used = mapped_column(BigInteger, nullable=False, default=0)
+    created_at = mapped_column(sa.DateTime, nullable=False, server_default=text("CURRENT_TIMESTAMP"))
+    updated_at = mapped_column(
+        sa.DateTime, nullable=False, server_default=func.current_timestamp(), onupdate=func.current_timestamp()
+    )
+
+    @property
+    def remaining_credits(self) -> int:
+        return max(0, self.quota_limit - self.quota_used)
+
+    def has_sufficient_credits(self, required_credits: int) -> bool:
+        return self.remaining_credits >= required_credits

api/models/workflow.py

@@ -1,5 +1,6 @@
 import json
 import logging
+import uuid
 from collections.abc import Mapping, Sequence
 from datetime import datetime
 from enum import StrEnum
@@ -11,10 +12,13 @@ from sqlalchemy import DateTime, Select, exists, orm, select
 
 from core.file.constants import maybe_file_object
 from core.file.models import File
+from core.memory.entities import MemoryBlockSpec
 from core.variables import utils as variable_utils
+from core.variables.segments import VersionedMemoryValue
 from core.variables.variables import FloatVariable, IntegerVariable, StringVariable
 from core.workflow.constants import (
     CONVERSATION_VARIABLE_NODE_ID,
+    MEMORY_BLOCK_VARIABLE_NODE_ID,
     SYSTEM_VARIABLE_NODE_ID,
 )
 from core.workflow.enums import NodeType, WorkflowExecutionStatus
@@ -152,6 +156,9 @@ class Workflow(Base):
     _rag_pipeline_variables: Mapped[str] = mapped_column(
         "rag_pipeline_variables", db.Text, nullable=False, server_default="{}"
     )
+    _memory_blocks: Mapped[str] = mapped_column(
+        "memory_blocks", sa.Text, nullable=False, server_default="[]"
+    )
 
     VERSION_DRAFT = "draft"
 
@@ -169,6 +176,7 @@ class Workflow(Base):
         environment_variables: Sequence[Variable],
         conversation_variables: Sequence[Variable],
         rag_pipeline_variables: list[dict],
+        memory_blocks: Sequence[MemoryBlockSpec] | None = None,
         marked_name: str = "",
         marked_comment: str = "",
     ) -> "Workflow":
@@ -184,6 +192,7 @@ class Workflow(Base):
         workflow.environment_variables = environment_variables or []
         workflow.conversation_variables = conversation_variables or []
         workflow.rag_pipeline_variables = rag_pipeline_variables or []
+        workflow.memory_blocks = memory_blocks or []
         workflow.marked_name = marked_name
         workflow.marked_comment = marked_comment
         workflow.created_at = naive_utc_now()
@@ -343,7 +352,7 @@ class Workflow(Base):
 
         :return: hash
         """
-        entity = {"graph": self.graph_dict, "features": self.features_dict}
+        entity = {"graph": self.graph_dict}
 
         return helper.generate_text_hash(json.dumps(entity, sort_keys=True))
 
@@ -452,7 +461,7 @@ class Workflow(Base):
             "features": self.features_dict,
             "environment_variables": [var.model_dump(mode="json") for var in environment_variables],
             "conversation_variables": [var.model_dump(mode="json") for var in self.conversation_variables],
-            "rag_pipeline_variables": self.rag_pipeline_variables,
+            "memory_blocks": [block.model_dump(mode="json") for block in self.memory_blocks],
         }
         return result
 
@@ -490,6 +499,27 @@ class Workflow(Base):
             ensure_ascii=False,
         )
 
+    @property
+    def memory_blocks(self) -> Sequence[MemoryBlockSpec]:
+        """Memory blocks configuration stored in database"""
+        if self._memory_blocks is None or self._memory_blocks == "":
+            self._memory_blocks = "[]"
+
+        memory_blocks_list: list[dict[str, Any]] = json.loads(self._memory_blocks)
+        results = [MemoryBlockSpec.model_validate(config) for config in memory_blocks_list]
+        return results
+
+    @memory_blocks.setter
+    def memory_blocks(self, value: Sequence[MemoryBlockSpec]):
+        if not value:
+            self._memory_blocks = "[]"
+            return
+
+        self._memory_blocks = json.dumps(
+            [block.model_dump() for block in value],
+            ensure_ascii=False,
+        )
+
     @staticmethod
     def version_from_datetime(d: datetime) -> str:
         return str(d)
@@ -1058,7 +1088,16 @@ class WorkflowAppLog(Base):
 
     @property
     def workflow_run(self):
-        return db.session.get(WorkflowRun, self.workflow_run_id)
+        if self.workflow_run_id:
+            from sqlalchemy.orm import sessionmaker
+
+            from repositories.factory import DifyAPIRepositoryFactory
+
+            session_maker = sessionmaker(bind=db.engine, expire_on_commit=False)
+            repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker)
+            return repo.get_workflow_run_by_id_without_tenant(run_id=self.workflow_run_id)
+
+        return None
 
     @property
     def created_by_account(self):
@@ -1512,6 +1551,31 @@ class WorkflowDraftVariable(Base):
         variable.editable = editable
         return variable
 
+    @staticmethod
+    def new_memory_block_variable(
+        *,
+        app_id: str,
+        node_id: str | None = None,
+        memory_id: str,
+        name: str,
+        value: VersionedMemoryValue,
+        description: str = "",
+    ) -> "WorkflowDraftVariable":
+        """Create a new memory block draft variable."""
+        return WorkflowDraftVariable(
+            id=str(uuid.uuid4()),
+            app_id=app_id,
+            node_id=MEMORY_BLOCK_VARIABLE_NODE_ID,
+            name=name,
+            value=value.model_dump_json(),
+            description=description,
+            selector=[MEMORY_BLOCK_VARIABLE_NODE_ID, memory_id] if node_id is None else
+            [MEMORY_BLOCK_VARIABLE_NODE_ID, memory_id, node_id],
+            value_type=SegmentType.VERSIONED_MEMORY,
+            visible=True,
+            editable=True,
+        )
+
     @property
     def edited(self):
         return self.last_edited_at is not None

api/pyproject.toml

@@ -20,6 +20,7 @@ dependencies = [
     "flask-orjson~=2.0.0",
     "flask-sqlalchemy~=3.1.1",
     "gevent~=25.9.1",
+    "gevent-websocket~=0.10.1",
     "gmpy2~=2.2.1",
     "google-api-core==2.18.0",
     "google-api-python-client==2.90.0",
@@ -37,6 +38,7 @@ dependencies = [
     "numpy~=1.26.4",
     "openpyxl~=3.1.5",
     "opik~=1.8.72",
+    "litellm==1.77.1",  # Pinned to avoid madoka dependency issue
     "opentelemetry-api==1.27.0",
     "opentelemetry-distro==0.48b0",
     "opentelemetry-exporter-otlp==1.27.0",
@@ -68,6 +70,7 @@ dependencies = [
     "pypdfium2==4.30.0",
     "python-docx~=1.1.0",
     "python-dotenv==1.0.1",
+    "python-socketio~=5.13.0",
     "pyyaml~=6.0.1",
     "readabilipy~=0.3.0",
     "redis[hiredis]~=6.1.0",
@@ -209,9 +212,9 @@ vdb = [
     "pgvector==0.2.5",
     "pymilvus~=2.5.0",
     "pymochow==2.2.9",
-    "pyobvector~=0.2.15",
+    "pyobvector~=0.2.17",
     "qdrant-client==1.9.0",
-    "tablestore==6.2.0",
+    "tablestore==6.3.7",
     "tcvectordb~=1.6.4",
     "tidb-vector==0.0.9",
     "upstash-vector==0.6.0",