if session unauthorized, rejoin

The collaboration feature increased workflow operator z-index from z-10 to z-[60].
This caused the AppInfo ContentDialog (z-30) to appear below the operator buttons.
Increased ContentDialog z-index to z-[70] to ensure proper layer hierarchy.

api/.env.example

@@ -30,6 +30,9 @@ INTERNAL_FILES_URL=http://127.0.0.1:5001
 # The time in seconds after the signature is rejected
 FILES_ACCESS_TIMEOUT=300
 
+# Collaboration mode toggle
+ENABLE_COLLABORATION_MODE=false
+
 # Access token expiration time in minutes
 ACCESS_TOKEN_EXPIRE_MINUTES=60
 
@@ -434,9 +437,6 @@ CODE_EXECUTION_SSL_VERIFY=True
 CODE_EXECUTION_POOL_MAX_CONNECTIONS=100
 CODE_EXECUTION_POOL_MAX_KEEPALIVE_CONNECTIONS=20
 CODE_EXECUTION_POOL_KEEPALIVE_EXPIRY=5.0
-CODE_EXECUTION_CONNECT_TIMEOUT=10
-CODE_EXECUTION_READ_TIMEOUT=60
-CODE_EXECUTION_WRITE_TIMEOUT=10
 CODE_MAX_NUMBER=9223372036854775807
 CODE_MIN_NUMBER=-9223372036854775808
 CODE_MAX_STRING_LENGTH=400000

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

@@ -1048,6 +1048,13 @@ class PositionConfig(BaseSettings):
         return {item.strip() for item in self.POSITION_TOOL_EXCLUDES.split(",") if item.strip() != ""}
 
 
+class CollaborationConfig(BaseSettings):
+    ENABLE_COLLABORATION_MODE: bool = Field(
+        description="Whether to enable collaboration mode features across the workspace",
+        default=False,
+    )
+
+
 class LoginConfig(BaseSettings):
     ENABLE_EMAIL_CODE_LOGIN: bool = Field(
         description="whether to enable email code login",
@@ -1136,6 +1143,7 @@ class FeatureConfig(
     WorkflowConfig,
     WorkflowNodeExecutionConfig,
     WorkspaceConfig,
+    CollaborationConfig,
     LoginConfig,
     AccountConfig,
     SwaggerUIConfig,

api/configs/middleware/__init__.py

@@ -145,7 +145,7 @@ class DatabaseConfig(BaseSettings):
         default="postgresql",
     )
 
-    @computed_field  # type: ignore[prop-decorator]
+    @computed_field  # type: ignore[misc]
     @property
     def SQLALCHEMY_DATABASE_URI(self) -> str:
         db_extras = (
@@ -198,7 +198,7 @@ class DatabaseConfig(BaseSettings):
         default=os.cpu_count() or 1,
     )
 
-    @computed_field  # type: ignore[prop-decorator]
+    @computed_field  # type: ignore[misc]
     @property
     def SQLALCHEMY_ENGINE_OPTIONS(self) -> dict[str, Any]:
         # Parse DB_EXTRAS for 'options'

api/constants/__init__.py

@@ -56,14 +56,10 @@ else:
     }
 DOCUMENT_EXTENSIONS: set[str] = convert_to_lower_and_upper_set(_doc_extensions)
 
-# console
 COOKIE_NAME_ACCESS_TOKEN = "access_token"
 COOKIE_NAME_REFRESH_TOKEN = "refresh_token"
-COOKIE_NAME_CSRF_TOKEN = "csrf_token"
-
-# webapp
-COOKIE_NAME_WEBAPP_ACCESS_TOKEN = "webapp_access_token"
 COOKIE_NAME_PASSPORT = "passport"
+COOKIE_NAME_CSRF_TOKEN = "csrf_token"
 
 HEADER_NAME_CSRF_TOKEN = "X-CSRF-Token"
 HEADER_NAME_APP_CODE = "X-App-Code"

api/constants/languages.py

@@ -31,9 +31,3 @@ def supported_language(lang):
 
     error = f"{lang} is not a valid language."
     raise ValueError(error)
-
-
-def get_valid_language(lang: str | None) -> str:
-    if lang and lang in languages:
-        return lang
-    return languages[0]

api/controllers/common/helpers.py

@@ -24,7 +24,7 @@ except ImportError:
         )
     else:
         warnings.warn("To use python-magic guess MIMETYPE, you need to install `libmagic`", stacklevel=2)
-    magic = None  # type: ignore[assignment]
+    magic = None  # type: ignore
 
 from pydantic import BaseModel
 

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,

api/controllers/console/app/online_user.py

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

api/controllers/console/app/workflow.py

@@ -9,6 +9,7 @@ from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
 
 import services
+from configs import dify_config
 from controllers.console import api, console_ns
 from controllers.console.app.error import ConversationCompletedError, DraftWorkflowNotExist, DraftWorkflowNotSync
 from controllers.console.app.wraps import get_app_model
@@ -21,7 +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
@@ -122,6 +125,7 @@ class DraftWorkflowApi(Resource):
                 .add_argument("hash", type=str, required=False, location="json")
                 .add_argument("environment_variables", type=list, required=True, location="json")
                 .add_argument("conversation_variables", type=list, required=False, location="json")
+                .add_argument("force_upload", type=bool, required=False, default=False, location="json")
             )
             args = parser.parse_args()
         elif "text/plain" in content_type:
@@ -139,6 +143,7 @@ class DraftWorkflowApi(Resource):
                     "hash": data.get("hash"),
                     "environment_variables": data.get("environment_variables"),
                     "conversation_variables": data.get("conversation_variables"),
+                    "force_upload": data.get("force_upload", False),
                 }
             except json.JSONDecodeError:
                 return {"message": "Invalid JSON data"}, 400
@@ -163,6 +168,7 @@ class DraftWorkflowApi(Resource):
                 account=current_user,
                 environment_variables=environment_variables,
                 conversation_variables=conversation_variables,
+                force_upload=args.get("force_upload", False),
             )
         except WorkflowHashNotEqualError:
             raise DraftWorkflowNotSync()
@@ -734,6 +740,46 @@ 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,
+        }
+
+
+@console_ns.route("/apps/<uuid:app_id>/workflows/draft/features")
+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):
+        current_user, _ = current_account_with_tenant()
+
+        parser = reqparse.RequestParser().add_argument("features", type=dict, required=True, location="json")
+        args = parser.parse_args()
+
+        features = args.get("features")
+
+        workflow_service = WorkflowService()
+        workflow_service.update_draft_workflow_features(app_model=app_model, features=features, account=current_user)
+
+        return {"result": "success"}
+
+
 @console_ns.route("/apps/<uuid:app_id>/workflows")
 class PublishedAllWorkflowApi(Resource):
     @api.doc("get_all_published_workflows")
@@ -915,3 +961,30 @@ class DraftWorkflowNodeLastRunApi(Resource):
         if node_exec is None:
             raise NotFound("last run not found")
         return node_exec
+
+
+@console_ns.route("/apps/workflows/online-users")
+class WorkflowOnlineUsersApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @marshal_with(online_user_list_fields)
+    def get(self):
+        parser = reqparse.RequestParser().add_argument("workflow_ids", type=str, required=True, location="args")
+        args = parser.parse_args()
+
+        workflow_ids = [workflow_id.strip() for workflow_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/controllers/console/app/workflow_comment.py

@@ -0,0 +1,240 @@
+import logging
+
+from flask_restx import Resource, fields, marshal_with, reqparse
+
+from controllers.console import api
+from controllers.console.app.wraps import get_app_model
+from controllers.console.wraps import account_initialization_required, setup_required
+from fields.member_fields import account_with_role_fields
+from fields.workflow_comment_fields import (
+    workflow_comment_basic_fields,
+    workflow_comment_create_fields,
+    workflow_comment_detail_fields,
+    workflow_comment_reply_create_fields,
+    workflow_comment_reply_update_fields,
+    workflow_comment_resolve_fields,
+    workflow_comment_update_fields,
+)
+from libs.login import current_user, login_required
+from models import App
+from services.account_service import TenantService
+from services.workflow_comment_service import WorkflowCommentService
+
+logger = logging.getLogger(__name__)
+
+
+class WorkflowCommentListApi(Resource):
+    """API for listing and creating workflow comments."""
+
+    @login_required
+    @setup_required
+    @account_initialization_required
+    @get_app_model
+    @marshal_with(workflow_comment_basic_fields, envelope="data")
+    def get(self, app_model: App):
+        """Get all comments for a workflow."""
+        comments = WorkflowCommentService.get_comments(tenant_id=current_user.current_tenant_id, app_id=app_model.id)
+
+        return comments
+
+    @login_required
+    @setup_required
+    @account_initialization_required
+    @get_app_model
+    @marshal_with(workflow_comment_create_fields)
+    def post(self, app_model: App):
+        """Create a new workflow comment."""
+        parser = reqparse.RequestParser()
+        parser.add_argument("position_x", type=float, required=True, location="json")
+        parser.add_argument("position_y", type=float, required=True, location="json")
+        parser.add_argument("content", type=str, required=True, location="json")
+        parser.add_argument("mentioned_user_ids", type=list, location="json", default=[])
+        args = parser.parse_args()
+
+        result = WorkflowCommentService.create_comment(
+            tenant_id=current_user.current_tenant_id,
+            app_id=app_model.id,
+            created_by=current_user.id,
+            content=args.content,
+            position_x=args.position_x,
+            position_y=args.position_y,
+            mentioned_user_ids=args.mentioned_user_ids,
+        )
+
+        return result, 201
+
+
+class WorkflowCommentDetailApi(Resource):
+    """API for managing individual workflow comments."""
+
+    @login_required
+    @setup_required
+    @account_initialization_required
+    @get_app_model
+    @marshal_with(workflow_comment_detail_fields)
+    def get(self, app_model: App, comment_id: str):
+        """Get a specific workflow comment."""
+        comment = WorkflowCommentService.get_comment(
+            tenant_id=current_user.current_tenant_id, app_id=app_model.id, comment_id=comment_id
+        )
+
+        return comment
+
+    @login_required
+    @setup_required
+    @account_initialization_required
+    @get_app_model
+    @marshal_with(workflow_comment_update_fields)
+    def put(self, app_model: App, comment_id: str):
+        """Update a workflow comment."""
+        parser = reqparse.RequestParser()
+        parser.add_argument("content", type=str, required=True, location="json")
+        parser.add_argument("position_x", type=float, required=False, location="json")
+        parser.add_argument("position_y", type=float, required=False, location="json")
+        parser.add_argument("mentioned_user_ids", type=list, location="json", default=[])
+        args = parser.parse_args()
+
+        result = WorkflowCommentService.update_comment(
+            tenant_id=current_user.current_tenant_id,
+            app_id=app_model.id,
+            comment_id=comment_id,
+            user_id=current_user.id,
+            content=args.content,
+            position_x=args.position_x,
+            position_y=args.position_y,
+            mentioned_user_ids=args.mentioned_user_ids,
+        )
+
+        return result
+
+    @login_required
+    @setup_required
+    @account_initialization_required
+    @get_app_model
+    def delete(self, app_model: App, comment_id: str):
+        """Delete a workflow comment."""
+        WorkflowCommentService.delete_comment(
+            tenant_id=current_user.current_tenant_id,
+            app_id=app_model.id,
+            comment_id=comment_id,
+            user_id=current_user.id,
+        )
+
+        return {"result": "success"}, 204
+
+
+class WorkflowCommentResolveApi(Resource):
+    """API for resolving and reopening workflow comments."""
+
+    @login_required
+    @setup_required
+    @account_initialization_required
+    @get_app_model
+    @marshal_with(workflow_comment_resolve_fields)
+    def post(self, app_model: App, comment_id: str):
+        """Resolve a workflow comment."""
+        comment = WorkflowCommentService.resolve_comment(
+            tenant_id=current_user.current_tenant_id,
+            app_id=app_model.id,
+            comment_id=comment_id,
+            user_id=current_user.id,
+        )
+
+        return comment
+
+
+class WorkflowCommentReplyApi(Resource):
+    """API for managing comment replies."""
+
+    @login_required
+    @setup_required
+    @account_initialization_required
+    @get_app_model
+    @marshal_with(workflow_comment_reply_create_fields)
+    def post(self, app_model: App, comment_id: str):
+        """Add a reply to a workflow comment."""
+        # Validate comment access first
+        WorkflowCommentService.validate_comment_access(
+            comment_id=comment_id, tenant_id=current_user.current_tenant_id, app_id=app_model.id
+        )
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("content", type=str, required=True, location="json")
+        parser.add_argument("mentioned_user_ids", type=list, location="json", default=[])
+        args = parser.parse_args()
+
+        result = WorkflowCommentService.create_reply(
+            comment_id=comment_id,
+            content=args.content,
+            created_by=current_user.id,
+            mentioned_user_ids=args.mentioned_user_ids,
+        )
+
+        return result, 201
+
+
+class WorkflowCommentReplyDetailApi(Resource):
+    """API for managing individual comment replies."""
+
+    @login_required
+    @setup_required
+    @account_initialization_required
+    @get_app_model
+    @marshal_with(workflow_comment_reply_update_fields)
+    def put(self, app_model: App, comment_id: str, reply_id: str):
+        """Update a comment reply."""
+        # Validate comment access first
+        WorkflowCommentService.validate_comment_access(
+            comment_id=comment_id, tenant_id=current_user.current_tenant_id, app_id=app_model.id
+        )
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("content", type=str, required=True, location="json")
+        parser.add_argument("mentioned_user_ids", type=list, location="json", default=[])
+        args = parser.parse_args()
+
+        reply = WorkflowCommentService.update_reply(
+            reply_id=reply_id, user_id=current_user.id, content=args.content, mentioned_user_ids=args.mentioned_user_ids
+        )
+
+        return reply
+
+    @login_required
+    @setup_required
+    @account_initialization_required
+    @get_app_model
+    def delete(self, app_model: App, comment_id: str, reply_id: str):
+        """Delete a comment reply."""
+        # Validate comment access first
+        WorkflowCommentService.validate_comment_access(
+            comment_id=comment_id, tenant_id=current_user.current_tenant_id, app_id=app_model.id
+        )
+
+        WorkflowCommentService.delete_reply(reply_id=reply_id, user_id=current_user.id)
+
+        return {"result": "success"}, 204
+
+
+class WorkflowCommentMentionUsersApi(Resource):
+    """API for getting mentionable users for workflow comments."""
+
+    @login_required
+    @setup_required
+    @account_initialization_required
+    @get_app_model
+    @marshal_with({"users": fields.List(fields.Nested(account_with_role_fields))})
+    def get(self, app_model: App):
+        """Get all users in current tenant for mentions."""
+        members = TenantService.get_tenant_members(current_user.current_tenant)
+        return {"users": members}
+
+
+# Register API routes
+api.add_resource(WorkflowCommentListApi, "/apps/<uuid:app_id>/workflow/comments")
+api.add_resource(WorkflowCommentDetailApi, "/apps/<uuid:app_id>/workflow/comments/<string:comment_id>")
+api.add_resource(WorkflowCommentResolveApi, "/apps/<uuid:app_id>/workflow/comments/<string:comment_id>/resolve")
+api.add_resource(WorkflowCommentReplyApi, "/apps/<uuid:app_id>/workflow/comments/<string:comment_id>/replies")
+api.add_resource(
+    WorkflowCommentReplyDetailApi, "/apps/<uuid:app_id>/workflow/comments/<string:comment_id>/replies/<string:reply_id>"
+)
+api.add_resource(WorkflowCommentMentionUsersApi, "/apps/<uuid:app_id>/workflow/comments/mention-users")

api/controllers/console/app/workflow_draft_variable.py

@@ -19,8 +19,8 @@ from core.variables.segments import ArrayFileSegment, FileSegment, Segment
 from core.variables.types import SegmentType
 from core.workflow.constants import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
 from extensions.ext_database import db
+from factories import variable_factory
 from factories.file_factory import build_from_mapping, build_from_mappings
-from factories.variable_factory import build_segment_with_type
 from libs.login import current_user, login_required
 from models import Account, App, AppMode
 from models.workflow import WorkflowDraftVariable
@@ -355,7 +355,7 @@ class VariableApi(Resource):
                 if len(raw_value) > 0 and not isinstance(raw_value[0], dict):
                     raise InvalidArgumentError(description=f"expected dict for files[0], got {type(raw_value)}")
                 raw_value = build_from_mappings(mappings=raw_value, tenant_id=app_model.tenant_id)
-            new_value = build_segment_with_type(variable.value_type, raw_value)
+            new_value = variable_factory.build_segment_with_type(variable.value_type, raw_value)
         draft_var_srv.update_variable(variable, name=new_name, value=new_value)
         db.session.commit()
         return variable
@@ -448,8 +448,35 @@ class ConversationVariableCollectionApi(Resource):
         db.session.commit()
         return _get_variable_list(app_model, CONVERSATION_VARIABLE_NODE_ID)
 
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=AppMode.ADVANCED_CHAT)
+    def post(self, app_model: App):
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not current_user.is_editor:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("conversation_variables", type=list, required=True, location="json")
+        args = parser.parse_args()
+
+        workflow_service = WorkflowService()
+
+        conversation_variables_list = args.get("conversation_variables") or []
+        conversation_variables = [
+            variable_factory.build_conversation_variable_from_mapping(obj) for obj in conversation_variables_list
+        ]
+
+        workflow_service.update_draft_workflow_conversation_variables(
+            app_model=app_model,
+            account=current_user,
+            conversation_variables=conversation_variables,
+        )
+
+        return {"result": "success"}
+
 
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/system-variables")
 class SystemVariableCollectionApi(Resource):
     @api.doc("get_system_variables")
     @api.doc(description="Get system variables for workflow")
