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

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

.github/workflows/deploy-dev.yml

@@ -18,7 +18,7 @@ jobs:
       - name: Deploy to server
         uses: appleboy/ssh-action@v0.1.8
         with:
-          host: ${{ secrets.RAG_SSH_HOST }}
+          host: ${{ secrets.SSH_HOST }}
           username: ${{ secrets.SSH_USER }}
           key: ${{ secrets.SSH_PRIVATE_KEY }}
           script: |

api/.ruff.toml

@@ -81,7 +81,6 @@ ignore = [
     "SIM113",  # enumerate-for-loop
     "SIM117",  # multiple-with-statements
     "SIM210",  # if-expr-with-true-false
-    "UP038",   # deprecated and not recommended by Ruff, https://docs.astral.sh/ruff/rules/non-pep604-isinstance/
 ]
 
 [lint.per-file-ignores]

api/app.py

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

api/app_factory.py

@@ -31,14 +31,22 @@ def create_flask_app_with_configs() -> DifyApp:
     return dify_app
 
 
-def create_app() -> DifyApp:
+def create_app() -> tuple[any, DifyApp]:
     start_time = time.perf_counter()
     app = create_flask_app_with_configs()
     initialize_extensions(app)
+
+    import socketio
+
+    from extensions.ext_socketio import sio
+
+    sio.app = app
+    socketio_app = socketio.WSGIApp(sio, app)
+
     end_time = time.perf_counter()
     if dify_config.DEBUG:
         logger.info("Finished create_app (%s ms)", round((end_time - start_time) * 1000, 2))
-    return app
+    return socketio_app, app
 
 
 def initialize_extensions(app: DifyApp):

api/configs/feature/__init__.py

@@ -362,11 +362,11 @@ class HttpConfig(BaseSettings):
     )
 
     HTTP_REQUEST_MAX_READ_TIMEOUT: int = Field(
-        ge=1, description="Maximum read timeout in seconds for HTTP requests", default=60
+        ge=1, description="Maximum read timeout in seconds for HTTP requests", default=600
     )
 
     HTTP_REQUEST_MAX_WRITE_TIMEOUT: int = Field(
-        ge=1, description="Maximum write timeout in seconds for HTTP requests", default=20
+        ge=1, description="Maximum write timeout in seconds for HTTP requests", default=600
     )
 
     HTTP_REQUEST_NODE_MAX_BINARY_SIZE: PositiveInt = Field(
@@ -771,7 +771,7 @@ class MailConfig(BaseSettings):
 
     MAIL_TEMPLATING_TIMEOUT: int = Field(
         description="""
-        Timeout for email templating in seconds. Used to prevent infinite loops in malicious templates. 
+        Timeout for email templating in seconds. Used to prevent infinite loops in malicious templates.
         Only available in sandbox mode.""",
         default=3,
     )
@@ -836,6 +836,16 @@ class MailConfig(BaseSettings):
         default=None,
     )
 
+    ENABLE_TRIAL_APP: bool = Field(
+        description="Enable trial app",
+        default=False,
+    )
+
+    ENABLE_EXPLORE_BANNER: bool = Field(
+        description="Enable explore banner",
+        default=False,
+    )
+
 
 class RagEtlConfig(BaseSettings):
     """

api/configs/feature/hosted_service/__init__.py

@@ -8,6 +8,11 @@ class HostedCreditConfig(BaseSettings):
         default="",
     )
 
+    HOSTED_POOL_CREDITS: int = Field(
+        description="Pool credits for hosted service",
+        default=200,
+    )
+
     def get_model_credits(self, model_name: str) -> int:
         """
         Get credit value for a specific model name.
@@ -70,11 +75,6 @@ class HostedOpenAiConfig(BaseSettings):
         "text-davinci-003",
     )
 
-    HOSTED_OPENAI_QUOTA_LIMIT: NonNegativeInt = Field(
-        description="Quota limit for hosted OpenAI service usage",
-        default=200,
-    )
-
     HOSTED_OPENAI_PAID_ENABLED: bool = Field(
         description="Enable paid access to hosted OpenAI service",
         default=False,
@@ -98,6 +98,129 @@ class HostedOpenAiConfig(BaseSettings):
     )
 
 
+class HostedGeminiConfig(BaseSettings):
+    """
+    Configuration for fetching Gemini service
+    """
+
+    HOSTED_GEMINI_API_KEY: str | None = Field(
+        description="API key for hosted Gemini service",
+        default=None,
+    )
+
+    HOSTED_GEMINI_API_BASE: str | None = Field(
+        description="Base URL for hosted Gemini API",
+        default=None,
+    )
+
+    HOSTED_GEMINI_API_ORGANIZATION: str | None = Field(
+        description="Organization ID for hosted Gemini service",
+        default=None,
+    )
+
+    HOSTED_GEMINI_TRIAL_ENABLED: bool = Field(
+        description="Enable trial access to hosted Gemini service",
+        default=False,
+    )
+
+    HOSTED_GEMINI_TRIAL_MODELS: str = Field(
+        description="Comma-separated list of available models for trial access",
+        default="gemini-2.5-flash,gemini-2.0-flash,gemini-2.0-flash-lite,",
+    )
+
+    HOSTED_GEMINI_PAID_ENABLED: bool = Field(
+        description="Enable paid access to hosted gemini service",
+        default=False,
+    )
+
+    HOSTED_GEMINI_PAID_MODELS: str = Field(
+        description="Comma-separated list of available models for paid access",
+        default="gemini-2.5-flash,gemini-2.0-flash,gemini-2.0-flash-lite,",
+    )
+
+
+class HostedXAIConfig(BaseSettings):
+    """
+    Configuration for fetching XAI service
+    """
+
+    HOSTED_XAI_API_KEY: str | None = Field(
+        description="API key for hosted XAI service",
+        default=None,
+    )
+
+    HOSTED_XAI_API_BASE: str | None = Field(
+        description="Base URL for hosted XAI API",
+        default=None,
+    )
+
+    HOSTED_XAI_API_ORGANIZATION: str | None = Field(
+        description="Organization ID for hosted XAI service",
+        default=None,
+    )
+
+    HOSTED_XAI_TRIAL_ENABLED: bool = Field(
+        description="Enable trial access to hosted XAI service",
+        default=False,
+    )
+
+    HOSTED_XAI_TRIAL_MODELS: str = Field(
+        description="Comma-separated list of available models for trial access",
+        default="grok-3,grok-3-mini,grok-3-mini-fast",
+    )
+
+    HOSTED_XAI_PAID_ENABLED: bool = Field(
+        description="Enable paid access to hosted XAI service",
+        default=False,
+    )
+
+    HOSTED_XAI_PAID_MODELS: str = Field(
+        description="Comma-separated list of available models for paid access",
+        default="grok-3,grok-3-mini,grok-3-mini-fast",
+    )
+
+
+class HostedDeepseekConfig(BaseSettings):
+    """
+    Configuration for fetching Deepseek service
+    """
+
+    HOSTED_DEEPSEEK_API_KEY: str | None = Field(
+        description="API key for hosted Deepseek service",
+        default=None,
+    )
+
+    HOSTED_DEEPSEEK_API_BASE: str | None = Field(
+        description="Base URL for hosted Deepseek API",
+        default=None,
+    )
+
+    HOSTED_DEEPSEEK_API_ORGANIZATION: str | None = Field(
+        description="Organization ID for hosted Deepseek service",
+        default=None,
+    )
+
+    HOSTED_DEEPSEEK_TRIAL_ENABLED: bool = Field(
+        description="Enable trial access to hosted Deepseek service",
+        default=False,
+    )
+
+    HOSTED_DEEPSEEK_TRIAL_MODELS: str = Field(
+        description="Comma-separated list of available models for trial access",
+        default="deepseek-chat,deepseek-reasoner",
+    )
+
+    HOSTED_DEEPSEEK_PAID_ENABLED: bool = Field(
+        description="Enable paid access to hosted XAI service",
+        default=False,
+    )
+
+    HOSTED_DEEPSEEK_PAID_MODELS: str = Field(
+        description="Comma-separated list of available models for paid access",
+        default="grok-3,grok-3-mini,grok-3-mini-fast",
+    )
+
+
 class HostedAzureOpenAiConfig(BaseSettings):
     """
     Configuration for hosted Azure OpenAI service
@@ -144,16 +267,32 @@ class HostedAnthropicConfig(BaseSettings):
         default=False,
     )
 
-    HOSTED_ANTHROPIC_QUOTA_LIMIT: NonNegativeInt = Field(
-        description="Quota limit for hosted Anthropic service usage",
-        default=600000,
-    )
-
     HOSTED_ANTHROPIC_PAID_ENABLED: bool = Field(
         description="Enable paid access to hosted Anthropic service",
         default=False,
     )
 
+    HOSTED_ANTHROPIC_TRIAL_MODELS: str = Field(
+        description="Comma-separated list of available models for paid access",
+        default="claude-opus-4-20250514,"
+        "claude-opus-4-20250514,"
+        "claude-sonnet-4-20250514,"
+        "claude-3-5-haiku-20241022,"
+        "claude-3-opus-20240229,"
+        "claude-3-7-sonnet-20250219,"
+        "claude-3-haiku-20240307",
+    )
+    HOSTED_ANTHROPIC_PAID_MODELS: str = Field(
+        description="Comma-separated list of available models for paid access",
+        default="claude-opus-4-20250514,"
+        "claude-opus-4-20250514,"
+        "claude-sonnet-4-20250514,"
+        "claude-3-5-haiku-20241022,"
+        "claude-3-opus-20240229,"
+        "claude-3-7-sonnet-20250219,"
+        "claude-3-haiku-20240307",
+    )
+
 
 class HostedMinmaxConfig(BaseSettings):
     """
@@ -250,5 +389,8 @@ class HostedServiceConfig(
     HostedModerationConfig,
     # credit config
     HostedCreditConfig,
+    HostedGeminiConfig,
+    HostedXAIConfig,
+    HostedDeepseekConfig,
 ):
     pass

api/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

@@ -15,7 +15,7 @@ from constants.languages import supported_language
 from controllers.console import api, console_ns
 from controllers.console.wraps import only_edition_cloud
 from extensions.ext_database import db
-from models.model import App, InstalledApp, RecommendedApp
+from models.model import App, ExporleBanner, InstalledApp, RecommendedApp, TrialApp
 
 
 def admin_required(view: Callable[P, R]):
@@ -61,6 +61,8 @@ class InsertExploreAppListApi(Resource):
                 "language": fields.String(required=True, description="Language code"),
                 "category": fields.String(required=True, description="App category"),
                 "position": fields.Integer(required=True, description="Display position"),
+                "can_trial": fields.Boolean(required=True, description="Can trial"),
+                "trial_limit": fields.Integer(required=True, description="Trial limit"),
             },
         )
     )
@@ -79,6 +81,8 @@ class InsertExploreAppListApi(Resource):
         parser.add_argument("language", type=supported_language, required=True, nullable=False, location="json")
         parser.add_argument("category", type=str, required=True, nullable=False, location="json")
         parser.add_argument("position", type=int, required=True, nullable=False, location="json")
+        parser.add_argument("can_trial", type=bool, required=True, nullable=False, location="json")
+        parser.add_argument("trial_limit", type=int, required=True, nullable=False, location="json")
         args = parser.parse_args()
 
         app = db.session.execute(select(App).where(App.id == args["app_id"])).scalar_one_or_none()
@@ -115,6 +119,20 @@ class InsertExploreAppListApi(Resource):
                 )
 
                 db.session.add(recommended_app)
+                if args["can_trial"]:
+                    trial_app = db.session.execute(
+                        select(TrialApp).where(TrialApp.app_id == args["app_id"])
+                    ).scalar_one_or_none()
+                    if not trial_app:
+                        db.session.add(
+                            TrialApp(
+                                app_id=args["app_id"],
+                                tenant_id=app.tenant_id,
+                                trial_limit=args["trial_limit"],
+                            )
+                        )
+                    else:
+                        trial_app.trial_limit = args["trial_limit"]
 
                 app.is_public = True
                 db.session.commit()
@@ -129,6 +147,20 @@ class InsertExploreAppListApi(Resource):
                 recommended_app.category = args["category"]
                 recommended_app.position = args["position"]
 
+                if args["can_trial"]:
+                    trial_app = db.session.execute(
+                        select(TrialApp).where(TrialApp.app_id == args["app_id"])
+                    ).scalar_one_or_none()
+                    if not trial_app:
+                        db.session.add(
+                            TrialApp(
+                                app_id=args["app_id"],
+                                tenant_id=app.tenant_id,
+                                trial_limit=args["trial_limit"],
+                            )
+                        )
+                    else:
+                        trial_app.trial_limit = args["trial_limit"]
                 app.is_public = True
 
                 db.session.commit()
@@ -174,7 +206,67 @@ class InsertExploreAppApi(Resource):
             for installed_app in installed_apps:
                 session.delete(installed_app)
 
+            trial_app = session.execute(
+                select(TrialApp).where(TrialApp.app_id == recommended_app.app_id)
+            ).scalar_one_or_none()
+            if trial_app:
+                session.delete(trial_app)
+
         db.session.delete(recommended_app)
         db.session.commit()
 
         return {"result": "success"}, 204
+
+
+@console_ns.route("/admin/insert-explore-banner")
+class InsertExploreBanner(Resource):
+    @api.doc("insert_explore_banner")
+    @api.doc(description="Insert an explore banner")
+    @api.expect(
+        api.model(
+            "InsertExploreBannerRequest",
+            {
+                "content": fields.String(required=True, description="Banner content"),
+                "link": fields.String(required=True, description="Banner link"),
+                "sort": fields.Integer(required=True, description="Banner sort"),
+            },
+        )
+    )
+    @api.response(200, "Banner inserted successfully")
+    @admin_required
+    @only_edition_cloud
+    def post(self):
+        parser = reqparse.RequestParser()
+        parser.add_argument("content", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("link", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("sort", type=int, required=True, nullable=False, location="json")
+
+        args = parser.parse_args()
+
+        banner = ExporleBanner(
+            content=args["content"],
+            link=args["link"],
+            sort=args["sort"],
+        )
+        db.session.add(banner)
+        db.session.commit()
+
+        return {"result": "success"}, 200
+
+
+@console_ns.route("/admin/delete-explore-banner/<uuid:banner_id>")
+class DeleteExploreBanner(Resource):
+    @api.doc("delete_explore_banner")
+    @api.doc(description="Delete an explore banner")
+    @api.response(204, "Banner deleted successfully")
+    @admin_required
+    @only_edition_cloud
+    def delete(self, banner_id):
+        banner = db.session.execute(select(ExporleBanner).where(ExporleBanner.id == banner_id)).scalar_one_or_none()
+        if not banner:
+            raise NotFound(f"Banner '{banner_id}' is not found")
+
+        db.session.delete(banner)
+        db.session.commit()
+
+        return {"result": "success"}, 204

api/controllers/console/app/model_config.py

@@ -90,7 +90,7 @@ class ModelConfigResource(Resource):
                 if not isinstance(tool, dict) or len(tool.keys()) <= 3:
                     continue
 
-                agent_tool_entity = AgentToolEntity(**tool)
+                agent_tool_entity = AgentToolEntity.model_validate(tool)
                 # get tool
                 try:
                     tool_runtime = ToolManager.get_agent_tool_runtime(
@@ -124,7 +124,7 @@ class ModelConfigResource(Resource):
             # encrypt agent tool parameters if it's secret-input
             agent_mode = new_app_model_config.agent_mode_dict
             for tool in agent_mode.get("tools") or []:
-                agent_tool_entity = AgentToolEntity(**tool)
+                agent_tool_entity = AgentToolEntity.model_validate(tool)
 
                 # get tool
                 key = f"{agent_tool_entity.provider_id}.{agent_tool_entity.provider_type}.{agent_tool_entity.tool_name}"

api/controllers/console/app/online_user.py

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

api/controllers/console/app/workflow.py

@@ -5,6 +5,7 @@ from typing import cast
 
 from flask import abort, request
 from flask_restx import Resource, fields, inputs, marshal_with, reqparse
+from pydantic_core import ValidationError
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
 
@@ -21,7 +22,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
@@ -103,6 +106,7 @@ class DraftWorkflowApi(Resource):
                 "hash": fields.String(description="Workflow hash for validation"),
                 "environment_variables": fields.List(fields.Raw, required=True, description="Environment variables"),
                 "conversation_variables": fields.List(fields.Raw, description="Conversation variables"),
+                "memory_blocks": fields.List(fields.Raw, description="Memory blocks"),
             },
         )
     )
@@ -127,6 +131,8 @@ class DraftWorkflowApi(Resource):
             parser.add_argument("hash", type=str, required=False, location="json")
             parser.add_argument("environment_variables", type=list, required=True, location="json")
             parser.add_argument("conversation_variables", type=list, required=False, location="json")
+            parser.add_argument("force_upload", type=bool, required=False, default=False, location="json")
+            parser.add_argument("memory_blocks", type=list, required=False, location="json")
             args = parser.parse_args()
         elif "text/plain" in content_type:
             try:
@@ -143,6 +149,8 @@ class DraftWorkflowApi(Resource):
                     "hash": data.get("hash"),
                     "environment_variables": data.get("environment_variables"),
                     "conversation_variables": data.get("conversation_variables"),
+                    "memory_blocks": data.get("memory_blocks"),
+                    "force_upload": data.get("force_upload", False),
                 }
             except json.JSONDecodeError:
                 return {"message": "Invalid JSON data"}, 400
@@ -163,6 +171,11 @@ class DraftWorkflowApi(Resource):
             conversation_variables = [
                 variable_factory.build_conversation_variable_from_mapping(obj) for obj in conversation_variables_list
             ]
+            memory_blocks_list = args.get("memory_blocks") or []
+            from core.memory.entities import MemoryBlockSpec
+            memory_blocks = [
+                MemoryBlockSpec.model_validate(obj) for obj in memory_blocks_list
+            ]
             workflow = workflow_service.sync_draft_workflow(
                 app_model=app_model,
                 graph=args["graph"],
@@ -171,9 +184,13 @@ class DraftWorkflowApi(Resource):
                 account=current_user,
                 environment_variables=environment_variables,
                 conversation_variables=conversation_variables,
+                force_upload=args.get("force_upload", False),
+                memory_blocks=memory_blocks,
             )
         except WorkflowHashNotEqualError:
             raise DraftWorkflowNotSync()
+        except ValidationError as e:
+            return {"message": str(e)}, 400
 
         return {
             "result": "success",
@@ -796,6 +813,45 @@ class ConvertToWorkflowApi(Resource):
         }
 
 
+@console_ns.route("/apps/<uuid:app_id>/workflows/draft/config")
+class WorkflowConfigApi(Resource):
+    """Resource for workflow configuration."""
+
+    @api.doc("get_workflow_config")
+    @api.doc(description="Get workflow configuration")
+    @api.doc(params={"app_id": "Application ID"})
+    @api.response(200, "Workflow configuration retrieved successfully")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    def get(self, app_model: App):
+        return {
+            "parallel_depth_limit": dify_config.WORKFLOW_PARALLEL_DEPTH_LIMIT,
+        }
+
+
+class WorkflowFeaturesApi(Resource):
+    """Update draft workflow features."""
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    def post(self, app_model: App):
+        parser = reqparse.RequestParser()
+        parser.add_argument("features", type=dict, required=True, location="json")
+        args = parser.parse_args()
+
+        features = args.get("features")
+
+        # Update draft workflow features
+        workflow_service = WorkflowService()
+        workflow_service.update_draft_workflow_features(app_model=app_model, features=features, account=current_user)
+
+        return {"result": "success"}
+
+
 @console_ns.route("/apps/<uuid:app_id>/workflows")
 class PublishedAllWorkflowApi(Resource):
     @api.doc("get_all_published_workflows")
@@ -985,3 +1041,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 App, AppMode
 from models.account import Account
@@ -353,7 +353,7 @@ class VariableApi(Resource):
                 if len(raw_value) > 0 and not isinstance(raw_value[0], dict):
                     raise InvalidArgumentError(description=f"expected dict for files[0], got {type(raw_value)}")
                 raw_value = build_from_mappings(mappings=raw_value, tenant_id=app_model.tenant_id)
-            new_value = build_segment_with_type(variable.value_type, raw_value)
+            new_value = variable_factory.build_segment_with_type(variable.value_type, raw_value)
         draft_var_srv.update_variable(variable, name=new_name, value=new_value)
         db.session.commit()
         return variable
@@ -446,8 +446,35 @@ class ConversationVariableCollectionApi(Resource):
         db.session.commit()
         return _get_variable_list(app_model, CONVERSATION_VARIABLE_NODE_ID)
 
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=AppMode.ADVANCED_CHAT)
+    def post(self, app_model: App):
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not current_user.is_editor:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("conversation_variables", type=list, required=True, location="json")
+        args = parser.parse_args()
+
+        workflow_service = WorkflowService()
+
+        conversation_variables_list = args.get("conversation_variables") or []
+        conversation_variables = [
+            variable_factory.build_conversation_variable_from_mapping(obj) for obj in conversation_variables_list
+        ]
+
+        workflow_service.update_draft_workflow_conversation_variables(
+            app_model=app_model,
+            account=current_user,
+            conversation_variables=conversation_variables,
+        )
+
+        return {"result": "success"}
+
 
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/system-variables")
 class SystemVariableCollectionApi(Resource):
     @api.doc("get_system_variables")
     @api.doc(description="Get system variables for workflow")
@@ -497,3 +524,44 @@ class EnvironmentVariableCollectionApi(Resource):
             )
 
         return {"items": env_vars_list}
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    def post(self, app_model: App):
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not current_user.is_editor:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("environment_variables", type=list, required=True, location="json")
+        args = parser.parse_args()
+
+        workflow_service = WorkflowService()
+
+        environment_variables_list = args.get("environment_variables") or []
+        environment_variables = [
+            variable_factory.build_environment_variable_from_mapping(obj) for obj in environment_variables_list
+        ]
+
+        workflow_service.update_draft_workflow_environment_variables(
+            app_model=app_model,
+            account=current_user,
+            environment_variables=environment_variables,
+        )
+
+        return {"result": "success"}
+
+
+api.add_resource(
+    WorkflowVariableCollectionApi,
+    "/apps/<uuid:app_id>/workflows/draft/variables",
+)
+api.add_resource(NodeVariableCollectionApi, "/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/variables")
+api.add_resource(VariableApi, "/apps/<uuid:app_id>/workflows/draft/variables/<uuid:variable_id>")
+api.add_resource(VariableResetApi, "/apps/<uuid:app_id>/workflows/draft/variables/<uuid:variable_id>/reset")
+
+api.add_resource(ConversationVariableCollectionApi, "/apps/<uuid:app_id>/workflows/draft/conversation-variables")
+api.add_resource(SystemVariableCollectionApi, "/apps/<uuid:app_id>/workflows/draft/system-variables")
+api.add_resource(EnvironmentVariableCollectionApi, "/apps/<uuid:app_id>/workflows/draft/environment-variables")

api/controllers/console/datasets/data_source.py

@@ -15,7 +15,7 @@ from core.datasource.entities.datasource_entities import DatasourceProviderType,
 from core.datasource.online_document.online_document_plugin import OnlineDocumentDatasourcePlugin
 from core.indexing_runner import IndexingRunner
 from core.rag.extractor.entity.datasource_type import DatasourceType
-from core.rag.extractor.entity.extract_setting import ExtractSetting
+from core.rag.extractor.entity.extract_setting import ExtractSetting, NotionInfo
 from core.rag.extractor.notion_extractor import NotionExtractor
 from extensions.ext_database import db
 from fields.data_source_fields import integrate_list_fields, integrate_notion_info_list_fields
@@ -257,13 +257,15 @@ class DataSourceNotionApi(Resource):
             for page in notion_info["pages"]:
                 extract_setting = ExtractSetting(
                     datasource_type=DatasourceType.NOTION.value,
-                    notion_info={
-                        "credential_id": credential_id,
-                        "notion_workspace_id": workspace_id,
-                        "notion_obj_id": page["page_id"],
-                        "notion_page_type": page["type"],
-                        "tenant_id": current_user.current_tenant_id,
-                    },
+                    notion_info=NotionInfo.model_validate(
+                        {
+                            "credential_id": credential_id,
+                            "notion_workspace_id": workspace_id,
+                            "notion_obj_id": page["page_id"],
+                            "notion_page_type": page["type"],
+                            "tenant_id": current_user.current_tenant_id,
+                        }
+                    ),
                     document_model=args["doc_form"],
                 )
                 extract_settings.append(extract_setting)

api/controllers/console/datasets/datasets.py

@@ -24,7 +24,7 @@ from core.model_runtime.entities.model_entities import ModelType
 from core.provider_manager import ProviderManager
 from core.rag.datasource.vdb.vector_type import VectorType
 from core.rag.extractor.entity.datasource_type import DatasourceType
-from core.rag.extractor.entity.extract_setting import ExtractSetting
+from core.rag.extractor.entity.extract_setting import ExtractSetting, NotionInfo, WebsiteInfo
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
 from extensions.ext_database import db
 from fields.app_fields import related_app_list
@@ -513,13 +513,15 @@ class DatasetIndexingEstimateApi(Resource):
                 for page in notion_info["pages"]:
                     extract_setting = ExtractSetting(
                         datasource_type=DatasourceType.NOTION.value,
-                        notion_info={
-                            "credential_id": credential_id,
-                            "notion_workspace_id": workspace_id,
-                            "notion_obj_id": page["page_id"],
-                            "notion_page_type": page["type"],
-                            "tenant_id": current_user.current_tenant_id,
-                        },
+                        notion_info=NotionInfo.model_validate(
+                            {
+                                "credential_id": credential_id,
+                                "notion_workspace_id": workspace_id,
+                                "notion_obj_id": page["page_id"],
+                                "notion_page_type": page["type"],
+                                "tenant_id": current_user.current_tenant_id,
+                            }
+                        ),
                         document_model=args["doc_form"],
                     )
                     extract_settings.append(extract_setting)
@@ -528,14 +530,16 @@ class DatasetIndexingEstimateApi(Resource):
             for url in website_info_list["urls"]:
                 extract_setting = ExtractSetting(
                     datasource_type=DatasourceType.WEBSITE.value,
-                    website_info={
-                        "provider": website_info_list["provider"],
-                        "job_id": website_info_list["job_id"],
-                        "url": url,
-                        "tenant_id": current_user.current_tenant_id,
-                        "mode": "crawl",
-                        "only_main_content": website_info_list["only_main_content"],
-                    },
+                    website_info=WebsiteInfo.model_validate(
+                        {
+                            "provider": website_info_list["provider"],
+                            "job_id": website_info_list["job_id"],
+                            "url": url,
+                            "tenant_id": current_user.current_tenant_id,
+                            "mode": "crawl",
+                            "only_main_content": website_info_list["only_main_content"],
+                        }
+                    ),
                     document_model=args["doc_form"],
                 )
                 extract_settings.append(extract_setting)

api/controllers/console/datasets/datasets_document.py

@@ -44,7 +44,7 @@ from core.model_runtime.entities.model_entities import ModelType
 from core.model_runtime.errors.invoke import InvokeAuthorizationError
 from core.plugin.impl.exc import PluginDaemonClientSideError
 from core.rag.extractor.entity.datasource_type import DatasourceType
-from core.rag.extractor.entity.extract_setting import ExtractSetting
+from core.rag.extractor.entity.extract_setting import ExtractSetting, NotionInfo, WebsiteInfo
 from extensions.ext_database import db
 from fields.document_fields import (
     dataset_and_document_fields,
@@ -305,7 +305,7 @@ class DatasetDocumentListApi(Resource):
             "doc_language", type=str, default="English", required=False, nullable=False, location="json"
         )
         args = parser.parse_args()
-        knowledge_config = KnowledgeConfig(**args)
+        knowledge_config = KnowledgeConfig.model_validate(args)
 
         if not dataset.indexing_technique and not knowledge_config.indexing_technique:
             raise ValueError("indexing_technique is required.")
@@ -395,7 +395,7 @@ class DatasetInitApi(Resource):
         parser.add_argument("embedding_model_provider", type=str, required=False, nullable=True, location="json")
         args = parser.parse_args()
 
-        knowledge_config = KnowledgeConfig(**args)
+        knowledge_config = KnowledgeConfig.model_validate(args)
         if knowledge_config.indexing_technique == "high_quality":
             if knowledge_config.embedding_model is None or knowledge_config.embedding_model_provider is None:
                 raise ValueError("embedding model and embedding model provider are required for high quality indexing.")
@@ -547,13 +547,15 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
                     continue
                 extract_setting = ExtractSetting(
                     datasource_type=DatasourceType.NOTION.value,
-                    notion_info={
-                        "credential_id": data_source_info["credential_id"],
-                        "notion_workspace_id": data_source_info["notion_workspace_id"],
-                        "notion_obj_id": data_source_info["notion_page_id"],
-                        "notion_page_type": data_source_info["type"],
-                        "tenant_id": current_user.current_tenant_id,
-                    },
+                    notion_info=NotionInfo.model_validate(
+                        {
+                            "credential_id": data_source_info["credential_id"],
+                            "notion_workspace_id": data_source_info["notion_workspace_id"],
+                            "notion_obj_id": data_source_info["notion_page_id"],
+                            "notion_page_type": data_source_info["type"],
+                            "tenant_id": current_user.current_tenant_id,
+                        }
+                    ),
                     document_model=document.doc_form,
                 )
                 extract_settings.append(extract_setting)
@@ -562,14 +564,16 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
                     continue
                 extract_setting = ExtractSetting(
                     datasource_type=DatasourceType.WEBSITE.value,
-                    website_info={
-                        "provider": data_source_info["provider"],
-                        "job_id": data_source_info["job_id"],
-                        "url": data_source_info["url"],
-                        "tenant_id": current_user.current_tenant_id,
-                        "mode": data_source_info["mode"],
-                        "only_main_content": data_source_info["only_main_content"],
-                    },
+                    website_info=WebsiteInfo.model_validate(
+                        {
+                            "provider": data_source_info["provider"],
+                            "job_id": data_source_info["job_id"],
+                            "url": data_source_info["url"],
+                            "tenant_id": current_user.current_tenant_id,
+                            "mode": data_source_info["mode"],
+                            "only_main_content": data_source_info["only_main_content"],
+                        }
+                    ),
                     document_model=document.doc_form,
                 )
                 extract_settings.append(extract_setting)

api/controllers/console/datasets/datasets_segments.py

@@ -309,7 +309,7 @@ class DatasetDocumentSegmentUpdateApi(Resource):
         )
         args = parser.parse_args()
         SegmentService.segment_create_args_validate(args, document)
-        segment = SegmentService.update_segment(SegmentUpdateArgs(**args), segment, document, dataset)
+        segment = SegmentService.update_segment(SegmentUpdateArgs.model_validate(args), segment, document, dataset)
         return {"data": marshal(segment, segment_fields), "doc_form": document.doc_form}, 200
 
     @setup_required
@@ -564,7 +564,7 @@ class ChildChunkAddApi(Resource):
         args = parser.parse_args()
         try:
             chunks_data = args["chunks"]
-            chunks = [ChildChunkUpdateArgs(**chunk) for chunk in chunks_data]
+            chunks = [ChildChunkUpdateArgs.model_validate(chunk) for chunk in chunks_data]
             child_chunks = SegmentService.update_child_chunks(chunks, segment, document, dataset)
         except ChildChunkIndexingServiceError as e:
             raise ChildChunkIndexingError(str(e))

api/controllers/console/datasets/metadata.py

@@ -28,7 +28,7 @@ class DatasetMetadataCreateApi(Resource):
         parser.add_argument("type", type=str, required=True, nullable=False, location="json")
         parser.add_argument("name", type=str, required=True, nullable=False, location="json")
         args = parser.parse_args()
-        metadata_args = MetadataArgs(**args)
+        metadata_args = MetadataArgs.model_validate(args)
 
         dataset_id_str = str(dataset_id)
         dataset = DatasetService.get_dataset(dataset_id_str)
@@ -137,7 +137,7 @@ class DocumentMetadataEditApi(Resource):
         parser = reqparse.RequestParser()
         parser.add_argument("operation_data", type=list, required=True, nullable=False, location="json")
         args = parser.parse_args()
-        metadata_args = MetadataOperationData(**args)
+        metadata_args = MetadataOperationData.model_validate(args)
 
         MetadataService.update_documents_metadata(dataset, metadata_args)
 

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

@@ -88,7 +88,7 @@ class CustomizedPipelineTemplateApi(Resource):
             nullable=True,
         )
         args = parser.parse_args()
-        pipeline_template_info = PipelineTemplateInfoEntity(**args)
+        pipeline_template_info = PipelineTemplateInfoEntity.model_validate(args)
         RagPipelineService.update_customized_pipeline_template(template_id, pipeline_template_info)
         return 200
 

api/controllers/console/explore/banner.py

@@ -0,0 +1,34 @@
+from flask_restx import Resource
+
+from controllers.console import api
+from controllers.console.explore.wraps import explore_banner_enabled
+from extensions.ext_database import db
+from models.model import ExporleBanner
+
+
+class BannerApi(Resource):
+    """Resource for banner list."""
+
+    @explore_banner_enabled
+    def get(self):
+        """Get banner list."""
+        banners = (
+            db.session.query(ExporleBanner).where(ExporleBanner.status == "enabled").order_by(ExporleBanner.sort).all()
+        )
+
+        # Convert banners to serializable format
+        result = []
+        for banner in banners:
+            banner_data = {
+                "content": banner.content,  # Already parsed as JSON by SQLAlchemy
+                "link": banner.link,
+                "sort": banner.sort,
+                "status": banner.status,
+                "created_at": banner.created_at.isoformat() if banner.created_at else None,
+            }
+            result.append(banner_data)
+
+        return result
+
+
+api.add_resource(BannerApi, "/explore/banners")

api/controllers/console/explore/error.py

@@ -29,3 +29,25 @@ class AppAccessDeniedError(BaseHTTPException):
     error_code = "access_denied"
     description = "App access denied."
     code = 403
+
+
+class TrialAppNotAllowed(BaseHTTPException):
+    """*403* `Trial App Not Allowed`
+
+    Raise if the user has reached the trial app limit.
+    """
+
+    error_code = "trial_app_not_allowed"
+    code = 403
+    description = "the app is not allowed to be trial."
+
+
+class TrialAppLimitExceeded(BaseHTTPException):
+    """*403* `Trial App Limit Exceeded`
+
+    Raise if the user has exceeded the trial app limit.
+    """
+
+    error_code = "trial_app_limit_exceeded"
+    code = 403
+    description = "The user has exceeded the trial app limit."

api/controllers/console/explore/installed_app.py

@@ -6,7 +6,7 @@ from flask_restx import Resource, inputs, marshal_with, reqparse
 from sqlalchemy import and_, select
 from werkzeug.exceptions import BadRequest, Forbidden, NotFound
 
-from controllers.console import api
+from controllers.console import console_ns
 from controllers.console.explore.wraps import InstalledAppResource
 from controllers.console.wraps import account_initialization_required, cloud_edition_billing_resource_check
 from extensions.ext_database import db
@@ -22,6 +22,7 @@ from services.feature_service import FeatureService
 logger = logging.getLogger(__name__)
 
 
+@console_ns.route("/installed-apps")
 class InstalledAppsListApi(Resource):
     @login_required
     @account_initialization_required
@@ -154,6 +155,7 @@ class InstalledAppsListApi(Resource):
         return {"message": "App installed successfully"}
 
 