@@ -499,3 +526,44 @@ class EnvironmentVariableCollectionApi(Resource):
             )
 
         return {"items": env_vars_list}
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    def post(self, app_model: App):
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not current_user.is_editor:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("environment_variables", type=list, required=True, location="json")
+        args = parser.parse_args()
+
+        workflow_service = WorkflowService()
+
+        environment_variables_list = args.get("environment_variables") or []
+        environment_variables = [
+            variable_factory.build_environment_variable_from_mapping(obj) for obj in environment_variables_list
+        ]
+
+        workflow_service.update_draft_workflow_environment_variables(
+            app_model=app_model,
+            account=current_user,
+            environment_variables=environment_variables,
+        )
+
+        return {"result": "success"}
+
+
+api.add_resource(
+    WorkflowVariableCollectionApi,
+    "/apps/<uuid:app_id>/workflows/draft/variables",
+)
+api.add_resource(NodeVariableCollectionApi, "/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/variables")
+api.add_resource(VariableApi, "/apps/<uuid:app_id>/workflows/draft/variables/<uuid:variable_id>")
+api.add_resource(VariableResetApi, "/apps/<uuid:app_id>/workflows/draft/variables/<uuid:variable_id>/reset")
+
+api.add_resource(ConversationVariableCollectionApi, "/apps/<uuid:app_id>/workflows/draft/conversation-variables")
+api.add_resource(SystemVariableCollectionApi, "/apps/<uuid:app_id>/workflows/draft/system-variables")
+api.add_resource(EnvironmentVariableCollectionApi, "/apps/<uuid:app_id>/workflows/draft/environment-variables")

api/controllers/console/auth/login.py

@@ -4,7 +4,7 @@ from flask_restx import Resource, reqparse
 
 import services
 from configs import dify_config
-from constants.languages import get_valid_language
+from constants.languages import languages
 from controllers.console import console_ns
 from controllers.console.auth.error import (
     AuthenticationFailedError,
@@ -29,7 +29,8 @@ from libs.token import (
     clear_access_token_from_cookie,
     clear_csrf_token_from_cookie,
     clear_refresh_token_from_cookie,
-    extract_refresh_token,
+    extract_access_token,
+    extract_csrf_token,
     set_access_token_to_cookie,
     set_csrf_token_to_cookie,
     set_refresh_token_to_cookie,
@@ -205,12 +206,10 @@ class EmailCodeLoginApi(Resource):
             .add_argument("email", type=str, required=True, location="json")
             .add_argument("code", type=str, required=True, location="json")
             .add_argument("token", type=str, required=True, location="json")
-            .add_argument("language", type=str, required=False, location="json")
         )
         args = parser.parse_args()
 
         user_email = args["email"]
-        language = args["language"]
 
         token_data = AccountService.get_email_code_login_data(args["token"])
         if token_data is None:
@@ -244,9 +243,7 @@ class EmailCodeLoginApi(Resource):
         if account is None:
             try:
                 account = AccountService.create_account_and_tenant(
-                    email=user_email,
-                    name=user_email,
-                    interface_language=get_valid_language(language),
+                    email=user_email, name=user_email, interface_language=languages[0]
                 )
             except WorkSpaceNotAllowedCreateError:
                 raise NotAllowedCreateWorkspace()
@@ -271,7 +268,7 @@ class EmailCodeLoginApi(Resource):
 class RefreshTokenApi(Resource):
     def post(self):
         # Get refresh token from cookie instead of request body
-        refresh_token = extract_refresh_token(request)
+        refresh_token = request.cookies.get("refresh_token")
 
         if not refresh_token:
             return {"result": "fail", "message": "No refresh token provided"}, 401
@@ -289,3 +286,13 @@ class RefreshTokenApi(Resource):
             return response
         except Exception as e:
             return {"result": "fail", "message": str(e)}, 401
+
+
+# this api helps frontend to check whether user is authenticated
+# TODO: remove in the future. frontend should redirect to login page by catching 401 status
+@console_ns.route("/login/status")
+class LoginStatus(Resource):
+    def get(self):
+        token = extract_access_token(request)
+        csrf_token = extract_csrf_token(request)
+        return {"logged_in": bool(token) and bool(csrf_token)}

api/controllers/console/explore/workflow.py

@@ -22,7 +22,7 @@ from core.errors.error import (
 from core.model_runtime.errors.invoke import InvokeError
 from core.workflow.graph_engine.manager import GraphEngineManager
 from libs import helper
-from libs.login import current_account_with_tenant
+from libs.login import current_user as current_user_
 from models.model import AppMode, InstalledApp
 from services.app_generate_service import AppGenerateService
 from services.errors.llm import InvokeRateLimitError
@@ -31,6 +31,8 @@ from .. import console_ns
 
 logger = logging.getLogger(__name__)
 
+current_user = current_user_._get_current_object()  # type: ignore
+
 
 @console_ns.route("/installed-apps/<uuid:installed_app_id>/workflows/run")
 class InstalledAppWorkflowRunApi(InstalledAppResource):
@@ -38,7 +40,6 @@ class InstalledAppWorkflowRunApi(InstalledAppResource):
         """
         Run workflow
         """
-        current_user, _ = current_account_with_tenant()
         app_model = installed_app.app
         if not app_model:
             raise NotWorkflowAppError()
@@ -52,6 +53,7 @@ class InstalledAppWorkflowRunApi(InstalledAppResource):
             .add_argument("files", type=list, required=False, location="json")
         )
         args = parser.parse_args()
+        assert current_user is not None
         try:
             response = AppGenerateService.generate(
                 app_model=app_model, user=current_user, args=args, invoke_from=InvokeFrom.EXPLORE, streaming=True
@@ -87,6 +89,7 @@ class InstalledAppWorkflowTaskStopApi(InstalledAppResource):
         app_mode = AppMode.value_of(app_model.mode)
         if app_mode != AppMode.WORKFLOW:
             raise NotWorkflowAppError()
+        assert current_user is not None
 
         # Stop using both mechanisms for backward compatibility
         # Legacy stop flag mechanism (without user check)

api/controllers/console/setup.py

@@ -74,17 +74,12 @@ class SetupApi(Resource):
             .add_argument("email", type=email, required=True, location="json")
             .add_argument("name", type=StrLen(30), required=True, location="json")
             .add_argument("password", type=valid_password, required=True, location="json")
-            .add_argument("language", type=str, required=False, location="json")
         )
         args = parser.parse_args()
 
         # setup
         RegisterService.setup(
-            email=args["email"],
-            name=args["name"],
-            password=args["password"],
-            ip_address=extract_remote_ip(request),
-            language=args["language"],
+            email=args["email"], name=args["name"], password=args["password"], ip_address=extract_remote_ip(request)
         )
 
         return {"result": "success"}, 201

api/controllers/console/workspace/account.py

@@ -32,6 +32,7 @@ from controllers.console.wraps import (
     only_edition_cloud,
     setup_required,
 )
+from core.file import helpers as file_helpers
 from extensions.ext_database import db
 from fields.member_fields import account_fields
 from libs.datetime_utils import naive_utc_now
@@ -128,6 +129,17 @@ class AccountNameApi(Resource):
 
 @console_ns.route("/account/avatar")
 class AccountAvatarApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self):
+        parser = reqparse.RequestParser()
+        parser.add_argument("avatar", type=str, required=True, location="args")
+        args = parser.parse_args()
+
+        avatar_url = file_helpers.get_signed_file_url(args["avatar"])
+        return {"avatar_url": avatar_url}
+    
     @setup_required
     @login_required
     @account_initialization_required

api/controllers/mcp/mcp.py

@@ -193,16 +193,15 @@ class MCPAppApi(Resource):
             except ValidationError as e:
                 raise MCPRequestError(mcp_types.INVALID_PARAMS, f"Invalid MCP request: {str(e)}")
 
-    def _retrieve_end_user(self, tenant_id: str, mcp_server_id: str) -> EndUser | None:
-        """Get end user - manages its own database session"""
-        with Session(db.engine, expire_on_commit=False) as session, session.begin():
-            return (
-                session.query(EndUser)
-                .where(EndUser.tenant_id == tenant_id)
-                .where(EndUser.session_id == mcp_server_id)
-                .where(EndUser.type == "mcp")
-                .first()
-            )
+    def _retrieve_end_user(self, tenant_id: str, mcp_server_id: str, session: Session) -> EndUser | None:
+        """Get end user from existing session - optimized query"""
+        return (
+            session.query(EndUser)
+            .where(EndUser.tenant_id == tenant_id)
+            .where(EndUser.session_id == mcp_server_id)
+            .where(EndUser.type == "mcp")
+            .first()
+        )
 
     def _create_end_user(
         self, client_name: str, tenant_id: str, app_id: str, mcp_server_id: str, session: Session
@@ -230,7 +229,7 @@ class MCPAppApi(Resource):
         request_id: Union[int, str],
     ) -> mcp_types.JSONRPCResponse | mcp_types.JSONRPCError | None:
         """Handle MCP request and return response"""
-        end_user = self._retrieve_end_user(mcp_server.tenant_id, mcp_server.id)
+        end_user = self._retrieve_end_user(mcp_server.tenant_id, mcp_server.id, session)
 
         if not end_user and isinstance(mcp_request.root, mcp_types.InitializeRequest):
             client_info = mcp_request.root.params.clientInfo

api/controllers/web/login.py

@@ -17,8 +17,8 @@ from libs.helper import email
 from libs.passport import PassportService
 from libs.password import valid_password
 from libs.token import (
-    clear_webapp_access_token_from_cookie,
-    extract_webapp_access_token,
+    clear_access_token_from_cookie,
+    extract_access_token,
 )
 from services.account_service import AccountService
 from services.app_service import AppService
@@ -81,7 +81,7 @@ class LoginStatusApi(Resource):
     )
     def get(self):
         app_code = request.args.get("app_code")
-        token = extract_webapp_access_token(request)
+        token = extract_access_token(request)
         if not app_code:
             return {
                 "logged_in": bool(token),
@@ -128,7 +128,7 @@ class LogoutApi(Resource):
         response = make_response({"result": "success"})
         # enterprise SSO sets same site to None in https deployment
         # so we need to logout by calling api
-        clear_webapp_access_token_from_cookie(response, samesite="None")
+        clear_access_token_from_cookie(response, samesite="None")
         return response
 
 

api/controllers/web/passport.py

@@ -12,8 +12,10 @@ from controllers.web import web_ns
 from controllers.web.error import WebAppAuthRequiredError
 from extensions.ext_database import db
 from libs.passport import PassportService
-from libs.token import extract_webapp_access_token
+from libs.token import extract_access_token
 from models.model import App, EndUser, Site
+from services.app_service import AppService
+from services.enterprise.enterprise_service import EnterpriseService
 from services.feature_service import FeatureService
 from services.webapp_auth_service import WebAppAuthService, WebAppAuthType
 
@@ -35,18 +37,23 @@ class PassportResource(Resource):
         system_features = FeatureService.get_system_features()
         app_code = request.headers.get(HEADER_NAME_APP_CODE)
         user_id = request.args.get("user_id")
-        access_token = extract_webapp_access_token(request)
+        access_token = extract_access_token(request)
+
         if app_code is None:
             raise Unauthorized("X-App-Code header is missing.")
+        app_id = AppService.get_app_id_by_code(app_code)
+        # exchange token for enterprise logined web user
+        enterprise_user_decoded = decode_enterprise_webapp_user_id(access_token)
+        if enterprise_user_decoded:
+            # a web user has already logged in, exchange a token for this app without redirecting to the login page
+            return exchange_token_for_existing_web_user(
+                app_code=app_code, enterprise_user_decoded=enterprise_user_decoded
+            )
+
         if system_features.webapp_auth.enabled:
-            enterprise_user_decoded = decode_enterprise_webapp_user_id(access_token)
-            app_auth_type = WebAppAuthService.get_app_auth_type(app_code=app_code)
-            if app_auth_type != WebAppAuthType.PUBLIC:
-                if not enterprise_user_decoded:
-                    raise WebAppAuthRequiredError()
-                return exchange_token_for_existing_web_user(
-                    app_code=app_code, enterprise_user_decoded=enterprise_user_decoded, auth_type=app_auth_type
-                )
+            app_settings = EnterpriseService.WebAppAuth.get_app_access_mode_by_id(app_id=app_id)
+            if not app_settings or not app_settings.access_mode == "public":
+                raise WebAppAuthRequiredError()
 
         # get site from db and check if it is normal
         site = db.session.scalar(select(Site).where(Site.code == app_code, Site.status == "normal"))
@@ -117,7 +124,7 @@ def decode_enterprise_webapp_user_id(jwt_token: str | None):
     return decoded
 
 
-def exchange_token_for_existing_web_user(app_code: str, enterprise_user_decoded: dict, auth_type: WebAppAuthType):
+def exchange_token_for_existing_web_user(app_code: str, enterprise_user_decoded: dict):
     """
     Exchange a token for an existing web user session.
     """
@@ -138,11 +145,13 @@ def exchange_token_for_existing_web_user(app_code: str, enterprise_user_decoded:
     if not app_model or app_model.status != "normal" or not app_model.enable_site:
         raise NotFound()
 
-    if auth_type == WebAppAuthType.PUBLIC:
+    app_auth_type = WebAppAuthService.get_app_auth_type(app_code=app_code)
+
+    if app_auth_type == WebAppAuthType.PUBLIC:
         return _exchange_for_public_app_token(app_model, site, enterprise_user_decoded)
-    elif auth_type == WebAppAuthType.EXTERNAL and user_auth_type != "external":
+    elif app_auth_type == WebAppAuthType.EXTERNAL and user_auth_type != "external":
         raise WebAppAuthRequiredError("Please login as external user.")
-    elif auth_type == WebAppAuthType.INTERNAL and user_auth_type != "internal":
+    elif app_auth_type == WebAppAuthType.INTERNAL and user_auth_type != "internal":
         raise WebAppAuthRequiredError("Please login as internal user.")
 
     end_user = None

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

@@ -211,7 +211,8 @@ class AgentChatAppGenerator(MessageBasedAppGenerator):
             user=user,
             stream=streaming,
         )
-        return AgentChatAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
+        # FIXME: Type hinting issue here, ignore it for now, will fix it later
+        return AgentChatAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)  # type: ignore
 
     def _generate_worker(
         self,

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

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

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

@@ -89,7 +89,7 @@ class WorkflowAppGenerateResponseConverter(AppGenerateResponseConverter):
                 data = cls._error_to_stream_response(sub_stream_response.err)
                 response_chunk.update(data)
             elif isinstance(sub_stream_response, NodeStartStreamResponse | NodeFinishStreamResponse):
-                response_chunk.update(sub_stream_response.to_ignore_detail_dict())
+                response_chunk.update(sub_stream_response.to_ignore_detail_dict())  # ty: ignore [unresolved-attribute]
             else:
                 response_chunk.update(sub_stream_response.model_dump(mode="json"))
             yield response_chunk

api/core/app/features/rate_limiting/rate_limit.py

@@ -98,7 +98,7 @@ class RateLimit:
         else:
             return RateLimitGenerator(
                 rate_limit=self,
-                generator=generator,
+                generator=generator,  # ty: ignore [invalid-argument-type]
                 request_id=request_id,
             )
 

api/core/app/task_pipeline/based_generate_task_pipeline.py

@@ -49,7 +49,7 @@ class BasedGenerateTaskPipeline:
         if isinstance(e, InvokeAuthorizationError):
             err = InvokeAuthorizationError("Incorrect API key provided")
         elif isinstance(e, InvokeError | ValueError):
-            err = e
+            err = e  # ty: ignore [invalid-assignment]
         else:
             description = getattr(e, "description", None)
             err = Exception(description if description is not None else str(e))

api/core/entities/provider_configuration.py

@@ -1868,7 +1868,7 @@ class ProviderConfigurations(BaseModel):
         if "/" not in key:
             key = str(ModelProviderID(key))
 
-        return self.configurations.get(key, default)
+        return self.configurations.get(key, default)  # type: ignore
 
 
 class ProviderModelBundle(BaseModel):

api/core/helper/module_import_helper.py

@@ -20,7 +20,7 @@ def import_module_from_source(*, module_name: str, py_file_path: AnyStr, use_laz
         else:
             # Refer to: https://docs.python.org/3/library/importlib.html#importing-a-source-file-directly
             # FIXME: mypy does not support the type of spec.loader
-            spec = importlib.util.spec_from_file_location(module_name, py_file_path)  # type: ignore[assignment]
+            spec = importlib.util.spec_from_file_location(module_name, py_file_path)  # type: ignore
             if not spec or not spec.loader:
                 raise Exception(f"Failed to load module {module_name} from {py_file_path!r}")
             if use_lazy_loader:

api/core/indexing_runner.py

@@ -415,6 +415,7 @@ class IndexingRunner:
             document_id=dataset_document.id,
             after_indexing_status="splitting",
             extra_update_params={
+                DatasetDocument.word_count: sum(len(text_doc.page_content) for text_doc in text_docs),
                 DatasetDocument.parsing_completed_at: naive_utc_now(),
             },
         )
@@ -754,7 +755,6 @@ class IndexingRunner:
             extra_update_params={
                 DatasetDocument.cleaning_completed_at: cur_time,
                 DatasetDocument.splitting_completed_at: cur_time,
-                DatasetDocument.word_count: sum(len(doc.page_content) for doc in documents),
             },
         )
 