+@console_ns.route("/installed-apps/<uuid:installed_app_id>")
 class InstalledAppApi(InstalledAppResource):
     """
     update and delete an installed app
@@ -185,7 +187,3 @@ class InstalledAppApi(InstalledAppResource):
             db.session.commit()
 
         return {"result": "success", "message": "App info updated successfully"}
-
-
-api.add_resource(InstalledAppsListApi, "/installed-apps")
-api.add_resource(InstalledAppApi, "/installed-apps/<uuid:installed_app_id>")

api/controllers/console/explore/parameter.py

@@ -1,7 +1,7 @@
 from flask_restx import marshal_with
 
 from controllers.common import fields
-from controllers.console import api
+from controllers.console import console_ns
 from controllers.console.app.error import AppUnavailableError
 from controllers.console.explore.wraps import InstalledAppResource
 from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
@@ -9,6 +9,7 @@ from models.model import AppMode, InstalledApp
 from services.app_service import AppService
 
 
+@console_ns.route("/installed-apps/<uuid:installed_app_id>/parameters", endpoint="installed_app_parameters")
 class AppParameterApi(InstalledAppResource):
     """Resource for app variables."""
 
@@ -39,6 +40,7 @@ class AppParameterApi(InstalledAppResource):
         return get_parameters_from_feature_dict(features_dict=features_dict, user_input_form=user_input_form)
 
 
+@console_ns.route("/installed-apps/<uuid:installed_app_id>/meta", endpoint="installed_app_meta")
 class ExploreAppMetaApi(InstalledAppResource):
     def get(self, installed_app: InstalledApp):
         """Get app meta"""
@@ -46,9 +48,3 @@ class ExploreAppMetaApi(InstalledAppResource):
         if not app_model:
             raise ValueError("App not found")
         return AppService().get_app_meta(app_model)
-
-
-api.add_resource(
-    AppParameterApi, "/installed-apps/<uuid:installed_app_id>/parameters", endpoint="installed_app_parameters"
-)
-api.add_resource(ExploreAppMetaApi, "/installed-apps/<uuid:installed_app_id>/meta", endpoint="installed_app_meta")

api/controllers/console/explore/recommended_app.py

@@ -1,7 +1,7 @@
 from flask_restx import Resource, fields, marshal_with, reqparse
 
 from constants.languages import languages
-from controllers.console import api
+from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required
 from libs.helper import AppIconUrlField
 from libs.login import current_user, login_required
@@ -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 = {
@@ -35,6 +36,7 @@ recommended_app_list_fields = {
 }
 
 
+@console_ns.route("/explore/apps")
 class RecommendedAppListApi(Resource):
     @login_required
     @account_initialization_required
@@ -56,13 +58,10 @@ class RecommendedAppListApi(Resource):
         return RecommendedAppService.get_recommended_apps_and_categories(language_prefix)
 
 
+@console_ns.route("/explore/apps/<uuid:app_id>")
 class RecommendedAppApi(Resource):
     @login_required
     @account_initialization_required
     def get(self, app_id):
         app_id = str(app_id)
         return RecommendedAppService.get_recommend_app_detail(app_id)
-
-
-api.add_resource(RecommendedAppListApi, "/explore/apps")
-api.add_resource(RecommendedAppApi, "/explore/apps/<uuid:app_id>")

api/controllers/console/explore/saved_message.py

@@ -2,7 +2,7 @@ from flask_restx import fields, marshal_with, reqparse
 from flask_restx.inputs import int_range
 from werkzeug.exceptions import NotFound
 
-from controllers.console import api
+from controllers.console import console_ns
 from controllers.console.explore.error import NotCompletionAppError
 from controllers.console.explore.wraps import InstalledAppResource
 from fields.conversation_fields import message_file_fields
@@ -25,6 +25,7 @@ message_fields = {
 }
 
 
+@console_ns.route("/installed-apps/<uuid:installed_app_id>/saved-messages", endpoint="installed_app_saved_messages")
 class SavedMessageListApi(InstalledAppResource):
     saved_message_infinite_scroll_pagination_fields = {
         "limit": fields.Integer,
@@ -66,6 +67,9 @@ class SavedMessageListApi(InstalledAppResource):
         return {"result": "success"}
 
 
+@console_ns.route(
+    "/installed-apps/<uuid:installed_app_id>/saved-messages/<uuid:message_id>", endpoint="installed_app_saved_message"
+)
 class SavedMessageApi(InstalledAppResource):
     def delete(self, installed_app, message_id):
         app_model = installed_app.app
@@ -80,15 +84,3 @@ class SavedMessageApi(InstalledAppResource):
         SavedMessageService.delete(app_model, current_user, message_id)
 
         return {"result": "success"}, 204
-
-
-api.add_resource(
-    SavedMessageListApi,
-    "/installed-apps/<uuid:installed_app_id>/saved-messages",
-    endpoint="installed_app_saved_messages",
-)
-api.add_resource(
-    SavedMessageApi,
-    "/installed-apps/<uuid:installed_app_id>/saved-messages/<uuid:message_id>",
-    endpoint="installed_app_saved_message",
-)

api/controllers/console/explore/trial.py

@@ -0,0 +1,375 @@
+import logging
+
+from flask import request
+from flask_restx import Resource, marshal_with, reqparse
+from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
+
+import services
+from controllers.common import fields
+from controllers.common.fields import build_site_model
+from controllers.console import api
+from controllers.console.app.error import (
+    AppUnavailableError,
+    AudioTooLargeError,
+    CompletionRequestError,
+    ConversationCompletedError,
+    NoAudioUploadedError,
+    ProviderModelCurrentlyNotSupportError,
+    ProviderNotInitializeError,
+    ProviderNotSupportSpeechToTextError,
+    ProviderQuotaExceededError,
+    UnsupportedAudioTypeError,
+)
+from controllers.console.app.wraps import get_app_model
+from controllers.console.explore.error import (
+    AppSuggestedQuestionsAfterAnswerDisabledError,
+    NotChatAppError,
+    NotCompletionAppError,
+)
+from controllers.console.explore.wraps import TrialAppResource, trial_feature_enable
+from controllers.service_api import service_api_ns
+from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
+from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
+from core.app.entities.app_invoke_entities import InvokeFrom
+from core.errors.error import (
+    ModelCurrentlyNotSupportError,
+    ProviderTokenNotInitError,
+    QuotaExceededError,
+)
+from core.model_runtime.errors.invoke import InvokeError
+from extensions.ext_database import db
+from fields.app_fields import app_detail_fields_with_site
+from libs import helper
+from libs.helper import uuid_value
+from libs.login import current_user
+from models import Account
+from models.account import TenantStatus
+from models.model import AppMode, Site
+from services.app_generate_service import AppGenerateService
+from services.app_service import AppService
+from services.audio_service import AudioService
+from services.errors.audio import (
+    AudioTooLargeServiceError,
+    NoAudioUploadedServiceError,
+    ProviderNotSupportSpeechToTextServiceError,
+    UnsupportedAudioTypeServiceError,
+)
+from services.errors.conversation import ConversationNotExistsError
+from services.errors.llm import InvokeRateLimitError
+from services.errors.message import (
+    MessageNotExistsError,
+    SuggestedQuestionsAfterAnswerDisabledError,
+)
+from services.message_service import MessageService
+from services.recommended_app_service import RecommendedAppService
+
+logger = logging.getLogger(__name__)
+
+
+class TrialChatApi(TrialAppResource):
+    @trial_feature_enable
+    def post(self, trial_app):
+        app_model = trial_app
+        app_mode = AppMode.value_of(app_model.mode)
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
+            raise NotChatAppError()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("inputs", type=dict, required=True, location="json")
+        parser.add_argument("query", type=str, required=True, location="json")
+        parser.add_argument("files", type=list, required=False, location="json")
+        parser.add_argument("conversation_id", type=uuid_value, location="json")
+        parser.add_argument("parent_message_id", type=uuid_value, required=False, location="json")
+        parser.add_argument("retriever_from", type=str, required=False, default="explore_app", location="json")
+        args = parser.parse_args()
+
+        args["auto_generate_name"] = False
+
+        try:
+            if not isinstance(current_user, Account):
+                raise ValueError("current_user must be an Account instance")
+            response = AppGenerateService.generate(
+                app_model=app_model, user=current_user, args=args, invoke_from=InvokeFrom.EXPLORE, streaming=True
+            )
+            RecommendedAppService.add_trial_app_record(app_model.id, current_user.id)
+            return helper.compact_generate_response(response)
+        except services.errors.conversation.ConversationNotExistsError:
+            raise NotFound("Conversation Not Exists.")
+        except services.errors.conversation.ConversationCompletedError:
+            raise ConversationCompletedError()
+        except services.errors.app_model_config.AppModelConfigBrokenError:
+            logger.exception("App model config broken.")
+            raise AppUnavailableError()
+        except ProviderTokenNotInitError as ex:
+            raise ProviderNotInitializeError(ex.description)
+        except QuotaExceededError:
+            raise ProviderQuotaExceededError()
+        except ModelCurrentlyNotSupportError:
+            raise ProviderModelCurrentlyNotSupportError()
+        except InvokeError as e:
+            raise CompletionRequestError(e.description)
+        except InvokeRateLimitError as ex:
+            raise InvokeRateLimitHttpError(ex.description)
+        except ValueError as e:
+            raise e
+        except Exception:
+            logger.exception("internal server error.")
+            raise InternalServerError()
+
+
+class TrialMessageSuggestedQuestionApi(TrialAppResource):
+    @trial_feature_enable
+    def get(self, trial_app, message_id):
+        app_model = trial_app
+        app_mode = AppMode.value_of(app_model.mode)
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
+            raise NotChatAppError()
+
+        message_id = str(message_id)
+
+        try:
+            if not isinstance(current_user, Account):
+                raise ValueError("current_user must be an Account instance")
+            questions = MessageService.get_suggested_questions_after_answer(
+                app_model=app_model, user=current_user, message_id=message_id, invoke_from=InvokeFrom.EXPLORE
+            )
+        except MessageNotExistsError:
+            raise NotFound("Message not found")
+        except ConversationNotExistsError:
+            raise NotFound("Conversation not found")
+        except SuggestedQuestionsAfterAnswerDisabledError:
+            raise AppSuggestedQuestionsAfterAnswerDisabledError()
+        except ProviderTokenNotInitError as ex:
+            raise ProviderNotInitializeError(ex.description)
+        except QuotaExceededError:
+            raise ProviderQuotaExceededError()
+        except ModelCurrentlyNotSupportError:
+            raise ProviderModelCurrentlyNotSupportError()
+        except InvokeError as e:
+            raise CompletionRequestError(e.description)
+        except Exception:
+            logger.exception("internal server error.")
+            raise InternalServerError()
+
+        return {"data": questions}
+
+
+class TrialChatAudioApi(TrialAppResource):
+    @trial_feature_enable
+    def post(self, trial_app):
+        app_model = trial_app
+
+        file = request.files["file"]
+
+        try:
+            if not isinstance(current_user, Account):
+                raise ValueError("current_user must be an Account instance")
+            response = AudioService.transcript_asr(app_model=app_model, file=file, end_user=None)
+            RecommendedAppService.add_trial_app_record(app_model.id, current_user.id)
+            return response
+        except services.errors.app_model_config.AppModelConfigBrokenError:
+            logger.exception("App model config broken.")
+            raise AppUnavailableError()
+        except NoAudioUploadedServiceError:
+            raise NoAudioUploadedError()
+        except AudioTooLargeServiceError as e:
+            raise AudioTooLargeError(str(e))
+        except UnsupportedAudioTypeServiceError:
+            raise UnsupportedAudioTypeError()
+        except ProviderNotSupportSpeechToTextServiceError:
+            raise ProviderNotSupportSpeechToTextError()
+        except ProviderTokenNotInitError as ex:
+            raise ProviderNotInitializeError(ex.description)
+        except QuotaExceededError:
+            raise ProviderQuotaExceededError()
+        except ModelCurrentlyNotSupportError:
+            raise ProviderModelCurrentlyNotSupportError()
+        except InvokeError as e:
+            raise CompletionRequestError(e.description)
+        except ValueError as e:
+            raise e
+        except Exception as e:
+            logger.exception("internal server error.")
+            raise InternalServerError()
+
+
+class TrialChatTextApi(TrialAppResource):
+    @trial_feature_enable
+    def post(self, trial_app):
+        app_model = trial_app
+        try:
+            parser = reqparse.RequestParser()
+            parser.add_argument("message_id", type=str, required=False, location="json")
+            parser.add_argument("voice", type=str, location="json")
+            parser.add_argument("text", type=str, location="json")
+            parser.add_argument("streaming", type=bool, location="json")
+            args = parser.parse_args()
+
+            message_id = args.get("message_id", None)
+            text = args.get("text", None)
+            voice = args.get("voice", None)
+            if not isinstance(current_user, Account):
+                raise ValueError("current_user must be an Account instance")
+            response = AudioService.transcript_tts(app_model=app_model, text=text, voice=voice, message_id=message_id)
+            RecommendedAppService.add_trial_app_record(app_model.id, current_user.id)
+            return response
+        except services.errors.app_model_config.AppModelConfigBrokenError:
+            logger.exception("App model config broken.")
+            raise AppUnavailableError()
+        except NoAudioUploadedServiceError:
+            raise NoAudioUploadedError()
+        except AudioTooLargeServiceError as e:
+            raise AudioTooLargeError(str(e))
+        except UnsupportedAudioTypeServiceError:
+            raise UnsupportedAudioTypeError()
+        except ProviderNotSupportSpeechToTextServiceError:
+            raise ProviderNotSupportSpeechToTextError()
+        except ProviderTokenNotInitError as ex:
+            raise ProviderNotInitializeError(ex.description)
+        except QuotaExceededError:
+            raise ProviderQuotaExceededError()
+        except ModelCurrentlyNotSupportError:
+            raise ProviderModelCurrentlyNotSupportError()
+        except InvokeError as e:
+            raise CompletionRequestError(e.description)
+        except ValueError as e:
+            raise e
+        except Exception as e:
+            logger.exception("internal server error.")
+            raise InternalServerError()
+
+
+class TrialCompletionApi(TrialAppResource):
+    @trial_feature_enable
+    def post(self, trial_app):
+        app_model = trial_app
+        if app_model.mode != "completion":
+            raise NotCompletionAppError()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("inputs", type=dict, required=True, location="json")
+        parser.add_argument("query", type=str, location="json", default="")
+        parser.add_argument("files", type=list, required=False, location="json")
+        parser.add_argument("response_mode", type=str, choices=["blocking", "streaming"], location="json")
+        parser.add_argument("retriever_from", type=str, required=False, default="explore_app", location="json")
+        args = parser.parse_args()
+
+        streaming = args["response_mode"] == "streaming"
+        args["auto_generate_name"] = False
+
+        try:
+            if not isinstance(current_user, Account):
+                raise ValueError("current_user must be an Account instance")
+            response = AppGenerateService.generate(
+                app_model=app_model, user=current_user, args=args, invoke_from=InvokeFrom.EXPLORE, streaming=streaming
+            )
+
+            RecommendedAppService.add_trial_app_record(app_model.id, current_user.id)
+            return helper.compact_generate_response(response)
+        except services.errors.conversation.ConversationNotExistsError:
+            raise NotFound("Conversation Not Exists.")
+        except services.errors.conversation.ConversationCompletedError:
+            raise ConversationCompletedError()
+        except services.errors.app_model_config.AppModelConfigBrokenError:
+            logger.exception("App model config broken.")
+            raise AppUnavailableError()
+        except ProviderTokenNotInitError as ex:
+            raise ProviderNotInitializeError(ex.description)
+        except QuotaExceededError:
+            raise ProviderQuotaExceededError()
+        except ModelCurrentlyNotSupportError:
+            raise ProviderModelCurrentlyNotSupportError()
+        except InvokeError as e:
+            raise CompletionRequestError(e.description)
+        except ValueError as e:
+            raise e
+        except Exception:
+            logger.exception("internal server error.")
+            raise InternalServerError()
+
+
+class TrialSitApi(Resource):
+    """Resource for trial app sites."""
+
+    @trial_feature_enable
+    @get_app_model
+    @service_api_ns.marshal_with(build_site_model(service_api_ns))
+    def get(self, app_model):
+        """Retrieve app site info.
+
+        Returns the site configuration for the application including theme, icons, and text.
+        """
+        site = db.session.query(Site).where(Site.app_id == app_model.id).first()
+
+        if not site:
+            raise Forbidden()
+
+        assert app_model.tenant
+        if app_model.tenant.status == TenantStatus.ARCHIVE:
+            raise Forbidden()
+
+        return site
+
+
+class TrialAppParameterApi(Resource):
+    """Resource for app variables."""
+
+    @trial_feature_enable
+    @get_app_model
+    @marshal_with(fields.parameters_fields)
+    def get(self, app_model):
+        """Retrieve app parameters."""
+
+        if app_model is None:
+            raise AppUnavailableError()
+
+        if app_model.mode in {AppMode.ADVANCED_CHAT, AppMode.WORKFLOW}:
+            workflow = app_model.workflow
+            if workflow is None:
+                raise AppUnavailableError()
+
+            features_dict = workflow.features_dict
+            user_input_form = workflow.user_input_form(to_old_structure=True)
+        else:
+            app_model_config = app_model.app_model_config
+            if app_model_config is None:
+                raise AppUnavailableError()
+
+            features_dict = app_model_config.to_dict()
+
+            user_input_form = features_dict.get("user_input_form", [])
+
+        return get_parameters_from_feature_dict(features_dict=features_dict, user_input_form=user_input_form)
+
+
+class AppApi(Resource):
+    @trial_feature_enable
+    @get_app_model
+    @marshal_with(app_detail_fields_with_site)
+    def get(self, app_model):
+        """Get app detail"""
+
+        app_service = AppService()
+        app_model = app_service.get_app(app_model)
+
+        return app_model
+
+
+api.add_resource(TrialChatApi, "/trial-apps/<uuid:app_id>/chat-messages", endpoint="trial_app_chat_completion")
+
+api.add_resource(
+    TrialMessageSuggestedQuestionApi,
+    "/trial-apps/<uuid:app_id>/messages/<uuid:message_id>/suggested-questions",
+    endpoint="trial_app_suggested_question",
+)
+
+api.add_resource(TrialChatAudioApi, "/trial-apps/<uuid:app_id>/audio-to-text", endpoint="trial_app_audio")
+api.add_resource(TrialChatTextApi, "/trial-apps/<uuid:app_id>/text-to-audio", endpoint="trial_app_text")
+
+api.add_resource(TrialCompletionApi, "/trial-apps/<uuid:app_id>/completion-messages", endpoint="trial_app_completion")
+
+api.add_resource(TrialSitApi, "/trial-apps/<uuid:app_id>/site")
+
+api.add_resource(TrialAppParameterApi, "/trial-apps/<uuid:app_id>/parameters", endpoint="trial_app_parameters")
+
+api.add_resource(AppApi, "/trial-apps/<uuid:app_id>", endpoint="trial_app")

api/controllers/console/explore/wraps.py

@@ -2,15 +2,16 @@ from collections.abc import Callable
 from functools import wraps
 from typing import Concatenate, ParamSpec, TypeVar
 
+from flask import abort
 from flask_login import current_user
 from flask_restx import Resource
 from werkzeug.exceptions import NotFound
 
-from controllers.console.explore.error import AppAccessDeniedError
+from controllers.console.explore.error import AppAccessDeniedError, TrialAppLimitExceeded, TrialAppNotAllowed
 from controllers.console.wraps import account_initialization_required
 from extensions.ext_database import db
 from libs.login import login_required
-from models import InstalledApp
+from models import AccountTrialAppRecord, App, InstalledApp, TrialApp
 from services.app_service import AppService
 from services.enterprise.enterprise_service import EnterpriseService
 from services.feature_service import FeatureService
@@ -74,6 +75,59 @@ def user_allowed_to_access_app(view: Callable[Concatenate[InstalledApp, P], R] |
     return decorator
 
 
+def trial_app_required(view: Callable[Concatenate[App, P], R] | None = None):
+    def decorator(view: Callable[Concatenate[App, P], R]):
+        @wraps(view)
+        def decorated(app_id: str, *args: P.args, **kwargs: P.kwargs):
+            trial_app = db.session.query(TrialApp).where(TrialApp.app_id == str(app_id)).first()
+
+            if trial_app is None:
+                raise TrialAppNotAllowed()
+            app = trial_app.app
+
+            if app is None:
+                raise TrialAppNotAllowed()
+
+            account_trial_app_record = (
+                db.session.query(AccountTrialAppRecord)
+                .where(AccountTrialAppRecord.account_id == current_user.id, AccountTrialAppRecord.app_id == app_id)
+                .first()
+            )
+            if account_trial_app_record:
+                if account_trial_app_record.count >= trial_app.trial_limit:
+                    raise TrialAppLimitExceeded()
+
+            return view(app, *args, **kwargs)
+
+        return decorated
+
+    if view:
+        return decorator(view)
+    return decorator
+
+
+def trial_feature_enable(view: Callable[..., R]) -> Callable[..., R]:
+    @wraps(view)
+    def decorated(*args, **kwargs):
+        features = FeatureService.get_system_features()
+        if not features.enable_trial_app:
+            abort(403, "Trial app feature is not enabled.")
+        return view(*args, **kwargs)
+
+    return decorated
+
+
+def explore_banner_enabled(view: Callable[..., R]) -> Callable[..., R]:
+    @wraps(view)
+    def decorated(*args, **kwargs):
+        features = FeatureService.get_system_features()
+        if not features.enable_explore_banner:
+            abort(403, "Explore banner feature is not enabled.")
+        return view(*args, **kwargs)
+
+    return decorated
+
+
 class InstalledAppResource(Resource):
     # must be reversed if there are multiple decorators
 
@@ -83,3 +137,13 @@ class InstalledAppResource(Resource):
         account_initialization_required,
         login_required,
     ]
+
+
+class TrialAppResource(Resource):
+    # must be reversed if there are multiple decorators
+
+    method_decorators = [
+        trial_app_required,
+        account_initialization_required,
+        login_required,
+    ]

api/controllers/console/workspace/account.py

@@ -9,7 +9,7 @@ from sqlalchemy.orm import Session
 
 from configs import dify_config
 from constants.languages import supported_language
-from controllers.console import api
+from controllers.console import console_ns
 from controllers.console.auth.error import (
     EmailAlreadyInUseError,
     EmailChangeLimitError,
@@ -33,6 +33,7 @@ from controllers.console.wraps import (
     only_edition_cloud,
     setup_required,
 )
+from core.file import helpers as file_helpers
 from extensions.ext_database import db
 from fields.member_fields import account_fields
 from libs.datetime_utils import naive_utc_now
@@ -45,6 +46,7 @@ from services.billing_service import BillingService
 from services.errors.account import CurrentPasswordIncorrectError as ServiceCurrentPasswordIncorrectError
 
 
+@console_ns.route("/account/init")
 class AccountInitApi(Resource):
     @setup_required
     @login_required
@@ -97,6 +99,7 @@ class AccountInitApi(Resource):
         return {"result": "success"}
 
 
+@console_ns.route("/account/profile")
 class AccountProfileApi(Resource):
     @setup_required
     @login_required
@@ -109,6 +112,7 @@ class AccountProfileApi(Resource):
         return current_user
 
 
+@console_ns.route("/account/name")
 class AccountNameApi(Resource):
     @setup_required
     @login_required
@@ -130,7 +134,19 @@ class AccountNameApi(Resource):
         return updated_account
 
 
+@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
@@ -147,6 +163,7 @@ class AccountAvatarApi(Resource):
         return updated_account
 
 
+@console_ns.route("/account/interface-language")
 class AccountInterfaceLanguageApi(Resource):
     @setup_required
     @login_required
@@ -164,6 +181,7 @@ class AccountInterfaceLanguageApi(Resource):
         return updated_account
 
 
+@console_ns.route("/account/interface-theme")
 class AccountInterfaceThemeApi(Resource):
     @setup_required
     @login_required
@@ -181,6 +199,7 @@ class AccountInterfaceThemeApi(Resource):
         return updated_account
 
 
+@console_ns.route("/account/timezone")
 class AccountTimezoneApi(Resource):
     @setup_required
     @login_required
@@ -202,6 +221,7 @@ class AccountTimezoneApi(Resource):
         return updated_account
 
 
+@console_ns.route("/account/password")
 class AccountPasswordApi(Resource):
     @setup_required
     @login_required
@@ -227,6 +247,7 @@ class AccountPasswordApi(Resource):
         return {"result": "success"}
 
 
+@console_ns.route("/account/integrates")
 class AccountIntegrateApi(Resource):
     integrate_fields = {
         "provider": fields.String,
@@ -283,6 +304,7 @@ class AccountIntegrateApi(Resource):
         return {"data": integrate_data}
 
 
+@console_ns.route("/account/delete/verify")
 class AccountDeleteVerifyApi(Resource):
     @setup_required
     @login_required
@@ -298,6 +320,7 @@ class AccountDeleteVerifyApi(Resource):
         return {"result": "success", "data": token}
 
 
+@console_ns.route("/account/delete")
 class AccountDeleteApi(Resource):
     @setup_required
     @login_required
@@ -320,6 +343,7 @@ class AccountDeleteApi(Resource):
         return {"result": "success"}
 
 
+@console_ns.route("/account/delete/feedback")
 class AccountDeleteUpdateFeedbackApi(Resource):
     @setup_required
     def post(self):
@@ -333,6 +357,7 @@ class AccountDeleteUpdateFeedbackApi(Resource):
         return {"result": "success"}
 
 
+@console_ns.route("/account/education/verify")
 class EducationVerifyApi(Resource):
     verify_fields = {
         "token": fields.String,
@@ -352,6 +377,7 @@ class EducationVerifyApi(Resource):
         return BillingService.EducationIdentity.verify(account.id, account.email)
 
 
+@console_ns.route("/account/education")
 class EducationApi(Resource):
     status_fields = {
         "result": fields.Boolean,
@@ -396,6 +422,7 @@ class EducationApi(Resource):
         return res
 
 
+@console_ns.route("/account/education/autocomplete")
 class EducationAutoCompleteApi(Resource):
     data_fields = {
         "data": fields.List(fields.String),
@@ -419,6 +446,7 @@ class EducationAutoCompleteApi(Resource):
         return BillingService.EducationIdentity.autocomplete(args["keywords"], args["page"], args["limit"])
 
 
+@console_ns.route("/account/change-email")
 class ChangeEmailSendEmailApi(Resource):
     @enable_change_email
     @setup_required
@@ -467,6 +495,7 @@ class ChangeEmailSendEmailApi(Resource):
         return {"result": "success", "data": token}
 
 
+@console_ns.route("/account/change-email/validity")
 class ChangeEmailCheckApi(Resource):
     @enable_change_email
     @setup_required
@@ -508,6 +537,7 @@ class ChangeEmailCheckApi(Resource):
         return {"is_valid": True, "email": token_data.get("email"), "token": new_token}
 
 
+@console_ns.route("/account/change-email/reset")
 class ChangeEmailResetApi(Resource):
     @enable_change_email
     @setup_required
@@ -547,6 +577,7 @@ class ChangeEmailResetApi(Resource):
         return updated_account
 
 
+@console_ns.route("/account/change-email/check-email-unique")
 class CheckEmailUnique(Resource):
     @setup_required
     def post(self):
@@ -558,28 +589,3 @@ class CheckEmailUnique(Resource):
         if not AccountService.check_email_unique(args["email"]):
             raise EmailAlreadyInUseError()
         return {"result": "success"}
-
-
-# Register API resources
-api.add_resource(AccountInitApi, "/account/init")
-api.add_resource(AccountProfileApi, "/account/profile")
-api.add_resource(AccountNameApi, "/account/name")
-api.add_resource(AccountAvatarApi, "/account/avatar")
-api.add_resource(AccountInterfaceLanguageApi, "/account/interface-language")
-api.add_resource(AccountInterfaceThemeApi, "/account/interface-theme")
-api.add_resource(AccountTimezoneApi, "/account/timezone")
-api.add_resource(AccountPasswordApi, "/account/password")
-api.add_resource(AccountIntegrateApi, "/account/integrates")
-api.add_resource(AccountDeleteVerifyApi, "/account/delete/verify")
-api.add_resource(AccountDeleteApi, "/account/delete")
-api.add_resource(AccountDeleteUpdateFeedbackApi, "/account/delete/feedback")
-api.add_resource(EducationVerifyApi, "/account/education/verify")
-api.add_resource(EducationApi, "/account/education")
-api.add_resource(EducationAutoCompleteApi, "/account/education/autocomplete")
-# Change email
-api.add_resource(ChangeEmailSendEmailApi, "/account/change-email")
-api.add_resource(ChangeEmailCheckApi, "/account/change-email/validity")
-api.add_resource(ChangeEmailResetApi, "/account/change-email/reset")
-api.add_resource(CheckEmailUnique, "/account/change-email/check-email-unique")
-# api.add_resource(AccountEmailApi, '/account/email')
-# api.add_resource(AccountEmailVerifyApi, '/account/email-verify')

api/controllers/console/workspace/load_balancing_config.py

@@ -1,7 +1,7 @@
 from flask_restx import Resource, reqparse
 from werkzeug.exceptions import Forbidden
 
-from controllers.console import api
+from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, setup_required
 from core.model_runtime.entities.model_entities import ModelType
 from core.model_runtime.errors.validate import CredentialsValidateFailedError
@@ -10,6 +10,9 @@ from models.account import Account, TenantAccountRole
 from services.model_load_balancing_service import ModelLoadBalancingService
 
 
+@console_ns.route(
+    "/workspaces/current/model-providers/<path:provider>/models/load-balancing-configs/credentials-validate"
+)
 class LoadBalancingCredentialsValidateApi(Resource):
     @setup_required
     @login_required
@@ -61,6 +64,9 @@ class LoadBalancingCredentialsValidateApi(Resource):
         return response
 
 
+@console_ns.route(
+    "/workspaces/current/model-providers/<path:provider>/models/load-balancing-configs/<string:config_id>/credentials-validate"
+)
 class LoadBalancingConfigCredentialsValidateApi(Resource):
     @setup_required
     @login_required
@@ -111,15 +117,3 @@ class LoadBalancingConfigCredentialsValidateApi(Resource):
             response["error"] = error
 
         return response
-
-
-# Load Balancing Config
-api.add_resource(
-    LoadBalancingCredentialsValidateApi,
-    "/workspaces/current/model-providers/<path:provider>/models/load-balancing-configs/credentials-validate",
-)
-
-api.add_resource(
-    LoadBalancingConfigCredentialsValidateApi,
-    "/workspaces/current/model-providers/<path:provider>/models/load-balancing-configs/<string:config_id>/credentials-validate",
-)

api/controllers/console/workspace/members.py

@@ -6,7 +6,7 @@ from flask_restx import Resource, marshal_with, reqparse
 
 import services
 from configs import dify_config
-from controllers.console import api
+from controllers.console import console_ns
 from controllers.console.auth.error import (
     CannotTransferOwnerToSelfError,
     EmailCodeError,
@@ -33,6 +33,7 @@ from services.errors.account import AccountAlreadyInTenantError
 from services.feature_service import FeatureService
 
 
+@console_ns.route("/workspaces/current/members")
 class MemberListApi(Resource):
     """List all members of current tenant."""
 
@@ -49,6 +50,7 @@ class MemberListApi(Resource):
         return {"result": "success", "accounts": members}, 200
 
 
+@console_ns.route("/workspaces/current/members/invite-email")
 class MemberInviteEmailApi(Resource):
     """Invite a new member by email."""
 
@@ -111,6 +113,7 @@ class MemberInviteEmailApi(Resource):
         }, 201
 
 
+@console_ns.route("/workspaces/current/members/<uuid:member_id>")
 class MemberCancelInviteApi(Resource):
     """Cancel an invitation by member id."""
 
@@ -143,6 +146,7 @@ class MemberCancelInviteApi(Resource):
         }, 200
 
 
+@console_ns.route("/workspaces/current/members/<uuid:member_id>/update-role")
 class MemberUpdateRoleApi(Resource):
     """Update member role."""
 
@@ -177,6 +181,7 @@ class MemberUpdateRoleApi(Resource):
         return {"result": "success"}
 
 
+@console_ns.route("/workspaces/current/dataset-operators")
 class DatasetOperatorMemberListApi(Resource):
     """List all members of current tenant."""
 
@@ -193,6 +198,7 @@ class DatasetOperatorMemberListApi(Resource):
         return {"result": "success", "accounts": members}, 200
 
 
+@console_ns.route("/workspaces/current/members/send-owner-transfer-confirm-email")
 class SendOwnerTransferEmailApi(Resource):
     """Send owner transfer email."""
 
@@ -233,6 +239,7 @@ class SendOwnerTransferEmailApi(Resource):
         return {"result": "success", "data": token}
 
 
+@console_ns.route("/workspaces/current/members/owner-transfer-check")
 class OwnerTransferCheckApi(Resource):
     @setup_required
     @login_required
@@ -278,6 +285,7 @@ class OwnerTransferCheckApi(Resource):
         return {"is_valid": True, "email": token_data.get("email"), "token": new_token}
 
 
+@console_ns.route("/workspaces/current/members/<uuid:member_id>/owner-transfer")
 class OwnerTransfer(Resource):
     @setup_required
     @login_required
@@ -339,14 +347,3 @@ class OwnerTransfer(Resource):
             raise ValueError(str(e))
 
         return {"result": "success"}
-
-
-api.add_resource(MemberListApi, "/workspaces/current/members")
-api.add_resource(MemberInviteEmailApi, "/workspaces/current/members/invite-email")
-api.add_resource(MemberCancelInviteApi, "/workspaces/current/members/<uuid:member_id>")
-api.add_resource(MemberUpdateRoleApi, "/workspaces/current/members/<uuid:member_id>/update-role")
-api.add_resource(DatasetOperatorMemberListApi, "/workspaces/current/dataset-operators")
-# owner transfer
-api.add_resource(SendOwnerTransferEmailApi, "/workspaces/current/members/send-owner-transfer-confirm-email")
-api.add_resource(OwnerTransferCheckApi, "/workspaces/current/members/owner-transfer-check")
-api.add_resource(OwnerTransfer, "/workspaces/current/members/<uuid:member_id>/owner-transfer")

api/controllers/console/workspace/model_providers.py

@@ -5,7 +5,7 @@ from flask_login import current_user
 from flask_restx import Resource, reqparse
 from werkzeug.exceptions import Forbidden
 
-from controllers.console import api
+from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, setup_required
 from core.model_runtime.entities.model_entities import ModelType
 from core.model_runtime.errors.validate import CredentialsValidateFailedError
@@ -17,6 +17,7 @@ from services.billing_service import BillingService
 from services.model_provider_service import ModelProviderService
 
 
+@console_ns.route("/workspaces/current/model-providers")
 class ModelProviderListApi(Resource):
     @setup_required
     @login_required
@@ -45,6 +46,7 @@ class ModelProviderListApi(Resource):
         return jsonable_encoder({"data": provider_list})
 
 
+@console_ns.route("/workspaces/current/model-providers/<path:provider>/credentials")
 class ModelProviderCredentialApi(Resource):
     @setup_required
     @login_required
@@ -151,6 +153,7 @@ class ModelProviderCredentialApi(Resource):
         return {"result": "success"}, 204
 
 
+@console_ns.route("/workspaces/current/model-providers/<path:provider>/credentials/switch")
 class ModelProviderCredentialSwitchApi(Resource):
     @setup_required
     @login_required
@@ -175,6 +178,7 @@ class ModelProviderCredentialSwitchApi(Resource):
         return {"result": "success"}
 
 
+@console_ns.route("/workspaces/current/model-providers/<path:provider>/credentials/validate")
 class ModelProviderValidateApi(Resource):
     @setup_required
     @login_required
@@ -211,6 +215,7 @@ class ModelProviderValidateApi(Resource):
         return response
 
 
+@console_ns.route("/workspaces/<string:tenant_id>/model-providers/<path:provider>/<string:icon_type>/<string:lang>")
 class ModelProviderIconApi(Resource):
     """
     Get model provider icon
@@ -229,6 +234,7 @@ class ModelProviderIconApi(Resource):
         return send_file(io.BytesIO(icon), mimetype=mimetype)
 
 
+@console_ns.route("/workspaces/current/model-providers/<path:provider>/preferred-provider-type")
 class PreferredProviderTypeUpdateApi(Resource):
     @setup_required
     @login_required
@@ -262,6 +268,7 @@ class PreferredProviderTypeUpdateApi(Resource):
         return {"result": "success"}
 
 
+@console_ns.route("/workspaces/current/model-providers/<path:provider>/checkout-url")
 class ModelProviderPaymentCheckoutUrlApi(Resource):
     @setup_required
     @login_required
@@ -281,21 +288,3 @@ class ModelProviderPaymentCheckoutUrlApi(Resource):
             prefilled_email=current_user.email,
         )
         return data
-
-
-api.add_resource(ModelProviderListApi, "/workspaces/current/model-providers")
-
-api.add_resource(ModelProviderCredentialApi, "/workspaces/current/model-providers/<path:provider>/credentials")
-api.add_resource(
-    ModelProviderCredentialSwitchApi, "/workspaces/current/model-providers/<path:provider>/credentials/switch"
-)
-api.add_resource(ModelProviderValidateApi, "/workspaces/current/model-providers/<path:provider>/credentials/validate")
-
-api.add_resource(
-    PreferredProviderTypeUpdateApi, "/workspaces/current/model-providers/<path:provider>/preferred-provider-type"
-)
-api.add_resource(ModelProviderPaymentCheckoutUrlApi, "/workspaces/current/model-providers/<path:provider>/checkout-url")
-api.add_resource(
-    ModelProviderIconApi,
-    "/workspaces/<string:tenant_id>/model-providers/<path:provider>/<string:icon_type>/<string:lang>",
-)

api/controllers/console/workspace/models.py

@@ -4,7 +4,7 @@ from flask_login import current_user
 from flask_restx import Resource, reqparse
 from werkzeug.exceptions import Forbidden
 
-from controllers.console import api
+from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, setup_required
 from core.model_runtime.entities.model_entities import ModelType
 from core.model_runtime.errors.validate import CredentialsValidateFailedError
@@ -17,6 +17,7 @@ from services.model_provider_service import ModelProviderService
 logger = logging.getLogger(__name__)
 
 
+@console_ns.route("/workspaces/current/default-model")
 class DefaultModelApi(Resource):
     @setup_required
     @login_required
@@ -85,6 +86,7 @@ class DefaultModelApi(Resource):
         return {"result": "success"}
 
 
+@console_ns.route("/workspaces/current/model-providers/<path:provider>/models")
 class ModelProviderModelApi(Resource):
     @setup_required
     @login_required
@@ -187,6 +189,7 @@ class ModelProviderModelApi(Resource):
         return {"result": "success"}, 204
 
 
+@console_ns.route("/workspaces/current/model-providers/<path:provider>/models/credentials")
 class ModelProviderModelCredentialApi(Resource):
     @setup_required
     @login_required
@@ -364,6 +367,7 @@ class ModelProviderModelCredentialApi(Resource):
         return {"result": "success"}, 204
 
 
+@console_ns.route("/workspaces/current/model-providers/<path:provider>/models/credentials/switch")
 class ModelProviderModelCredentialSwitchApi(Resource):
     @setup_required
     @login_required
@@ -395,6 +399,9 @@ class ModelProviderModelCredentialSwitchApi(Resource):
         return {"result": "success"}
 
 
+@console_ns.route(
+    "/workspaces/current/model-providers/<path:provider>/models/enable", endpoint="model-provider-model-enable"
+)
 class ModelProviderModelEnableApi(Resource):
     @setup_required
     @login_required
@@ -422,6 +429,9 @@ class ModelProviderModelEnableApi(Resource):
         return {"result": "success"}
 
 
+@console_ns.route(
+    "/workspaces/current/model-providers/<path:provider>/models/disable", endpoint="model-provider-model-disable"
+)
 class ModelProviderModelDisableApi(Resource):
     @setup_required
     @login_required
@@ -449,6 +459,7 @@ class ModelProviderModelDisableApi(Resource):
         return {"result": "success"}
 
 
+@console_ns.route("/workspaces/current/model-providers/<path:provider>/models/credentials/validate")
 class ModelProviderModelValidateApi(Resource):
     @setup_required
     @login_required
@@ -494,6 +505,7 @@ class ModelProviderModelValidateApi(Resource):
         return response
 
 
+@console_ns.route("/workspaces/current/model-providers/<path:provider>/models/parameter-rules")
 class ModelProviderModelParameterRuleApi(Resource):
     @setup_required
     @login_required
@@ -513,6 +525,7 @@ class ModelProviderModelParameterRuleApi(Resource):
         return jsonable_encoder({"data": parameter_rules})
 
 
+@console_ns.route("/workspaces/current/models/model-types/<string:model_type>")
 class ModelProviderAvailableModelApi(Resource):
     @setup_required
     @login_required
@@ -524,32 +537,3 @@ class ModelProviderAvailableModelApi(Resource):
         models = model_provider_service.get_models_by_model_type(tenant_id=tenant_id, model_type=model_type)
 
         return jsonable_encoder({"data": models})
-
-
-api.add_resource(ModelProviderModelApi, "/workspaces/current/model-providers/<path:provider>/models")
-api.add_resource(
-    ModelProviderModelEnableApi,
-    "/workspaces/current/model-providers/<path:provider>/models/enable",
-    endpoint="model-provider-model-enable",
-)
-api.add_resource(
-    ModelProviderModelDisableApi,
-    "/workspaces/current/model-providers/<path:provider>/models/disable",
-    endpoint="model-provider-model-disable",
-)
-api.add_resource(
-    ModelProviderModelCredentialApi, "/workspaces/current/model-providers/<path:provider>/models/credentials"
-)
-api.add_resource(
-    ModelProviderModelCredentialSwitchApi,
-    "/workspaces/current/model-providers/<path:provider>/models/credentials/switch",
-)
-api.add_resource(
-    ModelProviderModelValidateApi, "/workspaces/current/model-providers/<path:provider>/models/credentials/validate"
-)
-
-api.add_resource(
-    ModelProviderModelParameterRuleApi, "/workspaces/current/model-providers/<path:provider>/models/parameter-rules"
-)
-api.add_resource(ModelProviderAvailableModelApi, "/workspaces/current/models/model-types/<string:model_type>")
-api.add_resource(DefaultModelApi, "/workspaces/current/default-model")

api/controllers/console/workspace/plugin.py

@@ -6,7 +6,7 @@ from flask_restx import Resource, reqparse
 from werkzeug.exceptions import Forbidden
 
 from configs import dify_config
-from controllers.console import api
+from controllers.console import console_ns
 from controllers.console.workspace import plugin_permission_required
 from controllers.console.wraps import account_initialization_required, setup_required
 from core.model_runtime.utils.encoders import jsonable_encoder
@@ -19,6 +19,7 @@ from services.plugin.plugin_permission_service import PluginPermissionService
 from services.plugin.plugin_service import PluginService
 
 
+@console_ns.route("/workspaces/current/plugin/debugging-key")
 class PluginDebuggingKeyApi(Resource):
     @setup_required
     @login_required
@@ -37,6 +38,7 @@ class PluginDebuggingKeyApi(Resource):
             raise ValueError(e)
 
 
+@console_ns.route("/workspaces/current/plugin/list")
 class PluginListApi(Resource):
     @setup_required
     @login_required
@@ -55,6 +57,7 @@ class PluginListApi(Resource):
         return jsonable_encoder({"plugins": plugins_with_total.list, "total": plugins_with_total.total})
 
 
+@console_ns.route("/workspaces/current/plugin/list/latest-versions")
 class PluginListLatestVersionsApi(Resource):
     @setup_required
     @login_required
@@ -72,6 +75,7 @@ class PluginListLatestVersionsApi(Resource):
         return jsonable_encoder({"versions": versions})
 
 
+@console_ns.route("/workspaces/current/plugin/list/installations/ids")
 class PluginListInstallationsFromIdsApi(Resource):
     @setup_required
     @login_required
@@ -91,6 +95,7 @@ class PluginListInstallationsFromIdsApi(Resource):
         return jsonable_encoder({"plugins": plugins})
 
 
+@console_ns.route("/workspaces/current/plugin/icon")
 class PluginIconApi(Resource):
     @setup_required
     def get(self):
@@ -108,6 +113,7 @@ class PluginIconApi(Resource):
         return send_file(io.BytesIO(icon_bytes), mimetype=mimetype, max_age=icon_cache_max_age)
 
 
+@console_ns.route("/workspaces/current/plugin/upload/pkg")
 class PluginUploadFromPkgApi(Resource):
     @setup_required
     @login_required
@@ -131,6 +137,7 @@ class PluginUploadFromPkgApi(Resource):
         return jsonable_encoder(response)
 
 
+@console_ns.route("/workspaces/current/plugin/upload/github")
 class PluginUploadFromGithubApi(Resource):
     @setup_required
     @login_required
@@ -153,6 +160,7 @@ class PluginUploadFromGithubApi(Resource):
         return jsonable_encoder(response)
 
 
+@console_ns.route("/workspaces/current/plugin/upload/bundle")
 class PluginUploadFromBundleApi(Resource):
     @setup_required
     @login_required
@@ -176,6 +184,7 @@ class PluginUploadFromBundleApi(Resource):
         return jsonable_encoder(response)
 
 
+@console_ns.route("/workspaces/current/plugin/install/pkg")
 class PluginInstallFromPkgApi(Resource):
     @setup_required
     @login_required
@@ -201,6 +210,7 @@ class PluginInstallFromPkgApi(Resource):
         return jsonable_encoder(response)
 
 
+@console_ns.route("/workspaces/current/plugin/install/github")
 class PluginInstallFromGithubApi(Resource):
     @setup_required
     @login_required
@@ -230,6 +240,7 @@ class PluginInstallFromGithubApi(Resource):
         return jsonable_encoder(response)
 
 
+@console_ns.route("/workspaces/current/plugin/install/marketplace")
 class PluginInstallFromMarketplaceApi(Resource):
     @setup_required
     @login_required
@@ -255,6 +266,7 @@ class PluginInstallFromMarketplaceApi(Resource):
         return jsonable_encoder(response)
 
 
+@console_ns.route("/workspaces/current/plugin/marketplace/pkg")
 class PluginFetchMarketplacePkgApi(Resource):
     @setup_required
     @login_required
@@ -280,6 +292,7 @@ class PluginFetchMarketplacePkgApi(Resource):
             raise ValueError(e)
 
 
+@console_ns.route("/workspaces/current/plugin/fetch-manifest")
 class PluginFetchManifestApi(Resource):
     @setup_required
     @login_required
@@ -304,6 +317,7 @@ class PluginFetchManifestApi(Resource):
             raise ValueError(e)
 
 
+@console_ns.route("/workspaces/current/plugin/tasks")
 class PluginFetchInstallTasksApi(Resource):
     @setup_required
     @login_required
@@ -325,6 +339,7 @@ class PluginFetchInstallTasksApi(Resource):
             raise ValueError(e)
 
 
+@console_ns.route("/workspaces/current/plugin/tasks/<task_id>")
 class PluginFetchInstallTaskApi(Resource):
     @setup_required
     @login_required
@@ -339,6 +354,7 @@ class PluginFetchInstallTaskApi(Resource):
             raise ValueError(e)
 
 
+@console_ns.route("/workspaces/current/plugin/tasks/<task_id>/delete")
 class PluginDeleteInstallTaskApi(Resource):
     @setup_required
     @login_required
@@ -353,6 +369,7 @@ class PluginDeleteInstallTaskApi(Resource):
             raise ValueError(e)
 
 
+@console_ns.route("/workspaces/current/plugin/tasks/delete_all")
 class PluginDeleteAllInstallTaskItemsApi(Resource):
     @setup_required
     @login_required
@@ -367,6 +384,7 @@ class PluginDeleteAllInstallTaskItemsApi(Resource):
             raise ValueError(e)
 
 
+@console_ns.route("/workspaces/current/plugin/tasks/<task_id>/delete/<path:identifier>")
 class PluginDeleteInstallTaskItemApi(Resource):
     @setup_required
     @login_required
@@ -381,6 +399,7 @@ class PluginDeleteInstallTaskItemApi(Resource):
             raise ValueError(e)
 
 
+@console_ns.route("/workspaces/current/plugin/upgrade/marketplace")
 class PluginUpgradeFromMarketplaceApi(Resource):
     @setup_required
     @login_required
@@ -404,6 +423,7 @@ class PluginUpgradeFromMarketplaceApi(Resource):
             raise ValueError(e)
 
 
+@console_ns.route("/workspaces/current/plugin/upgrade/github")
 class PluginUpgradeFromGithubApi(Resource):
     @setup_required
     @login_required
@@ -435,6 +455,7 @@ class PluginUpgradeFromGithubApi(Resource):
             raise ValueError(e)
 
 
+@console_ns.route("/workspaces/current/plugin/uninstall")
 class PluginUninstallApi(Resource):
     @setup_required
     @login_required
@@ -453,6 +474,7 @@ class PluginUninstallApi(Resource):
             raise ValueError(e)
 
 
+@console_ns.route("/workspaces/current/plugin/permission/change")
 class PluginChangePermissionApi(Resource):
     @setup_required
     @login_required
@@ -475,6 +497,7 @@ class PluginChangePermissionApi(Resource):
         return {"success": PluginPermissionService.change_permission(tenant_id, install_permission, debug_permission)}
 
 
+@console_ns.route("/workspaces/current/plugin/permission/fetch")
 class PluginFetchPermissionApi(Resource):
     @setup_required
     @login_required
@@ -499,6 +522,7 @@ class PluginFetchPermissionApi(Resource):
         )
 
 
+@console_ns.route("/workspaces/current/plugin/parameters/dynamic-options")
 class PluginFetchDynamicSelectOptionsApi(Resource):
     @setup_required
     @login_required