api/core/ops/langfuse_trace/langfuse_trace.py

@@ -2,7 +2,7 @@ import logging
 import os
 from datetime import datetime, timedelta
 
-from langfuse import Langfuse
+from langfuse import Langfuse  # type: ignore
 from sqlalchemy.orm import sessionmaker
 
 from core.ops.base_trace_instance import BaseTraceInstance

api/core/plugin/entities/parameters.py

@@ -76,7 +76,7 @@ class PluginParameter(BaseModel):
     auto_generate: PluginParameterAutoGenerate | None = None
     template: PluginParameterTemplate | None = None
     required: bool = False
-    default: Union[float, int, str, bool] | None = None
+    default: Union[float, int, str] | None = None
     min: Union[float, int] | None = None
     max: Union[float, int] | None = None
     precision: int | None = None

api/core/plugin/impl/base.py

@@ -180,7 +180,7 @@ 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-value]
+        return type_(**response.json())  # type: ignore
 
     def _request_with_plugin_daemon_response(
         self,

api/core/plugin/impl/exc.py

@@ -40,7 +40,7 @@ class PluginDaemonBadRequestError(PluginDaemonClientSideError):
     description: str = "Bad Request"
 
 
-class PluginInvokeError(PluginDaemonClientSideError, ValueError):
+class PluginInvokeError(PluginDaemonClientSideError):
     description: str = "Invoke Error"
 
     def _get_error_object(self) -> Mapping:

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

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

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

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

api/core/rag/retrieval/dataset_retrieval.py

@@ -72,19 +72,6 @@ default_retrieval_model: dict[str, Any] = {
 class DatasetRetrieval:
     def __init__(self, application_generate_entity=None):
         self.application_generate_entity = application_generate_entity
-        self._llm_usage = LLMUsage.empty_usage()
-
-    @property
-    def llm_usage(self) -> LLMUsage:
-        return self._llm_usage.model_copy()
-
-    def _record_usage(self, usage: LLMUsage | None) -> None:
-        if usage is None or usage.total_tokens <= 0:
-            return
-        if self._llm_usage.total_tokens == 0:
-            self._llm_usage = usage
-        else:
-            self._llm_usage = self._llm_usage.plus(usage)
 
     def retrieve(
         self,
@@ -325,18 +312,15 @@ class DatasetRetrieval:
             )
             tools.append(message_tool)
         dataset_id = None
-        router_usage = LLMUsage.empty_usage()
         if planning_strategy == PlanningStrategy.REACT_ROUTER:
             react_multi_dataset_router = ReactMultiDatasetRouter()
-            dataset_id, router_usage = react_multi_dataset_router.invoke(
+            dataset_id = react_multi_dataset_router.invoke(
                 query, tools, model_config, model_instance, user_id, tenant_id
             )
 
         elif planning_strategy == PlanningStrategy.ROUTER:
             function_call_router = FunctionCallMultiDatasetRouter()
-            dataset_id, router_usage = function_call_router.invoke(query, tools, model_config, model_instance)
-
-        self._record_usage(router_usage)
+            dataset_id = function_call_router.invoke(query, tools, model_config, model_instance)
 
         if dataset_id:
             # get retrieval model config
@@ -999,8 +983,7 @@ class DatasetRetrieval:
             )
 
             # handle invoke result
-            result_text, usage = self._handle_invoke_result(invoke_result=invoke_result)
-            self._record_usage(usage)
+            result_text, _ = self._handle_invoke_result(invoke_result=invoke_result)
 
             result_text_json = parse_and_check_json_markdown(result_text, [])
             automatic_metadata_filters = []

api/core/rag/retrieval/router/multi_dataset_function_call_router.py

@@ -2,7 +2,7 @@ from typing import Union
 
 from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
 from core.model_manager import ModelInstance
-from core.model_runtime.entities.llm_entities import LLMResult, LLMUsage
+from core.model_runtime.entities.llm_entities import LLMResult
 from core.model_runtime.entities.message_entities import PromptMessageTool, SystemPromptMessage, UserPromptMessage
 
 
@@ -13,15 +13,15 @@ class FunctionCallMultiDatasetRouter:
         dataset_tools: list[PromptMessageTool],
         model_config: ModelConfigWithCredentialsEntity,
         model_instance: ModelInstance,
-    ) -> tuple[Union[str, None], LLMUsage]:
+    ) -> Union[str, None]:
         """Given input, decided what to do.
         Returns:
             Action specifying what tool to use.
         """
         if len(dataset_tools) == 0:
-            return None, LLMUsage.empty_usage()
+            return None
         elif len(dataset_tools) == 1:
-            return dataset_tools[0].name, LLMUsage.empty_usage()
+            return dataset_tools[0].name
 
         try:
             prompt_messages = [
@@ -34,10 +34,9 @@ class FunctionCallMultiDatasetRouter:
                 stream=False,
                 model_parameters={"temperature": 0.2, "top_p": 0.3, "max_tokens": 1500},
             )
-            usage = result.usage or LLMUsage.empty_usage()
             if result.message.tool_calls:
                 # get retrieval model config
-                return result.message.tool_calls[0].function.name, usage
-            return None, usage
+                return result.message.tool_calls[0].function.name
+            return None
         except Exception:
-            return None, LLMUsage.empty_usage()
+            return None

api/core/rag/retrieval/router/multi_dataset_react_route.py

@@ -58,15 +58,15 @@ class ReactMultiDatasetRouter:
         model_instance: ModelInstance,
         user_id: str,
         tenant_id: str,
-    ) -> tuple[Union[str, None], LLMUsage]:
+    ) -> Union[str, None]:
         """Given input, decided what to do.
         Returns:
             Action specifying what tool to use.
         """
         if len(dataset_tools) == 0:
-            return None, LLMUsage.empty_usage()
+            return None
         elif len(dataset_tools) == 1:
-            return dataset_tools[0].name, LLMUsage.empty_usage()
+            return dataset_tools[0].name
 
         try:
             return self._react_invoke(
@@ -78,7 +78,7 @@ class ReactMultiDatasetRouter:
                 tenant_id=tenant_id,
             )
         except Exception:
-            return None, LLMUsage.empty_usage()
+            return None
 
     def _react_invoke(
         self,
@@ -91,7 +91,7 @@ class ReactMultiDatasetRouter:
         prefix: str = PREFIX,
         suffix: str = SUFFIX,
         format_instructions: str = FORMAT_INSTRUCTIONS,
-    ) -> tuple[Union[str, None], LLMUsage]:
+    ) -> Union[str, None]:
         prompt: Union[list[ChatModelMessage], CompletionModelPromptTemplate]
         if model_config.mode == "chat":
             prompt = self.create_chat_prompt(
@@ -120,7 +120,7 @@ class ReactMultiDatasetRouter:
             memory=None,
             model_config=model_config,
         )