@@ -535,6 +559,7 @@ class PluginFetchDynamicSelectOptionsApi(Resource):
         return jsonable_encoder({"options": options})
 
 
+@console_ns.route("/workspaces/current/plugin/preferences/change")
 class PluginChangePreferencesApi(Resource):
     @setup_required
     @login_required
@@ -590,6 +615,7 @@ class PluginChangePreferencesApi(Resource):
         return jsonable_encoder({"success": True})
 
 
+@console_ns.route("/workspaces/current/plugin/preferences/fetch")
 class PluginFetchPreferencesApi(Resource):
     @setup_required
     @login_required
@@ -628,6 +654,7 @@ class PluginFetchPreferencesApi(Resource):
         return jsonable_encoder({"permission": permission_dict, "auto_upgrade": auto_upgrade_dict})
 
 
+@console_ns.route("/workspaces/current/plugin/preferences/autoupgrade/exclude")
 class PluginAutoUpgradeExcludePluginApi(Resource):
     @setup_required
     @login_required
@@ -641,35 +668,3 @@ class PluginAutoUpgradeExcludePluginApi(Resource):
         args = req.parse_args()
 
         return jsonable_encoder({"success": PluginAutoUpgradeService.exclude_plugin(tenant_id, args["plugin_id"])})
-
-
-api.add_resource(PluginDebuggingKeyApi, "/workspaces/current/plugin/debugging-key")
-api.add_resource(PluginListApi, "/workspaces/current/plugin/list")
-api.add_resource(PluginListLatestVersionsApi, "/workspaces/current/plugin/list/latest-versions")
-api.add_resource(PluginListInstallationsFromIdsApi, "/workspaces/current/plugin/list/installations/ids")
-api.add_resource(PluginIconApi, "/workspaces/current/plugin/icon")
-api.add_resource(PluginUploadFromPkgApi, "/workspaces/current/plugin/upload/pkg")
-api.add_resource(PluginUploadFromGithubApi, "/workspaces/current/plugin/upload/github")
-api.add_resource(PluginUploadFromBundleApi, "/workspaces/current/plugin/upload/bundle")
-api.add_resource(PluginInstallFromPkgApi, "/workspaces/current/plugin/install/pkg")
-api.add_resource(PluginInstallFromGithubApi, "/workspaces/current/plugin/install/github")
-api.add_resource(PluginUpgradeFromMarketplaceApi, "/workspaces/current/plugin/upgrade/marketplace")
-api.add_resource(PluginUpgradeFromGithubApi, "/workspaces/current/plugin/upgrade/github")
-api.add_resource(PluginInstallFromMarketplaceApi, "/workspaces/current/plugin/install/marketplace")
-api.add_resource(PluginFetchManifestApi, "/workspaces/current/plugin/fetch-manifest")
-api.add_resource(PluginFetchInstallTasksApi, "/workspaces/current/plugin/tasks")
-api.add_resource(PluginFetchInstallTaskApi, "/workspaces/current/plugin/tasks/<task_id>")
-api.add_resource(PluginDeleteInstallTaskApi, "/workspaces/current/plugin/tasks/<task_id>/delete")
-api.add_resource(PluginDeleteAllInstallTaskItemsApi, "/workspaces/current/plugin/tasks/delete_all")
-api.add_resource(PluginDeleteInstallTaskItemApi, "/workspaces/current/plugin/tasks/<task_id>/delete/<path:identifier>")
-api.add_resource(PluginUninstallApi, "/workspaces/current/plugin/uninstall")
-api.add_resource(PluginFetchMarketplacePkgApi, "/workspaces/current/plugin/marketplace/pkg")
-
-api.add_resource(PluginChangePermissionApi, "/workspaces/current/plugin/permission/change")
-api.add_resource(PluginFetchPermissionApi, "/workspaces/current/plugin/permission/fetch")
-
-api.add_resource(PluginFetchDynamicSelectOptionsApi, "/workspaces/current/plugin/parameters/dynamic-options")
-
-api.add_resource(PluginFetchPreferencesApi, "/workspaces/current/plugin/preferences/fetch")
-api.add_resource(PluginChangePreferencesApi, "/workspaces/current/plugin/preferences/change")
-api.add_resource(PluginAutoUpgradeExcludePluginApi, "/workspaces/current/plugin/preferences/autoupgrade/exclude")

api/controllers/console/workspace/tool_providers.py

@@ -10,7 +10,7 @@ from flask_restx import (
 from werkzeug.exceptions import Forbidden
 
 from configs import dify_config
-from controllers.console import api
+from controllers.console import console_ns
 from controllers.console.wraps import (
     account_initialization_required,
     enterprise_license_required,
@@ -47,6 +47,7 @@ def is_valid_url(url: str) -> bool:
         return False
 
 
+@console_ns.route("/workspaces/current/tool-providers")
 class ToolProviderListApi(Resource):
     @setup_required
     @login_required
@@ -71,6 +72,7 @@ class ToolProviderListApi(Resource):
         return ToolCommonService.list_tool_providers(user_id, tenant_id, args.get("type", None))
 
 
+@console_ns.route("/workspaces/current/tool-provider/builtin/<path:provider>/tools")
 class ToolBuiltinProviderListToolsApi(Resource):
     @setup_required
     @login_required
@@ -88,6 +90,7 @@ class ToolBuiltinProviderListToolsApi(Resource):
         )
 
 
+@console_ns.route("/workspaces/current/tool-provider/builtin/<path:provider>/info")
 class ToolBuiltinProviderInfoApi(Resource):
     @setup_required
     @login_required
@@ -100,6 +103,7 @@ class ToolBuiltinProviderInfoApi(Resource):
         return jsonable_encoder(BuiltinToolManageService.get_builtin_tool_provider_info(tenant_id, provider))
 
 
+@console_ns.route("/workspaces/current/tool-provider/builtin/<path:provider>/delete")
 class ToolBuiltinProviderDeleteApi(Resource):
     @setup_required
     @login_required
@@ -121,6 +125,7 @@ class ToolBuiltinProviderDeleteApi(Resource):
         )
 
 
+@console_ns.route("/workspaces/current/tool-provider/builtin/<path:provider>/add")
 class ToolBuiltinProviderAddApi(Resource):
     @setup_required
     @login_required
@@ -150,6 +155,7 @@ class ToolBuiltinProviderAddApi(Resource):
         )
 
 
+@console_ns.route("/workspaces/current/tool-provider/builtin/<path:provider>/update")
 class ToolBuiltinProviderUpdateApi(Resource):
     @setup_required
     @login_required
@@ -181,6 +187,7 @@ class ToolBuiltinProviderUpdateApi(Resource):
         return result
 
 
+@console_ns.route("/workspaces/current/tool-provider/builtin/<path:provider>/credentials")
 class ToolBuiltinProviderGetCredentialsApi(Resource):
     @setup_required
     @login_required
@@ -196,6 +203,7 @@ class ToolBuiltinProviderGetCredentialsApi(Resource):
         )
 
 
+@console_ns.route("/workspaces/current/tool-provider/builtin/<path:provider>/icon")
 class ToolBuiltinProviderIconApi(Resource):
     @setup_required
     def get(self, provider):
@@ -204,6 +212,7 @@ class ToolBuiltinProviderIconApi(Resource):
         return send_file(io.BytesIO(icon_bytes), mimetype=mimetype, max_age=icon_cache_max_age)
 
 
+@console_ns.route("/workspaces/current/tool-provider/api/add")
 class ToolApiProviderAddApi(Resource):
     @setup_required
     @login_required
@@ -243,6 +252,7 @@ class ToolApiProviderAddApi(Resource):
         )
 
 
+@console_ns.route("/workspaces/current/tool-provider/api/remote")
 class ToolApiProviderGetRemoteSchemaApi(Resource):
     @setup_required
     @login_required
@@ -266,6 +276,7 @@ class ToolApiProviderGetRemoteSchemaApi(Resource):
         )
 
 
+@console_ns.route("/workspaces/current/tool-provider/api/tools")
 class ToolApiProviderListToolsApi(Resource):
     @setup_required
     @login_required
@@ -291,6 +302,7 @@ class ToolApiProviderListToolsApi(Resource):
         )
 
 
+@console_ns.route("/workspaces/current/tool-provider/api/update")
 class ToolApiProviderUpdateApi(Resource):
     @setup_required
     @login_required
@@ -332,6 +344,7 @@ class ToolApiProviderUpdateApi(Resource):
         )
 
 
+@console_ns.route("/workspaces/current/tool-provider/api/delete")
 class ToolApiProviderDeleteApi(Resource):
     @setup_required
     @login_required
@@ -358,6 +371,7 @@ class ToolApiProviderDeleteApi(Resource):
         )
 
 
+@console_ns.route("/workspaces/current/tool-provider/api/get")
 class ToolApiProviderGetApi(Resource):
     @setup_required
     @login_required
@@ -381,6 +395,7 @@ class ToolApiProviderGetApi(Resource):
         )
 
 
+@console_ns.route("/workspaces/current/tool-provider/builtin/<path:provider>/credential/schema/<path:credential_type>")
 class ToolBuiltinProviderCredentialsSchemaApi(Resource):
     @setup_required
     @login_required
@@ -396,6 +411,7 @@ class ToolBuiltinProviderCredentialsSchemaApi(Resource):
         )
 
 
+@console_ns.route("/workspaces/current/tool-provider/api/schema")
 class ToolApiProviderSchemaApi(Resource):
     @setup_required
     @login_required
@@ -412,6 +428,7 @@ class ToolApiProviderSchemaApi(Resource):
         )
 
 
+@console_ns.route("/workspaces/current/tool-provider/api/test/pre")
 class ToolApiProviderPreviousTestApi(Resource):
     @setup_required
     @login_required
@@ -439,6 +456,7 @@ class ToolApiProviderPreviousTestApi(Resource):
         )
 
 
+@console_ns.route("/workspaces/current/tool-provider/workflow/create")
 class ToolWorkflowProviderCreateApi(Resource):
     @setup_required
     @login_required
@@ -478,6 +496,7 @@ class ToolWorkflowProviderCreateApi(Resource):
         )
 
 
+@console_ns.route("/workspaces/current/tool-provider/workflow/update")
 class ToolWorkflowProviderUpdateApi(Resource):
     @setup_required
     @login_required
@@ -520,6 +539,7 @@ class ToolWorkflowProviderUpdateApi(Resource):
         )
 
 
+@console_ns.route("/workspaces/current/tool-provider/workflow/delete")
 class ToolWorkflowProviderDeleteApi(Resource):
     @setup_required
     @login_required
@@ -545,6 +565,7 @@ class ToolWorkflowProviderDeleteApi(Resource):
         )
 
 
+@console_ns.route("/workspaces/current/tool-provider/workflow/get")
 class ToolWorkflowProviderGetApi(Resource):
     @setup_required
     @login_required
@@ -579,6 +600,7 @@ class ToolWorkflowProviderGetApi(Resource):
         return jsonable_encoder(tool)
 
 
+@console_ns.route("/workspaces/current/tool-provider/workflow/tools")
 class ToolWorkflowProviderListToolApi(Resource):
     @setup_required
     @login_required
@@ -603,6 +625,7 @@ class ToolWorkflowProviderListToolApi(Resource):
         )
 
 
+@console_ns.route("/workspaces/current/tools/builtin")
 class ToolBuiltinListApi(Resource):
     @setup_required
     @login_required
@@ -624,6 +647,7 @@ class ToolBuiltinListApi(Resource):
         )
 
 
+@console_ns.route("/workspaces/current/tools/api")
 class ToolApiListApi(Resource):
     @setup_required
     @login_required
@@ -642,6 +666,7 @@ class ToolApiListApi(Resource):
         )
 
 
+@console_ns.route("/workspaces/current/tools/workflow")
 class ToolWorkflowListApi(Resource):
     @setup_required
     @login_required
@@ -663,6 +688,7 @@ class ToolWorkflowListApi(Resource):
         )
 
 
+@console_ns.route("/workspaces/current/tool-labels")
 class ToolLabelsApi(Resource):
     @setup_required
     @login_required
@@ -672,6 +698,7 @@ class ToolLabelsApi(Resource):
         return jsonable_encoder(ToolLabelsService.list_tool_labels())
 
 
+@console_ns.route("/oauth/plugin/<path:provider>/tool/authorization-url")
 class ToolPluginOAuthApi(Resource):
     @setup_required
     @login_required
@@ -716,6 +743,7 @@ class ToolPluginOAuthApi(Resource):
         return response
 
 
+@console_ns.route("/oauth/plugin/<path:provider>/tool/callback")
 class ToolOAuthCallback(Resource):
     @setup_required
     def get(self, provider):
@@ -766,6 +794,7 @@ class ToolOAuthCallback(Resource):
         return redirect(f"{dify_config.CONSOLE_WEB_URL}/oauth-callback")
 
 
+@console_ns.route("/workspaces/current/tool-provider/builtin/<path:provider>/default-credential")
 class ToolBuiltinProviderSetDefaultApi(Resource):
     @setup_required
     @login_required
@@ -779,6 +808,7 @@ class ToolBuiltinProviderSetDefaultApi(Resource):
         )
 
 
+@console_ns.route("/workspaces/current/tool-provider/builtin/<path:provider>/oauth/custom-client")
 class ToolOAuthCustomClient(Resource):
     @setup_required
     @login_required
@@ -822,6 +852,7 @@ class ToolOAuthCustomClient(Resource):
         )
 
 
+@console_ns.route("/workspaces/current/tool-provider/builtin/<path:provider>/oauth/client-schema")
 class ToolBuiltinProviderGetOauthClientSchemaApi(Resource):
     @setup_required
     @login_required
@@ -834,6 +865,7 @@ class ToolBuiltinProviderGetOauthClientSchemaApi(Resource):
         )
 
 
+@console_ns.route("/workspaces/current/tool-provider/builtin/<path:provider>/credential/info")
 class ToolBuiltinProviderGetCredentialInfoApi(Resource):
     @setup_required
     @login_required
@@ -849,6 +881,7 @@ class ToolBuiltinProviderGetCredentialInfoApi(Resource):
         )
 
 
+@console_ns.route("/workspaces/current/tool-provider/mcp")
 class ToolProviderMCPApi(Resource):
     @setup_required
     @login_required
@@ -933,6 +966,7 @@ class ToolProviderMCPApi(Resource):
         return {"result": "success"}
 
 
+@console_ns.route("/workspaces/current/tool-provider/mcp/auth")
 class ToolMCPAuthApi(Resource):
     @setup_required
     @login_required
@@ -978,6 +1012,7 @@ class ToolMCPAuthApi(Resource):
             raise ValueError(f"Failed to connect to MCP server: {e}") from e
 
 
+@console_ns.route("/workspaces/current/tool-provider/mcp/tools/<path:provider_id>")
 class ToolMCPDetailApi(Resource):
     @setup_required
     @login_required
@@ -988,6 +1023,7 @@ class ToolMCPDetailApi(Resource):
         return jsonable_encoder(ToolTransformService.mcp_provider_to_user_provider(provider, for_list=True))
 
 
+@console_ns.route("/workspaces/current/tools/mcp")
 class ToolMCPListAllApi(Resource):
     @setup_required
     @login_required
@@ -1001,6 +1037,7 @@ class ToolMCPListAllApi(Resource):
         return [tool.to_dict() for tool in tools]
 
 
+@console_ns.route("/workspaces/current/tool-provider/mcp/update/<path:provider_id>")
 class ToolMCPUpdateApi(Resource):
     @setup_required
     @login_required
@@ -1014,6 +1051,7 @@ class ToolMCPUpdateApi(Resource):
         return jsonable_encoder(tools)
 
 
+@console_ns.route("/mcp/oauth/callback")
 class ToolMCPCallbackApi(Resource):
     def get(self):
         parser = reqparse.RequestParser()
@@ -1024,67 +1062,3 @@ class ToolMCPCallbackApi(Resource):
         authorization_code = args["code"]
         handle_callback(state_key, authorization_code)
         return redirect(f"{dify_config.CONSOLE_WEB_URL}/oauth-callback")
-
-
-# tool provider
-api.add_resource(ToolProviderListApi, "/workspaces/current/tool-providers")
-
-# tool oauth
-api.add_resource(ToolPluginOAuthApi, "/oauth/plugin/<path:provider>/tool/authorization-url")
-api.add_resource(ToolOAuthCallback, "/oauth/plugin/<path:provider>/tool/callback")
-api.add_resource(ToolOAuthCustomClient, "/workspaces/current/tool-provider/builtin/<path:provider>/oauth/custom-client")
-
-# builtin tool provider
-api.add_resource(ToolBuiltinProviderListToolsApi, "/workspaces/current/tool-provider/builtin/<path:provider>/tools")
-api.add_resource(ToolBuiltinProviderInfoApi, "/workspaces/current/tool-provider/builtin/<path:provider>/info")
-api.add_resource(ToolBuiltinProviderAddApi, "/workspaces/current/tool-provider/builtin/<path:provider>/add")
-api.add_resource(ToolBuiltinProviderDeleteApi, "/workspaces/current/tool-provider/builtin/<path:provider>/delete")
-api.add_resource(ToolBuiltinProviderUpdateApi, "/workspaces/current/tool-provider/builtin/<path:provider>/update")
-api.add_resource(
-    ToolBuiltinProviderSetDefaultApi, "/workspaces/current/tool-provider/builtin/<path:provider>/default-credential"
-)
-api.add_resource(
-    ToolBuiltinProviderGetCredentialInfoApi, "/workspaces/current/tool-provider/builtin/<path:provider>/credential/info"
-)
-api.add_resource(
-    ToolBuiltinProviderGetCredentialsApi, "/workspaces/current/tool-provider/builtin/<path:provider>/credentials"
-)
-api.add_resource(
-    ToolBuiltinProviderCredentialsSchemaApi,
-    "/workspaces/current/tool-provider/builtin/<path:provider>/credential/schema/<path:credential_type>",
-)
-api.add_resource(
-    ToolBuiltinProviderGetOauthClientSchemaApi,
-    "/workspaces/current/tool-provider/builtin/<path:provider>/oauth/client-schema",
-)
-api.add_resource(ToolBuiltinProviderIconApi, "/workspaces/current/tool-provider/builtin/<path:provider>/icon")
-
-# api tool provider
-api.add_resource(ToolApiProviderAddApi, "/workspaces/current/tool-provider/api/add")
-api.add_resource(ToolApiProviderGetRemoteSchemaApi, "/workspaces/current/tool-provider/api/remote")
-api.add_resource(ToolApiProviderListToolsApi, "/workspaces/current/tool-provider/api/tools")
-api.add_resource(ToolApiProviderUpdateApi, "/workspaces/current/tool-provider/api/update")
-api.add_resource(ToolApiProviderDeleteApi, "/workspaces/current/tool-provider/api/delete")
-api.add_resource(ToolApiProviderGetApi, "/workspaces/current/tool-provider/api/get")
-api.add_resource(ToolApiProviderSchemaApi, "/workspaces/current/tool-provider/api/schema")
-api.add_resource(ToolApiProviderPreviousTestApi, "/workspaces/current/tool-provider/api/test/pre")
-
-# workflow tool provider
-api.add_resource(ToolWorkflowProviderCreateApi, "/workspaces/current/tool-provider/workflow/create")
-api.add_resource(ToolWorkflowProviderUpdateApi, "/workspaces/current/tool-provider/workflow/update")
-api.add_resource(ToolWorkflowProviderDeleteApi, "/workspaces/current/tool-provider/workflow/delete")
-api.add_resource(ToolWorkflowProviderGetApi, "/workspaces/current/tool-provider/workflow/get")
-api.add_resource(ToolWorkflowProviderListToolApi, "/workspaces/current/tool-provider/workflow/tools")
-
-# mcp tool provider
-api.add_resource(ToolMCPDetailApi, "/workspaces/current/tool-provider/mcp/tools/<path:provider_id>")
-api.add_resource(ToolProviderMCPApi, "/workspaces/current/tool-provider/mcp")
-api.add_resource(ToolMCPUpdateApi, "/workspaces/current/tool-provider/mcp/update/<path:provider_id>")
-api.add_resource(ToolMCPAuthApi, "/workspaces/current/tool-provider/mcp/auth")
-api.add_resource(ToolMCPCallbackApi, "/mcp/oauth/callback")
-
-api.add_resource(ToolBuiltinListApi, "/workspaces/current/tools/builtin")
-api.add_resource(ToolApiListApi, "/workspaces/current/tools/api")
-api.add_resource(ToolMCPListAllApi, "/workspaces/current/tools/mcp")
-api.add_resource(ToolWorkflowListApi, "/workspaces/current/tools/workflow")
-api.add_resource(ToolLabelsApi, "/workspaces/current/tool-labels")

api/controllers/console/workspace/workspace.py

@@ -14,7 +14,7 @@ from controllers.common.errors import (
     TooManyFilesError,
     UnsupportedFileTypeError,
 )
-from controllers.console import api
+from controllers.console import console_ns
 from controllers.console.admin import admin_required
 from controllers.console.error import AccountNotLinkTenantError
 from controllers.console.wraps import (
@@ -51,6 +51,8 @@ tenant_fields = {
     "in_trial": fields.Boolean,
     "trial_end_reason": fields.String,
     "custom_config": fields.Raw(attribute="custom_config"),
+    "trial_credits": fields.Integer,
+    "trial_credits_used": fields.Integer,
 }
 
 tenants_fields = {
@@ -65,6 +67,7 @@ tenants_fields = {
 workspace_fields = {"id": fields.String, "name": fields.String, "status": fields.String, "created_at": TimestampField}
 
 
+@console_ns.route("/workspaces")
 class TenantListApi(Resource):
     @setup_required
     @login_required
@@ -93,6 +96,7 @@ class TenantListApi(Resource):
         return {"workspaces": marshal(tenant_dicts, tenants_fields)}, 200
 
 
+@console_ns.route("/all-workspaces")
 class WorkspaceListApi(Resource):
     @setup_required
     @admin_required
@@ -118,6 +122,8 @@ class WorkspaceListApi(Resource):
         }, 200
 
 
+@console_ns.route("/workspaces/current", endpoint="workspaces_current")
+@console_ns.route("/info", endpoint="info")  # Deprecated
 class TenantApi(Resource):
     @setup_required
     @login_required
@@ -143,11 +149,10 @@ class TenantApi(Resource):
             else:
                 raise Unauthorized("workspace is archived")
 
-        if not tenant:
-            raise ValueError("No tenant available")
         return WorkspaceService.get_tenant_info(tenant), 200
 
 
+@console_ns.route("/workspaces/switch")
 class SwitchWorkspaceApi(Resource):
     @setup_required
     @login_required
@@ -172,6 +177,7 @@ class SwitchWorkspaceApi(Resource):
         return {"result": "success", "new_tenant": marshal(WorkspaceService.get_tenant_info(new_tenant), tenant_fields)}
 
 
+@console_ns.route("/workspaces/custom-config")
 class CustomConfigWorkspaceApi(Resource):
     @setup_required
     @login_required
@@ -202,6 +208,7 @@ class CustomConfigWorkspaceApi(Resource):
         return {"result": "success", "tenant": marshal(WorkspaceService.get_tenant_info(tenant), tenant_fields)}
 
 
+@console_ns.route("/workspaces/custom-config/webapp-logo/upload")
 class WebappLogoWorkspaceApi(Resource):
     @setup_required
     @login_required
@@ -242,6 +249,7 @@ class WebappLogoWorkspaceApi(Resource):
         return {"id": upload_file.id}, 201
 
 
+@console_ns.route("/workspaces/info")
 class WorkspaceInfoApi(Resource):
     @setup_required
     @login_required
@@ -261,13 +269,3 @@ class WorkspaceInfoApi(Resource):
         db.session.commit()
 
         return {"result": "success", "tenant": marshal(WorkspaceService.get_tenant_info(tenant), tenant_fields)}
-
-
-api.add_resource(TenantListApi, "/workspaces")  # GET for getting all tenants
-api.add_resource(WorkspaceListApi, "/all-workspaces")  # GET for getting all tenants
-api.add_resource(TenantApi, "/workspaces/current", endpoint="workspaces_current")  # GET for getting current tenant info
-api.add_resource(TenantApi, "/info", endpoint="info")  # Deprecated
-api.add_resource(SwitchWorkspaceApi, "/workspaces/switch")  # POST for switching tenant
-api.add_resource(CustomConfigWorkspaceApi, "/workspaces/custom-config")
-api.add_resource(WebappLogoWorkspaceApi, "/workspaces/custom-config/webapp-logo/upload")
-api.add_resource(WorkspaceInfoApi, "/workspaces/info")  # POST for changing workspace info

api/controllers/inner_api/plugin/wraps.py

@@ -128,7 +128,7 @@ def plugin_data(view: Callable[P, R] | None = None, *, payload_type: type[BaseMo
                 raise ValueError("invalid json")
 
             try:
-                payload = payload_type(**data)
+                payload = payload_type.model_validate(data)
             except Exception as e:
                 raise ValueError(f"invalid payload: {str(e)}")
 

api/controllers/service_api/__init__.py

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

api/controllers/service_api/app/chatflow_memory.py

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

api/controllers/service_api/dataset/dataset.py

@@ -280,7 +280,7 @@ class DatasetListApi(DatasetApiResource):
                 external_knowledge_id=args["external_knowledge_id"],
                 embedding_model_provider=args["embedding_model_provider"],
                 embedding_model_name=args["embedding_model"],
-                retrieval_model=RetrievalModel(**args["retrieval_model"])
+                retrieval_model=RetrievalModel.model_validate(args["retrieval_model"])
                 if args["retrieval_model"] is not None
                 else None,
             )

api/controllers/service_api/dataset/document.py

@@ -136,7 +136,7 @@ class DocumentAddByTextApi(DatasetApiResource):
             "info_list": {"data_source_type": "upload_file", "file_info_list": {"file_ids": [upload_file.id]}},
         }
         args["data_source"] = data_source
-        knowledge_config = KnowledgeConfig(**args)
+        knowledge_config = KnowledgeConfig.model_validate(args)
         # validate args
         DocumentService.document_create_args_validate(knowledge_config)
 
@@ -221,7 +221,7 @@ class DocumentUpdateByTextApi(DatasetApiResource):
             args["data_source"] = data_source
         # validate args
         args["original_document_id"] = str(document_id)
-        knowledge_config = KnowledgeConfig(**args)
+        knowledge_config = KnowledgeConfig.model_validate(args)
         DocumentService.document_create_args_validate(knowledge_config)
 
         try:
@@ -328,7 +328,7 @@ class DocumentAddByFileApi(DatasetApiResource):
         }
         args["data_source"] = data_source
         # validate args
-        knowledge_config = KnowledgeConfig(**args)
+        knowledge_config = KnowledgeConfig.model_validate(args)
         DocumentService.document_create_args_validate(knowledge_config)
 
         dataset_process_rule = dataset.latest_process_rule if "process_rule" not in args else None
@@ -426,7 +426,7 @@ class DocumentUpdateByFileApi(DatasetApiResource):
         # validate args
         args["original_document_id"] = str(document_id)
 
-        knowledge_config = KnowledgeConfig(**args)
+        knowledge_config = KnowledgeConfig.model_validate(args)
         DocumentService.document_create_args_validate(knowledge_config)
 
         try:

api/controllers/service_api/dataset/metadata.py

@@ -51,7 +51,7 @@ class DatasetMetadataCreateServiceApi(DatasetApiResource):
     def post(self, tenant_id, dataset_id):
         """Create metadata for a dataset."""
         args = metadata_create_parser.parse_args()
-        metadata_args = MetadataArgs(**args)
+        metadata_args = MetadataArgs.model_validate(args)
 
         dataset_id_str = str(dataset_id)
         dataset = DatasetService.get_dataset(dataset_id_str)
@@ -200,7 +200,7 @@ class DocumentMetadataEditServiceApi(DatasetApiResource):
         DatasetService.check_dataset_permission(dataset, current_user)
 
         args = document_metadata_parser.parse_args()
-        metadata_args = MetadataOperationData(**args)
+        metadata_args = MetadataOperationData.model_validate(args)
 
         MetadataService.update_documents_metadata(dataset, metadata_args)
 

api/controllers/service_api/dataset/rag_pipeline/rag_pipeline_workflow.py

@@ -98,7 +98,7 @@ class DatasourceNodeRunApi(DatasetApiResource):
         parser.add_argument("is_published", type=bool, required=True, location="json")
         args: ParseResult = parser.parse_args()
 
-        datasource_node_run_api_entity: DatasourceNodeRunApiEntity = DatasourceNodeRunApiEntity(**args)
+        datasource_node_run_api_entity = DatasourceNodeRunApiEntity.model_validate(args)
         assert isinstance(current_user, Account)
         rag_pipeline_service: RagPipelineService = RagPipelineService()
         pipeline: Pipeline = rag_pipeline_service.get_pipeline(tenant_id=tenant_id, dataset_id=dataset_id)

api/controllers/service_api/dataset/segment.py

@@ -252,7 +252,7 @@ class DatasetSegmentApi(DatasetApiResource):
         args = segment_update_parser.parse_args()
 
         updated_segment = SegmentService.update_segment(
-            SegmentUpdateArgs(**args["segment"]), segment, document, dataset
+            SegmentUpdateArgs.model_validate(args["segment"]), segment, document, dataset
         )
         return {"data": marshal(updated_segment, segment_fields), "doc_form": document.doc_form}, 200
 

api/controllers/web/__init__.py

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

api/controllers/web/chatflow_memory.py

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

api/controllers/web/passport.py

@@ -126,6 +126,8 @@ def exchange_token_for_existing_web_user(app_code: str, enterprise_user_decoded:
     end_user_id = enterprise_user_decoded.get("end_user_id")
     session_id = enterprise_user_decoded.get("session_id")
     user_auth_type = enterprise_user_decoded.get("auth_type")
+    exchanged_token_expires_unix = enterprise_user_decoded.get("exp")
+
     if not user_auth_type:
         raise Unauthorized("Missing auth_type in the token.")
 
@@ -169,8 +171,11 @@ def exchange_token_for_existing_web_user(app_code: str, enterprise_user_decoded:
         )
         db.session.add(end_user)
         db.session.commit()
-    exp_dt = datetime.now(UTC) + timedelta(minutes=dify_config.ACCESS_TOKEN_EXPIRE_MINUTES)
-    exp = int(exp_dt.timestamp())
+
+    exp = int((datetime.now(UTC) + timedelta(minutes=dify_config.ACCESS_TOKEN_EXPIRE_MINUTES)).timestamp())
+    if exchanged_token_expires_unix:
+        exp = int(exchanged_token_expires_unix)
+
     payload = {
         "iss": site.id,
         "sub": "Web API Passport",

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

@@ -40,7 +40,7 @@ class AgentConfigManager:
                         "credential_id": tool.get("credential_id", None),
                     }
 
-                    agent_tools.append(AgentToolEntity(**agent_tool_properties))
+                    agent_tools.append(AgentToolEntity.model_validate(agent_tool_properties))
 
             if "strategy" in config["agent_mode"] and config["agent_mode"]["strategy"] not in {
                 "react_router",

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

@@ -1,10 +1,11 @@
 import logging
 import time
-from collections.abc import Mapping
+from collections.abc import Mapping, MutableMapping
 from typing import Any, cast
 
 from sqlalchemy import select
 from sqlalchemy.orm import Session
+from typing_extensions import override
 
 from core.app.apps.advanced_chat.app_config_manager import AdvancedChatAppConfig
 from core.app.apps.base_app_queue_manager import AppQueueManager
@@ -20,11 +21,14 @@ from core.app.entities.queue_entities import (
     QueueTextChunkEvent,
 )
 from core.app.features.annotation_reply.annotation_reply import AnnotationReplyFeature
+from core.memory.entities import MemoryCreatedBy, MemoryScope
+from core.model_runtime.entities import AssistantPromptMessage, UserPromptMessage
 from core.moderation.base import ModerationError
 from core.moderation.input_moderation import InputModeration
 from core.variables.variables import VariableUnion
 from core.workflow.entities import GraphRuntimeState, VariablePool
 from core.workflow.graph_engine.command_channels.redis_channel import RedisChannel
+from core.workflow.graph_events import GraphRunSucceededEvent
 from core.workflow.system_variable import SystemVariable
 from core.workflow.variable_loader import VariableLoader
 from core.workflow.workflow_entry import WorkflowEntry
@@ -34,6 +38,8 @@ from models import Workflow
 from models.enums import UserFrom
 from models.model import App, Conversation, Message, MessageAnnotation
 from models.workflow import ConversationVariable
+from services.chatflow_history_service import ChatflowHistoryService
+from services.chatflow_memory_service import ChatflowMemoryService
 
 logger = logging.getLogger(__name__)
 
@@ -70,6 +76,11 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
         self._app = app
 
     def run(self):
+        ChatflowMemoryService.wait_for_sync_memory_completion(
+            workflow=self._workflow,
+            conversation_id=self.conversation.id
+        )
+
         app_config = self.application_generate_entity.app_config
         app_config = cast(AdvancedChatAppConfig, app_config)
 
@@ -133,6 +144,7 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
                 # Based on the definition of `VariableUnion`,
                 # `list[Variable]` can be safely used as `list[VariableUnion]` since they are compatible.
                 conversation_variables=conversation_variables,
+                memory_blocks=self._fetch_memory_blocks(),
             )
 
             # init graph
@@ -177,6 +189,31 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
         for event in generator:
             self._handle_event(workflow_entry, event)
 
+        try:
+            self._check_app_memory_updates(variable_pool)
+        except Exception as e:
+            logger.exception("Failed to check app memory updates", exc_info=e)
+
+    @override
+    def _handle_event(self, workflow_entry: WorkflowEntry, event: Any) -> None:
+        super()._handle_event(workflow_entry, event)
+        if isinstance(event, GraphRunSucceededEvent):
+            workflow_outputs = event.outputs
+            if not workflow_outputs:
+                logger.warning("Chatflow output is empty.")
+                return
+            assistant_message = workflow_outputs.get('answer')
+            if not assistant_message:
+                logger.warning("Chatflow output does not contain 'answer'.")
+                return
+            if not isinstance(assistant_message, str):
+                logger.warning("Chatflow output 'answer' is not a string.")
+                return
+            try:
+                self._sync_conversation_to_chatflow_tables(assistant_message)
+            except Exception as e:
+                logger.exception("Failed to sync conversation to memory tables", exc_info=e)
+
     def handle_input_moderation(
         self,
         app_record: App,
@@ -374,3 +411,67 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
 
         # Return combined list
         return existing_variables + new_variables
+
+    def _fetch_memory_blocks(self) -> Mapping[str, str]:
+        """fetch all memory blocks for current app"""
+
+        memory_blocks_dict: MutableMapping[str, str] = {}
+        is_draft = (self.application_generate_entity.invoke_from == InvokeFrom.DEBUGGER)
+        conversation_id = self.conversation.id
+        memory_block_specs = self._workflow.memory_blocks
+        # Get runtime memory values
+        memories = ChatflowMemoryService.get_memories_by_specs(
+            memory_block_specs=memory_block_specs,
+            tenant_id=self._workflow.tenant_id,
+            app_id=self._workflow.app_id,
+            node_id=None,
+            conversation_id=conversation_id,
+            is_draft=is_draft,
+            created_by=self._get_created_by(),
+        )
+
+        # Build memory_id -> value mapping
+        for memory in memories:
+            if memory.spec.scope == MemoryScope.APP:
+                # App level: use memory_id directly
+                memory_blocks_dict[memory.spec.id] = memory.value
+            else:  # NODE scope
+                node_id = memory.node_id
+                if not node_id:
+                    logger.warning("Memory block %s has no node_id, skip.", memory.spec.id)
+                    continue
+                key = f"{node_id}.{memory.spec.id}"
+                memory_blocks_dict[key] = memory.value
+
+        return memory_blocks_dict
+
+    def _sync_conversation_to_chatflow_tables(self, assistant_message: str):
+        ChatflowHistoryService.save_app_message(
+            prompt_message=UserPromptMessage(content=(self.application_generate_entity.query)),
+            conversation_id=self.conversation.id,
+            app_id=self._workflow.app_id,
+            tenant_id=self._workflow.tenant_id
+        )
+        ChatflowHistoryService.save_app_message(
+            prompt_message=AssistantPromptMessage(content=assistant_message),
+            conversation_id=self.conversation.id,
+            app_id=self._workflow.app_id,
+            tenant_id=self._workflow.tenant_id
+        )
+
+    def _check_app_memory_updates(self, variable_pool: VariablePool):
+        is_draft = (self.application_generate_entity.invoke_from == InvokeFrom.DEBUGGER)
+
+        ChatflowMemoryService.update_app_memory_if_needed(
+            workflow=self._workflow,
+            conversation_id=self.conversation.id,
+            variable_pool=variable_pool,
+            is_draft=is_draft,
+            created_by=self._get_created_by()
+        )
+
+    def _get_created_by(self) -> MemoryCreatedBy:
+        if self.application_generate_entity.invoke_from in {InvokeFrom.DEBUGGER, InvokeFrom.EXPLORE}:
+            return MemoryCreatedBy(account_id=self.application_generate_entity.user_id)
+        else:
+            return MemoryCreatedBy(end_user_id=self.application_generate_entity.user_id)

api/core/app/apps/base_app_runner.py

@@ -61,9 +61,6 @@ class AppRunner:
         if model_context_tokens is None:
             return -1
 
-        if max_tokens is None:
-            max_tokens = 0
-
         prompt_tokens = model_instance.get_llm_num_tokens(prompt_messages)
 
         if prompt_tokens + max_tokens > model_context_tokens:

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

@@ -116,7 +116,7 @@ class PipelineRunner(WorkflowBasedAppRunner):
             rag_pipeline_variables = []
             if workflow.rag_pipeline_variables:
                 for v in workflow.rag_pipeline_variables:
-                    rag_pipeline_variable = RAGPipelineVariable(**v)
+                    rag_pipeline_variable = RAGPipelineVariable.model_validate(v)
                     if (
                         rag_pipeline_variable.belong_to_node_id
                         in (self.application_generate_entity.start_node_id, "shared")

api/core/datasource/entities/common_entities.py

@@ -1,4 +1,4 @@
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, model_validator
 
 
 class I18nObject(BaseModel):
@@ -11,11 +11,12 @@ class I18nObject(BaseModel):
     pt_BR: str | None = Field(default=None)
     ja_JP: str | None = Field(default=None)
 
-    def __init__(self, **data):
-        super().__init__(**data)
+    @model_validator(mode="after")
+    def _(self):
         self.zh_Hans = self.zh_Hans or self.en_US
         self.pt_BR = self.pt_BR or self.en_US
         self.ja_JP = self.ja_JP or self.en_US
+        return self
 
     def to_dict(self) -> dict:
         return {"zh_Hans": self.zh_Hans, "en_US": self.en_US, "pt_BR": self.pt_BR, "ja_JP": self.ja_JP}

api/core/entities/provider_configuration.py

@@ -5,7 +5,7 @@ from collections import defaultdict
 from collections.abc import Iterator, Sequence
 from json import JSONDecodeError
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, model_validator
 from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 
@@ -73,9 +73,8 @@ class ProviderConfiguration(BaseModel):
     # pydantic configs
     model_config = ConfigDict(protected_namespaces=())
 
-    def __init__(self, **data):
-        super().__init__(**data)
-
+    @model_validator(mode="after")
+    def _(self):
         if self.provider.provider not in original_provider_configurate_methods:
             original_provider_configurate_methods[self.provider.provider] = []
             for configurate_method in self.provider.configurate_methods:
@@ -90,6 +89,7 @@ class ProviderConfiguration(BaseModel):
                 and ConfigurateMethod.PREDEFINED_MODEL not in self.provider.configurate_methods
             ):
                 self.provider.configurate_methods.append(ConfigurateMethod.PREDEFINED_MODEL)
+        return self
 
     def get_current_credentials(self, model_type: ModelType, model: str) -> dict | None:
         """

api/core/helper/code_executor/code_executor.py

@@ -131,7 +131,7 @@ class CodeExecutor:
         if (code := response_data.get("code")) != 0:
             raise CodeExecutionError(f"Got error code: {code}. Got error msg: {response_data.get('message')}")
 
-        response_code = CodeExecutionResponse(**response_data)
+        response_code = CodeExecutionResponse.model_validate(response_data)
 
         if response_code.data.error:
             raise CodeExecutionError(response_code.data.error)

api/core/helper/marketplace.py

@@ -26,7 +26,7 @@ def batch_fetch_plugin_manifests(plugin_ids: list[str]) -> Sequence[MarketplaceP
     response = httpx.post(url, json={"plugin_ids": plugin_ids}, headers={"X-Dify-Version": dify_config.project.version})
     response.raise_for_status()
 
-    return [MarketplacePluginDeclaration(**plugin) for plugin in response.json()["data"]["plugins"]]
+    return [MarketplacePluginDeclaration.model_validate(plugin) for plugin in response.json()["data"]["plugins"]]
 
 
 def batch_fetch_plugin_manifests_ignore_deserialization_error(
@@ -41,7 +41,7 @@ def batch_fetch_plugin_manifests_ignore_deserialization_error(
     result: list[MarketplacePluginDeclaration] = []
     for plugin in response.json()["data"]["plugins"]:
         try:
-            result.append(MarketplacePluginDeclaration(**plugin))
+            result.append(MarketplacePluginDeclaration.model_validate(plugin))
         except Exception:
             pass
 

api/core/hosting_configuration.py

@@ -56,6 +56,9 @@ class HostingConfiguration:
         self.provider_map[f"{DEFAULT_PLUGIN_ID}/minimax/minimax"] = self.init_minimax()
         self.provider_map[f"{DEFAULT_PLUGIN_ID}/spark/spark"] = self.init_spark()
         self.provider_map[f"{DEFAULT_PLUGIN_ID}/zhipuai/zhipuai"] = self.init_zhipuai()
+        self.provider_map[f"{DEFAULT_PLUGIN_ID}/gemini/google"] = self.init_gemini()
+        self.provider_map[f"{DEFAULT_PLUGIN_ID}/x/x"] = self.init_xai()
+        self.provider_map[f"{DEFAULT_PLUGIN_ID}/deepseek/deepseek"] = self.init_deepseek()
 
         self.moderation_config = self.init_moderation_config()
 
@@ -128,7 +131,7 @@ class HostingConfiguration:
         quotas: list[HostingQuota] = []
 
         if dify_config.HOSTED_OPENAI_TRIAL_ENABLED:
-            hosted_quota_limit = dify_config.HOSTED_OPENAI_QUOTA_LIMIT
+            hosted_quota_limit = 0
             trial_models = self.parse_restrict_models_from_env("HOSTED_OPENAI_TRIAL_MODELS")
             trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit, restrict_models=trial_models)
             quotas.append(trial_quota)
@@ -156,18 +159,49 @@ class HostingConfiguration:
             quota_unit=quota_unit,
         )
 
-    @staticmethod
-    def init_anthropic() -> HostingProvider:
-        quota_unit = QuotaUnit.TOKENS
+    def init_gemini(self) -> HostingProvider:
+        quota_unit = QuotaUnit.CREDITS
+        quotas: list[HostingQuota] = []
+
+        if dify_config.HOSTED_GEMINI_TRIAL_ENABLED:
+            hosted_quota_limit = 0
+            trial_models = self.parse_restrict_models_from_env("HOSTED_GEMINI_TRIAL_MODELS")
+            trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit, restrict_models=trial_models)
+            quotas.append(trial_quota)
+
+        if dify_config.HOSTED_GEMINI_PAID_ENABLED:
+            paid_models = self.parse_restrict_models_from_env("HOSTED_GEMINI_PAID_MODELS")
+            paid_quota = PaidHostingQuota(restrict_models=paid_models)
+            quotas.append(paid_quota)
+
+        if len(quotas) > 0:
+            credentials = {
+                "google_api_key": dify_config.HOSTED_GEMINI_API_KEY,
+            }
+
+            if dify_config.HOSTED_GEMINI_API_BASE:
+                credentials["google_base_url"] = dify_config.HOSTED_GEMINI_API_BASE
+
+            return HostingProvider(enabled=True, credentials=credentials, quota_unit=quota_unit, quotas=quotas)
+
+        return HostingProvider(
+            enabled=False,
+            quota_unit=quota_unit,
+        )
+
+    def init_anthropic(self) -> HostingProvider:
+        quota_unit = QuotaUnit.CREDITS
         quotas: list[HostingQuota] = []
 
         if dify_config.HOSTED_ANTHROPIC_TRIAL_ENABLED:
-            hosted_quota_limit = dify_config.HOSTED_ANTHROPIC_QUOTA_LIMIT
-            trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit)
+            hosted_quota_limit = 0
+            trail_models = self.parse_restrict_models_from_env("HOSTED_ANTHROPIC_TRIAL_MODELS")
+            trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit, restrict_models=trail_models)
             quotas.append(trial_quota)
 
         if dify_config.HOSTED_ANTHROPIC_PAID_ENABLED:
-            paid_quota = PaidHostingQuota()
+            paid_models = self.parse_restrict_models_from_env("HOSTED_ANTHROPIC_PAID_MODELS")
+            paid_quota = PaidHostingQuota(restrict_models=paid_models)
             quotas.append(paid_quota)
 
         if len(quotas) > 0:
@@ -185,6 +219,66 @@ class HostingConfiguration:
             quota_unit=quota_unit,
         )
 
+    def init_xai(self) -> HostingProvider:
+        quota_unit = QuotaUnit.CREDITS
+        quotas: list[HostingQuota] = []
+
+        if dify_config.HOSTED_XAI_TRIAL_ENABLED:
+            hosted_quota_limit = 0
+            trail_models = self.parse_restrict_models_from_env("HOSTED_XAI_TRIAL_MODELS")
+            trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit, restrict_models=trail_models)
+            quotas.append(trial_quota)
+
+        if dify_config.HOSTED_XAI_PAID_ENABLED:
+            paid_models = self.parse_restrict_models_from_env("HOSTED_XAI_PAID_MODELS")
+            paid_quota = PaidHostingQuota(restrict_models=paid_models)
+            quotas.append(paid_quota)
+
+        if len(quotas) > 0:
+            credentials = {
+                "api_key": dify_config.HOSTED_XAI_API_KEY,
+            }
+
+            if dify_config.HOSTED_XAI_API_BASE:
+                credentials["endpoint_url"] = dify_config.HOSTED_XAI_API_BASE
+
+            return HostingProvider(enabled=True, credentials=credentials, quota_unit=quota_unit, quotas=quotas)
+
+        return HostingProvider(
+            enabled=False,
+            quota_unit=quota_unit,
+        )
+
+    def init_deepseek(self) -> HostingProvider:
+        quota_unit = QuotaUnit.CREDITS
+        quotas: list[HostingQuota] = []
+
+        if dify_config.HOSTED_DEEPSEEK_TRIAL_ENABLED:
+            hosted_quota_limit = 0
+            trail_models = self.parse_restrict_models_from_env("HOSTED_DEEPSEEK_TRIAL_MODELS")
+            trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit, restrict_models=trail_models)
+            quotas.append(trial_quota)
+
+        if dify_config.HOSTED_DEEPSEEK_PAID_ENABLED:
+            paid_models = self.parse_restrict_models_from_env("HOSTED_DEEPSEEK_PAID_MODELS")
+            paid_quota = PaidHostingQuota(restrict_models=paid_models)
+            quotas.append(paid_quota)
+
+        if len(quotas) > 0:
+            credentials = {
+                "api_key": dify_config.HOSTED_DEEPSEEK_API_KEY,
+            }
+
+            if dify_config.HOSTED_DEEPSEEK_API_BASE:
+                credentials["endpoint_url"] = dify_config.HOSTED_DEEPSEEK_API_BASE
+
+            return HostingProvider(enabled=True, credentials=credentials, quota_unit=quota_unit, quotas=quotas)
+
+        return HostingProvider(
+            enabled=False,
+            quota_unit=quota_unit,
+        )
+
     @staticmethod
     def init_minimax() -> HostingProvider:
         quota_unit = QuotaUnit.TOKENS

api/core/indexing_runner.py

@@ -20,7 +20,7 @@ from core.rag.cleaner.clean_processor import CleanProcessor
 from core.rag.datasource.keyword.keyword_factory import Keyword
 from core.rag.docstore.dataset_docstore import DatasetDocumentStore
 from core.rag.extractor.entity.datasource_type import DatasourceType
-from core.rag.extractor.entity.extract_setting import ExtractSetting
+from core.rag.extractor.entity.extract_setting import ExtractSetting, NotionInfo, WebsiteInfo
 from core.rag.index_processor.constant.index_type import IndexType
 from core.rag.index_processor.index_processor_base import BaseIndexProcessor
 from core.rag.index_processor.index_processor_factory import IndexProcessorFactory
@@ -357,14 +357,16 @@ class IndexingRunner:
                 raise ValueError("no notion import info found")
             extract_setting = ExtractSetting(
                 datasource_type=DatasourceType.NOTION.value,
-                notion_info={
-                    "credential_id": data_source_info["credential_id"],
-                    "notion_workspace_id": data_source_info["notion_workspace_id"],
-                    "notion_obj_id": data_source_info["notion_page_id"],
-                    "notion_page_type": data_source_info["type"],
-                    "document": dataset_document,
-                    "tenant_id": dataset_document.tenant_id,
-                },
+                notion_info=NotionInfo.model_validate(
+                    {
+                        "credential_id": data_source_info["credential_id"],
+                        "notion_workspace_id": data_source_info["notion_workspace_id"],
+                        "notion_obj_id": data_source_info["notion_page_id"],
+                        "notion_page_type": data_source_info["type"],
+                        "document": dataset_document,
+                        "tenant_id": dataset_document.tenant_id,
+                    }
+                ),
                 document_model=dataset_document.doc_form,
             )
             text_docs = index_processor.extract(extract_setting, process_rule_mode=process_rule["mode"])
@@ -378,14 +380,16 @@ class IndexingRunner:
                 raise ValueError("no website import info found")
             extract_setting = ExtractSetting(
                 datasource_type=DatasourceType.WEBSITE.value,
-                website_info={
-                    "provider": data_source_info["provider"],
-                    "job_id": data_source_info["job_id"],
-                    "tenant_id": dataset_document.tenant_id,
-                    "url": data_source_info["url"],
-                    "mode": data_source_info["mode"],
-                    "only_main_content": data_source_info["only_main_content"],
-                },
+                website_info=WebsiteInfo.model_validate(
+                    {
+                        "provider": data_source_info["provider"],
+                        "job_id": data_source_info["job_id"],
+                        "tenant_id": dataset_document.tenant_id,
+                        "url": data_source_info["url"],
+                        "mode": data_source_info["mode"],
+                        "only_main_content": data_source_info["only_main_content"],
+                    }
+                ),
                 document_model=dataset_document.doc_form,
             )
             text_docs = index_processor.extract(extract_setting, process_rule_mode=process_rule["mode"])

api/core/llm_generator/llm_generator.py

@@ -14,10 +14,12 @@ from core.llm_generator.prompts import (
     JAVASCRIPT_CODE_GENERATOR_PROMPT_TEMPLATE,
     LLM_MODIFY_CODE_SYSTEM,
     LLM_MODIFY_PROMPT_SYSTEM,
+    MEMORY_UPDATE_PROMPT,
     PYTHON_CODE_GENERATOR_PROMPT_TEMPLATE,
     SYSTEM_STRUCTURED_OUTPUT_GENERATE,
     WORKFLOW_RULE_CONFIG_PROMPT_GENERATE_TEMPLATE,
 )
+from core.memory.entities import MemoryBlock, MemoryBlockSpec
 from core.model_manager import ModelManager
 from core.model_runtime.entities.llm_entities import LLMResult
 from core.model_runtime.entities.message_entities import PromptMessage, SystemPromptMessage, UserPromptMessage
@@ -27,6 +29,7 @@ from core.ops.entities.trace_entity import TraceTaskName
 from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
 from core.ops.utils import measure_time
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
+from core.workflow.entities.variable_pool import VariablePool
 from core.workflow.entities.workflow_node_execution import WorkflowNodeExecutionMetadataKey
 from extensions.ext_database import db
 from extensions.ext_storage import storage
@@ -560,3 +563,35 @@ class LLMGenerator:
                 "Failed to invoke LLM model, model: %s", json.dumps(model_config.get("name")), exc_info=True
             )
             return {"error": f"An unexpected error occurred: {str(e)}"}
+
+    @staticmethod
+    def update_memory_block(
+        tenant_id: str,
+        visible_history: Sequence[tuple[str, str]],
+        variable_pool: VariablePool,
+        memory_block: MemoryBlock,
+        memory_spec: MemoryBlockSpec
+    ) -> str:
+        model_instance = ModelManager().get_model_instance(
+            tenant_id=tenant_id,
+            provider=memory_spec.model.provider,
+            model=memory_spec.model.name,
+            model_type=ModelType.LLM,
+        )
+        formatted_history = ""
+        for sender, message in visible_history:
+            formatted_history += f"{sender}: {message}\n"
+        filled_instruction = variable_pool.convert_template(memory_spec.instruction).text
+        formatted_prompt = PromptTemplateParser(MEMORY_UPDATE_PROMPT).format(
+            inputs={
+                "formatted_history": formatted_history,
+                "current_value": memory_block.value,
+                "instruction": filled_instruction,
+            }
+        )
+        llm_result = model_instance.invoke_llm(
+            prompt_messages=[UserPromptMessage(content=formatted_prompt)],
+            model_parameters=memory_spec.model.completion_params,
+            stream=False,
+        )
+        return llm_result.message.get_text_content()

api/core/llm_generator/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/mcp/session/client_session.py

@@ -294,7 +294,7 @@ class ClientSession(
                     method="completion/complete",
                     params=types.CompleteRequestParams(
                         ref=ref,
-                        argument=types.CompletionArgument(**argument),
+                        argument=types.CompletionArgument.model_validate(argument),
                     ),
                 )
             ),

api/core/memory/entities.py

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

api/core/memory/errors.py

@@ -0,0 +1,6 @@
+class MemorySyncTimeoutError(Exception):
+    def __init__(self, app_id: str, conversation_id: str):
+        self.app_id = app_id
+        self.conversation_id = conversation_id
+        self.message = "Memory synchronization timeout after 50 seconds"
+        super().__init__(self.message)

api/core/model_runtime/entities/common_entities.py

@@ -1,4 +1,4 @@
-from pydantic import BaseModel
+from pydantic import BaseModel, model_validator
 
 
 class I18nObject(BaseModel):
@@ -9,7 +9,8 @@ class I18nObject(BaseModel):
     zh_Hans: str | None = None
     en_US: str
 
-    def __init__(self, **data):
-        super().__init__(**data)
+    @model_validator(mode="after")
+    def _(self):
         if not self.zh_Hans:
             self.zh_Hans = self.en_US
+        return self

api/core/model_runtime/entities/provider_entities.py

@@ -1,7 +1,7 @@
 from collections.abc import Sequence
 from enum import Enum, StrEnum, auto
 
-from pydantic import BaseModel, ConfigDict, Field, field_validator
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
 from core.model_runtime.entities.common_entities import I18nObject
 from core.model_runtime.entities.model_entities import AIModelEntity, ModelType
@@ -46,10 +46,11 @@ class FormOption(BaseModel):
     value: str
     show_on: list[FormShowOnObject] = []
 
-    def __init__(self, **data):
-        super().__init__(**data)
+    @model_validator(mode="after")
+    def _(self):
         if not self.label:
             self.label = I18nObject(en_US=self.value)
+        return self
 
 
 class CredentialFormSchema(BaseModel):

api/core/model_runtime/model_providers/model_provider_factory.py

@@ -269,17 +269,17 @@ class ModelProviderFactory:
         }
 
         if model_type == ModelType.LLM:
-            return LargeLanguageModel(**init_params)  # type: ignore
+            return LargeLanguageModel.model_validate(init_params)
         elif model_type == ModelType.TEXT_EMBEDDING:
-            return TextEmbeddingModel(**init_params)  # type: ignore
+            return TextEmbeddingModel.model_validate(init_params)
         elif model_type == ModelType.RERANK:
-            return RerankModel(**init_params)  # type: ignore
+            return RerankModel.model_validate(init_params)
         elif model_type == ModelType.SPEECH2TEXT:
-            return Speech2TextModel(**init_params)  # type: ignore
+            return Speech2TextModel.model_validate(init_params)
         elif model_type == ModelType.MODERATION:
-            return ModerationModel(**init_params)  # type: ignore
+            return ModerationModel.model_validate(init_params)
         elif model_type == ModelType.TTS:
-            return TTSModel(**init_params)  # type: ignore
+            return TTSModel.model_validate(init_params)
 
     def get_provider_icon(self, provider: str, icon_type: str, lang: str) -> tuple[bytes, str]:
         """

api/core/moderation/api/api.py

@@ -51,7 +51,7 @@ class ApiModeration(Moderation):
             params = ModerationInputParams(app_id=self.app_id, inputs=inputs, query=query)
 
             result = self._get_config_by_requestor(APIBasedExtensionPoint.APP_MODERATION_INPUT, params.model_dump())
-            return ModerationInputsResult(**result)
+            return ModerationInputsResult.model_validate(result)
 
         return ModerationInputsResult(
             flagged=flagged, action=ModerationAction.DIRECT_OUTPUT, preset_response=preset_response
@@ -67,7 +67,7 @@ class ApiModeration(Moderation):
             params = ModerationOutputParams(app_id=self.app_id, text=text)
 
             result = self._get_config_by_requestor(APIBasedExtensionPoint.APP_MODERATION_OUTPUT, params.model_dump())
-            return ModerationOutputsResult(**result)
+            return ModerationOutputsResult.model_validate(result)
 
         return ModerationOutputsResult(
             flagged=flagged, action=ModerationAction.DIRECT_OUTPUT, preset_response=preset_response

api/core/plugin/entities/request.py

@@ -84,15 +84,15 @@ class RequestInvokeLLM(BaseRequestInvokeModel):
 
         for i in range(len(v)):
             if v[i]["role"] == PromptMessageRole.USER.value:
-                v[i] = UserPromptMessage(**v[i])
+                v[i] = UserPromptMessage.model_validate(v[i])
             elif v[i]["role"] == PromptMessageRole.ASSISTANT.value:
-                v[i] = AssistantPromptMessage(**v[i])
+                v[i] = AssistantPromptMessage.model_validate(v[i])
             elif v[i]["role"] == PromptMessageRole.SYSTEM.value:
-                v[i] = SystemPromptMessage(**v[i])
+                v[i] = SystemPromptMessage.model_validate(v[i])
             elif v[i]["role"] == PromptMessageRole.TOOL.value:
-                v[i] = ToolPromptMessage(**v[i])
+                v[i] = ToolPromptMessage.model_validate(v[i])
             else:
-                v[i] = PromptMessage(**v[i])
+                v[i] = PromptMessage.model_validate(v[i])
 
         return v
 

api/core/plugin/impl/base.py

@@ -94,7 +94,7 @@ class BasePluginClient:
         self,
         method: str,
         path: str,
-        type: type[T],
+        type_: type[T],
         headers: dict | None = None,
         data: bytes | dict | None = None,
         params: dict | None = None,
@@ -104,13 +104,13 @@ class BasePluginClient:
         Make a stream request to the plugin daemon inner API and yield the response as a model.
         """
         for line in self._stream_request(method, path, params, headers, data, files):
-            yield type(**json.loads(line))  # type: ignore
+            yield type_(**json.loads(line))  # type: ignore
 
     def _request_with_model(
         self,
         method: str,
         path: str,
-        type: type[T],
+        type_: type[T],
         headers: dict | None = None,
         data: bytes | None = None,
         params: dict | None = None,
@@ -120,13 +120,13 @@ class BasePluginClient:
         Make a request to the plugin daemon inner API and return the response as a model.
         """
         response = self._request(method, path, headers, data, params, files)
-        return type(**response.json())  # type: ignore
+        return type_(**response.json())  # type: ignore
 
     def _request_with_plugin_daemon_response(
         self,
         method: str,
         path: str,
-        type: type[T],
+        type_: type[T],
         headers: dict | None = None,
         data: bytes | dict | None = None,
         params: dict | None = None,
@@ -140,22 +140,22 @@ class BasePluginClient:
             response = self._request(method, path, headers, data, params, files)
             response.raise_for_status()
         except HTTPError as e:
-            msg = f"Failed to request plugin daemon, status: {e.response.status_code}, url: {path}"
-            logger.exception(msg)
+            logger.exception("Failed to request plugin daemon, status: %s, url: %s", e.response.status_code, path)
             raise e
         except Exception as e:
             msg = f"Failed to request plugin daemon, url: {path}"
-            logger.exception(msg)
+            logger.exception("Failed to request plugin daemon, url: %s", path)
             raise ValueError(msg) from e
 
         try:
             json_response = response.json()
             if transformer:
                 json_response = transformer(json_response)
-            rep = PluginDaemonBasicResponse[type](**json_response)  # type: ignore
+            # https://stackoverflow.com/questions/59634937/variable-foo-class-is-not-valid-as-type-but-why
+            rep = PluginDaemonBasicResponse[type_].model_validate(json_response)  # type: ignore
         except Exception:
             msg = (
-                f"Failed to parse response from plugin daemon to PluginDaemonBasicResponse [{str(type.__name__)}],"
+                f"Failed to parse response from plugin daemon to PluginDaemonBasicResponse [{str(type_.__name__)}],"
                 f" url: {path}"
             )
             logger.exception(msg)
@@ -163,7 +163,7 @@ class BasePluginClient:
 
         if rep.code != 0:
             try:
-                error = PluginDaemonError(**json.loads(rep.message))
+                error = PluginDaemonError.model_validate(json.loads(rep.message))
             except Exception:
                 raise ValueError(f"{rep.message}, code: {rep.code}")
 
@@ -178,7 +178,7 @@ class BasePluginClient:
         self,
         method: str,
         path: str,
-        type: type[T],
+        type_: type[T],
         headers: dict | None = None,
         data: bytes | dict | None = None,
         params: dict | None = None,
@@ -189,7 +189,7 @@ class BasePluginClient:
         """
         for line in self._stream_request(method, path, params, headers, data, files):
             try:
-                rep = PluginDaemonBasicResponse[type].model_validate_json(line)  # type: ignore
+                rep = PluginDaemonBasicResponse[type_].model_validate_json(line)  # type: ignore
             except (ValueError, TypeError):
                 # TODO modify this when line_data has code and message
                 try:
@@ -204,7 +204,7 @@ class BasePluginClient:
             if rep.code != 0:
                 if rep.code == -500:
                     try:
-                        error = PluginDaemonError(**json.loads(rep.message))
+                        error = PluginDaemonError.model_validate(json.loads(rep.message))
                     except Exception:
                         raise PluginDaemonInnerError(code=rep.code, message=rep.message)
 

api/core/plugin/impl/datasource.py

@@ -46,7 +46,9 @@ class PluginDatasourceManager(BasePluginClient):
             params={"page": 1, "page_size": 256},
             transformer=transformer,
         )
-        local_file_datasource_provider = PluginDatasourceProviderEntity(**self._get_local_file_datasource_provider())
+        local_file_datasource_provider = PluginDatasourceProviderEntity.model_validate(
+            self._get_local_file_datasource_provider()
+        )
 
         for provider in response:
             ToolTransformService.repack_provider(tenant_id=tenant_id, provider=provider)
@@ -104,7 +106,7 @@ class PluginDatasourceManager(BasePluginClient):
         Fetch datasource provider for the given tenant and plugin.
         """
         if provider_id == "langgenius/file/file":
-            return PluginDatasourceProviderEntity(**self._get_local_file_datasource_provider())
+            return PluginDatasourceProviderEntity.model_validate(self._get_local_file_datasource_provider())
 
         tool_provider_id = DatasourceProviderID(provider_id)
 

api/core/plugin/impl/model.py

@@ -162,7 +162,7 @@ class PluginModelClient(BasePluginClient):
         response = self._request_with_plugin_daemon_response_stream(
             method="POST",
             path=f"plugin/{tenant_id}/dispatch/llm/invoke",
-            type=LLMResultChunk,
+            type_=LLMResultChunk,
             data=jsonable_encoder(
                 {
                     "user_id": user_id,
@@ -208,7 +208,7 @@ class PluginModelClient(BasePluginClient):
         response = self._request_with_plugin_daemon_response_stream(
             method="POST",
             path=f"plugin/{tenant_id}/dispatch/llm/num_tokens",
-            type=PluginLLMNumTokensResponse,
+            type_=PluginLLMNumTokensResponse,
             data=jsonable_encoder(
                 {
                     "user_id": user_id,
@@ -250,7 +250,7 @@ class PluginModelClient(BasePluginClient):
         response = self._request_with_plugin_daemon_response_stream(
             method="POST",
             path=f"plugin/{tenant_id}/dispatch/text_embedding/invoke",
-            type=TextEmbeddingResult,
+            type_=TextEmbeddingResult,
             data=jsonable_encoder(
                 {
                     "user_id": user_id,
@@ -291,7 +291,7 @@ class PluginModelClient(BasePluginClient):
         response = self._request_with_plugin_daemon_response_stream(
             method="POST",
             path=f"plugin/{tenant_id}/dispatch/text_embedding/num_tokens",
-            type=PluginTextEmbeddingNumTokensResponse,
+            type_=PluginTextEmbeddingNumTokensResponse,
             data=jsonable_encoder(
                 {
                     "user_id": user_id,
@@ -334,7 +334,7 @@ class PluginModelClient(BasePluginClient):
         response = self._request_with_plugin_daemon_response_stream(
             method="POST",
             path=f"plugin/{tenant_id}/dispatch/rerank/invoke",
-            type=RerankResult,
+            type_=RerankResult,
             data=jsonable_encoder(
                 {
                     "user_id": user_id,
@@ -378,7 +378,7 @@ class PluginModelClient(BasePluginClient):
         response = self._request_with_plugin_daemon_response_stream(
             method="POST",
             path=f"plugin/{tenant_id}/dispatch/tts/invoke",
-            type=PluginStringResultResponse,
+            type_=PluginStringResultResponse,
             data=jsonable_encoder(
                 {
                     "user_id": user_id,
@@ -422,7 +422,7 @@ class PluginModelClient(BasePluginClient):
         response = self._request_with_plugin_daemon_response_stream(
             method="POST",
             path=f"plugin/{tenant_id}/dispatch/tts/model/voices",
-            type=PluginVoicesResponse,
+            type_=PluginVoicesResponse,
             data=jsonable_encoder(
                 {
                     "user_id": user_id,
@@ -466,7 +466,7 @@ class PluginModelClient(BasePluginClient):
         response = self._request_with_plugin_daemon_response_stream(
             method="POST",
             path=f"plugin/{tenant_id}/dispatch/speech2text/invoke",
-            type=PluginStringResultResponse,
+            type_=PluginStringResultResponse,
             data=jsonable_encoder(
                 {
                     "user_id": user_id,
@@ -506,7 +506,7 @@ class PluginModelClient(BasePluginClient):
         response = self._request_with_plugin_daemon_response_stream(
             method="POST",
             path=f"plugin/{tenant_id}/dispatch/moderation/invoke",
-            type=PluginBasicBooleanResponse,
+            type_=PluginBasicBooleanResponse,
             data=jsonable_encoder(
                 {
                     "user_id": user_id,

api/core/plugin/utils/chunk_merger.py

@@ -1,6 +1,6 @@
 from collections.abc import Generator
 from dataclasses import dataclass, field
-from typing import TypeVar, Union, cast
+from typing import TypeVar, Union
 
 from core.agent.entities import AgentInvokeMessage
 from core.tools.entities.tool_entities import ToolInvokeMessage
@@ -87,7 +87,8 @@ def merge_blob_chunks(
                     ),
                     meta=resp.meta,
                 )
-                yield cast(MessageType, merged_message)
+                assert isinstance(merged_message, (ToolInvokeMessage, AgentInvokeMessage))
+                yield merged_message  # type: ignore
                 # Clean up the buffer
                 del files[chunk_id]
         else:

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,9 +618,9 @@ 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(
@@ -628,8 +628,8 @@ class ProviderManager:
                                 # TODO: Use provider name with prefix after the data migration.
                                 provider_name=ModelProviderID(provider_name).provider_name,
                                 provider_type=ProviderType.SYSTEM.value,
-                                quota_type=ProviderQuotaType.TRIAL.value,
-                                quota_limit=quota.quota_limit,  # type: ignore
+                                quota_type=quota.quota_type,
+                                quota_limit=0,  # type: ignore
                                 quota_used=0,
                                 is_valid=True,
                             )
@@ -642,7 +642,7 @@ class ProviderManager:
                                 Provider.tenant_id == tenant_id,
                                 Provider.provider_name == ModelProviderID(provider_name).provider_name,
                                 Provider.provider_type == ProviderType.SYSTEM.value,
-                                Provider.quota_type == ProviderQuotaType.TRIAL.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,
+                    )
+                
+                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,
+                    )
 
-                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,
-                )
+                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/retrieval_service.py

@@ -134,7 +134,7 @@ class RetrievalService:
         if not dataset:
             return []
         metadata_condition = (
-            MetadataCondition(**metadata_filtering_conditions) if metadata_filtering_conditions else None
+            MetadataCondition.model_validate(metadata_filtering_conditions) if metadata_filtering_conditions else None
         )
         all_documents = ExternalDatasetService.fetch_external_knowledge_retrieval(
             dataset.tenant_id,

api/core/rag/extractor/entity/extract_setting.py

@@ -17,9 +17,6 @@ class NotionInfo(BaseModel):
     tenant_id: str
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
-    def __init__(self, **data):
-        super().__init__(**data)
-
 
 class WebsiteInfo(BaseModel):
     """
@@ -47,6 +44,3 @@ class ExtractSetting(BaseModel):
     website_info: WebsiteInfo | None = None
     document_model: str | None = None
     model_config = ConfigDict(arbitrary_types_allowed=True)
-
-    def __init__(self, **data):
-        super().__init__(**data)

api/core/rag/index_processor/processor/paragraph_index_processor.py

@@ -38,11 +38,11 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
             raise ValueError("No process rule found.")
         if process_rule.get("mode") == "automatic":
             automatic_rule = DatasetProcessRule.AUTOMATIC_RULES
-            rules = Rule(**automatic_rule)
+            rules = Rule.model_validate(automatic_rule)
         else:
             if not process_rule.get("rules"):
                 raise ValueError("No rules found in process rule.")
-            rules = Rule(**process_rule.get("rules"))
+            rules = Rule.model_validate(process_rule.get("rules"))
         # Split the text documents into nodes.
         if not rules.segmentation:
             raise ValueError("No segmentation found in rules.")

api/core/rag/index_processor/processor/parent_child_index_processor.py

@@ -40,7 +40,7 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
             raise ValueError("No process rule found.")
         if not process_rule.get("rules"):
             raise ValueError("No rules found in process rule.")
-        rules = Rule(**process_rule.get("rules"))
+        rules = Rule.model_validate(process_rule.get("rules"))
         all_documents: list[Document] = []
         if rules.parent_mode == ParentMode.PARAGRAPH:
             # Split the text documents into nodes.
@@ -110,7 +110,7 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
                 child_documents = document.children
                 if child_documents:
                     formatted_child_documents = [
-                        Document(**child_document.model_dump()) for child_document in child_documents
+                        Document.model_validate(child_document.model_dump()) for child_document in child_documents
                     ]
                     vector.create(formatted_child_documents)
 
@@ -224,7 +224,7 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
         return child_nodes
 
     def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any):
-        parent_childs = ParentChildStructureChunk(**chunks)
+        parent_childs = ParentChildStructureChunk.model_validate(chunks)
         documents = []
         for parent_child in parent_childs.parent_child_chunks:
             metadata = {
@@ -274,7 +274,7 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
                     vector.create(all_child_documents)
 
     def format_preview(self, chunks: Any) -> Mapping[str, Any]:
-        parent_childs = ParentChildStructureChunk(**chunks)
+        parent_childs = ParentChildStructureChunk.model_validate(chunks)
         preview = []
         for parent_child in parent_childs.parent_child_chunks:
             preview.append({"content": parent_child.parent_content, "child_chunks": parent_child.child_contents})

api/core/rag/index_processor/processor/qa_index_processor.py

@@ -47,7 +47,7 @@ class QAIndexProcessor(BaseIndexProcessor):
             raise ValueError("No process rule found.")
         if not process_rule.get("rules"):
             raise ValueError("No rules found in process rule.")
-        rules = Rule(**process_rule.get("rules"))
+        rules = Rule.model_validate(process_rule.get("rules"))
         splitter = self._get_splitter(
             processing_rule_mode=process_rule.get("mode"),
             max_tokens=rules.segmentation.max_tokens if rules.segmentation else 0,
@@ -168,7 +168,7 @@ class QAIndexProcessor(BaseIndexProcessor):
         return docs
 
     def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any):
-        qa_chunks = QAStructureChunk(**chunks)
+        qa_chunks = QAStructureChunk.model_validate(chunks)
         documents = []
         for qa_chunk in qa_chunks.qa_chunks:
             metadata = {
@@ -191,7 +191,7 @@ class QAIndexProcessor(BaseIndexProcessor):
                 raise ValueError("Indexing technique must be high quality.")
 
     def format_preview(self, chunks: Any) -> Mapping[str, Any]:
-        qa_chunks = QAStructureChunk(**chunks)
+        qa_chunks = QAStructureChunk.model_validate(chunks)
         preview = []
         for qa_chunk in qa_chunks.qa_chunks:
             preview.append({"question": qa_chunk.question, "answer": qa_chunk.answer})

api/core/rag/splitter/fixed_text_splitter.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import re
 from typing import Any
 
 from core.model_manager import ModelInstance
@@ -52,7 +53,7 @@ class FixedRecursiveCharacterTextSplitter(EnhanceRecursiveCharacterTextSplitter)
         """Create a new TextSplitter."""
         super().__init__(**kwargs)
         self._fixed_separator = fixed_separator
-        self._separators = separators or ["\n\n", "\n", " ", ""]
+        self._separators = separators or ["\n\n", "\n", "。", ". ", " ", ""]
 
     def split_text(self, text: str) -> list[str]:
         """Split incoming text and return chunks."""
@@ -90,16 +91,19 @@ class FixedRecursiveCharacterTextSplitter(EnhanceRecursiveCharacterTextSplitter)
         # Now that we have the separator, split the text
         if separator:
             if separator == " ":
-                splits = text.split()
+                splits = re.split(r" +", text)
             else:
                 splits = text.split(separator)
                 splits = [item + separator if i < len(splits) else item for i, item in enumerate(splits)]
         else:
             splits = list(text)
-        splits = [s for s in splits if (s not in {"", "\n"})]
+        if separator == "\n":
+            splits = [s for s in splits if s != ""]
+        else:
+            splits = [s for s in splits if (s not in {"", "\n"})]
         _good_splits = []
         _good_splits_lengths = []  # cache the lengths of the splits
-        _separator = "" if self._keep_separator else separator
+        _separator = separator if self._keep_separator else ""
         s_lens = self._length_function(splits)
         if separator != "":
             for s, s_len in zip(splits, s_lens):

api/core/tools/builtin_tool/provider.py

@@ -90,7 +90,7 @@ class BuiltinToolProviderController(ToolProviderController):
             tools.append(
                 assistant_tool_class(
                     provider=provider,
-                    entity=ToolEntity(**tool),
+                    entity=ToolEntity.model_validate(tool),
                     runtime=ToolRuntime(tenant_id=""),
                 )
             )

api/core/tools/entities/common_entities.py

@@ -1,4 +1,4 @@
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, model_validator
 
 
 class I18nObject(BaseModel):
@@ -11,11 +11,12 @@ class I18nObject(BaseModel):
     pt_BR: str | None = Field(default=None)
     ja_JP: str | None = Field(default=None)
 
-    def __init__(self, **data):
-        super().__init__(**data)
+    @model_validator(mode="after")
+    def _populate_missing_locales(self):
         self.zh_Hans = self.zh_Hans or self.en_US
         self.pt_BR = self.pt_BR or self.en_US
         self.ja_JP = self.ja_JP or self.en_US
+        return self
 
     def to_dict(self):
         return {"zh_Hans": self.zh_Hans, "en_US": self.en_US, "pt_BR": self.pt_BR, "ja_JP": self.ja_JP}

api/core/tools/mcp_tool/provider.py

@@ -54,7 +54,7 @@ class MCPToolProviderController(ToolProviderController):
         """
         tools = []
         tools_data = json.loads(db_provider.tools)
-        remote_mcp_tools = [RemoteMCPTool(**tool) for tool in tools_data]
+        remote_mcp_tools = [RemoteMCPTool.model_validate(tool) for tool in tools_data]
         user = db_provider.load_user()
         tools = [
             ToolEntity(

api/core/tools/tool_manager.py

@@ -1008,7 +1008,7 @@ class ToolManager:
                     config = tool_configurations.get(parameter.name, {})
                     if not (config and isinstance(config, dict) and config.get("value") is not None):
                         continue
-                    tool_input = ToolNodeData.ToolInput(**tool_configurations.get(parameter.name, {}))
+                    tool_input = ToolNodeData.ToolInput.model_validate(tool_configurations.get(parameter.name, {}))
                     if tool_input.type == "variable":
                         variable = variable_pool.get(tool_input.value)
                         if variable is None:

api/core/tools/utils/dataset_retriever/dataset_multi_retriever_tool.py

@@ -126,7 +126,7 @@ class DatasetMultiRetrieverTool(DatasetRetrieverBaseTool):
                             data_source_type=document.data_source_type,
                             segment_id=segment.id,
                             retriever_from=self.retriever_from,
-                            score=document_score_list.get(segment.index_node_id, None),
+                            score=document_score_list.get(segment.index_node_id),
                             doc_metadata=document.doc_metadata,
                         )
 

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] = None  # type: ignore
@@ -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/entities/variable_pool.py

@@ -8,11 +8,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,
 )
@@ -55,6 +56,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
@@ -75,6 +80,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/core/workflow/graph_engine/command_channels/redis_channel.py

@@ -105,10 +105,10 @@ class RedisChannel:
             command_type = CommandType(command_type_value)
 
             if command_type == CommandType.ABORT:
-                return AbortCommand(**data)
+                return AbortCommand.model_validate(data)
             else:
                 # For other command types, use base class
-                return GraphEngineCommand(**data)
+                return GraphEngineCommand.model_validate(data)
 
         except (ValueError, TypeError):
             return None

api/core/workflow/nodes/end/end_node.py

@@ -16,7 +16,7 @@ class EndNode(Node):
     _node_data: EndNodeData
 
     def init_node_data(self, data: Mapping[str, Any]):
-        self._node_data = EndNodeData(**data)
+        self._node_data = EndNodeData.model_validate(data)
 
     def _get_error_strategy(self) -> ErrorStrategy | None:
         return self._node_data.error_strategy

api/core/workflow/nodes/iteration/iteration_node.py

@@ -342,10 +342,13 @@ class IterationNode(Node):
         iterator_list_value: Sequence[object],
         iter_run_map: dict[str, float],
     ) -> Generator[NodeEventBase, None, None]:
+        # Flatten the list of lists if all outputs are lists
+        flattened_outputs = self._flatten_outputs_if_needed(outputs)
+
         yield IterationSucceededEvent(
             start_at=started_at,
             inputs=inputs,
-            outputs={"output": outputs},
+            outputs={"output": flattened_outputs},
             steps=len(iterator_list_value),
             metadata={
                 WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS: self.graph_runtime_state.total_tokens,
@@ -357,13 +360,39 @@ class IterationNode(Node):
         yield StreamCompletedEvent(
             node_run_result=NodeRunResult(
                 status=WorkflowNodeExecutionStatus.SUCCEEDED,
-                outputs={"output": outputs},
+                outputs={"output": flattened_outputs},
                 metadata={
                     WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS: self.graph_runtime_state.total_tokens,
                 },
             )
         )
 
+    def _flatten_outputs_if_needed(self, outputs: list[object]) -> list[object]:
+        """
+        Flatten the outputs list if all elements are lists.
+        This maintains backward compatibility with version 1.8.1 behavior.
+        """
+        if not outputs:
+            return outputs
+
+        # Check if all non-None outputs are lists
+        non_none_outputs = [output for output in outputs if output is not None]
+        if not non_none_outputs:
+            return outputs
+
+        if all(isinstance(output, list) for output in non_none_outputs):
+            # Flatten the list of lists
+            flattened: list[Any] = []
+            for output in outputs:
+                if isinstance(output, list):
+                    flattened.extend(output)
+                elif output is not None:
+                    # This shouldn't happen based on our check, but handle it gracefully
+                    flattened.append(output)
+            return flattened
+
+        return outputs
+
     def _handle_iteration_failure(
         self,
         started_at: datetime,
@@ -373,10 +402,13 @@ class IterationNode(Node):
         iter_run_map: dict[str, float],
         error: IterationNodeError,
     ) -> Generator[NodeEventBase, None, None]:
+        # Flatten the list of lists if all outputs are lists (even in failure case)
+        flattened_outputs = self._flatten_outputs_if_needed(outputs)
+
         yield IterationFailedEvent(
             start_at=started_at,
             inputs=inputs,
-            outputs={"output": outputs},
+            outputs={"output": flattened_outputs},
             steps=len(iterator_list_value),
             metadata={
                 WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS: self.graph_runtime_state.total_tokens,

api/core/workflow/nodes/iteration/iteration_start_node.py

@@ -18,7 +18,7 @@ class IterationStartNode(Node):
     _node_data: IterationStartNodeData
 
     def init_node_data(self, data: Mapping[str, Any]):
-        self._node_data = IterationStartNodeData(**data)
+        self._node_data = IterationStartNodeData.model_validate(data)
 
     def _get_error_strategy(self) -> ErrorStrategy | None:
         return self._node_data.error_strategy

api/core/workflow/nodes/knowledge_index/knowledge_index_node.py

@@ -2,7 +2,7 @@ import datetime
 import logging
 import time
 from collections.abc import Mapping
-from typing import Any, cast
+from typing import Any
 
 from sqlalchemy import func, select
 
@@ -62,7 +62,7 @@ class KnowledgeIndexNode(Node):
         return self._node_data
 
     def _run(self) -> NodeRunResult:  # type: ignore
-        node_data = cast(KnowledgeIndexNodeData, self._node_data)
+        node_data = self._node_data
         variable_pool = self.graph_runtime_state.variable_pool
         dataset_id = variable_pool.get(["sys", SystemVariableKey.DATASET_ID])
         if not dataset_id:

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

@@ -41,7 +41,7 @@ class ListOperatorNode(Node):
     _node_data: ListOperatorNodeData
 
     def init_node_data(self, data: Mapping[str, Any]):
-        self._node_data = ListOperatorNodeData(**data)
+        self._node_data = ListOperatorNodeData.model_validate(data)
 
     def _get_error_strategy(self) -> ErrorStrategy | None:
         return self._node_data.error_strategy

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.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(),
-                )
+
+        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

@@ -6,11 +6,15 @@ import re
 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 (
@@ -71,6 +75,8 @@ from core.workflow.node_events import (
 from core.workflow.nodes.base.entities import BaseNodeData, RetryConfig, VariableSelector
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.base.variable_template_parser import VariableTemplateParser
+from models import UserFrom, Workflow
+from models.engine import db
 
 from . import llm_utils
 from .entities import (
@@ -315,6 +321,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"],
@@ -1184,6 +1195,79 @@ 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/nodes/loop/loop_end_node.py

@@ -18,7 +18,7 @@ class LoopEndNode(Node):
     _node_data: LoopEndNodeData
 
     def init_node_data(self, data: Mapping[str, Any]):
-        self._node_data = LoopEndNodeData(**data)
+        self._node_data = LoopEndNodeData.model_validate(data)
 
     def _get_error_strategy(self) -> ErrorStrategy | None:
         return self._node_data.error_strategy

api/core/workflow/nodes/loop/loop_start_node.py

@@ -18,7 +18,7 @@ class LoopStartNode(Node):
     _node_data: LoopStartNodeData
 
     def init_node_data(self, data: Mapping[str, Any]):
-        self._node_data = LoopStartNodeData(**data)
+        self._node_data = LoopStartNodeData.model_validate(data)
 
     def _get_error_strategy(self) -> ErrorStrategy | None:
         return self._node_data.error_strategy

api/core/workflow/nodes/start/start_node.py

@@ -16,7 +16,7 @@ class StartNode(Node):
     _node_data: StartNodeData
 
     def init_node_data(self, data: Mapping[str, Any]):
-        self._node_data = StartNodeData(**data)
+        self._node_data = StartNodeData.model_validate(data)
 
     def _get_error_strategy(self) -> ErrorStrategy | None:
         return self._node_data.error_strategy

api/core/workflow/nodes/variable_aggregator/variable_aggregator_node.py

@@ -15,7 +15,7 @@ class VariableAggregatorNode(Node):
     _node_data: VariableAssignerNodeData
 
     def init_node_data(self, data: Mapping[str, Any]):
-        self._node_data = VariableAssignerNodeData(**data)
+        self._node_data = VariableAssignerNodeData.model_validate(data)
 
     def _get_error_strategy(self) -> ErrorStrategy | None:
         return self._node_data.error_strategy