-        result_text, usage = self._invoke_llm(
+        result_text, _ = self._invoke_llm(
             completion_param=model_config.parameters,
             model_instance=model_instance,
             prompt_messages=prompt_messages,
@@ -131,8 +131,8 @@ class ReactMultiDatasetRouter:
         output_parser = StructuredChatOutputParser()
         react_decision = output_parser.parse(result_text)
         if isinstance(react_decision, ReactAction):
-            return react_decision.tool, usage
-        return None, usage
+            return react_decision.tool
+        return None
 
     def _invoke_llm(
         self,

api/core/repositories/celery_workflow_execution_repository.py

@@ -74,7 +74,7 @@ class CeleryWorkflowExecutionRepository(WorkflowExecutionRepository):
         tenant_id = extract_tenant_id(user)
         if not tenant_id:
             raise ValueError("User must have a tenant_id or current_tenant_id")
-        self._tenant_id = tenant_id
+        self._tenant_id = tenant_id  # type: ignore[assignment]  # We've already checked tenant_id is not None
 
         # Store app context
         self._app_id = app_id

api/core/repositories/celery_workflow_node_execution_repository.py

@@ -81,7 +81,7 @@ class CeleryWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository):
         tenant_id = extract_tenant_id(user)
         if not tenant_id:
             raise ValueError("User must have a tenant_id or current_tenant_id")
-        self._tenant_id = tenant_id
+        self._tenant_id = tenant_id  # type: ignore[assignment]  # We've already checked tenant_id is not None
 
         # Store app context
         self._app_id = app_id

api/core/repositories/factory.py

@@ -60,7 +60,7 @@ class DifyCoreRepositoryFactory:
 
         try:
             repository_class = import_string(class_path)
-            return repository_class(
+            return repository_class(  # type: ignore[no-any-return]
                 session_factory=session_factory,
                 user=user,
                 app_id=app_id,
@@ -96,7 +96,7 @@ class DifyCoreRepositoryFactory:
 
         try:
             repository_class = import_string(class_path)
-            return repository_class(
+            return repository_class(  # type: ignore[no-any-return]
                 session_factory=session_factory,
                 user=user,
                 app_id=app_id,

api/core/tools/builtin_tool/provider.py

@@ -157,7 +157,7 @@ class BuiltinToolProviderController(ToolProviderController):
         """
         returns the tool that the provider can provide
         """
-        return next(filter(lambda x: x.entity.identity.name == tool_name, self.get_tools()), None)
+        return next(filter(lambda x: x.entity.identity.name == tool_name, self.get_tools()), None)  # type: ignore
 
     @property
     def need_credentials(self) -> bool:

api/core/tools/builtin_tool/providers/audio/tools/tts.py

@@ -43,7 +43,7 @@ class TTSTool(BuiltinTool):
             content_text=tool_parameters.get("text"),  # type: ignore
             user=user_id,
             tenant_id=self.runtime.tenant_id,
-            voice=voice,
+            voice=voice,  # type: ignore
         )
         buffer = io.BytesIO()
         for chunk in tts:

api/core/tools/builtin_tool/providers/time/tools/localtime_to_timestamp.py

@@ -34,7 +34,6 @@ class LocaltimeToTimestampTool(BuiltinTool):
 
         yield self.create_text_message(f"{timestamp}")
 
-    # TODO: this method's type is messy
     @staticmethod
     def localtime_to_timestamp(localtime: str, time_format: str, local_tz=None) -> int | None:
         try:

api/core/tools/builtin_tool/providers/time/tools/timezone_conversion.py

@@ -48,6 +48,6 @@ class TimezoneConversionTool(BuiltinTool):
             datetime_with_tz = input_timezone.localize(local_time)
             # timezone convert
             converted_datetime = datetime_with_tz.astimezone(output_timezone)
-            return converted_datetime.strftime(time_format)
+            return converted_datetime.strftime(format=time_format)  # type: ignore
         except Exception as e:
             raise ToolInvokeError(str(e))

api/core/tools/mcp_tool/provider.py

@@ -105,7 +105,7 @@ class MCPToolProviderController(ToolProviderController):
         """
         pass
 
-    def get_tool(self, tool_name: str) -> MCPTool:
+    def get_tool(self, tool_name: str) -> MCPTool:  # type: ignore
         """
         return tool with given name
         """
@@ -128,7 +128,7 @@ class MCPToolProviderController(ToolProviderController):
             sse_read_timeout=self.sse_read_timeout,
         )
 
-    def get_tools(self) -> list[MCPTool]:
+    def get_tools(self) -> list[MCPTool]:  # type: ignore
         """
         get all tools
         """

api/core/tools/tool_label_manager.py

@@ -26,7 +26,7 @@ class ToolLabelManager:
         labels = cls.filter_tool_labels(labels)
 
         if isinstance(controller, ApiToolProviderController | WorkflowToolProviderController):
-            provider_id = controller.provider_id
+            provider_id = controller.provider_id  # ty: ignore [unresolved-attribute]
         else:
             raise ValueError("Unsupported tool type")
 
@@ -51,7 +51,7 @@ class ToolLabelManager:
         Get tool labels
         """
         if isinstance(controller, ApiToolProviderController | WorkflowToolProviderController):
-            provider_id = controller.provider_id
+            provider_id = controller.provider_id  # ty: ignore [unresolved-attribute]
         elif isinstance(controller, BuiltinToolProviderController):
             return controller.tool_labels
         else:
@@ -85,7 +85,7 @@ class ToolLabelManager:
         provider_ids = []
         for controller in tool_providers:
             assert isinstance(controller, ApiToolProviderController | WorkflowToolProviderController)
-            provider_ids.append(controller.provider_id)
+            provider_ids.append(controller.provider_id)  # ty: ignore [unresolved-attribute]
 
         labels = db.session.scalars(select(ToolLabelBinding).where(ToolLabelBinding.tool_id.in_(provider_ids))).all()
 

api/core/tools/tool_manager.py

@@ -326,8 +326,7 @@ class ToolManager:
             workflow_provider_stmt = select(WorkflowToolProvider).where(
                 WorkflowToolProvider.tenant_id == tenant_id, WorkflowToolProvider.id == provider_id
             )
-            with Session(db.engine, expire_on_commit=False) as session, session.begin():
-                workflow_provider = session.scalar(workflow_provider_stmt)
+            workflow_provider = db.session.scalar(workflow_provider_stmt)
 
             if workflow_provider is None:
                 raise ToolProviderNotFoundError(f"workflow provider {provider_id} not found")

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

@@ -193,18 +193,18 @@ class DatasetRetrieverTool(DatasetRetrieverBaseTool):
                                 DatasetDocument.enabled == True,
                                 DatasetDocument.archived == False,
                             )
-                            document = db.session.scalar(dataset_document_stmt)
+                            document = db.session.scalar(dataset_document_stmt)  # type: ignore
                             if dataset and document:
                                 source = RetrievalSourceMetadata(
                                     dataset_id=dataset.id,
                                     dataset_name=dataset.name,
-                                    document_id=document.id,
-                                    document_name=document.name,
-                                    data_source_type=document.data_source_type,
+                                    document_id=document.id,  # type: ignore
+                                    document_name=document.name,  # type: ignore
+                                    data_source_type=document.data_source_type,  # type: ignore
                                     segment_id=segment.id,
                                     retriever_from=self.retriever_from,
                                     score=record.score or 0.0,
-                                    doc_metadata=document.doc_metadata,
+                                    doc_metadata=document.doc_metadata,  # type: ignore
                                 )
 
                                 if self.retriever_from == "dev":

api/core/tools/utils/parser.py

@@ -62,11 +62,6 @@ class ApiBasedToolSchemaParser:
                             root = root[ref]
                         interface["operation"]["parameters"][i] = root
                 for parameter in interface["operation"]["parameters"]:
-                    # Handle complex type defaults that are not supported by PluginParameter
-                    default_value = None
-                    if "schema" in parameter and "default" in parameter["schema"]:
-                        default_value = ApiBasedToolSchemaParser._sanitize_default_value(parameter["schema"]["default"])
-
                     tool_parameter = ToolParameter(
                         name=parameter["name"],
                         label=I18nObject(en_US=parameter["name"], zh_Hans=parameter["name"]),
@@ -77,7 +72,9 @@ class ApiBasedToolSchemaParser:
                         required=parameter.get("required", False),
                         form=ToolParameter.ToolParameterForm.LLM,
                         llm_description=parameter.get("description"),
-                        default=default_value,
+                        default=parameter["schema"]["default"]
+                        if "schema" in parameter and "default" in parameter["schema"]
+                        else None,
                         placeholder=I18nObject(
                             en_US=parameter.get("description", ""), zh_Hans=parameter.get("description", "")
                         ),
@@ -137,11 +134,6 @@ class ApiBasedToolSchemaParser:
                             required = body_schema.get("required", [])
                             properties = body_schema.get("properties", {})
                             for name, property in properties.items():
-                                # Handle complex type defaults that are not supported by PluginParameter
-                                default_value = ApiBasedToolSchemaParser._sanitize_default_value(
-                                    property.get("default", None)
-                                )
-
                                 tool = ToolParameter(
                                     name=name,
                                     label=I18nObject(en_US=name, zh_Hans=name),
@@ -152,11 +144,12 @@ class ApiBasedToolSchemaParser:
                                     required=name in required,
                                     form=ToolParameter.ToolParameterForm.LLM,
                                     llm_description=property.get("description", ""),
-                                    default=default_value,
+                                    default=property.get("default", None),
                                     placeholder=I18nObject(
                                         en_US=property.get("description", ""), zh_Hans=property.get("description", "")
                                     ),
                                 )
+
                                 # check if there is a type
                                 typ = ApiBasedToolSchemaParser._get_tool_parameter_type(property)
                                 if typ:
@@ -204,22 +197,6 @@ class ApiBasedToolSchemaParser:
 
         return bundles
 
-    @staticmethod
-    def _sanitize_default_value(value):
-        """
-        Sanitize default values for PluginParameter compatibility.
-        Complex types (list, dict) are converted to None to avoid validation errors.
-
-        Args:
-            value: The default value from OpenAPI schema
-
-        Returns:
-            None for complex types (list, dict), otherwise the original value
-        """
-        if isinstance(value, (list, dict)):
-            return None
-        return value
-
     @staticmethod
     def _get_tool_parameter_type(parameter: dict) -> ToolParameter.ToolParameterType | None:
         parameter = parameter or {}
@@ -240,11 +217,7 @@ class ApiBasedToolSchemaParser:
             return ToolParameter.ToolParameterType.STRING
         elif typ == "array":
             items = parameter.get("items") or parameter.get("schema", {}).get("items")
-            if items and items.get("format") == "binary":
-                return ToolParameter.ToolParameterType.FILES
-            else:
-                # For regular arrays, return ARRAY type instead of None
-                return ToolParameter.ToolParameterType.ARRAY
+            return ToolParameter.ToolParameterType.FILES if items and items.get("format") == "binary" else None
         else:
             return None
 

api/core/tools/utils/web_reader_tool.py

@@ -6,8 +6,8 @@ from typing import Any, cast
 from urllib.parse import unquote
 
 import chardet
-import cloudscraper
-from readabilipy import simple_json_from_html_string
+import cloudscraper  # type: ignore
+from readabilipy import simple_json_from_html_string  # type: ignore
 
 from core.helper import ssrf_proxy
 from core.rag.extractor import extract_processor
@@ -63,8 +63,8 @@ def get_url(url: str, user_agent: str | None = None) -> str:
         response = ssrf_proxy.get(url, headers=headers, follow_redirects=True, timeout=(120, 300))
     elif response.status_code == 403:
         scraper = cloudscraper.create_scraper()
-        scraper.perform_request = ssrf_proxy.make_request
-        response = scraper.get(url, headers=headers, timeout=(120, 300))
+        scraper.perform_request = ssrf_proxy.make_request  # type: ignore
+        response = scraper.get(url, headers=headers, follow_redirects=True, timeout=(120, 300))  # type: ignore
 
     if response.status_code != 200:
         return f"URL returned status code {response.status_code}."

api/core/tools/utils/yaml_utils.py

@@ -3,7 +3,7 @@ from functools import lru_cache
 from pathlib import Path
 from typing import Any
 
-import yaml
+import yaml  # type: ignore
 from yaml import YAMLError
 
 logger = logging.getLogger(__name__)

api/core/tools/workflow_as_tool/provider.py

@@ -1,7 +1,6 @@
 from collections.abc import Mapping
 
 from pydantic import Field
-from sqlalchemy.orm import Session
 
 from core.app.app_config.entities import VariableEntity, VariableEntityType
 from core.app.apps.workflow.app_config_manager import WorkflowAppConfigManager
@@ -21,7 +20,6 @@ from core.tools.entities.tool_entities import (
 from core.tools.utils.workflow_configuration_sync import WorkflowToolConfigurationUtils
 from core.tools.workflow_as_tool.tool import WorkflowTool
 from extensions.ext_database import db
-from models.account import Account
 from models.model import App, AppMode
 from models.tools import WorkflowToolProvider
 from models.workflow import Workflow
@@ -46,34 +44,29 @@ class WorkflowToolProviderController(ToolProviderController):
 
     @classmethod
     def from_db(cls, db_provider: WorkflowToolProvider) -> "WorkflowToolProviderController":
-        with Session(db.engine, expire_on_commit=False) as session, session.begin():
-            provider = session.get(WorkflowToolProvider, db_provider.id) if db_provider.id else None
-            if not provider:
-                raise ValueError("workflow provider not found")
-            app = session.get(App, provider.app_id)
-            if not app:
-                raise ValueError("app not found")
+        app = db_provider.app
 
-            user = session.get(Account, provider.user_id) if provider.user_id else None
+        if not app:
+            raise ValueError("app not found")
 
-            controller = WorkflowToolProviderController(
-                entity=ToolProviderEntity(
-                    identity=ToolProviderIdentity(
-                        author=user.name if user else "",
-                        name=provider.label,
-                        label=I18nObject(en_US=provider.label, zh_Hans=provider.label),
-                        description=I18nObject(en_US=provider.description, zh_Hans=provider.description),
-                        icon=provider.icon,
-                    ),
-                    credentials_schema=[],
-                    plugin_id=None,
+        controller = WorkflowToolProviderController(
+            entity=ToolProviderEntity(
+                identity=ToolProviderIdentity(
+                    author=db_provider.user.name if db_provider.user_id and db_provider.user else "",
+                    name=db_provider.label,
+                    label=I18nObject(en_US=db_provider.label, zh_Hans=db_provider.label),
+                    description=I18nObject(en_US=db_provider.description, zh_Hans=db_provider.description),
+                    icon=db_provider.icon,
                 ),
-                provider_id=provider.id or "",
-            )
+                credentials_schema=[],
+                plugin_id=None,
+            ),
+            provider_id=db_provider.id or "",
+        )
 
-            controller.tools = [
-                controller._get_db_provider_tool(provider, app, session=session, user=user),
-            ]
+        # init tools
+
+        controller.tools = [controller._get_db_provider_tool(db_provider, app)]
 
         return controller
 
@@ -81,14 +74,7 @@ class WorkflowToolProviderController(ToolProviderController):
     def provider_type(self) -> ToolProviderType:
         return ToolProviderType.WORKFLOW
 
-    def _get_db_provider_tool(
-        self,
-        db_provider: WorkflowToolProvider,
-        app: App,
-        *,
-        session: Session,
-        user: Account | None = None,
-    ) -> WorkflowTool:
+    def _get_db_provider_tool(self, db_provider: WorkflowToolProvider, app: App) -> WorkflowTool:
         """
         get db provider tool
         :param db_provider: the db provider
@@ -96,7 +82,7 @@ class WorkflowToolProviderController(ToolProviderController):
         :return: the tool
         """
         workflow: Workflow | None = (
-            session.query(Workflow)
+            db.session.query(Workflow)
             .where(Workflow.app_id == db_provider.app_id, Workflow.version == db_provider.version)
             .first()
         )
@@ -113,7 +99,9 @@ class WorkflowToolProviderController(ToolProviderController):
         variables = WorkflowToolConfigurationUtils.get_workflow_graph_variables(graph)
 
         def fetch_workflow_variable(variable_name: str) -> VariableEntity | None:
-            return next(filter(lambda x: x.variable == variable_name, variables), None)
+            return next(filter(lambda x: x.variable == variable_name, variables), None)  # type: ignore
+
+        user = db_provider.user
 
         workflow_tool_parameters = []
         for parameter in parameters:
@@ -199,25 +187,22 @@ class WorkflowToolProviderController(ToolProviderController):
         if self.tools is not None:
             return self.tools
 
-        with Session(db.engine, expire_on_commit=False) as session, session.begin():
-            db_provider: WorkflowToolProvider | None = (
-                session.query(WorkflowToolProvider)
-                .where(
-                    WorkflowToolProvider.tenant_id == tenant_id,
-                    WorkflowToolProvider.app_id == self.provider_id,
-                )
-                .first()
+        db_providers: WorkflowToolProvider | None = (
+            db.session.query(WorkflowToolProvider)
+            .where(
+                WorkflowToolProvider.tenant_id == tenant_id,
+                WorkflowToolProvider.app_id == self.provider_id,
             )
+            .first()
+        )
 
-            if not db_provider:
-                return []
+        if not db_providers:
+            return []
+        if not db_providers.app:
+            raise ValueError("app not found")
 
-            app = session.get(App, db_provider.app_id)
-            if not app:
-                raise ValueError("app not found")
-
-            user = session.get(Account, db_provider.user_id) if db_provider.user_id else None
-            self.tools = [self._get_db_provider_tool(db_provider, app, session=session, user=user)]
+        app = db_providers.app
+        self.tools = [self._get_db_provider_tool(db_providers, app)]
 
         return self.tools
 

api/core/tools/workflow_as_tool/tool.py

@@ -1,14 +1,12 @@
 import json
 import logging
-from collections.abc import Generator, Mapping, Sequence
-from typing import Any, cast
+from collections.abc import Generator
+from typing import Any
 
 from flask import has_request_context
 from sqlalchemy import select
-from sqlalchemy.orm import Session
 
 from core.file import FILE_MODEL_IDENTITY, File, FileTransferMethod
-from core.model_runtime.entities.llm_entities import LLMUsage, LLMUsageMetadata
 from core.tools.__base.tool import Tool
 from core.tools.__base.tool_runtime import ToolRuntime
 from core.tools.entities.tool_entities import (
@@ -50,7 +48,6 @@ class WorkflowTool(Tool):
         self.workflow_entities = workflow_entities
         self.workflow_call_depth = workflow_call_depth
         self.label = label
-        self._latest_usage = LLMUsage.empty_usage()
 
         super().__init__(entity=entity, runtime=runtime)
 
@@ -86,11 +83,10 @@ class WorkflowTool(Tool):
         assert self.runtime.invoke_from is not None
 
         user = self._resolve_user(user_id=user_id)
+
         if user is None:
             raise ToolInvokeError("User not found")
 
-        self._latest_usage = LLMUsage.empty_usage()
-
         result = generator.generate(
             app_model=app,
             workflow=workflow,
@@ -114,68 +110,9 @@ class WorkflowTool(Tool):
             for file in files:
                 yield self.create_file_message(file)  # type: ignore
 
-        self._latest_usage = self._derive_usage_from_result(data)
-
         yield self.create_text_message(json.dumps(outputs, ensure_ascii=False))
         yield self.create_json_message(outputs)
 
-    @property
-    def latest_usage(self) -> LLMUsage:
-        return self._latest_usage
-
-    @classmethod
-    def _derive_usage_from_result(cls, data: Mapping[str, Any]) -> LLMUsage:
-        usage_dict = cls._extract_usage_dict(data)
-        if usage_dict is not None:
-            return LLMUsage.from_metadata(cast(LLMUsageMetadata, dict(usage_dict)))
-
-        total_tokens = data.get("total_tokens")
-        total_price = data.get("total_price")
-        if total_tokens is None and total_price is None:
-            return LLMUsage.empty_usage()
-
-        usage_metadata: dict[str, Any] = {}
-        if total_tokens is not None:
-            try:
-                usage_metadata["total_tokens"] = int(str(total_tokens))
-            except (TypeError, ValueError):
-                pass
-        if total_price is not None:
-            usage_metadata["total_price"] = str(total_price)
-        currency = data.get("currency")
-        if currency is not None:
-            usage_metadata["currency"] = currency
-
-        if not usage_metadata:
-            return LLMUsage.empty_usage()
-
-        return LLMUsage.from_metadata(cast(LLMUsageMetadata, usage_metadata))
-
-    @classmethod
-    def _extract_usage_dict(cls, payload: Mapping[str, Any]) -> Mapping[str, Any] | None:
-        usage_candidate = payload.get("usage")
-        if isinstance(usage_candidate, Mapping):
-            return usage_candidate
-
-        metadata_candidate = payload.get("metadata")
-        if isinstance(metadata_candidate, Mapping):
-            usage_candidate = metadata_candidate.get("usage")
-            if isinstance(usage_candidate, Mapping):
-                return usage_candidate
-
-        for value in payload.values():
-            if isinstance(value, Mapping):
-                found = cls._extract_usage_dict(value)
-                if found is not None:
-                    return found
-            elif isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
-                for item in value:
-                    if isinstance(item, Mapping):
-                        found = cls._extract_usage_dict(item)
-                        if found is not None:
-                            return found
-        return None
-
     def fork_tool_runtime(self, runtime: ToolRuntime) -> "WorkflowTool":
         """
         fork a new tool with metadata
@@ -242,17 +179,16 @@ class WorkflowTool(Tool):
         """
         get the workflow by app id and version
         """
-        with Session(db.engine, expire_on_commit=False) as session, session.begin():
-            if not version:
-                stmt = (
-                    select(Workflow)
-                    .where(Workflow.app_id == app_id, Workflow.version != Workflow.VERSION_DRAFT)
-                    .order_by(Workflow.created_at.desc())
-                )
-                workflow = session.scalars(stmt).first()
-            else:
-                stmt = select(Workflow).where(Workflow.app_id == app_id, Workflow.version == version)
-                workflow = session.scalar(stmt)
+        if not version:
+            workflow = (
+                db.session.query(Workflow)
+                .where(Workflow.app_id == app_id, Workflow.version != Workflow.VERSION_DRAFT)
+                .order_by(Workflow.created_at.desc())
+                .first()
+            )
+        else:
+            stmt = select(Workflow).where(Workflow.app_id == app_id, Workflow.version == version)
+            workflow = db.session.scalar(stmt)
 
         if not workflow:
             raise ValueError("workflow not found or not published")
@@ -264,8 +200,7 @@ class WorkflowTool(Tool):
         get the app by app id
         """
         stmt = select(App).where(App.id == app_id)
-        with Session(db.engine, expire_on_commit=False) as session, session.begin():
-            app = session.scalar(stmt)
+        app = db.session.scalar(stmt)
         if not app:
             raise ValueError("app not found")
 

api/core/variables/segment_group.py

@@ -4,7 +4,7 @@ from .types import SegmentType
 
 class SegmentGroup(Segment):
     value_type: SegmentType = SegmentType.GROUP
-    value: list[Segment]
+    value: list[Segment] = None  # type: ignore
 
     @property
     def text(self):

api/core/variables/segments.py

@@ -19,7 +19,7 @@ class Segment(BaseModel):
     model_config = ConfigDict(frozen=True)
 
     value_type: SegmentType
-    value: Any
+    value: Any = None
 
     @field_validator("value_type")
     @classmethod
@@ -74,12 +74,12 @@ class NoneSegment(Segment):
 
 class StringSegment(Segment):
     value_type: SegmentType = SegmentType.STRING
-    value: str
+    value: str = None  # type: ignore
 
 
 class FloatSegment(Segment):
     value_type: SegmentType = SegmentType.FLOAT
-    value: float
+    value: float = None  # type: ignore
     # NOTE(QuantumGhost): seems that the equality for FloatSegment with `NaN` value has some problems.
     # The following tests cannot pass.
     #
@@ -98,12 +98,12 @@ class FloatSegment(Segment):
 
 class IntegerSegment(Segment):
     value_type: SegmentType = SegmentType.INTEGER
-    value: int
+    value: int = None  # type: ignore
 
 
 class ObjectSegment(Segment):
     value_type: SegmentType = SegmentType.OBJECT
-    value: Mapping[str, Any]
+    value: Mapping[str, Any] = None  # type: ignore
 
     @property
     def text(self) -> str:
@@ -136,7 +136,7 @@ class ArraySegment(Segment):
 
 class FileSegment(Segment):
     value_type: SegmentType = SegmentType.FILE
-    value: File
+    value: File = None  # type: ignore
 
     @property
     def markdown(self) -> str:
@@ -153,17 +153,17 @@ class FileSegment(Segment):
 
 class BooleanSegment(Segment):
     value_type: SegmentType = SegmentType.BOOLEAN
-    value: bool
+    value: bool = None  # type: ignore
 
 
 class ArrayAnySegment(ArraySegment):
     value_type: SegmentType = SegmentType.ARRAY_ANY
-    value: Sequence[Any]
+    value: Sequence[Any] = None  # type: ignore
 
 
 class ArrayStringSegment(ArraySegment):
     value_type: SegmentType = SegmentType.ARRAY_STRING
-    value: Sequence[str]
+    value: Sequence[str] = None  # type: ignore
 
     @property
     def text(self) -> str:
@@ -175,17 +175,17 @@ class ArrayStringSegment(ArraySegment):
 
 class ArrayNumberSegment(ArraySegment):
     value_type: SegmentType = SegmentType.ARRAY_NUMBER
-    value: Sequence[float | int]
+    value: Sequence[float | int] = None  # type: ignore
 
 
 class ArrayObjectSegment(ArraySegment):
     value_type: SegmentType = SegmentType.ARRAY_OBJECT
-    value: Sequence[Mapping[str, Any]]
+    value: Sequence[Mapping[str, Any]] = None  # type: ignore
 
 
 class ArrayFileSegment(ArraySegment):
     value_type: SegmentType = SegmentType.ARRAY_FILE
-    value: Sequence[File]
+    value: Sequence[File] = None  # type: ignore
 
     @property
     def markdown(self) -> str:
@@ -205,7 +205,7 @@ class ArrayFileSegment(ArraySegment):
 
 class ArrayBooleanSegment(ArraySegment):
     value_type: SegmentType = SegmentType.ARRAY_BOOLEAN
-    value: Sequence[bool]
+    value: Sequence[bool] = None  # type: ignore
 
 
 def get_segment_discriminator(v: Any) -> SegmentType | None:

api/core/workflow/entities/__init__.py

@@ -1,5 +1,3 @@
-from ..runtime.graph_runtime_state import GraphRuntimeState
-from ..runtime.variable_pool import VariablePool
 from .agent import AgentNodeStrategyInit
 from .graph_init_params import GraphInitParams
 from .workflow_execution import WorkflowExecution
@@ -8,8 +6,6 @@ from .workflow_node_execution import WorkflowNodeExecution
 __all__ = [
     "AgentNodeStrategyInit",
     "GraphInitParams",
-    "GraphRuntimeState",
-    "VariablePool",
     "WorkflowExecution",
     "WorkflowNodeExecution",
 ]

api/core/workflow/graph/graph.py

@@ -3,12 +3,11 @@ from collections import defaultdict
 from collections.abc import Mapping, Sequence
 from typing import Protocol, cast, final
 
-from core.workflow.enums import ErrorStrategy, NodeExecutionType, NodeState, NodeType
+from core.workflow.enums import NodeExecutionType, NodeState, NodeType
 from core.workflow.nodes.base.node import Node
 from libs.typing import is_str, is_str_dict
 
 from .edge import Edge
-from .validation import get_graph_validator
 
 logger = logging.getLogger(__name__)
 
@@ -202,17 +201,6 @@ class Graph:
 
         return GraphBuilder(graph_cls=cls)
 
-    @classmethod
-    def _promote_fail_branch_nodes(cls, nodes: dict[str, Node]) -> None:
-        """
-        Promote nodes configured with FAIL_BRANCH error strategy to branch execution type.
-
-        :param nodes: mapping of node ID to node instance
-        """
-        for node in nodes.values():
-            if node.error_strategy == ErrorStrategy.FAIL_BRANCH:
-                node.execution_type = NodeExecutionType.BRANCH
-
     @classmethod
     def _mark_inactive_root_branches(
         cls,
@@ -319,9 +307,6 @@ class Graph:
         # Create node instances
         nodes = cls._create_node_instances(node_configs_map, node_factory)
 
-        # Promote fail-branch nodes to branch execution type at graph level
-        cls._promote_fail_branch_nodes(nodes)
-
         # Get root node instance
         root_node = nodes[root_node_id]
 
@@ -329,7 +314,7 @@ class Graph:
         cls._mark_inactive_root_branches(nodes, edges, in_edges, out_edges, root_node_id)
 
         # Create and return the graph
-        graph = cls(
+        return cls(
             nodes=nodes,
             edges=edges,
             in_edges=in_edges,
@@ -337,11 +322,6 @@ class Graph:
             root_node=root_node,
         )
 
-        # Validate the graph structure using built-in validators
-        get_graph_validator().validate(graph)
-
-        return graph
-
     @property
     def node_ids(self) -> list[str]:
         """

api/core/workflow/graph/validation.py

@@ -1,125 +0,0 @@
-from __future__ import annotations
-
-from collections.abc import Sequence
-from dataclasses import dataclass
-from typing import TYPE_CHECKING, Protocol
-
-from core.workflow.enums import NodeExecutionType, NodeType
-
-if TYPE_CHECKING:
-    from .graph import Graph
-
-
-@dataclass(frozen=True, slots=True)
-class GraphValidationIssue:
-    """Immutable value object describing a single validation issue."""
-
-    code: str
-    message: str
-    node_id: str | None = None
-
-
-class GraphValidationError(ValueError):
-    """Raised when graph validation fails."""
-
-    def __init__(self, issues: Sequence[GraphValidationIssue]) -> None:
-        if not issues:
-            raise ValueError("GraphValidationError requires at least one issue.")
-        self.issues: tuple[GraphValidationIssue, ...] = tuple(issues)
-        message = "; ".join(f"[{issue.code}] {issue.message}" for issue in self.issues)
-        super().__init__(message)
-
-
-class GraphValidationRule(Protocol):
-    """Protocol that individual validation rules must satisfy."""
-
-    def validate(self, graph: Graph) -> Sequence[GraphValidationIssue]:
-        """Validate the provided graph and return any discovered issues."""
-        ...
-
-
-@dataclass(frozen=True, slots=True)
-class _EdgeEndpointValidator:
-    """Ensures all edges reference existing nodes."""
-
-    missing_node_code: str = "MISSING_NODE"
-
-    def validate(self, graph: Graph) -> Sequence[GraphValidationIssue]:
-        issues: list[GraphValidationIssue] = []
-        for edge in graph.edges.values():
-            if edge.tail not in graph.nodes:
-                issues.append(
-                    GraphValidationIssue(
-                        code=self.missing_node_code,
-                        message=f"Edge {edge.id} references unknown source node '{edge.tail}'.",
-                        node_id=edge.tail,
-                    )
-                )
-            if edge.head not in graph.nodes:
-                issues.append(
-                    GraphValidationIssue(
-                        code=self.missing_node_code,
-                        message=f"Edge {edge.id} references unknown target node '{edge.head}'.",
-                        node_id=edge.head,
-                    )
-                )
-        return issues
-
-
-@dataclass(frozen=True, slots=True)
-class _RootNodeValidator:
-    """Validates root node invariants."""
-
-    invalid_root_code: str = "INVALID_ROOT"
-    container_entry_types: tuple[NodeType, ...] = (NodeType.ITERATION_START, NodeType.LOOP_START)
-
-    def validate(self, graph: Graph) -> Sequence[GraphValidationIssue]:
-        root_node = graph.root_node
-        issues: list[GraphValidationIssue] = []
-        if root_node.id not in graph.nodes:
-            issues.append(
-                GraphValidationIssue(
-                    code=self.invalid_root_code,
-                    message=f"Root node '{root_node.id}' is missing from the node registry.",
-                    node_id=root_node.id,
-                )
-            )
-            return issues
-
-        node_type = getattr(root_node, "node_type", None)
-        if root_node.execution_type != NodeExecutionType.ROOT and node_type not in self.container_entry_types:
-            issues.append(
-                GraphValidationIssue(
-                    code=self.invalid_root_code,
-                    message=f"Root node '{root_node.id}' must declare execution type 'root'.",
-                    node_id=root_node.id,
-                )
-            )
-        return issues
-
-
-@dataclass(frozen=True, slots=True)
-class GraphValidator:
-    """Coordinates execution of graph validation rules."""
-
-    rules: tuple[GraphValidationRule, ...]
-
-    def validate(self, graph: Graph) -> None:
-        """Validate the graph against all configured rules."""
-        issues: list[GraphValidationIssue] = []
-        for rule in self.rules:
-            issues.extend(rule.validate(graph))
-
-        if issues:
-            raise GraphValidationError(issues)
-
-
-_DEFAULT_RULES: tuple[GraphValidationRule, ...] = (
-    _EdgeEndpointValidator(),
-    _RootNodeValidator(),
-)
-
-
-def get_graph_validator() -> GraphValidator:
-    """Construct the validator composed of default rules."""
-    return GraphValidator(_DEFAULT_RULES)

api/core/workflow/nodes/agent/entities.py

@@ -26,8 +26,8 @@ class AgentNodeData(BaseNodeData):
 
 
 class ParamsAutoGenerated(IntEnum):
-    CLOSE = 0
-    OPEN = 1
+    CLOSE = auto()
+    OPEN = auto()
 
 
 class AgentOldVersionModelFeatures(StrEnum):

api/core/workflow/nodes/base/__init__.py

@@ -1,5 +1,4 @@
 from .entities import BaseIterationNodeData, BaseIterationState, BaseLoopNodeData, BaseLoopState, BaseNodeData
-from .usage_tracking_mixin import LLMUsageTrackingMixin
 
 __all__ = [
     "BaseIterationNodeData",
@@ -7,5 +6,4 @@ __all__ = [
     "BaseLoopNodeData",
     "BaseLoopState",
     "BaseNodeData",
-    "LLMUsageTrackingMixin",
 ]

api/core/workflow/nodes/base/entities.py

@@ -1,6 +1,5 @@
 import json
 from abc import ABC
-from builtins import type as type_
 from collections.abc import Sequence
 from enum import StrEnum
 from typing import Any, Union
@@ -59,9 +58,10 @@ class DefaultValue(BaseModel):
             raise DefaultValueTypeError(f"Invalid JSON format for value: {value}")
 
     @staticmethod
-    def _validate_array(value: Any, element_type: type_ | tuple[type_, ...]) -> bool:
+    def _validate_array(value: Any, element_type: DefaultValueType) -> bool:
         """Unified array type validation"""
-        return isinstance(value, list) and all(isinstance(x, element_type) for x in value)
+        # FIXME, type ignore here for do not find the reason mypy complain, if find the root cause, please fix it
+        return isinstance(value, list) and all(isinstance(x, element_type) for x in value)  # type: ignore
 
     @staticmethod
     def _convert_number(value: str) -> float:

api/core/workflow/nodes/base/usage_tracking_mixin.py

@@ -1,28 +0,0 @@
-from core.model_runtime.entities.llm_entities import LLMUsage
-from core.workflow.runtime import GraphRuntimeState
-
-
-class LLMUsageTrackingMixin:
-    """Provides shared helpers for merging and recording LLM usage within workflow nodes."""
-
-    graph_runtime_state: GraphRuntimeState
-
-    @staticmethod
-    def _merge_usage(current: LLMUsage, new_usage: LLMUsage | None) -> LLMUsage:
-        """Return a combined usage snapshot, preserving zero-value inputs."""
-        if new_usage is None or new_usage.total_tokens <= 0:
-            return current
-        if current.total_tokens == 0:
-            return new_usage
-        return current.plus(new_usage)
-
-    def _accumulate_usage(self, usage: LLMUsage) -> None:
-        """Push usage into the graph runtime accumulator for downstream reporting."""
-        if usage.total_tokens <= 0:
-            return
-
-        current_usage = self.graph_runtime_state.llm_usage
-        if current_usage.total_tokens == 0:
-            self.graph_runtime_state.llm_usage = usage.model_copy()
-        else:
-            self.graph_runtime_state.llm_usage = current_usage.plus(usage)

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

@@ -10,10 +10,10 @@ from typing import Any
 import chardet
 import docx
 import pandas as pd
-import pypandoc
-import pypdfium2
-import webvtt
-import yaml
+import pypandoc  # type: ignore
+import pypdfium2  # type: ignore
+import webvtt  # type: ignore
+import yaml  # type: ignore
 from docx.document import Document
 from docx.oxml.table import CT_Tbl
 from docx.oxml.text.paragraph import CT_P

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

@@ -8,7 +8,6 @@ from typing import TYPE_CHECKING, Any, NewType, cast
 from flask import Flask, current_app
 from typing_extensions import TypeIs
 
-from core.model_runtime.entities.llm_entities import LLMUsage
 from core.variables import IntegerVariable, NoneSegment
 from core.variables.segments import ArrayAnySegment, ArraySegment
 from core.variables.variables import VariableUnion
@@ -35,7 +34,6 @@ from core.workflow.node_events import (
     NodeRunResult,
     StreamCompletedEvent,
 )
-from core.workflow.nodes.base import LLMUsageTrackingMixin
 from core.workflow.nodes.base.entities import BaseNodeData, RetryConfig
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.iteration.entities import ErrorHandleMode, IterationNodeData
@@ -60,7 +58,7 @@ logger = logging.getLogger(__name__)
 EmptyArraySegment = NewType("EmptyArraySegment", ArraySegment)
 
 
-class IterationNode(LLMUsageTrackingMixin, Node):
+class IterationNode(Node):
     """
     Iteration Node.
     """
@@ -120,7 +118,6 @@ class IterationNode(LLMUsageTrackingMixin, Node):
         started_at = naive_utc_now()
         iter_run_map: dict[str, float] = {}
         outputs: list[object] = []
-        usage_accumulator = [LLMUsage.empty_usage()]
 
         yield IterationStartedEvent(
             start_at=started_at,
@@ -133,27 +130,22 @@ class IterationNode(LLMUsageTrackingMixin, Node):
                 iterator_list_value=iterator_list_value,
                 outputs=outputs,
                 iter_run_map=iter_run_map,
-                usage_accumulator=usage_accumulator,
             )
 
-            self._accumulate_usage(usage_accumulator[0])
             yield from self._handle_iteration_success(
                 started_at=started_at,
                 inputs=inputs,
                 outputs=outputs,
                 iterator_list_value=iterator_list_value,
                 iter_run_map=iter_run_map,
-                usage=usage_accumulator[0],
             )
         except IterationNodeError as e:
-            self._accumulate_usage(usage_accumulator[0])
             yield from self._handle_iteration_failure(
                 started_at=started_at,
                 inputs=inputs,
                 outputs=outputs,
                 iterator_list_value=iterator_list_value,
                 iter_run_map=iter_run_map,
-                usage=usage_accumulator[0],
                 error=e,
             )
 
@@ -204,7 +196,6 @@ class IterationNode(LLMUsageTrackingMixin, Node):
         iterator_list_value: Sequence[object],
         outputs: list[object],
         iter_run_map: dict[str, float],
-        usage_accumulator: list[LLMUsage],
     ) -> Generator[GraphNodeEventBase | NodeEventBase, None, None]:
         if self._node_data.is_parallel:
             # Parallel mode execution
@@ -212,7 +203,6 @@ class IterationNode(LLMUsageTrackingMixin, Node):
                 iterator_list_value=iterator_list_value,
                 outputs=outputs,
                 iter_run_map=iter_run_map,
-                usage_accumulator=usage_accumulator,
             )
         else:
             # Sequential mode execution
@@ -238,9 +228,6 @@ class IterationNode(LLMUsageTrackingMixin, Node):
 
                 # Update the total tokens from this iteration
                 self.graph_runtime_state.total_tokens += graph_engine.graph_runtime_state.total_tokens
-                usage_accumulator[0] = self._merge_usage(
-                    usage_accumulator[0], graph_engine.graph_runtime_state.llm_usage
-                )
                 iter_run_map[str(index)] = (datetime.now(UTC).replace(tzinfo=None) - iter_start_at).total_seconds()
 
     def _execute_parallel_iterations(
@@ -248,7 +235,6 @@ class IterationNode(LLMUsageTrackingMixin, Node):
         iterator_list_value: Sequence[object],
         outputs: list[object],
         iter_run_map: dict[str, float],
-        usage_accumulator: list[LLMUsage],
     ) -> Generator[GraphNodeEventBase | NodeEventBase, None, None]:
         # Initialize outputs list with None values to maintain order
         outputs.extend([None] * len(iterator_list_value))
@@ -259,16 +245,7 @@ class IterationNode(LLMUsageTrackingMixin, Node):
         with ThreadPoolExecutor(max_workers=max_workers) as executor:
             # Submit all iteration tasks
             future_to_index: dict[
-                Future[
-                    tuple[
-                        datetime,
-                        list[GraphNodeEventBase],
-                        object | None,
-                        int,
-                        dict[str, VariableUnion],
-                        LLMUsage,
-                    ]
-                ],
+                Future[tuple[datetime, list[GraphNodeEventBase], object | None, int, dict[str, VariableUnion]]],
                 int,
             ] = {}
             for index, item in enumerate(iterator_list_value):
@@ -287,14 +264,7 @@ class IterationNode(LLMUsageTrackingMixin, Node):
                 index = future_to_index[future]
                 try:
                     result = future.result()
-                    (
-                        iter_start_at,
-                        events,
-                        output_value,
-                        tokens_used,
-                        conversation_snapshot,
-                        iteration_usage,
-                    ) = result
+                    iter_start_at, events, output_value, tokens_used, conversation_snapshot = result
 
                     # Update outputs at the correct index
                     outputs[index] = output_value
@@ -306,8 +276,6 @@ class IterationNode(LLMUsageTrackingMixin, Node):
                     self.graph_runtime_state.total_tokens += tokens_used
                     iter_run_map[str(index)] = (datetime.now(UTC).replace(tzinfo=None) - iter_start_at).total_seconds()
 
-                    usage_accumulator[0] = self._merge_usage(usage_accumulator[0], iteration_usage)
-
                     # Sync conversation variables after iteration completion
                     self._sync_conversation_variables_from_snapshot(conversation_snapshot)
 
@@ -335,7 +303,7 @@ class IterationNode(LLMUsageTrackingMixin, Node):
         item: object,
         flask_app: Flask,
         context_vars: contextvars.Context,
-    ) -> tuple[datetime, list[GraphNodeEventBase], object | None, int, dict[str, VariableUnion], LLMUsage]:
+    ) -> tuple[datetime, list[GraphNodeEventBase], object | None, int, dict[str, VariableUnion]]:
         """Execute a single iteration in parallel mode and return results."""
         with preserve_flask_contexts(flask_app=flask_app, context_vars=context_vars):
             iter_start_at = datetime.now(UTC).replace(tzinfo=None)
@@ -364,7 +332,6 @@ class IterationNode(LLMUsageTrackingMixin, Node):
                 output_value,
                 graph_engine.graph_runtime_state.total_tokens,
                 conversation_snapshot,
-                graph_engine.graph_runtime_state.llm_usage,
             )
 
     def _handle_iteration_success(
@@ -374,8 +341,6 @@ class IterationNode(LLMUsageTrackingMixin, Node):
         outputs: list[object],
         iterator_list_value: Sequence[object],
         iter_run_map: dict[str, float],
-        *,
-        usage: LLMUsage,
     ) -> Generator[NodeEventBase, None, None]:
         # Flatten the list of lists if all outputs are lists
         flattened_outputs = self._flatten_outputs_if_needed(outputs)
@@ -386,9 +351,7 @@ class IterationNode(LLMUsageTrackingMixin, Node):
             outputs={"output": flattened_outputs},
             steps=len(iterator_list_value),
             metadata={
-                WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS: usage.total_tokens,
-                WorkflowNodeExecutionMetadataKey.TOTAL_PRICE: usage.total_price,
-                WorkflowNodeExecutionMetadataKey.CURRENCY: usage.currency,
+                WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS: self.graph_runtime_state.total_tokens,
                 WorkflowNodeExecutionMetadataKey.ITERATION_DURATION_MAP: iter_run_map,
             },
         )
@@ -399,11 +362,8 @@ class IterationNode(LLMUsageTrackingMixin, Node):
                 status=WorkflowNodeExecutionStatus.SUCCEEDED,
                 outputs={"output": flattened_outputs},
                 metadata={
-                    WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS: usage.total_tokens,
-                    WorkflowNodeExecutionMetadataKey.TOTAL_PRICE: usage.total_price,
-                    WorkflowNodeExecutionMetadataKey.CURRENCY: usage.currency,
+                    WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS: self.graph_runtime_state.total_tokens,
                 },
-                llm_usage=usage,
             )
         )
 
@@ -440,8 +400,6 @@ class IterationNode(LLMUsageTrackingMixin, Node):
         outputs: list[object],
         iterator_list_value: Sequence[object],
         iter_run_map: dict[str, float],
-        *,
-        usage: LLMUsage,
         error: IterationNodeError,
     ) -> Generator[NodeEventBase, None, None]:
         # Flatten the list of lists if all outputs are lists (even in failure case)
@@ -453,9 +411,7 @@ class IterationNode(LLMUsageTrackingMixin, Node):
             outputs={"output": flattened_outputs},
             steps=len(iterator_list_value),
             metadata={
-                WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS: usage.total_tokens,
-                WorkflowNodeExecutionMetadataKey.TOTAL_PRICE: usage.total_price,
-                WorkflowNodeExecutionMetadataKey.CURRENCY: usage.currency,
+                WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS: self.graph_runtime_state.total_tokens,
                 WorkflowNodeExecutionMetadataKey.ITERATION_DURATION_MAP: iter_run_map,
             },
             error=str(error),
@@ -464,12 +420,6 @@ class IterationNode(LLMUsageTrackingMixin, Node):
             node_run_result=NodeRunResult(
                 status=WorkflowNodeExecutionStatus.FAILED,
                 error=str(error),
-                metadata={
-                    WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS: usage.total_tokens,
-                    WorkflowNodeExecutionMetadataKey.TOTAL_PRICE: usage.total_price,
-                    WorkflowNodeExecutionMetadataKey.CURRENCY: usage.currency,
-                },
-                llm_usage=usage,
             )
         )
 

api/core/workflow/nodes/knowledge_retrieval/knowledge_retrieval_node.py

@@ -15,11 +15,14 @@ from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEnti
 from core.entities.agent_entities import PlanningStrategy
 from core.entities.model_entities import ModelStatus
 from core.model_manager import ModelInstance, ModelManager
-from core.model_runtime.entities.llm_entities import LLMUsage
-from core.model_runtime.entities.message_entities import PromptMessageRole
-from core.model_runtime.entities.model_entities import ModelFeature, ModelType
+from core.model_runtime.entities.message_entities import (
+    PromptMessageRole,
+)
+from core.model_runtime.entities.model_entities import (
+    ModelFeature,
+    ModelType,
+)
 from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
-from core.model_runtime.utils.encoders import jsonable_encoder
 from core.prompt.simple_prompt_transform import ModelMode
 from core.rag.datasource.retrieval_service import RetrievalService
 from core.rag.entities.metadata_entities import Condition, MetadataCondition
@@ -30,14 +33,8 @@ from core.variables import (
 )
 from core.variables.segments import ArrayObjectSegment
 from core.workflow.entities import GraphInitParams
-from core.workflow.enums import (
-    ErrorStrategy,
-    NodeType,
-    WorkflowNodeExecutionMetadataKey,
-    WorkflowNodeExecutionStatus,
-)
+from core.workflow.enums import ErrorStrategy, NodeType, WorkflowNodeExecutionStatus
 from core.workflow.node_events import ModelInvokeCompletedEvent, NodeRunResult
-from core.workflow.nodes.base import LLMUsageTrackingMixin
 from core.workflow.nodes.base.entities import BaseNodeData, RetryConfig
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.knowledge_retrieval.template_prompts import (
@@ -83,7 +80,7 @@ default_retrieval_model = {
 }
 
 
-class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node):
+class KnowledgeRetrievalNode(Node):
     node_type = NodeType.KNOWLEDGE_RETRIEVAL
 
     _node_data: KnowledgeRetrievalNodeData
@@ -144,7 +141,7 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node):
     def version(cls):
         return "1"
 
-    def _run(self) -> NodeRunResult:
+    def _run(self) -> NodeRunResult:  # type: ignore
         # extract variables
         variable = self.graph_runtime_state.variable_pool.get(self._node_data.query_variable_selector)
         if not isinstance(variable, StringSegment):
@@ -185,21 +182,14 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node):
                 )
 
         # retrieve knowledge
-        usage = LLMUsage.empty_usage()
         try:
-            results, usage = self._fetch_dataset_retriever(node_data=self._node_data, query=query)
+            results = self._fetch_dataset_retriever(node_data=self._node_data, query=query)
             outputs = {"result": ArrayObjectSegment(value=results)}
             return NodeRunResult(
                 status=WorkflowNodeExecutionStatus.SUCCEEDED,
                 inputs=variables,
-                process_data={"usage": jsonable_encoder(usage)},
+                process_data={},
                 outputs=outputs,  # type: ignore
-                metadata={
-                    WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS: usage.total_tokens,
-                    WorkflowNodeExecutionMetadataKey.TOTAL_PRICE: usage.total_price,
-                    WorkflowNodeExecutionMetadataKey.CURRENCY: usage.currency,
-                },
-                llm_usage=usage,
             )
 
         except KnowledgeRetrievalNodeError as e:
@@ -209,7 +199,6 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node):
                 inputs=variables,
                 error=str(e),
                 error_type=type(e).__name__,
-                llm_usage=usage,
             )
         # Temporary handle all exceptions from DatasetRetrieval class here.
         except Exception as e:
@@ -218,15 +207,11 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node):
                 inputs=variables,
                 error=str(e),
                 error_type=type(e).__name__,
-                llm_usage=usage,
             )
         finally:
             db.session.close()
 
-    def _fetch_dataset_retriever(
-        self, node_data: KnowledgeRetrievalNodeData, query: str
-    ) -> tuple[list[dict[str, Any]], LLMUsage]:
-        usage = LLMUsage.empty_usage()
+    def _fetch_dataset_retriever(self, node_data: KnowledgeRetrievalNodeData, query: str) -> list[dict[str, Any]]:
         available_datasets = []
         dataset_ids = node_data.dataset_ids
 
@@ -260,10 +245,9 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node):
             if not dataset:
                 continue
             available_datasets.append(dataset)
-        metadata_filter_document_ids, metadata_condition, metadata_usage = self._get_metadata_filter_condition(
+        metadata_filter_document_ids, metadata_condition = self._get_metadata_filter_condition(
             [dataset.id for dataset in available_datasets], query, node_data
         )
-        usage = self._merge_usage(usage, metadata_usage)
         all_documents = []
         dataset_retrieval = DatasetRetrieval()
         if node_data.retrieval_mode == DatasetRetrieveConfigEntity.RetrieveStrategy.SINGLE:
@@ -346,8 +330,6 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node):
                 metadata_filter_document_ids=metadata_filter_document_ids,
                 metadata_condition=metadata_condition,
             )
-        usage = self._merge_usage(usage, dataset_retrieval.llm_usage)
-
         dify_documents = [item for item in all_documents if item.provider == "dify"]
         external_documents = [item for item in all_documents if item.provider == "external"]
         retrieval_resource_list = []
@@ -424,12 +406,11 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node):
             )
             for position, item in enumerate(retrieval_resource_list, start=1):
                 item["metadata"]["position"] = position
-        return retrieval_resource_list, usage
+        return retrieval_resource_list
 
     def _get_metadata_filter_condition(
         self, dataset_ids: list, query: str, node_data: KnowledgeRetrievalNodeData
-    ) -> tuple[dict[str, list[str]] | None, MetadataCondition | None, LLMUsage]:
-        usage = LLMUsage.empty_usage()
+    ) -> tuple[dict[str, list[str]] | None, MetadataCondition | None]:
         document_query = db.session.query(Document).where(
             Document.dataset_id.in_(dataset_ids),
             Document.indexing_status == "completed",
@@ -439,12 +420,9 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node):
         filters: list[Any] = []
         metadata_condition = None
         if node_data.metadata_filtering_mode == "disabled":
-            return None, None, usage
+            return None, None
         elif node_data.metadata_filtering_mode == "automatic":
-            automatic_metadata_filters, automatic_usage = self._automatic_metadata_filter_func(
-                dataset_ids, query, node_data
-            )
-            usage = self._merge_usage(usage, automatic_usage)
+            automatic_metadata_filters = self._automatic_metadata_filter_func(dataset_ids, query, node_data)
             if automatic_metadata_filters:
                 conditions = []
                 for sequence, filter in enumerate(automatic_metadata_filters):
@@ -465,7 +443,7 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node):
                 metadata_condition = MetadataCondition(
                     logical_operator=node_data.metadata_filtering_conditions.logical_operator
                     if node_data.metadata_filtering_conditions
-                    else "or",
+                    else "or",  # type: ignore
                     conditions=conditions,
                 )
         elif node_data.metadata_filtering_mode == "manual":
@@ -479,10 +457,10 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node):
                             expected_value = self.graph_runtime_state.variable_pool.convert_template(
                                 expected_value
                             ).value[0]
-                            if expected_value.value_type in {"number", "integer", "float"}:
-                                expected_value = expected_value.value
-                            elif expected_value.value_type == "string":
-                                expected_value = re.sub(r"[\r\n\t]+", " ", expected_value.text).strip()
+                            if expected_value.value_type in {"number", "integer", "float"}:  # type: ignore
+                                expected_value = expected_value.value  # type: ignore
+                            elif expected_value.value_type == "string":  # type: ignore
+                                expected_value = re.sub(r"[\r\n\t]+", " ", expected_value.text).strip()  # type: ignore
                             else:
                                 raise ValueError("Invalid expected metadata value type")
                     conditions.append(
@@ -509,7 +487,7 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node):
             if (
                 node_data.metadata_filtering_conditions
                 and node_data.metadata_filtering_conditions.logical_operator == "and"
-            ):
+            ):  # type: ignore
                 document_query = document_query.where(and_(*filters))
             else:
                 document_query = document_query.where(or_(*filters))
@@ -518,12 +496,11 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node):
         metadata_filter_document_ids = defaultdict(list) if documents else None  # type: ignore
         for document in documents:
             metadata_filter_document_ids[document.dataset_id].append(document.id)  # type: ignore
-        return metadata_filter_document_ids, metadata_condition, usage
+        return metadata_filter_document_ids, metadata_condition
 
     def _automatic_metadata_filter_func(
         self, dataset_ids: list, query: str, node_data: KnowledgeRetrievalNodeData
-    ) -> tuple[list[dict[str, Any]], LLMUsage]:
-        usage = LLMUsage.empty_usage()
+    ) -> list[dict[str, Any]]:
         # get all metadata field
         stmt = select(DatasetMetadata).where(DatasetMetadata.dataset_id.in_(dataset_ids))
         metadata_fields = db.session.scalars(stmt).all()
@@ -571,7 +548,6 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node):
             for event in generator:
                 if isinstance(event, ModelInvokeCompletedEvent):
                     result_text = event.text
-                    usage = self._merge_usage(usage, event.usage)
                     break
 
             result_text_json = parse_and_check_json_markdown(result_text, [])
@@ -588,8 +564,8 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node):
                             }
                         )
         except Exception:
-            return [], usage
-        return automatic_metadata_filters, usage
+            return []
+        return automatic_metadata_filters
 
     def _process_metadata_filter_func(
         self, sequence: int, condition: str, metadata_name: str, value: Any, filters: list[Any]

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

@@ -5,7 +5,6 @@ from collections.abc import Callable, Generator, Mapping, Sequence
 from datetime import datetime
 from typing import TYPE_CHECKING, Any, Literal, cast
 
-from core.model_runtime.entities.llm_entities import LLMUsage
 from core.variables import Segment, SegmentType
 from core.workflow.enums import (
     ErrorStrategy,
@@ -28,7 +27,6 @@ from core.workflow.node_events import (
     NodeRunResult,
     StreamCompletedEvent,
 )
-from core.workflow.nodes.base import LLMUsageTrackingMixin
 from core.workflow.nodes.base.entities import BaseNodeData, RetryConfig
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.loop.entities import LoopNodeData, LoopVariableData
@@ -42,7 +40,7 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
-class LoopNode(LLMUsageTrackingMixin, Node):
+class LoopNode(Node):
     """
     Loop Node.
     """
@@ -110,7 +108,7 @@ class LoopNode(LLMUsageTrackingMixin, Node):
                     raise ValueError(f"Invalid value for loop variable {loop_variable.label}")
                 variable_selector = [self._node_id, loop_variable.label]
                 variable = segment_to_variable(segment=processed_segment, selector=variable_selector)
-                self.graph_runtime_state.variable_pool.add(variable_selector, variable.value)
+                self.graph_runtime_state.variable_pool.add(variable_selector, variable)
                 loop_variable_selectors[loop_variable.label] = variable_selector
                 inputs[loop_variable.label] = processed_segment.value
 
@@ -119,7 +117,6 @@ class LoopNode(LLMUsageTrackingMixin, Node):
 
         loop_duration_map: dict[str, float] = {}
         single_loop_variable_map: dict[str, dict[str, Any]] = {}  # single loop variable output
-        loop_usage = LLMUsage.empty_usage()
 
         # Start Loop event
         yield LoopStartedEvent(
@@ -166,9 +163,6 @@ class LoopNode(LLMUsageTrackingMixin, Node):
                 # Update the total tokens from this iteration
                 cost_tokens += graph_engine.graph_runtime_state.total_tokens
 
-                # Accumulate usage from the sub-graph execution
-                loop_usage = self._merge_usage(loop_usage, graph_engine.graph_runtime_state.llm_usage)
-
                 # Collect loop variable values after iteration
                 single_loop_variable = {}
                 for key, selector in loop_variable_selectors.items():
@@ -195,7 +189,6 @@ class LoopNode(LLMUsageTrackingMixin, Node):
                 )
 
             self.graph_runtime_state.total_tokens += cost_tokens
-            self._accumulate_usage(loop_usage)
             # Loop completed successfully
             yield LoopSucceededEvent(
                 start_at=start_at,
@@ -203,9 +196,7 @@ class LoopNode(LLMUsageTrackingMixin, Node):
                 outputs=self._node_data.outputs,
                 steps=loop_count,
                 metadata={
-                    WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS: loop_usage.total_tokens,
-                    WorkflowNodeExecutionMetadataKey.TOTAL_PRICE: loop_usage.total_price,
-                    WorkflowNodeExecutionMetadataKey.CURRENCY: loop_usage.currency,
+                    WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS: cost_tokens,
                     "completed_reason": "loop_break" if reach_break_condition else "loop_completed",
                     WorkflowNodeExecutionMetadataKey.LOOP_DURATION_MAP: loop_duration_map,
                     WorkflowNodeExecutionMetadataKey.LOOP_VARIABLE_MAP: single_loop_variable_map,
@@ -216,28 +207,22 @@ class LoopNode(LLMUsageTrackingMixin, Node):
                 node_run_result=NodeRunResult(
                     status=WorkflowNodeExecutionStatus.SUCCEEDED,
                     metadata={
-                        WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS: loop_usage.total_tokens,
-                        WorkflowNodeExecutionMetadataKey.TOTAL_PRICE: loop_usage.total_price,
-                        WorkflowNodeExecutionMetadataKey.CURRENCY: loop_usage.currency,
+                        WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS: self.graph_runtime_state.total_tokens,
                         WorkflowNodeExecutionMetadataKey.LOOP_DURATION_MAP: loop_duration_map,
                         WorkflowNodeExecutionMetadataKey.LOOP_VARIABLE_MAP: single_loop_variable_map,
                     },
                     outputs=self._node_data.outputs,
                     inputs=inputs,
-                    llm_usage=loop_usage,
                 )
             )
 
         except Exception as e:
-            self._accumulate_usage(loop_usage)
             yield LoopFailedEvent(
                 start_at=start_at,
                 inputs=inputs,
                 steps=loop_count,
                 metadata={
-                    WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS: loop_usage.total_tokens,
-                    WorkflowNodeExecutionMetadataKey.TOTAL_PRICE: loop_usage.total_price,
-                    WorkflowNodeExecutionMetadataKey.CURRENCY: loop_usage.currency,
+                    WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS: self.graph_runtime_state.total_tokens,
                     "completed_reason": "error",
                     WorkflowNodeExecutionMetadataKey.LOOP_DURATION_MAP: loop_duration_map,
                     WorkflowNodeExecutionMetadataKey.LOOP_VARIABLE_MAP: single_loop_variable_map,
@@ -250,13 +235,10 @@ class LoopNode(LLMUsageTrackingMixin, Node):
                     status=WorkflowNodeExecutionStatus.FAILED,
                     error=str(e),
                     metadata={
-                        WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS: loop_usage.total_tokens,
-                        WorkflowNodeExecutionMetadataKey.TOTAL_PRICE: loop_usage.total_price,
-                        WorkflowNodeExecutionMetadataKey.CURRENCY: loop_usage.currency,
+                        WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS: self.graph_runtime_state.total_tokens,
                         WorkflowNodeExecutionMetadataKey.LOOP_DURATION_MAP: loop_duration_map,
                         WorkflowNodeExecutionMetadataKey.LOOP_VARIABLE_MAP: single_loop_variable_map,
                     },
-                    llm_usage=loop_usage,
                 )
             )
 

api/core/workflow/nodes/node_factory.py

@@ -2,7 +2,7 @@ from typing import TYPE_CHECKING, final
 
 from typing_extensions import override
 
-from core.workflow.enums import NodeType
+from core.workflow.enums import ErrorStrategy, NodeExecutionType, NodeType
 from core.workflow.graph import NodeFactory
 from core.workflow.nodes.base.node import Node
 from libs.typing import is_str, is_str_dict
@@ -82,4 +82,8 @@ class DifyNodeFactory(NodeFactory):
             raise ValueError(f"Node {node_id} missing data information")
         node_instance.init_node_data(node_data)
 
+        # If node has fail branch, change execution type to branch
+        if node_instance.error_strategy == ErrorStrategy.FAIL_BRANCH:
+            node_instance.execution_type = NodeExecutionType.BRANCH
+
         return node_instance

api/core/workflow/nodes/question_classifier/question_classifier_node.py

@@ -193,19 +193,15 @@ class QuestionClassifierNode(Node):
                     finish_reason = event.finish_reason
                     break
 
-            rendered_classes = [
-                c.model_copy(update={"name": variable_pool.convert_template(c.name).text}) for c in node_data.classes
-            ]
-
-            category_name = rendered_classes[0].name
-            category_id = rendered_classes[0].id
+            category_name = node_data.classes[0].name
+            category_id = node_data.classes[0].id
             if "<think>" in result_text:
                 result_text = re.sub(r"<think[^>]*>[\s\S]*?</think>", "", result_text, flags=re.IGNORECASE)
             result_text_json = parse_and_check_json_markdown(result_text, [])
             # result_text_json = json.loads(result_text.strip('```JSON\n'))
             if "category_name" in result_text_json and "category_id" in result_text_json:
                 category_id_result = result_text_json["category_id"]
-                classes = rendered_classes
+                classes = node_data.classes
                 classes_map = {class_.id: class_.name for class_ in classes}
                 category_ids = [_class.id for _class in classes]
                 if category_id_result in category_ids:

api/core/workflow/nodes/tool/tool_node.py

@@ -6,13 +6,10 @@ from sqlalchemy.orm import Session
 
 from core.callback_handler.workflow_tool_callback_handler import DifyWorkflowCallbackHandler
 from core.file import File, FileTransferMethod
-from core.model_runtime.entities.llm_entities import LLMUsage
-from core.tools.__base.tool import Tool
 from core.tools.entities.tool_entities import ToolInvokeMessage, ToolParameter
 from core.tools.errors import ToolInvokeError
 from core.tools.tool_engine import ToolEngine
 from core.tools.utils.message_transformer import ToolFileMessageTransformer
-from core.tools.workflow_as_tool.tool import WorkflowTool
 from core.variables.segments import ArrayAnySegment, ArrayFileSegment
 from core.variables.variables import ArrayAnyVariable
 from core.workflow.enums import (
@@ -139,14 +136,13 @@ class ToolNode(Node):
 
         try:
             # convert tool messages
-            _ = yield from self._transform_message(
+            yield from self._transform_message(
                 messages=message_stream,
                 tool_info=tool_info,
                 parameters_for_log=parameters_for_log,
                 user_id=self.user_id,
                 tenant_id=self.tenant_id,
                 node_id=self._node_id,
-                tool_runtime=tool_runtime,
             )
         except ToolInvokeError as e:
             yield StreamCompletedEvent(
@@ -240,8 +236,7 @@ class ToolNode(Node):
         user_id: str,
         tenant_id: str,
         node_id: str,
-        tool_runtime: Tool,
-    ) -> Generator[NodeEventBase, None, LLMUsage]:
+    ) -> Generator:
         """
         Convert ToolInvokeMessages into tuple[plain_text, files]
         """
@@ -429,34 +424,17 @@ class ToolNode(Node):
                 is_final=True,
             )
 
-        usage = self._extract_tool_usage(tool_runtime)
-
-        metadata: dict[WorkflowNodeExecutionMetadataKey, Any] = {
-            WorkflowNodeExecutionMetadataKey.TOOL_INFO: tool_info,
-        }
-        if usage.total_tokens > 0:
-            metadata[WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS] = usage.total_tokens
-            metadata[WorkflowNodeExecutionMetadataKey.TOTAL_PRICE] = usage.total_price
-            metadata[WorkflowNodeExecutionMetadataKey.CURRENCY] = usage.currency
-
         yield StreamCompletedEvent(
             node_run_result=NodeRunResult(
                 status=WorkflowNodeExecutionStatus.SUCCEEDED,
                 outputs={"text": text, "files": ArrayFileSegment(value=files), "json": json_output, **variables},
-                metadata=metadata,
+                metadata={
+                    WorkflowNodeExecutionMetadataKey.TOOL_INFO: tool_info,
+                },
                 inputs=parameters_for_log,
-                llm_usage=usage,
             )
         )
 
-        return usage
-
-    @staticmethod
-    def _extract_tool_usage(tool_runtime: Tool) -> LLMUsage:
-        if isinstance(tool_runtime, WorkflowTool):
-            return tool_runtime.latest_usage
-        return LLMUsage.empty_usage()
-
     @classmethod
     def _extract_variable_selector_to_variable_mapping(
         cls,

api/core/workflow/runtime/graph_runtime_state.py

@@ -5,7 +5,6 @@ import json
 from collections.abc import Mapping, Sequence
 from collections.abc import Mapping as TypingMapping
 from copy import deepcopy
-from dataclasses import dataclass
 from typing import Any, Protocol
 
 from pydantic.json import pydantic_encoder
@@ -107,23 +106,6 @@ class GraphProtocol(Protocol):
     def get_outgoing_edges(self, node_id: str) -> Sequence[object]: ...
 
 
-@dataclass(slots=True)
-class _GraphRuntimeStateSnapshot:
-    """Immutable view of a serialized runtime state snapshot."""
-
-    start_at: float
-    total_tokens: int
-    node_run_steps: int
-    llm_usage: LLMUsage
-    outputs: dict[str, Any]
-    variable_pool: VariablePool
-    has_variable_pool: bool
-    ready_queue_dump: str | None
-    graph_execution_dump: str | None
-    response_coordinator_dump: str | None
-    paused_nodes: tuple[str, ...]
-
-
 class GraphRuntimeState:
     """Mutable runtime state shared across graph execution components."""
 
@@ -311,28 +293,69 @@ class GraphRuntimeState:
 
         return json.dumps(snapshot, default=pydantic_encoder)
 
-    @classmethod
-    def from_snapshot(cls, data: str | Mapping[str, Any]) -> GraphRuntimeState:
+    def loads(self, data: str | Mapping[str, Any]) -> None:
         """Restore runtime state from a serialized snapshot."""
 
-        snapshot = cls._parse_snapshot_payload(data)
+        payload: dict[str, Any]
+        if isinstance(data, str):
+            payload = json.loads(data)
+        else:
+            payload = dict(data)
 
-        state = cls(
-            variable_pool=snapshot.variable_pool,
-            start_at=snapshot.start_at,
-            total_tokens=snapshot.total_tokens,
-            llm_usage=snapshot.llm_usage,
-            outputs=snapshot.outputs,
-            node_run_steps=snapshot.node_run_steps,
-        )
-        state._apply_snapshot(snapshot)
-        return state
+        version = payload.get("version")
+        if version != "1.0":
+            raise ValueError(f"Unsupported GraphRuntimeState snapshot version: {version}")
 
-    def loads(self, data: str | Mapping[str, Any]) -> None:
-        """Restore runtime state from a serialized snapshot (legacy API)."""
+        self._start_at = float(payload.get("start_at", 0.0))
+        total_tokens = int(payload.get("total_tokens", 0))
+        if total_tokens < 0:
+            raise ValueError("total_tokens must be non-negative")
+        self._total_tokens = total_tokens
 
-        snapshot = self._parse_snapshot_payload(data)
-        self._apply_snapshot(snapshot)
+        node_run_steps = int(payload.get("node_run_steps", 0))
+        if node_run_steps < 0:
+            raise ValueError("node_run_steps must be non-negative")
+        self._node_run_steps = node_run_steps
+
+        llm_usage_payload = payload.get("llm_usage", {})
+        self._llm_usage = LLMUsage.model_validate(llm_usage_payload)
+
+        self._outputs = deepcopy(payload.get("outputs", {}))
+
+        variable_pool_payload = payload.get("variable_pool")
+        if variable_pool_payload is not None:
+            self._variable_pool = VariablePool.model_validate(variable_pool_payload)
+
+        ready_queue_payload = payload.get("ready_queue")
+        if ready_queue_payload is not None:
+            self._ready_queue = self._build_ready_queue()
+            self._ready_queue.loads(ready_queue_payload)
+        else:
+            self._ready_queue = None
+
+        graph_execution_payload = payload.get("graph_execution")
+        self._graph_execution = None
+        self._pending_graph_execution_workflow_id = None
+        if graph_execution_payload is not None:
+            try:
+                execution_payload = json.loads(graph_execution_payload)
+                self._pending_graph_execution_workflow_id = execution_payload.get("workflow_id")
+            except (json.JSONDecodeError, TypeError, AttributeError):
+                self._pending_graph_execution_workflow_id = None
+            self.graph_execution.loads(graph_execution_payload)
+
+        response_payload = payload.get("response_coordinator")
+        if response_payload is not None:
+            if self._graph is not None:
+                self.response_coordinator.loads(response_payload)
+            else:
+                self._pending_response_coordinator_dump = response_payload
+        else:
+            self._pending_response_coordinator_dump = None
+            self._response_coordinator = None
+
+        paused_nodes_payload = payload.get("paused_nodes", [])
+        self._paused_nodes = set(map(str, paused_nodes_payload))
 
     def register_paused_node(self, node_id: str) -> None:
         """Record a node that should resume when execution is continued."""
@@ -368,106 +391,3 @@ class GraphRuntimeState:
         module = importlib.import_module("core.workflow.graph_engine.response_coordinator")
         coordinator_cls = module.ResponseStreamCoordinator
         return coordinator_cls(variable_pool=self.variable_pool, graph=graph)
-
-    # ------------------------------------------------------------------
-    # Snapshot helpers
-    # ------------------------------------------------------------------
-    @classmethod
-    def _parse_snapshot_payload(cls, data: str | Mapping[str, Any]) -> _GraphRuntimeStateSnapshot:
-        payload: dict[str, Any]
-        if isinstance(data, str):
-            payload = json.loads(data)
-        else:
-            payload = dict(data)
-
-        version = payload.get("version")
-        if version != "1.0":
-            raise ValueError(f"Unsupported GraphRuntimeState snapshot version: {version}")
-
-        start_at = float(payload.get("start_at", 0.0))
-
-        total_tokens = int(payload.get("total_tokens", 0))
-        if total_tokens < 0:
-            raise ValueError("total_tokens must be non-negative")
-
-        node_run_steps = int(payload.get("node_run_steps", 0))
-        if node_run_steps < 0:
-            raise ValueError("node_run_steps must be non-negative")
-
-        llm_usage_payload = payload.get("llm_usage", {})
-        llm_usage = LLMUsage.model_validate(llm_usage_payload)
-
-        outputs_payload = deepcopy(payload.get("outputs", {}))
-
-        variable_pool_payload = payload.get("variable_pool")
-        has_variable_pool = variable_pool_payload is not None
-        variable_pool = VariablePool.model_validate(variable_pool_payload) if has_variable_pool else VariablePool()
-
-        ready_queue_payload = payload.get("ready_queue")
-        graph_execution_payload = payload.get("graph_execution")
-        response_payload = payload.get("response_coordinator")
-        paused_nodes_payload = payload.get("paused_nodes", [])
-
-        return _GraphRuntimeStateSnapshot(
-            start_at=start_at,
-            total_tokens=total_tokens,
-            node_run_steps=node_run_steps,
-            llm_usage=llm_usage,
-            outputs=outputs_payload,
-            variable_pool=variable_pool,
-            has_variable_pool=has_variable_pool,
-            ready_queue_dump=ready_queue_payload,
-            graph_execution_dump=graph_execution_payload,
-            response_coordinator_dump=response_payload,
-            paused_nodes=tuple(map(str, paused_nodes_payload)),
-        )
-
-    def _apply_snapshot(self, snapshot: _GraphRuntimeStateSnapshot) -> None:
-        self._start_at = snapshot.start_at
-        self._total_tokens = snapshot.total_tokens
-        self._node_run_steps = snapshot.node_run_steps
-        self._llm_usage = snapshot.llm_usage.model_copy()
-        self._outputs = deepcopy(snapshot.outputs)
-        if snapshot.has_variable_pool or self._variable_pool is None:
-            self._variable_pool = snapshot.variable_pool
-
-        self._restore_ready_queue(snapshot.ready_queue_dump)
-        self._restore_graph_execution(snapshot.graph_execution_dump)
-        self._restore_response_coordinator(snapshot.response_coordinator_dump)
-        self._paused_nodes = set(snapshot.paused_nodes)
-
-    def _restore_ready_queue(self, payload: str | None) -> None:
-        if payload is not None:
-            self._ready_queue = self._build_ready_queue()
-            self._ready_queue.loads(payload)
-        else:
-            self._ready_queue = None
-
-    def _restore_graph_execution(self, payload: str | None) -> None:
-        self._graph_execution = None
-        self._pending_graph_execution_workflow_id = None
-
-        if payload is None:
-            return
-
-        try:
-            execution_payload = json.loads(payload)
-            self._pending_graph_execution_workflow_id = execution_payload.get("workflow_id")
-        except (json.JSONDecodeError, TypeError, AttributeError):
-            self._pending_graph_execution_workflow_id = None
-
-        self.graph_execution.loads(payload)
-
-    def _restore_response_coordinator(self, payload: str | None) -> None:
-        if payload is None:
-            self._pending_response_coordinator_dump = None
-            self._response_coordinator = None
-            return
-
-        if self._graph is not None:
-            self.response_coordinator.loads(payload)
-            self._pending_response_coordinator_dump = None
-            return
-
-        self._pending_response_coordinator_dump = payload
-        self._response_coordinator = None

api/core/workflow/runtime/variable_pool.py

@@ -260,7 +260,7 @@ class VariablePool(BaseModel):
             # This ensures that we can keep the id of the system variables intact.
             if self._has(selector):
                 continue
-            self.add(selector, value)
+            self.add(selector, value)  # type: ignore
 
     @classmethod
     def empty(cls) -> "VariablePool":

api/docker/entrypoint.sh

@@ -32,21 +32,22 @@ if [[ "${MODE}" == "worker" ]]; then
 
   exec celery -A celery_entrypoint.celery worker -P ${CELERY_WORKER_CLASS:-gevent} $CONCURRENCY_OPTION \
     --max-tasks-per-child ${MAX_TASKS_PER_CHILD:-50} --loglevel ${LOG_LEVEL:-INFO} \
-    -Q ${CELERY_QUEUES:-dataset,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation} \
-    --prefetch-multiplier=1
+    -Q ${CELERY_QUEUES:-dataset,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation}
 
 elif [[ "${MODE}" == "beat" ]]; then
   exec celery -A app.celery beat --loglevel ${LOG_LEVEL:-INFO}
 else
   if [[ "${DEBUG}" == "true" ]]; then
-    exec flask run --host=${DIFY_BIND_ADDRESS:-0.0.0.0} --port=${DIFY_PORT:-5001} --debug
+    export HOST=${DIFY_BIND_ADDRESS:-0.0.0.0}
+    export PORT=${DIFY_PORT:-5001}
+    exec python -m app
   else
     exec gunicorn \
       --bind "${DIFY_BIND_ADDRESS:-0.0.0.0}:${DIFY_PORT:-5001}" \
       --workers ${SERVER_WORKER_AMOUNT:-1} \
-      --worker-class ${SERVER_WORKER_CLASS:-gevent} \
+      --worker-class ${SERVER_WORKER_CLASS:-geventwebsocket.gunicorn.workers.GeventWebSocketWorker} \
       --worker-connections ${SERVER_WORKER_CONNECTIONS:-10} \
       --timeout ${GUNICORN_TIMEOUT:-200} \
-      app:app
+      app:socketio_app
   fi
 fi

api/extensions/ext_compress.py

@@ -7,7 +7,7 @@ def is_enabled() -> bool:
 
 
 def init_app(app: DifyApp):
-    from flask_compress import Compress
+    from flask_compress import Compress  # type: ignore
 
     compress = Compress()
     compress.init_app(app)

api/extensions/ext_login.py

@@ -1,16 +1,15 @@
 import json
 
-import flask_login
+import flask_login  # type: ignore
 from flask import Response, request
 from flask_login import user_loaded_from_request, user_logged_in
 from werkzeug.exceptions import NotFound, Unauthorized
 
 from configs import dify_config
-from constants import HEADER_NAME_APP_CODE
 from dify_app import DifyApp
 from extensions.ext_database import db
 from libs.passport import PassportService
-from libs.token import extract_access_token, extract_webapp_passport
+from libs.token import extract_access_token
 from models import Account, Tenant, TenantAccountJoin
 from models.model import AppMCPServer, EndUser
 from services.account_service import AccountService
@@ -62,30 +61,14 @@ def load_user_from_request(request_from_flask_login):
         logged_in_account = AccountService.load_logged_in_account(account_id=user_id)
         return logged_in_account
     elif request.blueprint == "web":
-        app_code = request.headers.get(HEADER_NAME_APP_CODE)
-        webapp_token = extract_webapp_passport(app_code, request) if app_code else None
-
-        if webapp_token:
-            decoded = PassportService().verify(webapp_token)
-            end_user_id = decoded.get("end_user_id")
-            if not end_user_id:
-                raise Unauthorized("Invalid Authorization token.")
-            end_user = db.session.query(EndUser).where(EndUser.id == end_user_id).first()
-            if not end_user:
-                raise NotFound("End user not found.")
-            return end_user
-        else:
-            if not auth_token:
-                raise Unauthorized("Invalid Authorization token.")
-            decoded = PassportService().verify(auth_token)
-            end_user_id = decoded.get("end_user_id")
-            if end_user_id:
-                end_user = db.session.query(EndUser).where(EndUser.id == end_user_id).first()
-                if not end_user:
-                    raise NotFound("End user not found.")
-                return end_user
-            else:
-                raise Unauthorized("Invalid Authorization token for web API.")
+        decoded = PassportService().verify(auth_token)
+        end_user_id = decoded.get("end_user_id")
+        if not end_user_id:
+            raise Unauthorized("Invalid Authorization token.")
+        end_user = db.session.query(EndUser).where(EndUser.id == decoded["end_user_id"]).first()
+        if not end_user:
+            raise NotFound("End user not found.")
+        return end_user
     elif request.blueprint == "mcp":
         server_code = request.view_args.get("server_code") if request.view_args else None
         if not server_code:

api/extensions/ext_migrate.py

@@ -2,7 +2,7 @@ from dify_app import DifyApp
 
 
 def init_app(app: DifyApp):
-    import flask_migrate
+    import flask_migrate  # type: ignore
 
     from extensions.ext_database import db
 

api/extensions/ext_otel.py

@@ -103,7 +103,7 @@ def init_app(app: DifyApp):
     def shutdown_tracer():
         provider = trace.get_tracer_provider()
         if hasattr(provider, "force_flush"):
-            provider.force_flush()
+            provider.force_flush()  # ty: ignore [call-non-callable]
 
     class ExceptionLoggingHandler(logging.Handler):
         """Custom logging handler that creates spans for logging.exception() calls"""

api/extensions/ext_proxy_fix.py

@@ -6,4 +6,4 @@ def init_app(app: DifyApp):
     if dify_config.RESPECT_XFORWARD_HEADERS_ENABLED:
         from werkzeug.middleware.proxy_fix import ProxyFix
 
-        app.wsgi_app = ProxyFix(app.wsgi_app, x_port=1)  # type: ignore[method-assign]
+        app.wsgi_app = ProxyFix(app.wsgi_app, x_port=1)  # type: ignore

api/extensions/ext_sentry.py

@@ -5,7 +5,7 @@ from dify_app import DifyApp
 def init_app(app: DifyApp):
     if dify_config.SENTRY_DSN:
         import sentry_sdk
-        from langfuse import parse_error
+        from langfuse import parse_error  # type: ignore
         from sentry_sdk.integrations.celery import CeleryIntegration
         from sentry_sdk.integrations.flask import FlaskIntegration
         from werkzeug.exceptions import HTTPException

api/extensions/ext_socketio.py

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

api/extensions/storage/aliyun_oss_storage.py

@@ -1,7 +1,7 @@
 import posixpath
 from collections.abc import Generator
 
-import oss2 as aliyun_s3
+import oss2 as aliyun_s3  # type: ignore
 
 from configs import dify_config
 from extensions.storage.base_storage import BaseStorage

api/extensions/storage/baidu_obs_storage.py

@@ -2,9 +2,9 @@ import base64
 import hashlib
 from collections.abc import Generator
 
-from baidubce.auth.bce_credentials import BceCredentials
-from baidubce.bce_client_configuration import BceClientConfiguration
-from baidubce.services.bos.bos_client import BosClient
+from baidubce.auth.bce_credentials import BceCredentials  # type: ignore
+from baidubce.bce_client_configuration import BceClientConfiguration  # type: ignore
+from baidubce.services.bos.bos_client import BosClient  # type: ignore
 
 from configs import dify_config
 from extensions.storage.base_storage import BaseStorage

api/extensions/storage/clickzetta_volume/clickzetta_volume_storage.py

@@ -11,7 +11,7 @@ from collections.abc import Generator
 from io import BytesIO
 from pathlib import Path
 
-import clickzetta
+import clickzetta  # type: ignore[import]
 from pydantic import BaseModel, model_validator
 
 from extensions.storage.base_storage import BaseStorage

api/extensions/storage/clickzetta_volume/volume_permissions.py

@@ -34,7 +34,7 @@ class VolumePermissionManager:
         # Support two initialization methods: connection object or configuration dictionary
         if isinstance(connection_or_config, dict):
             # Create connection from configuration dictionary
-            import clickzetta
+            import clickzetta  # type: ignore[import-untyped]
 
             config = connection_or_config
             self._connection = clickzetta.connect(

api/extensions/storage/google_cloud_storage.py

@@ -3,7 +3,7 @@ import io
 import json
 from collections.abc import Generator
 
-from google.cloud import storage as google_cloud_storage
+from google.cloud import storage as google_cloud_storage  # type: ignore
 
 from configs import dify_config
 from extensions.storage.base_storage import BaseStorage

api/extensions/storage/huawei_obs_storage.py

@@ -1,6 +1,6 @@
 from collections.abc import Generator
 
-from obs import ObsClient
+from obs import ObsClient  # type: ignore
 
 from configs import dify_config
 from extensions.storage.base_storage import BaseStorage

api/extensions/storage/oracle_oci_storage.py

@@ -1,7 +1,7 @@
 from collections.abc import Generator
 
-import boto3
-from botocore.exceptions import ClientError
+import boto3  # type: ignore
+from botocore.exceptions import ClientError  # type: ignore
 
 from configs import dify_config
 from extensions.storage.base_storage import BaseStorage

api/extensions/storage/tencent_cos_storage.py

@@ -1,6 +1,6 @@
 from collections.abc import Generator
 
-from qcloud_cos import CosConfig, CosS3Client
+from qcloud_cos import CosConfig, CosS3Client  # type: ignore
 
 from configs import dify_config
 from extensions.storage.base_storage import BaseStorage

api/extensions/storage/volcengine_tos_storage.py

@@ -1,6 +1,6 @@
 from collections.abc import Generator
 
-import tos
+import tos  # type: ignore
 
 from configs import dify_config
 from extensions.storage.base_storage import BaseStorage

api/fields/online_user_fields.py

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

api/fields/workflow_comment_fields.py

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

api/libs/external_api.py

@@ -146,6 +146,6 @@ class ExternalApi(Api):
         kwargs["doc"] = dify_config.SWAGGER_UI_PATH if dify_config.SWAGGER_UI_ENABLED else False
 
         # manual separate call on construction and init_app to ensure configs in kwargs effective
-        super().__init__(app=None, *args, **kwargs)
+        super().__init__(app=None, *args, **kwargs)  # type: ignore
         self.init_app(app, **kwargs)
         register_external_error_handlers(self)

api/libs/gmpy2_pkcs10aep_cipher.py

@@ -23,7 +23,7 @@ from hashlib import sha1
 
 import Crypto.Hash.SHA1
 import Crypto.Util.number
-import gmpy2
+import gmpy2  # type: ignore
 from Crypto import Random
 from Crypto.Signature.pss import MGF1
 from Crypto.Util.number import bytes_to_long, ceil_div, long_to_bytes
@@ -136,7 +136,7 @@ class PKCS1OAepCipher:
         # Step 3a (OS2IP)
         em_int = bytes_to_long(em)
         # Step 3b (RSAEP)
-        m_int = gmpy2.powmod(em_int, self._key.e, self._key.n)
+        m_int = gmpy2.powmod(em_int, self._key.e, self._key.n)  # ty: ignore [unresolved-attribute]
         # Step 3c (I2OSP)
         c = long_to_bytes(m_int, k)
         return c
@@ -169,7 +169,7 @@ class PKCS1OAepCipher:
         ct_int = bytes_to_long(ciphertext)
         # Step 2b (RSADP)
         # m_int = self._key._decrypt(ct_int)
-        m_int = gmpy2.powmod(ct_int, self._key.d, self._key.n)
+        m_int = gmpy2.powmod(ct_int, self._key.d, self._key.n)  # ty: ignore [unresolved-attribute]
         # Complete step 2c (I2OSP)
         em = long_to_bytes(m_int, k)
         # Step 3a
@@ -191,12 +191,12 @@ class PKCS1OAepCipher:
         # Step 3g
         one_pos = hLen + db[hLen:].find(b"\x01")
         lHash1 = db[:hLen]
-        invalid = bord(y) | int(one_pos < hLen)  # type: ignore[arg-type]
+        invalid = bord(y) | int(one_pos < hLen)  # type: ignore
         hash_compare = strxor(lHash1, lHash)
         for x in hash_compare:
-            invalid |= bord(x)  # type: ignore[arg-type]
+            invalid |= bord(x)  # type: ignore
         for x in db[hLen:one_pos]:
-            invalid |= bord(x)  # type: ignore[arg-type]
+            invalid |= bord(x)  # type: ignore
         if invalid != 0:
             raise ValueError("Incorrect decryption.")
         # Step 4

api/libs/helper.py

@@ -81,8 +81,6 @@ class AvatarUrlField(fields.Raw):
         from models import Account
 
         if isinstance(obj, Account) and obj.avatar is not None:
-            if obj.avatar.startswith(("http://", "https://")):
-                return obj.avatar
             return file_helpers.get_signed_file_url(obj.avatar)
         return None
 

api/libs/login.py

@@ -3,7 +3,7 @@ from functools import wraps
 from typing import Any
 
 from flask import current_app, g, has_request_context, request
-from flask_login.config import EXEMPT_METHODS
+from flask_login.config import EXEMPT_METHODS  # type: ignore
 from werkzeug.local import LocalProxy
 
 from configs import dify_config
@@ -87,7 +87,7 @@ def _get_user() -> EndUser | Account | None:
         if "_login_user" not in g:
             current_app.login_manager._load_user()  # type: ignore
 
-        return g._login_user
+        return g._login_user  # type: ignore
 
     return None
 

api/libs/sendgrid.py

@@ -1,8 +1,8 @@
 import logging
 
-import sendgrid
+import sendgrid  # type: ignore
 from python_http_client.exceptions import ForbiddenError, UnauthorizedError
-from sendgrid.helpers.mail import Content, Email, Mail, To
+from sendgrid.helpers.mail import Content, Email, Mail, To  # type: ignore
 
 logger = logging.getLogger(__name__)
 

api/libs/token.py

@@ -12,7 +12,6 @@ from constants import (
     COOKIE_NAME_CSRF_TOKEN,
     COOKIE_NAME_PASSPORT,
     COOKIE_NAME_REFRESH_TOKEN,
-    COOKIE_NAME_WEBAPP_ACCESS_TOKEN,
     HEADER_NAME_CSRF_TOKEN,
     HEADER_NAME_PASSPORT,
 )
@@ -38,6 +37,9 @@ def _real_cookie_name(cookie_name: str) -> str:
 
 
 def _try_extract_from_header(request: Request) -> str | None:
+    """
+    Try to extract access token from header
+    """
     auth_header = request.headers.get("Authorization")
     if auth_header:
         if " " not in auth_header:
@@ -52,30 +54,40 @@ def _try_extract_from_header(request: Request) -> str | None:
     return None
 
 
-def extract_refresh_token(request: Request) -> str | None:
-    return request.cookies.get(_real_cookie_name(COOKIE_NAME_REFRESH_TOKEN))
-
-
 def extract_csrf_token(request: Request) -> str | None:
+    """
+    Try to extract CSRF token from header or cookie.
+    """
     return request.headers.get(HEADER_NAME_CSRF_TOKEN)
 
 
 def extract_csrf_token_from_cookie(request: Request) -> str | None:
+    """
+    Try to extract CSRF token from cookie.
+    """
     return request.cookies.get(_real_cookie_name(COOKIE_NAME_CSRF_TOKEN))
 
 
 def extract_access_token(request: Request) -> str | None:
+    """
+    Try to extract access token from cookie, header or params.
+
+    Access token is either for console session or webapp passport exchange.
+    """
+
     def _try_extract_from_cookie(request: Request) -> str | None:
         return request.cookies.get(_real_cookie_name(COOKIE_NAME_ACCESS_TOKEN))
 
     return _try_extract_from_cookie(request) or _try_extract_from_header(request)
 
 
-def extract_webapp_access_token(request: Request) -> str | None:
-    return request.cookies.get(_real_cookie_name(COOKIE_NAME_WEBAPP_ACCESS_TOKEN)) or _try_extract_from_header(request)
-
-
 def extract_webapp_passport(app_code: str, request: Request) -> str | None:
+    """
+    Try to extract app token from header or params.
+
+    Webapp access token (part of passport) is only used for webapp session.
+    """
+
     def _try_extract_passport_token_from_cookie(request: Request) -> str | None:
         return request.cookies.get(_real_cookie_name(COOKIE_NAME_PASSPORT + "-" + app_code))
 
@@ -143,10 +155,6 @@ def clear_access_token_from_cookie(response: Response, samesite: str = "Lax"):
     _clear_cookie(response, COOKIE_NAME_ACCESS_TOKEN, samesite)
 
 
-def clear_webapp_access_token_from_cookie(response: Response, samesite: str = "Lax"):
-    _clear_cookie(response, COOKIE_NAME_WEBAPP_ACCESS_TOKEN, samesite)
-
-
 def clear_refresh_token_from_cookie(response: Response):
     _clear_cookie(response, COOKIE_NAME_REFRESH_TOKEN)
 

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

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

api/migrations/versions/2025_10_21_1430-ae662b25d9bc_remove_builtin_template_user.py

@@ -1,36 +0,0 @@
-"""remove-builtin-template-user
-
-Revision ID: ae662b25d9bc
-Revises: d98acf217d43
-Create Date: 2025-10-21 14:30:28.566192
-
-"""
-from alembic import op
-import models as models
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = 'ae662b25d9bc'
-down_revision = 'd98acf217d43'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    # ### commands auto generated by Alembic - please adjust! ###
-
-    with op.batch_alter_table('pipeline_built_in_templates', schema=None) as batch_op:
-        batch_op.drop_column('updated_by')
-        batch_op.drop_column('created_by')
-
-    # ### end Alembic commands ###
-
-
-def downgrade():
-    # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('pipeline_built_in_templates', schema=None) as batch_op:
-        batch_op.add_column(sa.Column('created_by', sa.UUID(), autoincrement=False, nullable=False))
-        batch_op.add_column(sa.Column('updated_by', sa.UUID(), autoincrement=False, nullable=True))
-
-    # ### end Alembic commands ###

api/models/__init__.py

@@ -9,6 +9,11 @@ from .account import (
     TenantStatus,
 )
 from .api_based_extension import APIBasedExtension, APIBasedExtensionPoint
+from .comment import (
+    WorkflowComment,
+    WorkflowCommentMention,
+    WorkflowCommentReply,
+)
 from .dataset import (
     AppDatasetJoin,
     Dataset,
@@ -174,6 +179,9 @@ __all__ = [
     "Workflow",
     "WorkflowAppLog",
     "WorkflowAppLogCreatedFrom",
+    "WorkflowComment",
+    "WorkflowCommentMention",
+    "WorkflowCommentReply",
     "WorkflowNodeExecutionModel",
     "WorkflowNodeExecutionOffload",
     "WorkflowNodeExecutionTriggeredFrom",

api/models/account.py

@@ -5,7 +5,7 @@ from datetime import datetime
 from typing import Any, Optional
 
 import sqlalchemy as sa
-from flask_login import UserMixin
+from flask_login import UserMixin  # type: ignore[import-untyped]
 from sqlalchemy import DateTime, String, func, select
 from sqlalchemy.orm import Mapped, Session, mapped_column
 from typing_extensions import deprecated