remove forceUpload

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Cursor Agent <cursoragent@cursor.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

api/.env.example

@@ -33,6 +33,9 @@ TRIGGER_URL=http://localhost: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
 

api/agent-notes/controllers/console/datasets/datasets_document.py.md

@@ -0,0 +1,52 @@
+## Purpose
+
+`api/controllers/console/datasets/datasets_document.py` contains the console (authenticated) APIs for managing dataset documents (list/create/update/delete, processing controls, estimates, etc.).
+
+## Storage model (uploaded files)
+
+- For local file uploads into a knowledge base, the binary is stored via `extensions.ext_storage.storage` under the key:
+  - `upload_files/<tenant_id>/<uuid>.<ext>`
+- File metadata is stored in the `upload_files` table (`UploadFile` model), keyed by `UploadFile.id`.
+- Dataset `Document` records reference the uploaded file via:
+  - `Document.data_source_info.upload_file_id`
+
+## Download endpoint
+
+- `GET /datasets/<dataset_id>/documents/<document_id>/download`
+
+  - Only supported when `Document.data_source_type == "upload_file"`.
+  - Performs dataset permission + tenant checks via `DocumentResource.get_document(...)`.
+  - Delegates `Document -> UploadFile` validation and signed URL generation to `DocumentService.get_document_download_url(...)`.
+  - Applies `cloud_edition_billing_rate_limit_check("knowledge")` to match other KB operations.
+  - Response body is **only**: `{ "url": "<signed-url>" }`.
+
+- `POST /datasets/<dataset_id>/documents/download-zip`
+
+  - Accepts `{ "document_ids": ["..."] }` (upload-file only).
+  - Returns `application/zip` as a single attachment download.
+  - Rationale: browsers often block multiple automatic downloads; a ZIP avoids that limitation.
+  - Applies `cloud_edition_billing_rate_limit_check("knowledge")`.
+  - Delegates dataset permission checks, document/upload-file validation, and download-name generation to
+    `DocumentService.prepare_document_batch_download_zip(...)` before streaming the ZIP.
+
+## Verification plan
+
+- Upload a document from a local file into a dataset.
+- Call the download endpoint and confirm it returns a signed URL.
+- Open the URL and confirm:
+  - Response headers force download (`Content-Disposition`), and
+  - Downloaded bytes match the uploaded file.
+- Select multiple uploaded-file documents and download as ZIP; confirm all selected files exist in the archive.
+
+## Shared helper
+
+- `DocumentService.get_document_download_url(document)` resolves the `UploadFile` and signs a download URL.
+- `DocumentService.prepare_document_batch_download_zip(...)` performs dataset permission checks, batches
+  document + upload file lookups, preserves request order, and generates the client-visible ZIP filename.
+- Internal helpers now live in `DocumentService` (`_get_upload_file_id_for_upload_file_document(...)`,
+  `_get_upload_file_for_upload_file_document(...)`, `_get_upload_files_by_document_id_for_zip_download(...)`).
+- ZIP packing is handled by `FileService.build_upload_files_zip_tempfile(...)`, which also:
+  - sanitizes entry names to avoid path traversal, and
+  - deduplicates names while preserving extensions (e.g., `doc.txt` → `doc (1).txt`).
+    Streaming the response and deferring cleanup is handled by the route via `send_file(path, ...)` + `ExitStack` +
+    `response.call_on_close(...)` (the file is deleted when the response is closed).

api/agent-notes/services/dataset_service.py.md

@@ -0,0 +1,18 @@
+## Purpose
+
+`api/services/dataset_service.py` hosts dataset/document service logic used by console and API controllers.
+
+## Batch document operations
+
+- Batch document workflows should avoid N+1 database queries by using set-based lookups.
+- Tenant checks must be enforced consistently across dataset/document operations.
+- `DocumentService.get_documents_by_ids(...)` fetches documents for a dataset using `id.in_(...)`.
+- `FileService.get_upload_files_by_ids(...)` performs tenant-scoped batch lookup for `UploadFile` (dedupes ids with `set(...)`).
+- `DocumentService.get_document_download_url(...)` and `prepare_document_batch_download_zip(...)` handle
+  dataset/document permission checks plus `Document -> UploadFile` validation for download endpoints.
+
+## Verification plan
+
+- Exercise document list and download endpoints that use the service helpers.
+- Confirm batch download uses constant query count for documents + upload files.
+- Request a ZIP with a missing document id and confirm a 404 is returned.

api/agent-notes/services/file_service.py.md

@@ -0,0 +1,35 @@
+## Purpose
+
+`api/services/file_service.py` owns business logic around `UploadFile` objects: upload validation, storage persistence,
+previews/generators, and deletion.
+
+## Key invariants
+
+- All storage I/O goes through `extensions.ext_storage.storage`.
+- Uploaded file keys follow: `upload_files/<tenant_id>/<uuid>.<ext>`.
+- Upload validation is enforced in `FileService.upload_file(...)` (blocked extensions, size limits, dataset-only types).
+
+## Batch lookup helpers
+
+- `FileService.get_upload_files_by_ids(tenant_id, upload_file_ids)` is the canonical tenant-scoped batch loader for
+  `UploadFile`.
+
+## Dataset document download helpers
+
+The dataset document download/ZIP endpoints now delegate “Document → UploadFile” validation and permission checks to
+`DocumentService` (`api/services/dataset_service.py`). `FileService` stays focused on generic `UploadFile` operations
+(uploading, previews, deletion), plus generic ZIP serving.
+
+### ZIP serving
+
+- `FileService.build_upload_files_zip_tempfile(...)` builds a ZIP from `UploadFile` objects and yields a seeked
+  tempfile **path** so callers can stream it (e.g., `send_file(path, ...)`) without hitting "read of closed file"
+  issues from file-handle lifecycle during streamed responses.
+- Flask `send_file(...)` and the `ExitStack`/`call_on_close(...)` cleanup pattern are handled in the route layer.
+
+## Verification plan
+
+- Unit: `api/tests/unit_tests/controllers/console/datasets/test_datasets_document_download.py`
+  - Verify signed URL generation for upload-file documents and ZIP download behavior for multiple documents.
+- Unit: `api/tests/unit_tests/services/test_file_service_zip_and_lookup.py`
+  - Verify ZIP packing produces a valid, openable archive and preserves file content.

api/agent-notes/tests/unit_tests/controllers/console/datasets/test_datasets_document_download.py.md

@@ -0,0 +1,28 @@
+## Purpose
+
+Unit tests for the console dataset document download endpoint:
+
+- `GET /datasets/<dataset_id>/documents/<document_id>/download`
+
+## Testing approach
+
+- Uses `Flask.test_request_context()` and calls the `Resource.get(...)` method directly.
+- Monkeypatches console decorators (`login_required`, `setup_required`, rate limit) to no-ops to keep the test focused.
+- Mocks:
+  - `DatasetService.get_dataset` / `check_dataset_permission`
+  - `DocumentService.get_document` for single-file download tests
+  - `DocumentService.get_documents_by_ids` + `FileService.get_upload_files_by_ids` for ZIP download tests
+  - `FileService.get_upload_files_by_ids` for `UploadFile` lookups in single-file tests
+  - `services.dataset_service.file_helpers.get_signed_file_url` to return a deterministic URL
+- Document mocks include `id` fields so batch lookups can map documents by id.
+
+## Covered cases
+
+- Success returns `{ "url": "<signed>" }` for upload-file documents.
+- 404 when document is not `upload_file`.
+- 404 when `upload_file_id` is missing.
+- 404 when referenced `UploadFile` row does not exist.
+- 403 when document tenant does not match current tenant.
+- Batch ZIP download returns `application/zip` for upload-file documents.
+- Batch ZIP download rejects non-upload-file documents.
+- Batch ZIP download uses a random `.zip` attachment name (`download_name`), so tests only assert the suffix.

api/agent-notes/tests/unit_tests/services/test_file_service_zip_and_lookup.py.md

@@ -0,0 +1,18 @@
+## Purpose
+
+Unit tests for `api/services/file_service.py` helper methods that are not covered by higher-level controller tests.
+
+## What’s covered
+
+- `FileService.build_upload_files_zip_tempfile(...)`
+  - ZIP entry name sanitization (no directory components / traversal)
+  - name deduplication while preserving extensions
+  - writing streamed bytes from `storage.load(...)` into ZIP entries
+  - yields a tempfile path so callers can open/stream the ZIP without holding a live file handle
+- `FileService.get_upload_files_by_ids(...)`
+  - returns `{}` for empty id lists
+  - returns an id-keyed mapping for non-empty lists
+
+## Notes
+
+- These tests intentionally stub `storage.load` and `db.session.scalars(...).all()` to avoid needing a real DB/storage.

api/app.py

@@ -1,3 +1,4 @@
+import os
 import sys
 
 
@@ -8,10 +9,15 @@ def is_db_command() -> bool:
 
 
 # create app
+flask_app = None
+socketio_app = None
+
 if is_db_command():
     from app_factory import create_migrations_app
 
     app = create_migrations_app()
+    socketio_app = app
+    flask_app = app
 else:
     # Gunicorn and Celery handle monkey patching automatically in production by
     # specifying the `gevent` worker class. Manual monkey patching is not required here.
@@ -22,8 +28,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  # type: ignore[reportMissingTypeStubs]
+
+    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

@@ -1,6 +1,7 @@
 import logging
 import time
 
+import socketio  # type: ignore[reportMissingTypeStubs]
 from opentelemetry.trace import get_current_span
 from opentelemetry.trace.span import INVALID_SPAN_ID, INVALID_TRACE_ID
 
@@ -8,6 +9,7 @@ from configs import dify_config
 from contexts.wrapper import RecyclableContextVar
 from core.logging.context import init_request_context
 from dify_app import DifyApp
+from extensions.ext_socketio import sio
 
 logger = logging.getLogger(__name__)
 
@@ -60,14 +62,18 @@ def create_flask_app_with_configs() -> DifyApp:
     return dify_app
 
 
-def create_app() -> DifyApp:
+def create_app() -> tuple[socketio.WSGIApp, DifyApp]:
     start_time = time.perf_counter()
     app = create_flask_app_with_configs()
     initialize_extensions(app)
+
+    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

@@ -1219,6 +1219,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",
@@ -1333,6 +1340,7 @@ class FeatureConfig(
     WorkflowConfig,
     WorkflowNodeExecutionConfig,
     WorkspaceConfig,
+    CollaborationConfig,
     LoginConfig,
     AccountConfig,
     SwaggerUIConfig,

api/controllers/console/__init__.py

@@ -63,6 +63,7 @@ from .app import (
     statistic,
     workflow,
     workflow_app_log,
+    workflow_comment,
     workflow_draft_variable,
     workflow_run,
     workflow_statistic,
@@ -112,6 +113,7 @@ from .explore import (
     recommended_app,
     saved_message,
 )
+from .socketio import workflow as socketio_workflow  # pyright: ignore[reportUnusedImport]
 
 # Import tag controllers
 from .tag import tags
@@ -203,6 +205,7 @@ __all__ = [
     "website",
     "workflow",
     "workflow_app_log",
+    "workflow_comment",
     "workflow_draft_variable",
     "workflow_run",
     "workflow_statistic",

api/controllers/console/app/app.py

@@ -1,4 +1,3 @@
-import re
 import uuid
 from datetime import datetime
 from typing import Any, Literal, TypeAlias
@@ -68,48 +67,6 @@ class AppListQuery(BaseModel):
             raise ValueError("Invalid UUID format in tag_ids.") from exc
 
 
-# XSS prevention: patterns that could lead to XSS attacks
-# Includes: script tags, iframe tags, javascript: protocol, SVG with onload, etc.
-_XSS_PATTERNS = [
-    r"<script[^>]*>.*?</script>",  # Script tags
-    r"<iframe\b[^>]*?(?:/>|>.*?</iframe>)",  # Iframe tags (including self-closing)
-    r"javascript:",  # JavaScript protocol
-    r"<svg[^>]*?\s+onload\s*=[^>]*>",  # SVG with onload handler (attribute-aware, flexible whitespace)
-    r"<.*?on\s*\w+\s*=",  # Event handlers like onclick, onerror, etc.
-    r"<object\b[^>]*(?:\s*/>|>.*?</object\s*>)",  # Object tags (opening tag)
-    r"<embed[^>]*>",  # Embed tags (self-closing)
-    r"<link[^>]*>",  # Link tags with javascript
-]
-
-
-def _validate_xss_safe(value: str | None, field_name: str = "Field") -> str | None:
-    """
-    Validate that a string value doesn't contain potential XSS payloads.
-
-    Args:
-        value: The string value to validate
-        field_name: Name of the field for error messages
-
-    Returns:
-        The original value if safe
-
-    Raises:
-        ValueError: If the value contains XSS patterns
-    """
-    if value is None:
-        return None
-
-    value_lower = value.lower()
-    for pattern in _XSS_PATTERNS:
-        if re.search(pattern, value_lower, re.DOTALL | re.IGNORECASE):
-            raise ValueError(
-                f"{field_name} contains invalid characters or patterns. "
-                "HTML tags, JavaScript, and other potentially dangerous content are not allowed."
-            )
-
-    return value
-
-
 class CreateAppPayload(BaseModel):
     name: str = Field(..., min_length=1, description="App name")
     description: str | None = Field(default=None, description="App description (max 400 chars)", max_length=400)
@@ -118,11 +75,6 @@ class CreateAppPayload(BaseModel):
     icon: str | None = Field(default=None, description="Icon")
     icon_background: str | None = Field(default=None, description="Icon background color")
 
-    @field_validator("name", "description", mode="before")
-    @classmethod
-    def validate_xss_safe(cls, value: str | None, info) -> str | None:
-        return _validate_xss_safe(value, info.field_name)
-
 
 class UpdateAppPayload(BaseModel):
     name: str = Field(..., min_length=1, description="App name")
@@ -133,11 +85,6 @@ class UpdateAppPayload(BaseModel):
     use_icon_as_answer_icon: bool | None = Field(default=None, description="Use icon as answer icon")
     max_active_requests: int | None = Field(default=None, description="Maximum active requests")
 
-    @field_validator("name", "description", mode="before")
-    @classmethod
-    def validate_xss_safe(cls, value: str | None, info) -> str | None:
-        return _validate_xss_safe(value, info.field_name)
-
 
 class CopyAppPayload(BaseModel):
     name: str | None = Field(default=None, description="Name for the copied app")
@@ -146,11 +93,6 @@ class CopyAppPayload(BaseModel):
     icon: str | None = Field(default=None, description="Icon")
     icon_background: str | None = Field(default=None, description="Icon background color")
 
-    @field_validator("name", "description", mode="before")
-    @classmethod
-    def validate_xss_safe(cls, value: str | None, info) -> str | None:
-        return _validate_xss_safe(value, info.field_name)
-
 
 class AppExportQuery(BaseModel):
     include_secret: bool = Field(default=False, description="Include secrets in export")

api/controllers/console/app/workflow.py

@@ -32,8 +32,10 @@ from core.trigger.debug.event_selectors import (
 from core.workflow.enums import NodeType
 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.member_fields import simple_account_fields
+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
@@ -43,6 +45,7 @@ from libs.login import current_account_with_tenant, login_required
 from models import App
 from models.model import AppMode
 from models.workflow import Workflow
+from repositories.workflow_collaboration_repository import WORKFLOW_ONLINE_USERS_PREFIX
 from services.app_generate_service import AppGenerateService
 from services.errors.app import WorkflowHashNotEqualError
 from services.errors.llm import InvokeRateLimitError
@@ -180,6 +183,14 @@ class WorkflowUpdatePayload(BaseModel):
     marked_comment: str | None = Field(default=None, max_length=100)
 
 
+class WorkflowFeaturesPayload(BaseModel):
+    features: dict[str, Any] = Field(..., description="Workflow feature configuration")
+
+
+class WorkflowOnlineUsersQuery(BaseModel):
+    workflow_ids: str = Field(..., description="Comma-separated workflow IDs")
+
+
 class DraftWorkflowTriggerRunPayload(BaseModel):
     node_id: str
 
@@ -203,6 +214,8 @@ reg(DefaultBlockConfigQuery)
 reg(ConvertToWorkflowPayload)
 reg(WorkflowListQuery)
 reg(WorkflowUpdatePayload)
+reg(WorkflowFeaturesPayload)
+reg(WorkflowOnlineUsersQuery)
 reg(DraftWorkflowTriggerRunPayload)
 reg(DraftWorkflowTriggerRunAllPayload)
 
@@ -791,6 +804,31 @@ class ConvertToWorkflowApi(Resource):
         }
 
 
+@console_ns.route("/apps/<uuid:app_id>/workflows/draft/features")
+class WorkflowFeaturesApi(Resource):
+    """Update draft workflow features."""
+
+    @console_ns.expect(console_ns.models[WorkflowFeaturesPayload.__name__])
+    @console_ns.doc("update_workflow_features")
+    @console_ns.doc(description="Update draft workflow features")
+    @console_ns.doc(params={"app_id": "Application ID"})
+    @console_ns.response(200, "Workflow features updated successfully")
+    @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()
+
+        args = WorkflowFeaturesPayload.model_validate(console_ns.payload or {})
+        features = args.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):
     @console_ns.expect(console_ns.models[WorkflowListQuery.__name__])
@@ -1166,3 +1204,32 @@ class DraftWorkflowTriggerRunAllApi(Resource):
                     "status": "error",
                 }
             ), 400
+
+
+@console_ns.route("/apps/workflows/online-users")
+class WorkflowOnlineUsersApi(Resource):
+    @console_ns.expect(console_ns.models[WorkflowOnlineUsersQuery.__name__])
+    @console_ns.doc("get_workflow_online_users")
+    @console_ns.doc(description="Get workflow online users")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @marshal_with(online_user_list_fields)
+    def get(self):
+        args = WorkflowOnlineUsersQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+
+        workflow_ids = [workflow_id.strip() for workflow_id in args.workflow_ids.split(",") if workflow_id.strip()]
+
+        results = []
+        for workflow_id in workflow_ids:
+            users_json = redis_client.hgetall(f"{WORKFLOW_ONLINE_USERS_PREFIX}{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,317 @@
+import logging
+
+from flask_restx import Resource, fields, marshal_with
+from pydantic import BaseModel, Field
+
+from controllers.console import console_ns
+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__)
+DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
+
+
+class WorkflowCommentCreatePayload(BaseModel):
+    position_x: float = Field(..., description="Comment X position")
+    position_y: float = Field(..., description="Comment Y position")
+    content: str = Field(..., description="Comment content")
+    mentioned_user_ids: list[str] = Field(default_factory=list, description="Mentioned user IDs")
+
+
+class WorkflowCommentUpdatePayload(BaseModel):
+    content: str = Field(..., description="Comment content")
+    position_x: float | None = Field(default=None, description="Comment X position")
+    position_y: float | None = Field(default=None, description="Comment Y position")
+    mentioned_user_ids: list[str] = Field(default_factory=list, description="Mentioned user IDs")
+
+
+class WorkflowCommentReplyCreatePayload(BaseModel):
+    content: str = Field(..., description="Reply content")
+    mentioned_user_ids: list[str] = Field(default_factory=list, description="Mentioned user IDs")
+
+
+class WorkflowCommentReplyUpdatePayload(BaseModel):
+    content: str = Field(..., description="Reply content")
+    mentioned_user_ids: list[str] = Field(default_factory=list, description="Mentioned user IDs")
+
+
+for model in (
+    WorkflowCommentCreatePayload,
+    WorkflowCommentUpdatePayload,
+    WorkflowCommentReplyCreatePayload,
+    WorkflowCommentReplyUpdatePayload,
+):
+    console_ns.schema_model(model.__name__, model.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
+
+workflow_comment_basic_model = console_ns.model("WorkflowCommentBasic", workflow_comment_basic_fields)
+workflow_comment_detail_model = console_ns.model("WorkflowCommentDetail", workflow_comment_detail_fields)
+workflow_comment_create_model = console_ns.model("WorkflowCommentCreate", workflow_comment_create_fields)
+workflow_comment_update_model = console_ns.model("WorkflowCommentUpdate", workflow_comment_update_fields)
+workflow_comment_resolve_model = console_ns.model("WorkflowCommentResolve", workflow_comment_resolve_fields)
+workflow_comment_reply_create_model = console_ns.model(
+    "WorkflowCommentReplyCreate", workflow_comment_reply_create_fields
+)
+workflow_comment_reply_update_model = console_ns.model(
+    "WorkflowCommentReplyUpdate", workflow_comment_reply_update_fields
+)
+workflow_comment_mention_users_model = console_ns.model(
+    "WorkflowCommentMentionUsers",
+    {"users": fields.List(fields.Nested(account_with_role_fields))},
+)
+
+
+@console_ns.route("/apps/<uuid:app_id>/workflow/comments")
+class WorkflowCommentListApi(Resource):
+    """API for listing and creating workflow comments."""
+
+    @console_ns.doc("list_workflow_comments")
+    @console_ns.doc(description="Get all comments for a workflow")
+    @console_ns.doc(params={"app_id": "Application ID"})
+    @console_ns.response(200, "Comments retrieved successfully", workflow_comment_basic_model)
+    @login_required
+    @setup_required
+    @account_initialization_required
+    @get_app_model()
+    @marshal_with(workflow_comment_basic_model, 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
+
+    @console_ns.doc("create_workflow_comment")
+    @console_ns.doc(description="Create a new workflow comment")
+    @console_ns.doc(params={"app_id": "Application ID"})
+    @console_ns.expect(console_ns.models[WorkflowCommentCreatePayload.__name__])
+    @console_ns.response(201, "Comment created successfully", workflow_comment_create_model)
+    @login_required
+    @setup_required
+    @account_initialization_required
+    @get_app_model()
+    @marshal_with(workflow_comment_create_model)
+    def post(self, app_model: App):
+        """Create a new workflow comment."""
+        payload = WorkflowCommentCreatePayload.model_validate(console_ns.payload or {})
+
+        result = WorkflowCommentService.create_comment(
+            tenant_id=current_user.current_tenant_id,
+            app_id=app_model.id,
+            created_by=current_user.id,
+            content=payload.content,
+            position_x=payload.position_x,
+            position_y=payload.position_y,
+            mentioned_user_ids=payload.mentioned_user_ids,
+        )
+
+        return result, 201
+
+
+@console_ns.route("/apps/<uuid:app_id>/workflow/comments/<string:comment_id>")
+class WorkflowCommentDetailApi(Resource):
+    """API for managing individual workflow comments."""
+
+    @console_ns.doc("get_workflow_comment")
+    @console_ns.doc(description="Get a specific workflow comment")
+    @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID"})
+    @console_ns.response(200, "Comment retrieved successfully", workflow_comment_detail_model)
+    @login_required
+    @setup_required
+    @account_initialization_required
+    @get_app_model()
+    @marshal_with(workflow_comment_detail_model)
+    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
+
+    @console_ns.doc("update_workflow_comment")
+    @console_ns.doc(description="Update a workflow comment")
+    @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID"})
+    @console_ns.expect(console_ns.models[WorkflowCommentUpdatePayload.__name__])
+    @console_ns.response(200, "Comment updated successfully", workflow_comment_update_model)
+    @login_required
+    @setup_required
+    @account_initialization_required
+    @get_app_model()
+    @marshal_with(workflow_comment_update_model)
+    def put(self, app_model: App, comment_id: str):
+        """Update a workflow comment."""
+        payload = WorkflowCommentUpdatePayload.model_validate(console_ns.payload or {})
+
+        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=payload.content,
+            position_x=payload.position_x,
+            position_y=payload.position_y,
+            mentioned_user_ids=payload.mentioned_user_ids,
+        )
+
+        return result
+
+    @console_ns.doc("delete_workflow_comment")
+    @console_ns.doc(description="Delete a workflow comment")
+    @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID"})
+    @console_ns.response(204, "Comment deleted successfully")
+    @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
+
+
+@console_ns.route("/apps/<uuid:app_id>/workflow/comments/<string:comment_id>/resolve")
+class WorkflowCommentResolveApi(Resource):
+    """API for resolving and reopening workflow comments."""
+
+    @console_ns.doc("resolve_workflow_comment")
+    @console_ns.doc(description="Resolve a workflow comment")
+    @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID"})
+    @console_ns.response(200, "Comment resolved successfully", workflow_comment_resolve_model)
+    @login_required
+    @setup_required
+    @account_initialization_required
+    @get_app_model()
+    @marshal_with(workflow_comment_resolve_model)
+    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
+
+
+@console_ns.route("/apps/<uuid:app_id>/workflow/comments/<string:comment_id>/replies")
+class WorkflowCommentReplyApi(Resource):
+    """API for managing comment replies."""
+
+    @console_ns.doc("create_workflow_comment_reply")
+    @console_ns.doc(description="Add a reply to a workflow comment")
+    @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID"})
+    @console_ns.expect(console_ns.models[WorkflowCommentReplyCreatePayload.__name__])
+    @console_ns.response(201, "Reply created successfully", workflow_comment_reply_create_model)
+    @login_required
+    @setup_required
+    @account_initialization_required
+    @get_app_model()
+    @marshal_with(workflow_comment_reply_create_model)
+    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
+        )
+
+        payload = WorkflowCommentReplyCreatePayload.model_validate(console_ns.payload or {})
+
+        result = WorkflowCommentService.create_reply(
+            comment_id=comment_id,
+            content=payload.content,
+            created_by=current_user.id,
+            mentioned_user_ids=payload.mentioned_user_ids,
+        )
+
+        return result, 201
+
+
+@console_ns.route("/apps/<uuid:app_id>/workflow/comments/<string:comment_id>/replies/<string:reply_id>")
+class WorkflowCommentReplyDetailApi(Resource):
+    """API for managing individual comment replies."""
+
+    @console_ns.doc("update_workflow_comment_reply")
+    @console_ns.doc(description="Update a comment reply")
+    @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID", "reply_id": "Reply ID"})
+    @console_ns.expect(console_ns.models[WorkflowCommentReplyUpdatePayload.__name__])
+    @console_ns.response(200, "Reply updated successfully", workflow_comment_reply_update_model)
+    @login_required
+    @setup_required
+    @account_initialization_required
+    @get_app_model()
+    @marshal_with(workflow_comment_reply_update_model)
+    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
+        )
+
+        payload = WorkflowCommentReplyUpdatePayload.model_validate(console_ns.payload or {})
+
+        reply = WorkflowCommentService.update_reply(
+            reply_id=reply_id,
+            user_id=current_user.id,
+            content=payload.content,
+            mentioned_user_ids=payload.mentioned_user_ids,
+        )
+
+        return reply
+
+    @console_ns.doc("delete_workflow_comment_reply")
+    @console_ns.doc(description="Delete a comment reply")
+    @console_ns.doc(params={"app_id": "Application ID", "comment_id": "Comment ID", "reply_id": "Reply ID"})
+    @console_ns.response(204, "Reply deleted successfully")
+    @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
+
+
+@console_ns.route("/apps/<uuid:app_id>/workflow/comments/mention-users")
+class WorkflowCommentMentionUsersApi(Resource):
+    """API for getting mentionable users for workflow comments."""
+
+    @console_ns.doc("workflow_comment_mention_users")
+    @console_ns.doc(description="Get all users in current tenant for mentions")
+    @console_ns.doc(params={"app_id": "Application ID"})
+    @console_ns.response(200, "Mentionable users retrieved successfully", workflow_comment_mention_users_model)
+    @login_required
+    @setup_required
+    @account_initialization_required
+    @get_app_model()
+    @marshal_with(workflow_comment_mention_users_model)
+    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}

api/controllers/console/app/workflow_draft_variable.py

@@ -21,9 +21,9 @@ 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 login_required
+from libs.login import current_user, login_required
 from models import App, AppMode
 from models.workflow import WorkflowDraftVariable
 from services.workflow_draft_variable_service import WorkflowDraftVariableList, WorkflowDraftVariableService
@@ -43,6 +43,16 @@ class WorkflowDraftVariableUpdatePayload(BaseModel):
     value: Any | None = Field(default=None, description="Variable value")
 
 
+class ConversationVariableUpdatePayload(BaseModel):
+    conversation_variables: list[dict[str, Any]] = Field(
+        ..., description="Conversation variables for the draft workflow"
+    )
+
+
+class EnvironmentVariableUpdatePayload(BaseModel):
+    environment_variables: list[dict[str, Any]] = Field(..., description="Environment variables for the draft workflow")
+
+
 console_ns.schema_model(
     WorkflowDraftVariableListQuery.__name__,
     WorkflowDraftVariableListQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
@@ -51,6 +61,14 @@ console_ns.schema_model(
     WorkflowDraftVariableUpdatePayload.__name__,
     WorkflowDraftVariableUpdatePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
 )
+console_ns.schema_model(
+    ConversationVariableUpdatePayload.__name__,
+    ConversationVariableUpdatePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+console_ns.schema_model(
+    EnvironmentVariableUpdatePayload.__name__,
+    EnvironmentVariableUpdatePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
 
 
 def _convert_values_to_json_serializable_object(value: Segment):
@@ -383,7 +401,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
@@ -476,6 +494,34 @@ class ConversationVariableCollectionApi(Resource):
         db.session.commit()
         return _get_variable_list(app_model, CONVERSATION_VARIABLE_NODE_ID)
 
+    @console_ns.expect(console_ns.models[ConversationVariableUpdatePayload.__name__])
+    @console_ns.doc("update_conversation_variables")
+    @console_ns.doc(description="Update conversation variables for workflow draft")
+    @console_ns.doc(params={"app_id": "Application ID"})
+    @console_ns.response(200, "Conversation variables updated successfully")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    @get_app_model(mode=AppMode.ADVANCED_CHAT)
+    def post(self, app_model: App):
+        payload = ConversationVariableUpdatePayload.model_validate(console_ns.payload or {})
+
+        workflow_service = WorkflowService()
+
+        conversation_variables_list = payload.conversation_variables
+        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):
@@ -527,3 +573,31 @@ class EnvironmentVariableCollectionApi(Resource):
             )
 
         return {"items": env_vars_list}
+
+    @console_ns.expect(console_ns.models[EnvironmentVariableUpdatePayload.__name__])
+    @console_ns.doc("update_environment_variables")
+    @console_ns.doc(description="Update environment variables for workflow draft")
+    @console_ns.doc(params={"app_id": "Application ID"})
+    @console_ns.response(200, "Environment variables updated successfully")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    def post(self, app_model: App):
+        payload = EnvironmentVariableUpdatePayload.model_validate(console_ns.payload or {})
+
+        workflow_service = WorkflowService()
+
+        environment_variables_list = payload.environment_variables
+        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/controllers/console/datasets/datasets_document.py

@@ -2,10 +2,12 @@ import json
 import logging
 from argparse import ArgumentTypeError
 from collections.abc import Sequence
-from typing import Literal, cast
+from contextlib import ExitStack
+from typing import Any, Literal, cast
+from uuid import UUID
 
 import sqlalchemy as sa
-from flask import request
+from flask import request, send_file
 from flask_restx import Resource, fields, marshal, marshal_with
 from pydantic import BaseModel, Field
 from sqlalchemy import asc, desc, select
@@ -42,6 +44,7 @@ from models import DatasetProcessRule, Document, DocumentSegment, UploadFile
 from models.dataset import DocumentPipelineExecutionLog
 from services.dataset_service import DatasetService, DocumentService
 from services.entities.knowledge_entities.knowledge_entities import KnowledgeConfig, ProcessRule, RetrievalModel
+from services.file_service import FileService
 
 from ..app.error import (
     ProviderModelCurrentlyNotSupportError,
@@ -65,6 +68,9 @@ from ..wraps import (
 
 logger = logging.getLogger(__name__)
 
+# NOTE: Keep constants near the top of the module for discoverability.
+DOCUMENT_BATCH_DOWNLOAD_ZIP_MAX_DOCS = 100
+
 
 def _get_or_create_model(model_name: str, field_def):
     existing = console_ns.models.get(model_name)
@@ -104,6 +110,12 @@ class DocumentRenamePayload(BaseModel):
     name: str
 
 
+class DocumentBatchDownloadZipPayload(BaseModel):
+    """Request payload for bulk downloading documents as a zip archive."""
+
+    document_ids: list[UUID] = Field(..., min_length=1, max_length=DOCUMENT_BATCH_DOWNLOAD_ZIP_MAX_DOCS)
+
+
 class DocumentDatasetListParam(BaseModel):
     page: int = Field(1, title="Page", description="Page number.")
     limit: int = Field(20, title="Limit", description="Page size.")
@@ -120,6 +132,7 @@ register_schema_models(
     RetrievalModel,
     DocumentRetryPayload,
     DocumentRenamePayload,
+    DocumentBatchDownloadZipPayload,
 )
 
 
@@ -853,6 +866,62 @@ class DocumentApi(DocumentResource):
         return {"result": "success"}, 204
 
 
+@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/download")
+class DocumentDownloadApi(DocumentResource):
+    """Return a signed download URL for a dataset document's original uploaded file."""
+
+    @console_ns.doc("get_dataset_document_download_url")
+    @console_ns.doc(description="Get a signed download URL for a dataset document's original uploaded file")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @cloud_edition_billing_rate_limit_check("knowledge")
+    def get(self, dataset_id: str, document_id: str) -> dict[str, Any]:
+        # Reuse the shared permission/tenant checks implemented in DocumentResource.
+        document = self.get_document(str(dataset_id), str(document_id))
+        return {"url": DocumentService.get_document_download_url(document)}
+
+
+@console_ns.route("/datasets/<uuid:dataset_id>/documents/download-zip")
+class DocumentBatchDownloadZipApi(DocumentResource):
+    """Download multiple uploaded-file documents as a single ZIP (avoids browser multi-download limits)."""
+
+    @console_ns.doc("download_dataset_documents_as_zip")
+    @console_ns.doc(description="Download selected dataset documents as a single ZIP archive (upload-file only)")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @cloud_edition_billing_rate_limit_check("knowledge")
+    @console_ns.expect(console_ns.models[DocumentBatchDownloadZipPayload.__name__])
+    def post(self, dataset_id: str):
+        """Stream a ZIP archive containing the requested uploaded documents."""
+        # Parse and validate request payload.
+        payload = DocumentBatchDownloadZipPayload.model_validate(console_ns.payload or {})
+
+        current_user, current_tenant_id = current_account_with_tenant()
+        dataset_id = str(dataset_id)
+        document_ids: list[str] = [str(document_id) for document_id in payload.document_ids]
+        upload_files, download_name = DocumentService.prepare_document_batch_download_zip(
+            dataset_id=dataset_id,
+            document_ids=document_ids,
+            tenant_id=current_tenant_id,
+            current_user=current_user,
+        )
+
+        # Delegate ZIP packing to FileService, but keep Flask response+cleanup in the route.
+        with ExitStack() as stack:
+            zip_path = stack.enter_context(FileService.build_upload_files_zip_tempfile(upload_files=upload_files))
+            response = send_file(
+                zip_path,
+                mimetype="application/zip",
+                as_attachment=True,
+                download_name=download_name,
+            )
+            cleanup = stack.pop_all()
+            response.call_on_close(cleanup.close)
+        return response
+
+
 @console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/processing/<string:action>")
 class DocumentProcessingApi(DocumentResource):
     @console_ns.doc("update_document_processing")

api/controllers/console/socketio/__init__.py

@@ -0,0 +1 @@
+

api/controllers/console/socketio/workflow.py

@@ -0,0 +1,112 @@
+import logging
+from collections.abc import Callable
+from typing import cast
+
+from flask import Request as FlaskRequest
+
+from extensions.ext_socketio import sio
+from libs.passport import PassportService
+from libs.token import extract_access_token
+from repositories.workflow_collaboration_repository import WorkflowCollaborationRepository
+from services.account_service import AccountService
+from services.workflow_collaboration_service import WorkflowCollaborationService
+
+repository = WorkflowCollaborationRepository()
+collaboration_service = WorkflowCollaborationService(repository, sio)
+
+
+def _sio_on(event: str) -> Callable[[Callable[..., object]], Callable[..., object]]:
+    return cast(Callable[[Callable[..., object]], Callable[..., object]], sio.on(event))
+
+
+@_sio_on("connect")
+def socket_connect(sid, environ, auth):
+    """
+    WebSocket connect event, do authentication here.
+    """
+    try:
+        request_environ = FlaskRequest(environ)
+        token = extract_access_token(request_environ)
+    except Exception:
+        logging.exception("Failed to extract token")
+        token = None
+
+    if not token:
+        logging.warning("Socket connect rejected: missing token (sid=%s)", sid)
+        return False
+
+    try:
+        decoded = PassportService().verify(token)
+        user_id = decoded.get("user_id")
+        if not user_id:
+            logging.warning("Socket connect rejected: missing user_id (sid=%s)", sid)
+            return False
+
+        with sio.app.app_context():
+            user = AccountService.load_logged_in_account(account_id=user_id)
+            if not user:
+                logging.warning(
+                    "Socket connect rejected: user not found (user_id=%s, sid=%s)", user_id, sid
+                )
+                return False
+            if not user.has_edit_permission:
+                logging.warning(
+                    "Socket connect rejected: no edit permission (user_id=%s, sid=%s)", user_id, sid
+                )
+                return False
+
+            collaboration_service.save_session(sid, user)
+            return True
+
+    except Exception:
+        logging.exception("Socket authentication failed")
+        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
+
+    result = collaboration_service.register_session(workflow_id, sid)
+    if not result:
+        return {"msg": "unauthorized"}, 401
+
+    user_id, is_leader = result
+    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.
+    """
+    collaboration_service.disconnect_session(sid)
+
+
+@_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
+    """
+    return collaboration_service.relay_collaboration_event(sid, data)
+
+
+@_sio_on("graph_event")
+def handle_graph_event(sid, data):
+    """
+    Handle graph events - simple broadcast relay.
+    """
+    return collaboration_service.relay_graph_event(sid, data)

api/controllers/console/workspace/account.py

@@ -36,6 +36,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
@@ -73,6 +74,10 @@ class AccountAvatarPayload(BaseModel):
     avatar: str
 
 
+class AccountAvatarQuery(BaseModel):
+    avatar: str = Field(..., description="Avatar file ID")
+
+
 class AccountInterfaceLanguagePayload(BaseModel):
     interface_language: str
 
@@ -158,6 +163,7 @@ def reg(cls: type[BaseModel]):
 reg(AccountInitPayload)
 reg(AccountNamePayload)
 reg(AccountAvatarPayload)
+reg(AccountAvatarQuery)
 reg(AccountInterfaceLanguagePayload)
 reg(AccountInterfaceThemePayload)
 reg(AccountTimezonePayload)
@@ -248,6 +254,18 @@ class AccountNameApi(Resource):
 
 @console_ns.route("/account/avatar")
 class AccountAvatarApi(Resource):
+    @console_ns.expect(console_ns.models[AccountAvatarQuery.__name__])
+    @console_ns.doc("get_account_avatar")
+    @console_ns.doc(description="Get account avatar url")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self):
+        args = AccountAvatarQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+
+        avatar_url = file_helpers.get_signed_file_url(args.avatar)
+        return {"avatar_url": avatar_url}
+
     @console_ns.expect(console_ns.models[AccountAvatarPayload.__name__])
     @setup_required
     @login_required

api/core/plugin/impl/base.py

@@ -320,18 +320,17 @@ class BasePluginClient:
             case PluginInvokeError.__name__:
                 error_object = json.loads(message)
                 invoke_error_type = error_object.get("error_type")
-                args = error_object.get("args")
                 match invoke_error_type:
                     case InvokeRateLimitError.__name__:
-                        raise InvokeRateLimitError(description=args.get("description"))
+                        raise InvokeRateLimitError(description=error_object.get("message"))
                     case InvokeAuthorizationError.__name__:
-                        raise InvokeAuthorizationError(description=args.get("description"))
+                        raise InvokeAuthorizationError(description=error_object.get("message"))
                     case InvokeBadRequestError.__name__:
-                        raise InvokeBadRequestError(description=args.get("description"))
+                        raise InvokeBadRequestError(description=error_object.get("message"))
                     case InvokeConnectionError.__name__:
-                        raise InvokeConnectionError(description=args.get("description"))
+                        raise InvokeConnectionError(description=error_object.get("message"))
                     case InvokeServerUnavailableError.__name__:
-                        raise InvokeServerUnavailableError(description=args.get("description"))
+                        raise InvokeServerUnavailableError(description=error_object.get("message"))
                     case CredentialsValidateFailedError.__name__:
                         raise CredentialsValidateFailedError(error_object.get("message"))
                     case EndpointSetupFailedError.__name__:
@@ -339,11 +338,11 @@ class BasePluginClient:
                     case TriggerProviderCredentialValidationError.__name__:
                         raise TriggerProviderCredentialValidationError(error_object.get("message"))
                     case TriggerPluginInvokeError.__name__:
-                        raise TriggerPluginInvokeError(description=error_object.get("description"))
+                        raise TriggerPluginInvokeError(description=error_object.get("message"))
                     case TriggerInvokeError.__name__:
                         raise TriggerInvokeError(error_object.get("message"))
                     case EventIgnoreError.__name__:
-                        raise EventIgnoreError(description=error_object.get("description"))
+                        raise EventIgnoreError(description=error_object.get("message"))
                     case _:
                         raise PluginInvokeError(description=message)
             case PluginDaemonInternalServerError.__name__:

api/docker/entrypoint.sh

@@ -119,14 +119,16 @@ elif [[ "${MODE}" == "job" ]]; then
 
 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_socketio.py

@@ -0,0 +1,5 @@
+import socketio  # type: ignore[reportMissingTypeStubs]
+
+from configs import dify_config
+
+sio = socketio.Server(async_mode="gevent", cors_allowed_origins=dify_config.CONSOLE_CORS_ALLOW_ORIGINS)

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/migrations/versions/2026_01_17_1726-227822d22895_add_workflow_comments_table.py

@@ -0,0 +1,90 @@
+"""Add workflow comments table
+
+Revision ID: 227822d22895
+Revises: 288345cd01d1
+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 = '288345cd01d1'
+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('uuidv7()'), 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('uuidv7()'), 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('uuidv7()'), 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/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,
@@ -197,6 +202,9 @@ __all__ = [
     "Workflow",
     "WorkflowAppLog",
     "WorkflowAppLogCreatedFrom",
+    "WorkflowComment",
+    "WorkflowCommentMention",
+    "WorkflowCommentReply",
     "WorkflowNodeExecutionModel",
     "WorkflowNodeExecutionOffload",
     "WorkflowNodeExecutionTriggeredFrom",

api/models/comment.py

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

api/models/workflow.py

@@ -401,7 +401,7 @@ class Workflow(Base):  # bug
 
         :return: hash
         """
-        entity = {"graph": self.graph_dict, "features": self.features_dict}
+        entity = {"graph": self.graph_dict}
 
         return helper.generate_text_hash(json.dumps(entity, sort_keys=True))
 

api/pyproject.toml

@@ -21,6 +21,7 @@ dependencies = [
     "flask-orjson~=2.0.0",
     "flask-sqlalchemy~=3.1.1",
     "gevent~=25.9.1",
+    "gevent-websocket~=0.10.1",
     "gmpy2~=2.2.1",
     "google-api-core==2.18.0",
     "google-api-python-client==2.90.0",
@@ -72,6 +73,7 @@ dependencies = [
     "pypdfium2==5.2.0",
     "python-docx~=1.1.0",
     "python-dotenv==1.0.1",
+    "python-socketio~=5.13.0",
     "pyyaml~=6.0.1",
     "readabilipy~=0.3.0",
     "redis[hiredis]~=6.1.0",

api/repositories/workflow_collaboration_repository.py

@@ -0,0 +1,147 @@
+from __future__ import annotations
+
+import json
+from typing import TypedDict
+
+from extensions.ext_redis import redis_client
+
+SESSION_STATE_TTL_SECONDS = 3600
+WORKFLOW_ONLINE_USERS_PREFIX = "workflow_online_users:"
+WORKFLOW_LEADER_PREFIX = "workflow_leader:"
+WS_SID_MAP_PREFIX = "ws_sid_map:"
+
+
+class WorkflowSessionInfo(TypedDict):
+    user_id: str
+    username: str
+    avatar: str | None
+    sid: str
+    connected_at: int
+
+
+class SidMapping(TypedDict):
+    workflow_id: str
+    user_id: str
+
+
+class WorkflowCollaborationRepository:
+    def __init__(self) -> None:
+        self._redis = redis_client
+
+    def __repr__(self) -> str:
+        return f"{self.__class__.__name__}(redis_client={self._redis})"
+
+    @staticmethod
+    def workflow_key(workflow_id: str) -> str:
+        return f"{WORKFLOW_ONLINE_USERS_PREFIX}{workflow_id}"
+
+    @staticmethod
+    def leader_key(workflow_id: str) -> str:
+        return f"{WORKFLOW_LEADER_PREFIX}{workflow_id}"
+
+    @staticmethod
+    def sid_key(sid: str) -> str:
+        return f"{WS_SID_MAP_PREFIX}{sid}"
+
+    @staticmethod
+    def _decode(value: str | bytes | None) -> str | None:
+        if value is None:
+            return None
+        if isinstance(value, bytes):
+            return value.decode("utf-8")
+        return value
+
+    def refresh_session_state(self, workflow_id: str, sid: str) -> None:
+        workflow_key = self.workflow_key(workflow_id)
+        sid_key = self.sid_key(sid)
+        if self._redis.exists(workflow_key):
+            self._redis.expire(workflow_key, SESSION_STATE_TTL_SECONDS)
+        if self._redis.exists(sid_key):
+            self._redis.expire(sid_key, SESSION_STATE_TTL_SECONDS)
+
+    def set_session_info(self, workflow_id: str, session_info: WorkflowSessionInfo) -> None:
+        workflow_key = self.workflow_key(workflow_id)
+        self._redis.hset(workflow_key, session_info["sid"], json.dumps(session_info))
+        self._redis.set(
+            self.sid_key(session_info["sid"]),
+            json.dumps({"workflow_id": workflow_id, "user_id": session_info["user_id"]}),
+            ex=SESSION_STATE_TTL_SECONDS,
+        )
+        self.refresh_session_state(workflow_id, session_info["sid"])
+
+    def get_sid_mapping(self, sid: str) -> SidMapping | None:
+        raw = self._redis.get(self.sid_key(sid))
+        if not raw:
+            return None
+        value = self._decode(raw)
+        if not value:
+            return None
+        try:
+            return json.loads(value)
+        except (TypeError, json.JSONDecodeError):
+            return None
+
+    def delete_session(self, workflow_id: str, sid: str) -> None:
+        self._redis.hdel(self.workflow_key(workflow_id), sid)
+        self._redis.delete(self.sid_key(sid))
+
+    def session_exists(self, workflow_id: str, sid: str) -> bool:
+        return bool(self._redis.hexists(self.workflow_key(workflow_id), sid))
+
+    def sid_mapping_exists(self, sid: str) -> bool:
+        return bool(self._redis.exists(self.sid_key(sid)))
+
+    def get_session_sids(self, workflow_id: str) -> list[str]:
+        raw_sids = self._redis.hkeys(self.workflow_key(workflow_id))
+        decoded_sids: list[str] = []
+        for sid in raw_sids:
+            decoded = self._decode(sid)
+            if decoded:
+                decoded_sids.append(decoded)
+        return decoded_sids
+
+    def list_sessions(self, workflow_id: str) -> list[WorkflowSessionInfo]:
+        sessions_json = self._redis.hgetall(self.workflow_key(workflow_id))
+        users: list[WorkflowSessionInfo] = []
+
+        for session_info_json in sessions_json.values():
+            value = self._decode(session_info_json)
+            if not value:
+                continue
+            try:
+                session_info = json.loads(value)
+            except (TypeError, json.JSONDecodeError):
+                continue
+
+            if not isinstance(session_info, dict):
+                continue
+            if "user_id" not in session_info or "username" not in session_info or "sid" not in session_info:
+                continue
+
+            users.append(
+                {
+                    "user_id": str(session_info["user_id"]),
+                    "username": str(session_info["username"]),
+                    "avatar": session_info.get("avatar"),
+                    "sid": str(session_info["sid"]),
+                    "connected_at": int(session_info.get("connected_at") or 0),
+                }
+            )
+
+        return users
+
+    def get_current_leader(self, workflow_id: str) -> str | None:
+        raw = self._redis.get(self.leader_key(workflow_id))
+        return self._decode(raw)
+
+    def set_leader_if_absent(self, workflow_id: str, sid: str) -> bool:
+        return bool(self._redis.set(self.leader_key(workflow_id), sid, nx=True, ex=SESSION_STATE_TTL_SECONDS))
+
+    def set_leader(self, workflow_id: str, sid: str) -> None:
+        self._redis.set(self.leader_key(workflow_id), sid, ex=SESSION_STATE_TTL_SECONDS)
+
+    def delete_leader(self, workflow_id: str) -> None:
+        self._redis.delete(self.leader_key(workflow_id))
+
+    def expire_leader(self, workflow_id: str) -> None:
+        self._redis.expire(self.leader_key(workflow_id), SESSION_STATE_TTL_SECONDS)

api/services/dataset_service.py

@@ -13,10 +13,11 @@ import sqlalchemy as sa
 from redis.exceptions import LockNotOwnedError
 from sqlalchemy import exists, func, select
 from sqlalchemy.orm import Session
-from werkzeug.exceptions import NotFound
+from werkzeug.exceptions import Forbidden, NotFound
 
 from configs import dify_config
 from core.errors.error import LLMBadRequestError, ProviderTokenNotInitError
+from core.file import helpers as file_helpers
 from core.helper.name_generator import generate_incremental_name
 from core.model_manager import ModelManager
 from core.model_runtime.entities.model_entities import ModelFeature, ModelType
@@ -73,6 +74,7 @@ from services.errors.document import DocumentIndexingError
 from services.errors.file import FileNotExistsError
 from services.external_knowledge_service import ExternalDatasetService
 from services.feature_service import FeatureModel, FeatureService
+from services.file_service import FileService
 from services.rag_pipeline.rag_pipeline import RagPipelineService
 from services.tag_service import TagService
 from services.vector_service import VectorService
@@ -1162,6 +1164,7 @@ class DocumentService:
             Document.archived.is_(True),
         ),
     }
+    DOCUMENT_BATCH_DOWNLOAD_ZIP_FILENAME_EXTENSION = ".zip"
 
     @classmethod
     def normalize_display_status(cls, status: str | None) -> str | None:
@@ -1288,6 +1291,143 @@ class DocumentService:
         else:
             return None
 
+    @staticmethod
+    def get_documents_by_ids(dataset_id: str, document_ids: Sequence[str]) -> Sequence[Document]:
+        """Fetch documents for a dataset in a single batch query."""
+        if not document_ids:
+            return []
+        document_id_list: list[str] = [str(document_id) for document_id in document_ids]
+        # Fetch all requested documents in one query to avoid N+1 lookups.
+        documents: Sequence[Document] = db.session.scalars(
+            select(Document).where(
+                Document.dataset_id == dataset_id,
+                Document.id.in_(document_id_list),
+            )
+        ).all()
+        return documents
+
+    @staticmethod
+    def get_document_download_url(document: Document) -> str:
+        """
+        Return a signed download URL for an upload-file document.
+        """
+        upload_file = DocumentService._get_upload_file_for_upload_file_document(document)
+        return file_helpers.get_signed_file_url(upload_file_id=upload_file.id, as_attachment=True)
+
+    @staticmethod
+    def prepare_document_batch_download_zip(
+        *,
+        dataset_id: str,
+        document_ids: Sequence[str],
+        tenant_id: str,
+        current_user: Account,
+    ) -> tuple[list[UploadFile], str]:
+        """
+        Resolve upload files for batch ZIP downloads and generate a client-visible filename.
+        """
+        dataset = DatasetService.get_dataset(dataset_id)
+        if not dataset:
+            raise NotFound("Dataset not found.")
+        try:
+            DatasetService.check_dataset_permission(dataset, current_user)
+        except NoPermissionError as e:
+            raise Forbidden(str(e))
+
+        upload_files_by_document_id = DocumentService._get_upload_files_by_document_id_for_zip_download(
+            dataset_id=dataset_id,
+            document_ids=document_ids,
+            tenant_id=tenant_id,
+        )
+        upload_files = [upload_files_by_document_id[document_id] for document_id in document_ids]
+        download_name = DocumentService._generate_document_batch_download_zip_filename()
+        return upload_files, download_name
+
+    @staticmethod
+    def _generate_document_batch_download_zip_filename() -> str:
+        """
+        Generate a random attachment filename for the batch download ZIP.
+        """
+        return f"{uuid.uuid4().hex}{DocumentService.DOCUMENT_BATCH_DOWNLOAD_ZIP_FILENAME_EXTENSION}"
+
+    @staticmethod
+    def _get_upload_file_id_for_upload_file_document(
+        document: Document,
+        *,
+        invalid_source_message: str,
+        missing_file_message: str,
+    ) -> str:
+        """
+        Normalize and validate `Document -> UploadFile` linkage for download flows.
+        """
+        if document.data_source_type != "upload_file":
+            raise NotFound(invalid_source_message)
+
+        data_source_info: dict[str, Any] = document.data_source_info_dict or {}
+        upload_file_id: str | None = data_source_info.get("upload_file_id")
+        if not upload_file_id:
+            raise NotFound(missing_file_message)
+
+        return str(upload_file_id)
+
+    @staticmethod
+    def _get_upload_file_for_upload_file_document(document: Document) -> UploadFile:
+        """
+        Load the `UploadFile` row for an upload-file document.
+        """
+        upload_file_id = DocumentService._get_upload_file_id_for_upload_file_document(
+            document,
+            invalid_source_message="Document does not have an uploaded file to download.",
+            missing_file_message="Uploaded file not found.",
+        )
+        upload_files_by_id = FileService.get_upload_files_by_ids(document.tenant_id, [upload_file_id])
+        upload_file = upload_files_by_id.get(upload_file_id)
+        if not upload_file:
+            raise NotFound("Uploaded file not found.")
+        return upload_file
+
+    @staticmethod
+    def _get_upload_files_by_document_id_for_zip_download(
+        *,
+        dataset_id: str,
+        document_ids: Sequence[str],
+        tenant_id: str,
+    ) -> dict[str, UploadFile]:
+        """
+        Batch load upload files keyed by document id for ZIP downloads.
+        """
+        document_id_list: list[str] = [str(document_id) for document_id in document_ids]
+
+        documents = DocumentService.get_documents_by_ids(dataset_id, document_id_list)
+        documents_by_id: dict[str, Document] = {str(document.id): document for document in documents}
+
+        missing_document_ids: set[str] = set(document_id_list) - set(documents_by_id.keys())
+        if missing_document_ids:
+            raise NotFound("Document not found.")
+
+        upload_file_ids: list[str] = []
+        upload_file_ids_by_document_id: dict[str, str] = {}
+        for document_id, document in documents_by_id.items():
+            if document.tenant_id != tenant_id:
+                raise Forbidden("No permission.")
+
+            upload_file_id = DocumentService._get_upload_file_id_for_upload_file_document(
+                document,
+                invalid_source_message="Only uploaded-file documents can be downloaded as ZIP.",
+                missing_file_message="Only uploaded-file documents can be downloaded as ZIP.",
+            )
+            upload_file_ids.append(upload_file_id)
+            upload_file_ids_by_document_id[document_id] = upload_file_id
+
+        upload_files_by_id = FileService.get_upload_files_by_ids(tenant_id, upload_file_ids)
+        missing_upload_file_ids: set[str] = set(upload_file_ids) - set(upload_files_by_id.keys())
+        if missing_upload_file_ids:
+            raise NotFound("Only uploaded-file documents can be downloaded as ZIP.")
+
+        return {
+            document_id: upload_files_by_id[upload_file_id]
+            for document_id, upload_file_id in upload_file_ids_by_document_id.items()
+        }
+
     @staticmethod
     def get_document_by_id(document_id: str) -> Document | None:
         document = db.session.query(Document).where(Document.id == document_id).first()

api/services/feature_service.py

@@ -161,6 +161,7 @@ class SystemFeatureModel(BaseModel):
     enable_email_code_login: bool = False
     enable_email_password_login: bool = True
     enable_social_oauth_login: bool = False
+    enable_collaboration_mode: bool = False
     is_allow_register: bool = False
     is_allow_create_workspace: bool = False
     is_email_setup: bool = False
@@ -222,6 +223,7 @@ class FeatureService:
         system_features.enable_email_code_login = dify_config.ENABLE_EMAIL_CODE_LOGIN
         system_features.enable_email_password_login = dify_config.ENABLE_EMAIL_PASSWORD_LOGIN
         system_features.enable_social_oauth_login = dify_config.ENABLE_SOCIAL_OAUTH_LOGIN
+        system_features.enable_collaboration_mode = dify_config.ENABLE_COLLABORATION_MODE
         system_features.is_allow_register = dify_config.ALLOW_REGISTER
         system_features.is_allow_create_workspace = dify_config.ALLOW_CREATE_WORKSPACE
         system_features.is_email_setup = dify_config.MAIL_TYPE is not None and dify_config.MAIL_TYPE != ""

api/services/file_service.py

@@ -2,7 +2,11 @@ import base64
 import hashlib
 import os
 import uuid
+from collections.abc import Iterator, Sequence
+from contextlib import contextmanager, suppress
+from tempfile import NamedTemporaryFile
 from typing import Literal, Union
+from zipfile import ZIP_DEFLATED, ZipFile
 
 from sqlalchemy import Engine, select
 from sqlalchemy.orm import Session, sessionmaker
@@ -17,6 +21,7 @@ from constants import (
 )
 from core.file import helpers as file_helpers
 from core.rag.extractor.extract_processor import ExtractProcessor
+from extensions.ext_database import db
 from extensions.ext_storage import storage
 from libs.datetime_utils import naive_utc_now
 from libs.helper import extract_tenant_id
@@ -167,6 +172,9 @@ class FileService:
         return upload_file
 
     def get_file_preview(self, file_id: str):
+        """
+        Return a short text preview extracted from a document file.
+        """
         with self._session_maker(expire_on_commit=False) as session:
             upload_file = session.query(UploadFile).where(UploadFile.id == file_id).first()
 
@@ -253,3 +261,101 @@ class FileService:
                 return
             storage.delete(upload_file.key)
             session.delete(upload_file)
+
+    @staticmethod
+    def get_upload_files_by_ids(tenant_id: str, upload_file_ids: Sequence[str]) -> dict[str, UploadFile]:
+        """
+        Fetch `UploadFile` rows for a tenant in a single batch query.
+
+        This is a generic `UploadFile` lookup helper (not dataset/document specific), so it lives in `FileService`.
+        """
+        if not upload_file_ids:
+            return {}
+
+        # Normalize and deduplicate ids before using them in the IN clause.
+        upload_file_id_list: list[str] = [str(upload_file_id) for upload_file_id in upload_file_ids]
+        unique_upload_file_ids: list[str] = list(set(upload_file_id_list))
+
+        # Fetch upload files in one query for efficient batch access.
+        upload_files: Sequence[UploadFile] = db.session.scalars(
+            select(UploadFile).where(
+                UploadFile.tenant_id == tenant_id,
+                UploadFile.id.in_(unique_upload_file_ids),
+            )
+        ).all()
+        return {str(upload_file.id): upload_file for upload_file in upload_files}
+
+    @staticmethod
+    def _sanitize_zip_entry_name(name: str) -> str:
+        """
+        Sanitize a ZIP entry name to avoid path traversal and weird separators.
+
+        We keep this conservative: the upload flow already rejects `/` and `\\`, but older rows (or imported data)
+        could still contain unsafe names.
+        """
+        # Drop any directory components and prevent empty names.
+        base = os.path.basename(name).strip() or "file"
+
+        # ZIP uses forward slashes as separators; remove any residual separator characters.
+        return base.replace("/", "_").replace("\\", "_")
+
+    @staticmethod
+    def _dedupe_zip_entry_name(original_name: str, used_names: set[str]) -> str:
+        """
+        Return a unique ZIP entry name, inserting suffixes before the extension.
+        """
+        # Keep the original name when it's not already used.
+        if original_name not in used_names:
+            return original_name
+
+        # Insert suffixes before the extension (e.g., "doc.txt" -> "doc (1).txt").
+        stem, extension = os.path.splitext(original_name)
+        suffix = 1
+        while True:
+            candidate = f"{stem} ({suffix}){extension}"
+            if candidate not in used_names:
+                return candidate
+            suffix += 1
+
+    @staticmethod
+    @contextmanager
+    def build_upload_files_zip_tempfile(
+        *,
+        upload_files: Sequence[UploadFile],
+    ) -> Iterator[str]:
+        """
+        Build a ZIP from `UploadFile`s and yield a tempfile path.
+
+        We yield a path (rather than an open file handle) to avoid "read of closed file" issues when Flask/Werkzeug
+        streams responses. The caller is expected to keep this context open until the response is fully sent, then
+        close it (e.g., via `response.call_on_close(...)`) to delete the tempfile.
+        """
+        used_names: set[str] = set()
+
+        # Build a ZIP in a temp file and keep it on disk until the caller finishes streaming it.
+        tmp_path: str | None = None
+        try:
+            with NamedTemporaryFile(mode="w+b", suffix=".zip", delete=False) as tmp:
+                tmp_path = tmp.name
+                with ZipFile(tmp, mode="w", compression=ZIP_DEFLATED) as zf:
+                    for upload_file in upload_files:
+                        # Ensure the entry name is safe and unique.
+                        safe_name = FileService._sanitize_zip_entry_name(upload_file.name)
+                        arcname = FileService._dedupe_zip_entry_name(safe_name, used_names)
+                        used_names.add(arcname)
+
+                        # Stream file bytes from storage into the ZIP entry.
+                        with zf.open(arcname, "w") as entry:
+                            for chunk in storage.load(upload_file.key, stream=True):
+                                entry.write(chunk)
+
+                # Flush so `send_file(path, ...)` can re-open it safely on all platforms.
+                tmp.flush()
+
+            assert tmp_path is not None
+            yield tmp_path
+        finally:
+            # Remove the temp file when the context is closed (typically after the response finishes streaming).
+            if tmp_path is not None:
+                with suppress(FileNotFoundError):
+                    os.remove(tmp_path)

api/services/workflow_collaboration_service.py

@@ -0,0 +1,196 @@
+from __future__ import annotations
+
+import logging
+import time
+from collections.abc import Mapping
+
+from models.account import Account
+from repositories.workflow_collaboration_repository import WorkflowCollaborationRepository, WorkflowSessionInfo
+
+
+class WorkflowCollaborationService:
+    def __init__(self, repository: WorkflowCollaborationRepository, socketio) -> None:
+        self._repository = repository
+        self._socketio = socketio
+
+    def __repr__(self) -> str:
+        return f"{self.__class__.__name__}(repository={self._repository})"
+
+    def save_session(self, sid: str, user: Account) -> None:
+        self._socketio.save_session(
+            sid,
+            {
+                "user_id": user.id,
+                "username": user.name,
+                "avatar": user.avatar,
+            },
+        )
+
+    def register_session(self, workflow_id: str, sid: str) -> tuple[str, bool] | None:
+        session = self._socketio.get_session(sid)
+        user_id = session.get("user_id")
+        if not user_id:
+            return None
+
+        session_info: WorkflowSessionInfo = {
+            "user_id": str(user_id),
+            "username": str(session.get("username", "Unknown")),
+            "avatar": session.get("avatar"),
+            "sid": sid,
+            "connected_at": int(time.time()),
+        }
+
+        self._repository.set_session_info(workflow_id, session_info)
+
+        leader_sid = self.get_or_set_leader(workflow_id, sid)
+        is_leader = leader_sid == sid
+
+        self._socketio.enter_room(sid, workflow_id)
+        self.broadcast_online_users(workflow_id)
+
+        self._socketio.emit("status", {"isLeader": is_leader}, room=sid)
+
+        return str(user_id), is_leader
+
+    def disconnect_session(self, sid: str) -> None:
+        mapping = self._repository.get_sid_mapping(sid)
+        if not mapping:
+            return
+
+        workflow_id = mapping["workflow_id"]
+        self._repository.delete_session(workflow_id, sid)
+
+        self.handle_leader_disconnect(workflow_id, sid)
+        self.broadcast_online_users(workflow_id)
+
+    def relay_collaboration_event(self, sid: str, data: Mapping[str, object]) -> tuple[dict[str, str], int]:
+        mapping = self._repository.get_sid_mapping(sid)
+        if not mapping:
+            return {"msg": "unauthorized"}, 401
+
+        workflow_id = mapping["workflow_id"]
+        user_id = mapping["user_id"]
+        self.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
+
+        self._socketio.emit(
+            "collaboration_update",
+            {"type": event_type, "userId": user_id, "data": event_data, "timestamp": timestamp},
+            room=workflow_id,
+            skip_sid=sid,
+        )
+
+        return {"msg": "event_broadcasted"}, 200
+
+    def relay_graph_event(self, sid: str, data: object) -> tuple[dict[str, str], int]:
+        mapping = self._repository.get_sid_mapping(sid)
+        if not mapping:
+            return {"msg": "unauthorized"}, 401
+
+        workflow_id = mapping["workflow_id"]
+        self.refresh_session_state(workflow_id, sid)
+
+        self._socketio.emit("graph_update", data, room=workflow_id, skip_sid=sid)
+
+        return {"msg": "graph_update_broadcasted"}, 200
+
+    def get_or_set_leader(self, workflow_id: str, sid: str) -> str:
+        current_leader = self._repository.get_current_leader(workflow_id)
+
+        if current_leader:
+            if self.is_session_active(workflow_id, current_leader):
+                return current_leader
+            self._repository.delete_session(workflow_id, current_leader)
+            self._repository.delete_leader(workflow_id)
+
+        was_set = self._repository.set_leader_if_absent(workflow_id, sid)
+
+        if was_set:
+            if current_leader:
+                self.broadcast_leader_change(workflow_id, sid)
+            return sid
+
+        current_leader = self._repository.get_current_leader(workflow_id)
+        if current_leader:
+            return current_leader
+
+        return sid
+
+    def handle_leader_disconnect(self, workflow_id: str, disconnected_sid: str) -> None:
+        current_leader = self._repository.get_current_leader(workflow_id)
+        if not current_leader:
+            return
+
+        if current_leader != disconnected_sid:
+            return
+
+        session_sids = self._repository.get_session_sids(workflow_id)
+        if session_sids:
+            new_leader_sid = session_sids[0]
+            self._repository.set_leader(workflow_id, new_leader_sid)
+            self.broadcast_leader_change(workflow_id, new_leader_sid)
+        else:
+            self._repository.delete_leader(workflow_id)
+
+    def broadcast_leader_change(self, workflow_id: str, new_leader_sid: str) -> None:
+        for sid in self._repository.get_session_sids(workflow_id):
+            try:
+                is_leader = sid == new_leader_sid
+                self._socketio.emit("status", {"isLeader": is_leader}, room=sid)
+            except Exception:
+                logging.exception("Failed to emit leader status to session %s", sid)
+
+    def get_current_leader(self, workflow_id: str) -> str | None:
+        return self._repository.get_current_leader(workflow_id)
+
+    def broadcast_online_users(self, workflow_id: str) -> None:
+        users = self._repository.list_sessions(workflow_id)
+        users.sort(key=lambda x: x.get("connected_at") or 0)
+
+        leader_sid = self.get_current_leader(workflow_id)
+
+        self._socketio.emit(
+            "online_users",
+            {"workflow_id": workflow_id, "users": users, "leader": leader_sid},
+            room=workflow_id,
+        )
+
+    def refresh_session_state(self, workflow_id: str, sid: str) -> None:
+        self._repository.refresh_session_state(workflow_id, sid)
+        self._ensure_leader(workflow_id, sid)
+
+    def _ensure_leader(self, workflow_id: str, sid: str) -> None:
+        current_leader = self._repository.get_current_leader(workflow_id)
+        if current_leader and self.is_session_active(workflow_id, current_leader):
+            self._repository.expire_leader(workflow_id)
+            return
+
+        if current_leader:
+            self._repository.delete_leader(workflow_id)
+
+        self._repository.set_leader(workflow_id, sid)
+        self.broadcast_leader_change(workflow_id, sid)
+
+    def is_session_active(self, workflow_id: str, sid: str) -> bool:
+        if not sid:
+            return False
+
+        try:
+            if not self._socketio.manager.is_connected(sid, "/"):
+                return False
+        except AttributeError:
+            return False
+
+        if not self._repository.session_exists(workflow_id, sid):
+            return False
+
+        if not self._repository.sid_mapping_exists(sid):
+            return False
+
+        return True

api/services/workflow_comment_service.py

@@ -0,0 +1,345 @@
+import logging
+from collections.abc import Sequence
+
+from sqlalchemy import desc, select
+from sqlalchemy.orm import Session, selectinload
+from werkzeug.exceptions import Forbidden, NotFound
+
+from extensions.ext_database import db
+from libs.datetime_utils import naive_utc_now
+from libs.helper import uuid_value
+from models import WorkflowComment, WorkflowCommentMention, WorkflowCommentReply
+from models.account import Account
+
+logger = logging.getLogger(__name__)
+
+
+class WorkflowCommentService:
+    """Service for managing workflow comments."""
+
+    @staticmethod
+    def _validate_content(content: str) -> None:
+        if len(content.strip()) == 0:
+            raise ValueError("Comment content cannot be empty")
+
+        if len(content) > 1000:
+            raise ValueError("Comment content cannot exceed 1000 characters")
+
+    @staticmethod
+    def get_comments(tenant_id: str, app_id: str) -> Sequence[WorkflowComment]:
+        """Get all comments for a workflow."""
+        with Session(db.engine) as session:
+            # Get all comments with eager loading
+            stmt = (
+                select(WorkflowComment)
+                .options(selectinload(WorkflowComment.replies), selectinload(WorkflowComment.mentions))
+                .where(WorkflowComment.tenant_id == tenant_id, WorkflowComment.app_id == app_id)
+                .order_by(desc(WorkflowComment.created_at))
+            )
+
+            comments = session.scalars(stmt).all()
+
+            # Batch preload all Account objects to avoid N+1 queries
+            WorkflowCommentService._preload_accounts(session, comments)
+
+            return comments
+
+    @staticmethod
+    def _preload_accounts(session: Session, comments: Sequence[WorkflowComment]) -> None:
+        """Batch preload Account objects for comments, replies, and mentions."""
+        # Collect all user IDs
+        user_ids: set[str] = set()
+        for comment in comments:
+            user_ids.add(comment.created_by)
+            if comment.resolved_by:
+                user_ids.add(comment.resolved_by)
+            user_ids.update(reply.created_by for reply in comment.replies)
+            user_ids.update(mention.mentioned_user_id for mention in comment.mentions)
+
+        if not user_ids:
+            return
+
+        # Batch query all accounts
+        accounts = session.scalars(select(Account).where(Account.id.in_(user_ids))).all()
+        account_map = {str(account.id): account for account in accounts}
+
+        # Cache accounts on objects
+        for comment in comments:
+            comment.cache_created_by_account(account_map.get(comment.created_by))
+            comment.cache_resolved_by_account(account_map.get(comment.resolved_by) if comment.resolved_by else None)
+            for reply in comment.replies:
+                reply.cache_created_by_account(account_map.get(reply.created_by))
+            for mention in comment.mentions:
+                mention.cache_mentioned_user_account(account_map.get(mention.mentioned_user_id))
+
+    @staticmethod
+    def get_comment(tenant_id: str, app_id: str, comment_id: str, session: Session | None = None) -> WorkflowComment:
+        """Get a specific comment."""
+
+        def _get_comment(session: Session) -> WorkflowComment:
+            stmt = (
+                select(WorkflowComment)
+                .options(selectinload(WorkflowComment.replies), selectinload(WorkflowComment.mentions))
+                .where(
+                    WorkflowComment.id == comment_id,
+                    WorkflowComment.tenant_id == tenant_id,
+                    WorkflowComment.app_id == app_id,
+                )
+            )
+            comment = session.scalar(stmt)
+
+            if not comment:
+                raise NotFound("Comment not found")
+
+            # Preload accounts to avoid N+1 queries
+            WorkflowCommentService._preload_accounts(session, [comment])
+
+            return comment
+
+        if session is not None:
+            return _get_comment(session)
+        else:
+            with Session(db.engine, expire_on_commit=False) as session:
+                return _get_comment(session)
+
+    @staticmethod
+    def create_comment(
+        tenant_id: str,
+        app_id: str,
+        created_by: str,
+        content: str,
+        position_x: float,
+        position_y: float,
+        mentioned_user_ids: list[str] | None = None,
+    ) -> dict:
+        """Create a new workflow comment."""
+        WorkflowCommentService._validate_content(content)
+
+        with Session(db.engine) as session:
+            comment = WorkflowComment(
+                tenant_id=tenant_id,
+                app_id=app_id,
+                position_x=position_x,
+                position_y=position_y,
+                content=content,
+                created_by=created_by,
+            )
+
+            session.add(comment)
+            session.flush()  # Get the comment ID for mentions
+
+            # Create mentions if specified
+            mentioned_user_ids = mentioned_user_ids or []
+            for user_id in mentioned_user_ids:
+                if isinstance(user_id, str) and uuid_value(user_id):
+                    mention = WorkflowCommentMention(
+                        comment_id=comment.id,
+                        reply_id=None,  # This is a comment mention, not reply mention
+                        mentioned_user_id=user_id,
+                    )
+                    session.add(mention)
+
+            session.commit()
+
+            # Return only what we need - id and created_at
+            return {"id": comment.id, "created_at": comment.created_at}
+
+    @staticmethod
+    def update_comment(
+        tenant_id: str,
+        app_id: str,
+        comment_id: str,
+        user_id: str,
+        content: str,
+        position_x: float | None = None,
+        position_y: float | None = None,
+        mentioned_user_ids: list[str] | None = None,
+    ) -> dict:
+        """Update a workflow comment."""
+        WorkflowCommentService._validate_content(content)
+
+        with Session(db.engine, expire_on_commit=False) as session:
+            # Get comment with validation
+            stmt = select(WorkflowComment).where(
+                WorkflowComment.id == comment_id,
+                WorkflowComment.tenant_id == tenant_id,
+                WorkflowComment.app_id == app_id,
+            )
+            comment = session.scalar(stmt)
+
+            if not comment:
+                raise NotFound("Comment not found")
+
+            # Only the creator can update the comment
+            if comment.created_by != user_id:
+                raise Forbidden("Only the comment creator can update it")
+
+            # Update comment fields
+            comment.content = content
+            if position_x is not None:
+                comment.position_x = position_x
+            if position_y is not None:
+                comment.position_y = position_y
+
+            # Update mentions - first remove existing mentions for this comment only (not replies)
+            existing_mentions = session.scalars(
+                select(WorkflowCommentMention).where(
+                    WorkflowCommentMention.comment_id == comment.id,
+                    WorkflowCommentMention.reply_id.is_(None),  # Only comment mentions, not reply mentions
+                )
+            ).all()
+            for mention in existing_mentions:
+                session.delete(mention)
+
+            # Add new mentions
+            mentioned_user_ids = mentioned_user_ids or []
+            for user_id_str in mentioned_user_ids:
+                if isinstance(user_id_str, str) and uuid_value(user_id_str):
+                    mention = WorkflowCommentMention(
+                        comment_id=comment.id,
+                        reply_id=None,  # This is a comment mention
+                        mentioned_user_id=user_id_str,
+                    )
+                    session.add(mention)
+
+            session.commit()
+
+            return {"id": comment.id, "updated_at": comment.updated_at}
+
+    @staticmethod
+    def delete_comment(tenant_id: str, app_id: str, comment_id: str, user_id: str) -> None:
+        """Delete a workflow comment."""
+        with Session(db.engine, expire_on_commit=False) as session:
+            comment = WorkflowCommentService.get_comment(tenant_id, app_id, comment_id, session)
+
+            # Only the creator can delete the comment
+            if comment.created_by != user_id:
+                raise Forbidden("Only the comment creator can delete it")
+
+            # Delete associated mentions (both comment and reply mentions)
+            mentions = session.scalars(
+                select(WorkflowCommentMention).where(WorkflowCommentMention.comment_id == comment_id)
+            ).all()
+            for mention in mentions:
+                session.delete(mention)
+
+            # Delete associated replies
+            replies = session.scalars(
+                select(WorkflowCommentReply).where(WorkflowCommentReply.comment_id == comment_id)
+            ).all()
+            for reply in replies:
+                session.delete(reply)
+
+            session.delete(comment)
+            session.commit()
+
+    @staticmethod
+    def resolve_comment(tenant_id: str, app_id: str, comment_id: str, user_id: str) -> WorkflowComment:
+        """Resolve a workflow comment."""
+        with Session(db.engine, expire_on_commit=False) as session:
+            comment = WorkflowCommentService.get_comment(tenant_id, app_id, comment_id, session)
+            if comment.resolved:
+                return comment
+
+            comment.resolved = True
+            comment.resolved_at = naive_utc_now()
+            comment.resolved_by = user_id
+            session.commit()
+
+        return comment
+
+    @staticmethod
+    def create_reply(
+        comment_id: str, content: str, created_by: str, mentioned_user_ids: list[str] | None = None
+    ) -> dict:
+        """Add a reply to a workflow comment."""
+        WorkflowCommentService._validate_content(content)
+
+        with Session(db.engine, expire_on_commit=False) as session:
+            # Check if comment exists
+            comment = session.get(WorkflowComment, comment_id)
+            if not comment:
+                raise NotFound("Comment not found")
+
+            reply = WorkflowCommentReply(comment_id=comment_id, content=content, created_by=created_by)
+
+            session.add(reply)
+            session.flush()  # Get the reply ID for mentions
+
+            # Create mentions if specified
+            mentioned_user_ids = mentioned_user_ids or []
+            for user_id in mentioned_user_ids:
+                if isinstance(user_id, str) and uuid_value(user_id):
+                    # Create mention linking to specific reply
+                    mention = WorkflowCommentMention(
+                        comment_id=comment_id, reply_id=reply.id, mentioned_user_id=user_id
+                    )
+                    session.add(mention)
+
+            session.commit()
+
+            return {"id": reply.id, "created_at": reply.created_at}
+
+    @staticmethod
+    def update_reply(reply_id: str, user_id: str, content: str, mentioned_user_ids: list[str] | None = None) -> dict:
+        """Update a comment reply."""
+        WorkflowCommentService._validate_content(content)
+
+        with Session(db.engine, expire_on_commit=False) as session:
+            reply = session.get(WorkflowCommentReply, reply_id)
+            if not reply:
+                raise NotFound("Reply not found")
+
+            # Only the creator can update the reply
+            if reply.created_by != user_id:
+                raise Forbidden("Only the reply creator can update it")
+
+            reply.content = content
+
+            # Update mentions - first remove existing mentions for this reply
+            existing_mentions = session.scalars(
+                select(WorkflowCommentMention).where(WorkflowCommentMention.reply_id == reply.id)
+            ).all()
+            for mention in existing_mentions:
+                session.delete(mention)
+
+            # Add mentions
+            mentioned_user_ids = mentioned_user_ids or []
+            for user_id_str in mentioned_user_ids:
+                if isinstance(user_id_str, str) and uuid_value(user_id_str):
+                    mention = WorkflowCommentMention(
+                        comment_id=reply.comment_id, reply_id=reply.id, mentioned_user_id=user_id_str
+                    )
+                    session.add(mention)
+
+            session.commit()
+            session.refresh(reply)  # Refresh to get updated timestamp
+
+            return {"id": reply.id, "updated_at": reply.updated_at}
+
+    @staticmethod
+    def delete_reply(reply_id: str, user_id: str) -> None:
+        """Delete a comment reply."""
+        with Session(db.engine, expire_on_commit=False) as session:
+            reply = session.get(WorkflowCommentReply, reply_id)
+            if not reply:
+                raise NotFound("Reply not found")
+
+            # Only the creator can delete the reply
+            if reply.created_by != user_id:
+                raise Forbidden("Only the reply creator can delete it")
+
+            # Delete associated mentions first
+            mentions = session.scalars(
+                select(WorkflowCommentMention).where(WorkflowCommentMention.reply_id == reply_id)
+            ).all()
+            for mention in mentions:
+                session.delete(mention)
+
+            session.delete(reply)
+            session.commit()
+
+    @staticmethod
+    def validate_comment_access(comment_id: str, tenant_id: str, app_id: str) -> WorkflowComment:
+        """Validate that a comment belongs to the specified tenant and app."""
+        return WorkflowCommentService.get_comment(tenant_id, app_id, comment_id)

api/services/workflow_service.py

@@ -249,6 +249,78 @@ class WorkflowService:
         # return draft workflow
         return workflow
 
+    def update_draft_workflow_environment_variables(
+        self,
+        *,
+        app_model: App,
+        environment_variables: Sequence[VariableBase],
+        account: Account,
+    ):
+        """
+        Update draft workflow environment variables
+        """
+        # fetch draft workflow by app_model
+        workflow = self.get_draft_workflow(app_model=app_model)
+
+        if not workflow:
+            raise ValueError("No draft workflow found.")
+
+        workflow.environment_variables = environment_variables
+        workflow.updated_by = account.id
+        workflow.updated_at = naive_utc_now()
+
+        # commit db session changes
+        db.session.commit()
+
+    def update_draft_workflow_conversation_variables(
+        self,
+        *,
+        app_model: App,
+        conversation_variables: Sequence[VariableBase],
+        account: Account,
+    ):
+        """
+        Update draft workflow conversation variables
+        """
+        # fetch draft workflow by app_model
+        workflow = self.get_draft_workflow(app_model=app_model)
+
+        if not workflow:
+            raise ValueError("No draft workflow found.")
+
+        workflow.conversation_variables = conversation_variables
+        workflow.updated_by = account.id
+        workflow.updated_at = naive_utc_now()
+
+        # commit db session changes
+        db.session.commit()
+
+    def update_draft_workflow_features(
+        self,
+        *,
+        app_model: App,
+        features: dict,
+        account: Account,
+    ):
+        """
+        Update draft workflow features
+        """
+        # fetch draft workflow by app_model
+        workflow = self.get_draft_workflow(app_model=app_model)
+
+        if not workflow:
+            raise ValueError("No draft workflow found.")
+
+        # validate features structure
+        self.validate_features_structure(app_model=app_model, features=features)
+
+        workflow.features = json.dumps(features)
+        workflow.updated_by = account.id
+        workflow.updated_at = naive_utc_now()
+
+        # commit db session changes
+        db.session.commit()
+
     def publish_workflow(
         self,
         *,

api/tests/integration_tests/conftest.py

@@ -38,7 +38,7 @@ os.environ["OPENDAL_FS_ROOT"] = "/tmp/dify-storage"
 os.environ.setdefault("STORAGE_TYPE", "opendal")
 os.environ.setdefault("OPENDAL_SCHEME", "fs")
 
-_CACHED_APP = create_app()
+_SIO_APP, _CACHED_APP = create_app()
 
 
 @pytest.fixture

api/tests/test_containers_integration_tests/conftest.py

@@ -364,7 +364,7 @@ def _create_app_with_containers() -> Flask:
 
     # Create and configure the Flask application
     logger.info("Initializing Flask application...")
-    app = create_app()
+    sio_app, app = create_app()
     logger.info("Flask application created successfully")
 
     # Initialize database schema

api/tests/test_containers_integration_tests/services/test_feature_service.py

@@ -268,6 +268,7 @@ class TestFeatureService:
             mock_config.ENABLE_EMAIL_CODE_LOGIN = True
             mock_config.ENABLE_EMAIL_PASSWORD_LOGIN = True
             mock_config.ENABLE_SOCIAL_OAUTH_LOGIN = False
+            mock_config.ENABLE_COLLABORATION_MODE = True
             mock_config.ALLOW_REGISTER = False
             mock_config.ALLOW_CREATE_WORKSPACE = False
             mock_config.MAIL_TYPE = "smtp"
@@ -292,6 +293,7 @@ class TestFeatureService:
         # Verify authentication settings
         assert result.enable_email_code_login is True
         assert result.enable_email_password_login is False
+        assert result.enable_collaboration_mode is True
         assert result.is_allow_register is False
         assert result.is_allow_create_workspace is False
 
@@ -341,6 +343,7 @@ class TestFeatureService:
             mock_config.ENABLE_EMAIL_CODE_LOGIN = True
             mock_config.ENABLE_EMAIL_PASSWORD_LOGIN = True
             mock_config.ENABLE_SOCIAL_OAUTH_LOGIN = False
+            mock_config.ENABLE_COLLABORATION_MODE = False
             mock_config.ALLOW_REGISTER = True
             mock_config.ALLOW_CREATE_WORKSPACE = True
             mock_config.MAIL_TYPE = "smtp"
@@ -362,6 +365,7 @@ class TestFeatureService:
             assert result.enable_email_code_login is True
             assert result.enable_email_password_login is True
             assert result.enable_social_oauth_login is False
+            assert result.enable_collaboration_mode is False
             assert result.is_allow_register is True
             assert result.is_allow_create_workspace is True
             assert result.is_email_setup is True

api/tests/unit_tests/controllers/console/app/test_xss_prevention.py

@@ -1,254 +0,0 @@
-"""
-Unit tests for XSS prevention in App payloads.
-
-This test module validates that HTML tags, JavaScript, and other potentially
-dangerous content are rejected in App names and descriptions.
-"""
-
-import pytest
-
-from controllers.console.app.app import CopyAppPayload, CreateAppPayload, UpdateAppPayload
-
-
-class TestXSSPreventionUnit:
-    """Unit tests for XSS prevention in App payloads."""
-
-    def test_create_app_valid_names(self):
-        """Test CreateAppPayload with valid app names."""
-        # Normal app names should be valid
-        valid_names = [
-            "My App",
-            "Test App 123",
-            "App with - dash",
-            "App with _ underscore",
-            "App with + plus",
-            "App with () parentheses",
-            "App with [] brackets",
-            "App with {} braces",
-            "App with ! exclamation",
-            "App with @ at",
-            "App with # hash",
-            "App with $ dollar",
-            "App with % percent",
-            "App with ^ caret",
-            "App with & ampersand",
-            "App with * asterisk",
-            "Unicode: 测试应用",
-            "Emoji: 🤖",
-            "Mixed: Test 测试 123",
-        ]
-
-        for name in valid_names:
-            payload = CreateAppPayload(
-                name=name,
-                mode="chat",
-            )
-            assert payload.name == name
-
-    def test_create_app_xss_script_tags(self):
-        """Test CreateAppPayload rejects script tags."""
-        xss_payloads = [
-            "<script>alert(document.cookie)</script>",
-            "<Script>alert(1)</Script>",
-            "<SCRIPT>alert('XSS')</SCRIPT>",
-            "<script>alert(String.fromCharCode(88,83,83))</script>",
-            "<script src='evil.js'></script>",
-            "<script>document.location='http://evil.com'</script>",
-        ]
-
-        for name in xss_payloads:
-            with pytest.raises(ValueError) as exc_info:
-                CreateAppPayload(name=name, mode="chat")
-            assert "invalid characters or patterns" in str(exc_info.value).lower()
-
-    def test_create_app_xss_iframe_tags(self):
-        """Test CreateAppPayload rejects iframe tags."""
-        xss_payloads = [
-            "<iframe src='evil.com'></iframe>",
-            "<Iframe srcdoc='<script>alert(1)</script>'></iframe>",
-            "<IFRAME src='javascript:alert(1)'></iframe>",
-        ]
-
-        for name in xss_payloads:
-            with pytest.raises(ValueError) as exc_info:
-                CreateAppPayload(name=name, mode="chat")
-            assert "invalid characters or patterns" in str(exc_info.value).lower()
-
-    def test_create_app_xss_javascript_protocol(self):
-        """Test CreateAppPayload rejects javascript: protocol."""
-        xss_payloads = [
-            "javascript:alert(1)",
-            "JAVASCRIPT:alert(1)",
-            "JavaScript:alert(document.cookie)",
-            "javascript:void(0)",
-            "javascript://comment%0Aalert(1)",
-        ]
-
-        for name in xss_payloads:
-            with pytest.raises(ValueError) as exc_info:
-                CreateAppPayload(name=name, mode="chat")
-            assert "invalid characters or patterns" in str(exc_info.value).lower()
-
-    def test_create_app_xss_svg_onload(self):
-        """Test CreateAppPayload rejects SVG with onload."""
-        xss_payloads = [
-            "<svg onload=alert(1)>",
-            "<SVG ONLOAD=alert(1)>",
-            "<svg/x/onload=alert(1)>",
-        ]
-
-        for name in xss_payloads:
-            with pytest.raises(ValueError) as exc_info:
-                CreateAppPayload(name=name, mode="chat")
-            assert "invalid characters or patterns" in str(exc_info.value).lower()
-
-    def test_create_app_xss_event_handlers(self):
-        """Test CreateAppPayload rejects HTML event handlers."""
-        xss_payloads = [
-            "<div onclick=alert(1)>",
-            "<img onerror=alert(1)>",
-            "<body onload=alert(1)>",
-            "<input onfocus=alert(1)>",
-            "<a onmouseover=alert(1)>",
-            "<DIV ONCLICK=alert(1)>",
-            "<img src=x onerror=alert(1)>",
-        ]
-
-        for name in xss_payloads:
-            with pytest.raises(ValueError) as exc_info:
-                CreateAppPayload(name=name, mode="chat")
-            assert "invalid characters or patterns" in str(exc_info.value).lower()
-
-    def test_create_app_xss_object_embed(self):
-        """Test CreateAppPayload rejects object and embed tags."""
-        xss_payloads = [
-            "<object data='evil.swf'></object>",
-            "<embed src='evil.swf'>",
-            "<OBJECT data='javascript:alert(1)'></OBJECT>",
-        ]
-
-        for name in xss_payloads:
-            with pytest.raises(ValueError) as exc_info:
-                CreateAppPayload(name=name, mode="chat")
-            assert "invalid characters or patterns" in str(exc_info.value).lower()
-
-    def test_create_app_xss_link_javascript(self):
-        """Test CreateAppPayload rejects link tags with javascript."""
-        xss_payloads = [
-            "<link href='javascript:alert(1)'>",
-            "<LINK HREF='javascript:alert(1)'>",
-        ]
-
-        for name in xss_payloads:
-            with pytest.raises(ValueError) as exc_info:
-                CreateAppPayload(name=name, mode="chat")
-            assert "invalid characters or patterns" in str(exc_info.value).lower()
-
-    def test_create_app_xss_in_description(self):
-        """Test CreateAppPayload rejects XSS in description."""
-        xss_descriptions = [
-            "<script>alert(1)</script>",
-            "javascript:alert(1)",
-            "<img onerror=alert(1)>",
-        ]
-
-        for description in xss_descriptions:
-            with pytest.raises(ValueError) as exc_info:
-                CreateAppPayload(
-                    name="Valid Name",
-                    mode="chat",
-                    description=description,
-                )
-            assert "invalid characters or patterns" in str(exc_info.value).lower()
-
-    def test_create_app_valid_descriptions(self):
-        """Test CreateAppPayload with valid descriptions."""
-        valid_descriptions = [
-            "A simple description",
-            "Description with < and > symbols",
-            "Description with & ampersand",
-            "Description with 'quotes' and \"double quotes\"",
-            "Description with / slashes",
-            "Description with \\ backslashes",
-            "Description with ; semicolons",
-            "Unicode: 这是一个描述",
-            "Emoji: 🎉🚀",
-        ]
-
-        for description in valid_descriptions:
-            payload = CreateAppPayload(
-                name="Valid App Name",
-                mode="chat",
-                description=description,
-            )
-            assert payload.description == description
-
-    def test_create_app_none_description(self):
-        """Test CreateAppPayload with None description."""
-        payload = CreateAppPayload(
-            name="Valid App Name",
-            mode="chat",
-            description=None,
-        )
-        assert payload.description is None
-
-    def test_update_app_xss_prevention(self):
-        """Test UpdateAppPayload also prevents XSS."""
-        xss_names = [
-            "<script>alert(1)</script>",
-            "javascript:alert(1)",
-            "<img onerror=alert(1)>",
-        ]
-
-        for name in xss_names:
-            with pytest.raises(ValueError) as exc_info:
-                UpdateAppPayload(name=name)
-            assert "invalid characters or patterns" in str(exc_info.value).lower()
-
-    def test_update_app_valid_names(self):
-        """Test UpdateAppPayload with valid names."""
-        payload = UpdateAppPayload(name="Valid Updated Name")
-        assert payload.name == "Valid Updated Name"
-
-    def test_copy_app_xss_prevention(self):
-        """Test CopyAppPayload also prevents XSS."""
-        xss_names = [
-            "<script>alert(1)</script>",
-            "javascript:alert(1)",
-            "<img onerror=alert(1)>",
-        ]
-
-        for name in xss_names:
-            with pytest.raises(ValueError) as exc_info:
-                CopyAppPayload(name=name)
-            assert "invalid characters or patterns" in str(exc_info.value).lower()
-
-    def test_copy_app_valid_names(self):
-        """Test CopyAppPayload with valid names."""
-        payload = CopyAppPayload(name="Valid Copy Name")
-        assert payload.name == "Valid Copy Name"
-
-    def test_copy_app_none_name(self):
-        """Test CopyAppPayload with None name (should be allowed)."""
-        payload = CopyAppPayload(name=None)
-        assert payload.name is None
-
-    def test_edge_case_angle_brackets_content(self):
-        """Test that angle brackets with actual content are rejected."""
-        # Angle brackets without valid HTML-like patterns should be checked
-        # The regex pattern <.*?on\w+\s*= should catch event handlers
-        # But let's verify other patterns too
-
-        # Valid: angle brackets used as symbols (not matched by our patterns)
-        # Our patterns specifically look for dangerous constructs
-
-        # Invalid: actual HTML tags with event handlers
-        invalid_names = [
-            "<div onclick=xss>",
-            "<img src=x onerror=alert(1)>",
-        ]
-
-        for name in invalid_names:
-            with pytest.raises(ValueError) as exc_info:
-                CreateAppPayload(name=name, mode="chat")
-            assert "invalid characters or patterns" in str(exc_info.value).lower()

api/tests/unit_tests/controllers/console/datasets/test_datasets_document_download.py

@@ -0,0 +1,430 @@
+"""
+Unit tests for the dataset document download endpoint.
+
+These tests validate that the controller returns a signed download URL for
+upload-file documents, and rejects unsupported or missing file cases.
+"""
+
+from __future__ import annotations
+
+import importlib
+import sys
+from collections import UserDict
+from io import BytesIO
+from types import SimpleNamespace
+from typing import Any
+from zipfile import ZipFile
+
+import pytest
+from flask import Flask
+from werkzeug.exceptions import Forbidden, NotFound
+
+
+@pytest.fixture
+def app() -> Flask:
+    """Create a minimal Flask app for request-context based controller tests."""
+    app = Flask(__name__)
+    app.config["TESTING"] = True
+    return app
+
+
+@pytest.fixture
+def datasets_document_module(monkeypatch: pytest.MonkeyPatch):
+    """
+    Reload `controllers.console.datasets.datasets_document` with lightweight decorators.
+
+    We patch auth / setup / rate-limit decorators to no-ops so we can unit test the
+    controller logic without requiring the full console stack.
+    """
+
+    from controllers.console import console_ns, wraps
+    from libs import login
+
+    def _noop(func):  # type: ignore[no-untyped-def]
+        return func
+
+    # Bypass login/setup/account checks in unit tests.
+    monkeypatch.setattr(login, "login_required", _noop)
+    monkeypatch.setattr(wraps, "setup_required", _noop)
+    monkeypatch.setattr(wraps, "account_initialization_required", _noop)
+
+    # Bypass billing-related decorators used by other endpoints in this module.
+    monkeypatch.setattr(wraps, "cloud_edition_billing_resource_check", lambda *_args, **_kwargs: (lambda f: f))
+    monkeypatch.setattr(wraps, "cloud_edition_billing_rate_limit_check", lambda *_args, **_kwargs: (lambda f: f))
+
+    # Avoid Flask-RESTX route registration side effects during import.
+    def _noop_route(*_args, **_kwargs):  # type: ignore[override]
+        def _decorator(cls):
+            return cls
+
+        return _decorator
+
+    monkeypatch.setattr(console_ns, "route", _noop_route)
+
+    module_name = "controllers.console.datasets.datasets_document"
+    sys.modules.pop(module_name, None)
+    return importlib.import_module(module_name)
+
+
+def _mock_user(*, is_dataset_editor: bool = True) -> SimpleNamespace:
+    """Build a minimal user object compatible with dataset permission checks."""
+    return SimpleNamespace(is_dataset_editor=is_dataset_editor, id="user-123")
+
+
+def _mock_document(
+    *,
+    document_id: str,
+    tenant_id: str,
+    data_source_type: str,
+    upload_file_id: str | None,
+) -> SimpleNamespace:
+    """Build a minimal document object used by the controller."""
+    data_source_info_dict: dict[str, Any] | None = None
+    if upload_file_id is not None:
+        data_source_info_dict = {"upload_file_id": upload_file_id}
+    else:
+        data_source_info_dict = {}
+
+    return SimpleNamespace(
+        id=document_id,
+        tenant_id=tenant_id,
+        data_source_type=data_source_type,
+        data_source_info_dict=data_source_info_dict,
+    )
+
+
+def _wire_common_success_mocks(
+    *,
+    module,
+    monkeypatch: pytest.MonkeyPatch,
+    current_tenant_id: str,
+    document_tenant_id: str,
+    data_source_type: str,
+    upload_file_id: str | None,
+    upload_file_exists: bool,
+    signed_url: str,
+) -> None:
+    """Patch controller dependencies to create a deterministic test environment."""
+    import services.dataset_service as dataset_service_module
+
+    # Make `current_account_with_tenant()` return a known user + tenant id.
+    monkeypatch.setattr(module, "current_account_with_tenant", lambda: (_mock_user(), current_tenant_id))
+
+    # Return a dataset object and allow permission checks to pass.
+    monkeypatch.setattr(module.DatasetService, "get_dataset", lambda _dataset_id: SimpleNamespace(id="ds-1"))
+    monkeypatch.setattr(module.DatasetService, "check_dataset_permission", lambda *_args, **_kwargs: None)
+
+    # Return a document that will be validated inside DocumentResource.get_document.
+    document = _mock_document(
+        document_id="doc-1",
+        tenant_id=document_tenant_id,
+        data_source_type=data_source_type,
+        upload_file_id=upload_file_id,
+    )
+    monkeypatch.setattr(module.DocumentService, "get_document", lambda *_args, **_kwargs: document)
+
+    # Mock UploadFile lookup via FileService batch helper.
+    upload_files_by_id: dict[str, Any] = {}
+    if upload_file_exists and upload_file_id is not None:
+        upload_files_by_id[str(upload_file_id)] = SimpleNamespace(id=str(upload_file_id))
+    monkeypatch.setattr(module.FileService, "get_upload_files_by_ids", lambda *_args, **_kwargs: upload_files_by_id)
+
+    # Mock signing helper so the returned URL is deterministic.
+    monkeypatch.setattr(dataset_service_module.file_helpers, "get_signed_file_url", lambda **_kwargs: signed_url)
+
+
+def _mock_send_file(obj, **kwargs):  # type: ignore[no-untyped-def]
+    """Return a lightweight representation of `send_file(...)` for unit tests."""
+
+    class _ResponseMock(UserDict):
+        def __init__(self, sent_file: object, send_file_kwargs: dict[str, object]) -> None:
+            super().__init__({"_sent_file": sent_file, "_send_file_kwargs": send_file_kwargs})
+            self._on_close: object | None = None
+
+        def call_on_close(self, func):  # type: ignore[no-untyped-def]
+            self._on_close = func
+            return func
+
+    return _ResponseMock(obj, kwargs)
+
+
+def test_batch_download_zip_returns_send_file(
+    app: Flask, datasets_document_module, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Ensure batch ZIP download returns a zip attachment via `send_file`."""
+
+    # Arrange common permission mocks.
+    monkeypatch.setattr(datasets_document_module, "current_account_with_tenant", lambda: (_mock_user(), "tenant-123"))
+    monkeypatch.setattr(
+        datasets_document_module.DatasetService, "get_dataset", lambda _dataset_id: SimpleNamespace(id="ds-1")
+    )
+    monkeypatch.setattr(
+        datasets_document_module.DatasetService, "check_dataset_permission", lambda *_args, **_kwargs: None
+    )
+
+    # Two upload-file documents, each referencing an UploadFile.
+    doc1 = _mock_document(
+        document_id="11111111-1111-1111-1111-111111111111",
+        tenant_id="tenant-123",
+        data_source_type="upload_file",
+        upload_file_id="file-1",
+    )
+    doc2 = _mock_document(
+        document_id="22222222-2222-2222-2222-222222222222",
+        tenant_id="tenant-123",
+        data_source_type="upload_file",
+        upload_file_id="file-2",
+    )
+    monkeypatch.setattr(
+        datasets_document_module.DocumentService,
+        "get_documents_by_ids",
+        lambda *_args, **_kwargs: [doc1, doc2],
+    )
+    monkeypatch.setattr(
+        datasets_document_module.FileService,
+        "get_upload_files_by_ids",
+        lambda *_args, **_kwargs: {
+            "file-1": SimpleNamespace(id="file-1", name="a.txt", key="k1"),
+            "file-2": SimpleNamespace(id="file-2", name="b.txt", key="k2"),
+        },
+    )
+
+    # Mock storage streaming content.
+    import services.file_service as file_service_module
+
+    monkeypatch.setattr(file_service_module.storage, "load", lambda _key, stream=True: [b"hello"])
+
+    # Replace send_file used by the controller to avoid a real Flask response object.
+    monkeypatch.setattr(datasets_document_module, "send_file", _mock_send_file)
+
+    # Act
+    with app.test_request_context(
+        "/datasets/ds-1/documents/download-zip",
+        method="POST",
+        json={"document_ids": ["11111111-1111-1111-1111-111111111111", "22222222-2222-2222-2222-222222222222"]},
+    ):
+        api = datasets_document_module.DocumentBatchDownloadZipApi()
+        result = api.post(dataset_id="ds-1")
+
+    # Assert: we returned via send_file with correct mime type and attachment.
+    assert result["_send_file_kwargs"]["mimetype"] == "application/zip"
+    assert result["_send_file_kwargs"]["as_attachment"] is True
+    assert isinstance(result["_send_file_kwargs"]["download_name"], str)
+    assert result["_send_file_kwargs"]["download_name"].endswith(".zip")
+    # Ensure our cleanup hook is registered and execute it to avoid temp file leaks in unit tests.
+    assert getattr(result, "_on_close", None) is not None
+    result._on_close()  # type: ignore[attr-defined]
+
+
+def test_batch_download_zip_response_is_openable_zip(
+    app: Flask, datasets_document_module, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Ensure the real Flask `send_file` response body is a valid ZIP that can be opened."""
+
+    # Arrange: same controller mocks as the lightweight send_file test, but we keep the real `send_file`.
+    monkeypatch.setattr(datasets_document_module, "current_account_with_tenant", lambda: (_mock_user(), "tenant-123"))
+    monkeypatch.setattr(
+        datasets_document_module.DatasetService, "get_dataset", lambda _dataset_id: SimpleNamespace(id="ds-1")
+    )
+    monkeypatch.setattr(
+        datasets_document_module.DatasetService, "check_dataset_permission", lambda *_args, **_kwargs: None
+    )
+
+    doc1 = _mock_document(
+        document_id="33333333-3333-3333-3333-333333333333",
+        tenant_id="tenant-123",
+        data_source_type="upload_file",
+        upload_file_id="file-1",
+    )
+    doc2 = _mock_document(
+        document_id="44444444-4444-4444-4444-444444444444",
+        tenant_id="tenant-123",
+        data_source_type="upload_file",
+        upload_file_id="file-2",
+    )
+    monkeypatch.setattr(
+        datasets_document_module.DocumentService,
+        "get_documents_by_ids",
+        lambda *_args, **_kwargs: [doc1, doc2],
+    )
+    monkeypatch.setattr(
+        datasets_document_module.FileService,
+        "get_upload_files_by_ids",
+        lambda *_args, **_kwargs: {
+            "file-1": SimpleNamespace(id="file-1", name="a.txt", key="k1"),
+            "file-2": SimpleNamespace(id="file-2", name="b.txt", key="k2"),
+        },
+    )
+
+    # Stream distinct bytes per key so we can verify both ZIP entries.
+    import services.file_service as file_service_module
+
+    monkeypatch.setattr(
+        file_service_module.storage, "load", lambda key, stream=True: [b"one"] if key == "k1" else [b"two"]
+    )
+
+    # Act
+    with app.test_request_context(
+        "/datasets/ds-1/documents/download-zip",
+        method="POST",
+        json={"document_ids": ["33333333-3333-3333-3333-333333333333", "44444444-4444-4444-4444-444444444444"]},
+    ):
+        api = datasets_document_module.DocumentBatchDownloadZipApi()
+        response = api.post(dataset_id="ds-1")
+
+    # Assert: response body is a valid ZIP and contains the expected entries.
+    response.direct_passthrough = False
+    data = response.get_data()
+    response.close()
+
+    with ZipFile(BytesIO(data), mode="r") as zf:
+        assert zf.namelist() == ["a.txt", "b.txt"]
+        assert zf.read("a.txt") == b"one"
+        assert zf.read("b.txt") == b"two"
+
+
+def test_batch_download_zip_rejects_non_upload_file_document(
+    app: Flask, datasets_document_module, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Ensure batch ZIP download rejects non upload-file documents."""
+
+    monkeypatch.setattr(datasets_document_module, "current_account_with_tenant", lambda: (_mock_user(), "tenant-123"))
+    monkeypatch.setattr(
+        datasets_document_module.DatasetService, "get_dataset", lambda _dataset_id: SimpleNamespace(id="ds-1")
+    )
+    monkeypatch.setattr(
+        datasets_document_module.DatasetService, "check_dataset_permission", lambda *_args, **_kwargs: None
+    )
+
+    doc = _mock_document(
+        document_id="55555555-5555-5555-5555-555555555555",
+        tenant_id="tenant-123",
+        data_source_type="website_crawl",
+        upload_file_id="file-1",
+    )
+    monkeypatch.setattr(
+        datasets_document_module.DocumentService,
+        "get_documents_by_ids",
+        lambda *_args, **_kwargs: [doc],
+    )
+
+    with app.test_request_context(
+        "/datasets/ds-1/documents/download-zip",
+        method="POST",
+        json={"document_ids": ["55555555-5555-5555-5555-555555555555"]},
+    ):
+        api = datasets_document_module.DocumentBatchDownloadZipApi()
+        with pytest.raises(NotFound):
+            api.post(dataset_id="ds-1")
+
+
+def test_document_download_returns_url_for_upload_file_document(
+    app: Flask, datasets_document_module, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Ensure upload-file documents return a `{url}` JSON payload."""
+
+    _wire_common_success_mocks(
+        module=datasets_document_module,
+        monkeypatch=monkeypatch,
+        current_tenant_id="tenant-123",
+        document_tenant_id="tenant-123",
+        data_source_type="upload_file",
+        upload_file_id="file-123",
+        upload_file_exists=True,
+        signed_url="https://example.com/signed",
+    )
+
+    # Build a request context then call the resource method directly.
+    with app.test_request_context("/datasets/ds-1/documents/doc-1/download", method="GET"):
+        api = datasets_document_module.DocumentDownloadApi()
+        result = api.get(dataset_id="ds-1", document_id="doc-1")
+
+    assert result == {"url": "https://example.com/signed"}
+
+
+def test_document_download_rejects_non_upload_file_document(
+    app: Flask, datasets_document_module, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Ensure non-upload documents raise 404 (no file to download)."""
+
+    _wire_common_success_mocks(
+        module=datasets_document_module,
+        monkeypatch=monkeypatch,
+        current_tenant_id="tenant-123",
+        document_tenant_id="tenant-123",
+        data_source_type="website_crawl",
+        upload_file_id="file-123",
+        upload_file_exists=True,
+        signed_url="https://example.com/signed",
+    )
+
+    with app.test_request_context("/datasets/ds-1/documents/doc-1/download", method="GET"):
+        api = datasets_document_module.DocumentDownloadApi()
+        with pytest.raises(NotFound):
+            api.get(dataset_id="ds-1", document_id="doc-1")
+
+
+def test_document_download_rejects_missing_upload_file_id(
+    app: Flask, datasets_document_module, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Ensure missing `upload_file_id` raises 404."""
+
+    _wire_common_success_mocks(
+        module=datasets_document_module,
+        monkeypatch=monkeypatch,
+        current_tenant_id="tenant-123",
+        document_tenant_id="tenant-123",
+        data_source_type="upload_file",
+        upload_file_id=None,
+        upload_file_exists=False,
+        signed_url="https://example.com/signed",
+    )
+
+    with app.test_request_context("/datasets/ds-1/documents/doc-1/download", method="GET"):
+        api = datasets_document_module.DocumentDownloadApi()
+        with pytest.raises(NotFound):
+            api.get(dataset_id="ds-1", document_id="doc-1")
+
+
+def test_document_download_rejects_when_upload_file_record_missing(
+    app: Flask, datasets_document_module, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Ensure missing UploadFile row raises 404."""
+
+    _wire_common_success_mocks(
+        module=datasets_document_module,
+        monkeypatch=monkeypatch,
+        current_tenant_id="tenant-123",
+        document_tenant_id="tenant-123",
+        data_source_type="upload_file",
+        upload_file_id="file-123",
+        upload_file_exists=False,
+        signed_url="https://example.com/signed",
+    )
+
+    with app.test_request_context("/datasets/ds-1/documents/doc-1/download", method="GET"):
+        api = datasets_document_module.DocumentDownloadApi()
+        with pytest.raises(NotFound):
+            api.get(dataset_id="ds-1", document_id="doc-1")
+
+
+def test_document_download_rejects_tenant_mismatch(
+    app: Flask, datasets_document_module, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Ensure tenant mismatch is rejected by the shared `get_document()` permission check."""
+
+    _wire_common_success_mocks(
+        module=datasets_document_module,
+        monkeypatch=monkeypatch,
+        current_tenant_id="tenant-123",
+        document_tenant_id="tenant-999",
+        data_source_type="upload_file",
+        upload_file_id="file-123",
+        upload_file_exists=True,
+        signed_url="https://example.com/signed",
+    )
+
+    with app.test_request_context("/datasets/ds-1/documents/doc-1/download", method="GET"):
+        api = datasets_document_module.DocumentDownloadApi()
+        with pytest.raises(Forbidden):
+            api.get(dataset_id="ds-1", document_id="doc-1")

api/tests/unit_tests/core/plugin/test_plugin_runtime.py

@@ -346,6 +346,7 @@ class TestPluginRuntimeErrorHandling:
         mock_response.status_code = 200
         invoke_error = {
             "error_type": "InvokeRateLimitError",
+            "message": "Rate limit exceeded",
             "args": {"description": "Rate limit exceeded"},
         }
         error_message = json.dumps({"error_type": "PluginInvokeError", "message": json.dumps(invoke_error)})
@@ -364,6 +365,7 @@ class TestPluginRuntimeErrorHandling:
         mock_response.status_code = 200
         invoke_error = {
             "error_type": "InvokeAuthorizationError",
+            "message": "Invalid credentials",
             "args": {"description": "Invalid credentials"},
         }
         error_message = json.dumps({"error_type": "PluginInvokeError", "message": json.dumps(invoke_error)})
@@ -382,6 +384,7 @@ class TestPluginRuntimeErrorHandling:
         mock_response.status_code = 200
         invoke_error = {
             "error_type": "InvokeBadRequestError",
+            "message": "Invalid parameters",
             "args": {"description": "Invalid parameters"},
         }
         error_message = json.dumps({"error_type": "PluginInvokeError", "message": json.dumps(invoke_error)})
@@ -400,6 +403,7 @@ class TestPluginRuntimeErrorHandling:
         mock_response.status_code = 200
         invoke_error = {
             "error_type": "InvokeConnectionError",
+            "message": "Connection to external service failed",
             "args": {"description": "Connection to external service failed"},
         }
         error_message = json.dumps({"error_type": "PluginInvokeError", "message": json.dumps(invoke_error)})
@@ -418,6 +422,7 @@ class TestPluginRuntimeErrorHandling:
         mock_response.status_code = 200
         invoke_error = {
             "error_type": "InvokeServerUnavailableError",
+            "message": "Service temporarily unavailable",
             "args": {"description": "Service temporarily unavailable"},
         }
         error_message = json.dumps({"error_type": "PluginInvokeError", "message": json.dumps(invoke_error)})

api/tests/unit_tests/repositories/test_workflow_collaboration_repository.py

@@ -0,0 +1,121 @@
+import json
+from unittest.mock import Mock
+
+import pytest
+
+from repositories import workflow_collaboration_repository as repo_module
+from repositories.workflow_collaboration_repository import WorkflowCollaborationRepository
+
+
+class TestWorkflowCollaborationRepository:
+    @pytest.fixture
+    def mock_redis(self, monkeypatch: pytest.MonkeyPatch) -> Mock:
+        mock_redis = Mock()
+        monkeypatch.setattr(repo_module, "redis_client", mock_redis)
+        return mock_redis
+
+    def test_get_sid_mapping_returns_mapping(self, mock_redis: Mock) -> None:
+        # Arrange
+        mock_redis.get.return_value = b'{"workflow_id":"wf-1","user_id":"u-1"}'
+        repository = WorkflowCollaborationRepository()
+
+        # Act
+        result = repository.get_sid_mapping("sid-1")
+
+        # Assert
+        assert result == {"workflow_id": "wf-1", "user_id": "u-1"}
+
+    def test_list_sessions_filters_invalid_entries(self, mock_redis: Mock) -> None:
+        # Arrange
+        mock_redis.hgetall.return_value = {
+            b"sid-1": b'{"user_id":"u-1","username":"Jane","sid":"sid-1","connected_at":2}',
+            b"sid-2": b'{"username":"Missing","sid":"sid-2"}',
+            b"sid-3": b"not-json",
+        }
+        repository = WorkflowCollaborationRepository()
+
+        # Act
+        result = repository.list_sessions("wf-1")
+
+        # Assert
+        assert result == [
+            {
+                "user_id": "u-1",
+                "username": "Jane",
+                "avatar": None,
+                "sid": "sid-1",
+                "connected_at": 2,
+            }
+        ]
+
+    def test_set_session_info_persists_payload(self, mock_redis: Mock) -> None:
+        # Arrange
+        mock_redis.exists.return_value = True
+        repository = WorkflowCollaborationRepository()
+        payload = {
+            "user_id": "u-1",
+            "username": "Jane",
+            "avatar": None,
+            "sid": "sid-1",
+            "connected_at": 1,
+        }
+
+        # Act
+        repository.set_session_info("wf-1", payload)
+
+        # Assert
+        assert mock_redis.hset.called
+        workflow_key, sid, session_json = mock_redis.hset.call_args.args
+        assert workflow_key == "workflow_online_users:wf-1"
+        assert sid == "sid-1"
+        assert json.loads(session_json)["user_id"] == "u-1"
+        assert mock_redis.set.called
+
+    def test_refresh_session_state_expires_keys(self, mock_redis: Mock) -> None:
+        # Arrange
+        mock_redis.exists.return_value = True
+        repository = WorkflowCollaborationRepository()
+
+        # Act
+        repository.refresh_session_state("wf-1", "sid-1")
+
+        # Assert
+        assert mock_redis.expire.call_count == 2
+
+    def test_get_current_leader_decodes_bytes(self, mock_redis: Mock) -> None:
+        # Arrange
+        mock_redis.get.return_value = b"sid-1"
+        repository = WorkflowCollaborationRepository()
+
+        # Act
+        result = repository.get_current_leader("wf-1")
+
+        # Assert
+        assert result == "sid-1"
+
+    def test_set_leader_if_absent_uses_nx(self, mock_redis: Mock) -> None:
+        # Arrange
+        mock_redis.set.return_value = True
+        repository = WorkflowCollaborationRepository()
+
+        # Act
+        result = repository.set_leader_if_absent("wf-1", "sid-1")
+
+        # Assert
+        assert result is True
+        _key, _value = mock_redis.set.call_args.args
+        assert _key == "workflow_leader:wf-1"
+        assert _value == "sid-1"
+        assert mock_redis.set.call_args.kwargs["nx"] is True
+        assert "ex" in mock_redis.set.call_args.kwargs
+
+    def test_get_session_sids_decodes(self, mock_redis: Mock) -> None:
+        # Arrange
+        mock_redis.hkeys.return_value = [b"sid-1", "sid-2"]
+        repository = WorkflowCollaborationRepository()
+
+        # Act
+        result = repository.get_session_sids("wf-1")
+
+        # Assert
+        assert result == ["sid-1", "sid-2"]

api/tests/unit_tests/services/test_file_service_zip_and_lookup.py

@@ -0,0 +1,99 @@
+"""
+Unit tests for `services.file_service.FileService` helpers.
+
+We keep these tests focused on:
+- ZIP tempfile building (sanitization + deduplication + content writes)
+- tenant-scoped batch lookup behavior (`get_upload_files_by_ids`)
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from typing import Any
+from zipfile import ZipFile
+
+import pytest
+
+import services.file_service as file_service_module
+from services.file_service import FileService
+
+
+def test_build_upload_files_zip_tempfile_sanitizes_and_dedupes_names(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Ensure ZIP entry names are safe and unique while preserving extensions."""
+
+    # Arrange: three upload files that all sanitize down to the same basename ("b.txt").
+    upload_files: list[Any] = [
+        SimpleNamespace(name="a/b.txt", key="k1"),
+        SimpleNamespace(name="c/b.txt", key="k2"),
+        SimpleNamespace(name="../b.txt", key="k3"),
+    ]
+
+    # Stream distinct bytes per key so we can verify content is written to the right entry.
+    data_by_key: dict[str, list[bytes]] = {"k1": [b"one"], "k2": [b"two"], "k3": [b"three"]}
+
+    def _load(key: str, stream: bool = True) -> list[bytes]:
+        # Return the corresponding chunks for this key (the production code iterates chunks).
+        assert stream is True
+        return data_by_key[key]
+
+    monkeypatch.setattr(file_service_module.storage, "load", _load)
+
+    # Act: build zip in a tempfile.
+    with FileService.build_upload_files_zip_tempfile(upload_files=upload_files) as tmp:
+        with ZipFile(tmp, mode="r") as zf:
+            # Assert: names are sanitized (no directory components) and deduped with suffixes.
+            assert zf.namelist() == ["b.txt", "b (1).txt", "b (2).txt"]
+
+            # Assert: each entry contains the correct bytes from storage.
+            assert zf.read("b.txt") == b"one"
+            assert zf.read("b (1).txt") == b"two"
+            assert zf.read("b (2).txt") == b"three"
+
+
+def test_get_upload_files_by_ids_returns_empty_when_no_ids(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Ensure empty input returns an empty mapping without hitting the database."""
+
+    class _Session:
+        def scalars(self, _stmt):  # type: ignore[no-untyped-def]
+            raise AssertionError("db.session.scalars should not be called for empty id lists")
+
+    monkeypatch.setattr(file_service_module, "db", SimpleNamespace(session=_Session()))
+
+    assert FileService.get_upload_files_by_ids("tenant-1", []) == {}
+
+
+def test_get_upload_files_by_ids_returns_id_keyed_mapping(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Ensure batch lookup returns a dict keyed by stringified UploadFile ids."""
+
+    upload_files: list[Any] = [
+        SimpleNamespace(id="file-1", tenant_id="tenant-1"),
+        SimpleNamespace(id="file-2", tenant_id="tenant-1"),
+    ]
+
+    class _ScalarResult:
+        def __init__(self, items: list[Any]) -> None:
+            self._items = items
+
+        def all(self) -> list[Any]:
+            return self._items
+
+    class _Session:
+        def __init__(self, items: list[Any]) -> None:
+            self._items = items
+            self.calls: list[object] = []
+
+        def scalars(self, stmt):  # type: ignore[no-untyped-def]
+            # Capture the statement so we can at least assert the query path is taken.
+            self.calls.append(stmt)
+            return _ScalarResult(self._items)
+
+    session = _Session(upload_files)
+    monkeypatch.setattr(file_service_module, "db", SimpleNamespace(session=session))
+
+    # Provide duplicates to ensure callers can safely pass repeated ids.
+    result = FileService.get_upload_files_by_ids("tenant-1", ["file-1", "file-1", "file-2"])
+
+    assert set(result.keys()) == {"file-1", "file-2"}
+    assert result["file-1"].id == "file-1"
+    assert result["file-2"].id == "file-2"
+    assert len(session.calls) == 1

api/tests/unit_tests/services/test_workflow_collaboration_service.py

@@ -0,0 +1,271 @@
+from unittest.mock import Mock, patch
+
+import pytest
+
+from repositories.workflow_collaboration_repository import WorkflowCollaborationRepository
+from services.workflow_collaboration_service import WorkflowCollaborationService
+
+
+class TestWorkflowCollaborationService:
+    @pytest.fixture
+    def service(self) -> tuple[WorkflowCollaborationService, Mock, Mock]:
+        repository = Mock(spec=WorkflowCollaborationRepository)
+        socketio = Mock()
+        return WorkflowCollaborationService(repository, socketio), repository, socketio
+
+    def test_register_session_returns_leader_status(
+        self, service: tuple[WorkflowCollaborationService, Mock, Mock]
+    ) -> None:
+        # Arrange
+        collaboration_service, repository, socketio = service
+        socketio.get_session.return_value = {"user_id": "u-1", "username": "Jane", "avatar": None}
+
+        with (
+            patch.object(collaboration_service, "get_or_set_leader", return_value="sid-1"),
+            patch.object(collaboration_service, "broadcast_online_users"),
+        ):
+            # Act
+            result = collaboration_service.register_session("wf-1", "sid-1")
+
+        # Assert
+        assert result == ("u-1", True)
+        repository.set_session_info.assert_called_once()
+        socketio.enter_room.assert_called_once_with("sid-1", "wf-1")
+        socketio.emit.assert_called_once_with("status", {"isLeader": True}, room="sid-1")
+
+    def test_register_session_returns_none_when_missing_user(
+        self, service: tuple[WorkflowCollaborationService, Mock, Mock]
+    ) -> None:
+        # Arrange
+        collaboration_service, _repository, socketio = service
+        socketio.get_session.return_value = {}
+
+        # Act
+        result = collaboration_service.register_session("wf-1", "sid-1")
+
+        # Assert
+        assert result is None
+
+    def test_relay_collaboration_event_unauthorized(
+        self, service: tuple[WorkflowCollaborationService, Mock, Mock]
+    ) -> None:
+        # Arrange
+        collaboration_service, repository, _socketio = service
+        repository.get_sid_mapping.return_value = None
+
+        # Act
+        result = collaboration_service.relay_collaboration_event("sid-1", {})
+
+        # Assert
+        assert result == ({"msg": "unauthorized"}, 401)
+
+    def test_relay_collaboration_event_emits_update(
+        self, service: tuple[WorkflowCollaborationService, Mock, Mock]
+    ) -> None:
+        # Arrange
+        collaboration_service, repository, socketio = service
+        repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
+        payload = {"type": "mouse_move", "data": {"x": 1}, "timestamp": 123}
+
+        # Act
+        result = collaboration_service.relay_collaboration_event("sid-1", payload)
+
+        # Assert
+        assert result == ({"msg": "event_broadcasted"}, 200)
+        socketio.emit.assert_called_once_with(
+            "collaboration_update",
+            {"type": "mouse_move", "userId": "u-1", "data": {"x": 1}, "timestamp": 123},
+            room="wf-1",
+            skip_sid="sid-1",
+        )
+
+    def test_relay_graph_event_unauthorized(self, service: tuple[WorkflowCollaborationService, Mock, Mock]) -> None:
+        # Arrange
+        collaboration_service, repository, _socketio = service
+        repository.get_sid_mapping.return_value = None
+
+        # Act
+        result = collaboration_service.relay_graph_event("sid-1", {"nodes": []})
+
+        # Assert
+        assert result == ({"msg": "unauthorized"}, 401)
+
+    def test_disconnect_session_no_mapping(self, service: tuple[WorkflowCollaborationService, Mock, Mock]) -> None:
+        # Arrange
+        collaboration_service, repository, _socketio = service
+        repository.get_sid_mapping.return_value = None
+
+        # Act
+        collaboration_service.disconnect_session("sid-1")
+
+        # Assert
+        repository.delete_session.assert_not_called()
+
+    def test_disconnect_session_cleans_up(self, service: tuple[WorkflowCollaborationService, Mock, Mock]) -> None:
+        # Arrange
+        collaboration_service, repository, _socketio = service
+        repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
+
+        with (
+            patch.object(collaboration_service, "handle_leader_disconnect") as handle_leader_disconnect,
+            patch.object(collaboration_service, "broadcast_online_users") as broadcast_online_users,
+        ):
+            # Act
+            collaboration_service.disconnect_session("sid-1")
+
+        # Assert
+        repository.delete_session.assert_called_once_with("wf-1", "sid-1")
+        handle_leader_disconnect.assert_called_once_with("wf-1", "sid-1")
+        broadcast_online_users.assert_called_once_with("wf-1")
+
+    def test_get_or_set_leader_returns_active_leader(
+        self, service: tuple[WorkflowCollaborationService, Mock, Mock]
+    ) -> None:
+        # Arrange
+        collaboration_service, repository, _socketio = service
+        repository.get_current_leader.return_value = "sid-1"
+
+        with patch.object(collaboration_service, "is_session_active", return_value=True):
+            # Act
+            result = collaboration_service.get_or_set_leader("wf-1", "sid-2")
+
+        # Assert
+        assert result == "sid-1"
+        repository.set_leader_if_absent.assert_not_called()
+
+    def test_get_or_set_leader_replaces_dead_leader(
+        self, service: tuple[WorkflowCollaborationService, Mock, Mock]
+    ) -> None:
+        # Arrange
+        collaboration_service, repository, _socketio = service
+        repository.get_current_leader.return_value = "sid-1"
+        repository.set_leader_if_absent.return_value = True
+
+        with (
+            patch.object(collaboration_service, "is_session_active", return_value=False),
+            patch.object(collaboration_service, "broadcast_leader_change") as broadcast_leader_change,
+        ):
+            # Act
+            result = collaboration_service.get_or_set_leader("wf-1", "sid-2")
+
+        # Assert
+        assert result == "sid-2"
+        repository.delete_session.assert_called_once_with("wf-1", "sid-1")
+        repository.delete_leader.assert_called_once_with("wf-1")
+        broadcast_leader_change.assert_called_once_with("wf-1", "sid-2")
+
+    def test_get_or_set_leader_falls_back_to_existing(
+        self, service: tuple[WorkflowCollaborationService, Mock, Mock]
+    ) -> None:
+        # Arrange
+        collaboration_service, repository, _socketio = service
+        repository.get_current_leader.side_effect = [None, "sid-3"]
+        repository.set_leader_if_absent.return_value = False
+
+        # Act
+        result = collaboration_service.get_or_set_leader("wf-1", "sid-2")
+
+        # Assert
+        assert result == "sid-3"
+
+    def test_handle_leader_disconnect_elects_new(
+        self, service: tuple[WorkflowCollaborationService, Mock, Mock]
+    ) -> None:
+        # Arrange
+        collaboration_service, repository, _socketio = service
+        repository.get_current_leader.return_value = "sid-1"
+        repository.get_session_sids.return_value = ["sid-2"]
+
+        with patch.object(collaboration_service, "broadcast_leader_change") as broadcast_leader_change:
+            # Act
+            collaboration_service.handle_leader_disconnect("wf-1", "sid-1")
+
+        # Assert
+        repository.set_leader.assert_called_once_with("wf-1", "sid-2")
+        broadcast_leader_change.assert_called_once_with("wf-1", "sid-2")
+
+    def test_handle_leader_disconnect_clears_when_empty(
+        self, service: tuple[WorkflowCollaborationService, Mock, Mock]
+    ) -> None:
+        # Arrange
+        collaboration_service, repository, _socketio = service
+        repository.get_current_leader.return_value = "sid-1"
+        repository.get_session_sids.return_value = []
+
+        # Act
+        collaboration_service.handle_leader_disconnect("wf-1", "sid-1")
+
+        # Assert
+        repository.delete_leader.assert_called_once_with("wf-1")
+
+    def test_broadcast_online_users_sorts_and_emits(
+        self, service: tuple[WorkflowCollaborationService, Mock, Mock]
+    ) -> None:
+        # Arrange
+        collaboration_service, repository, socketio = service
+        repository.list_sessions.return_value = [
+            {"user_id": "u-1", "username": "A", "avatar": None, "sid": "sid-1", "connected_at": 3},
+            {"user_id": "u-2", "username": "B", "avatar": None, "sid": "sid-2", "connected_at": 1},
+        ]
+        repository.get_current_leader.return_value = "sid-1"
+
+        # Act
+        collaboration_service.broadcast_online_users("wf-1")
+
+        # Assert
+        socketio.emit.assert_called_once_with(
+            "online_users",
+            {
+                "workflow_id": "wf-1",
+                "users": [
+                    {"user_id": "u-2", "username": "B", "avatar": None, "sid": "sid-2", "connected_at": 1},
+                    {"user_id": "u-1", "username": "A", "avatar": None, "sid": "sid-1", "connected_at": 3},
+                ],
+                "leader": "sid-1",
+            },
+            room="wf-1",
+        )
+
+    def test_refresh_session_state_expires_active_leader(
+        self, service: tuple[WorkflowCollaborationService, Mock, Mock]
+    ) -> None:
+        # Arrange
+        collaboration_service, repository, _socketio = service
+        repository.get_current_leader.return_value = "sid-1"
+
+        with patch.object(collaboration_service, "is_session_active", return_value=True):
+            # Act
+            collaboration_service.refresh_session_state("wf-1", "sid-1")
+
+        # Assert
+        repository.refresh_session_state.assert_called_once_with("wf-1", "sid-1")
+        repository.expire_leader.assert_called_once_with("wf-1")
+        repository.set_leader.assert_not_called()
+
+    def test_refresh_session_state_sets_leader_when_missing(
+        self, service: tuple[WorkflowCollaborationService, Mock, Mock]
+    ) -> None:
+        # Arrange
+        collaboration_service, repository, _socketio = service
+        repository.get_current_leader.return_value = None
+
+        with patch.object(collaboration_service, "broadcast_leader_change") as broadcast_leader_change:
+            # Act
+            collaboration_service.refresh_session_state("wf-1", "sid-2")
+
+        # Assert
+        repository.set_leader.assert_called_once_with("wf-1", "sid-2")
+        broadcast_leader_change.assert_called_once_with("wf-1", "sid-2")
+
+    def test_relay_graph_event_emits_update(self, service: tuple[WorkflowCollaborationService, Mock, Mock]) -> None:
+        # Arrange
+        collaboration_service, repository, socketio = service
+        repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
+
+        # Act
+        result = collaboration_service.relay_graph_event("sid-1", {"nodes": []})
+
+        # Assert
+        assert result == ({"msg": "graph_update_broadcasted"}, 200)
+        repository.refresh_session_state.assert_called_once_with("wf-1", "sid-1")
+        socketio.emit.assert_called_once_with("graph_update", {"nodes": []}, room="wf-1", skip_sid="sid-1")

api/tests/unit_tests/services/test_workflow_comment_service.py

@@ -0,0 +1,245 @@
+from unittest.mock import MagicMock, Mock, patch
+
+import pytest
+from werkzeug.exceptions import Forbidden, NotFound
+
+from services import workflow_comment_service as service_module
+from services.workflow_comment_service import WorkflowCommentService
+
+
+@pytest.fixture
+def mock_session(monkeypatch: pytest.MonkeyPatch) -> Mock:
+    session = Mock()
+    context_manager = MagicMock()
+    context_manager.__enter__.return_value = session
+    context_manager.__exit__.return_value = False
+    mock_db = MagicMock()
+    mock_db.engine = Mock()
+    monkeypatch.setattr(service_module, "Session", Mock(return_value=context_manager))
+    monkeypatch.setattr(service_module, "db", mock_db)
+    return session
+
+
+def _mock_scalars(result_list: list[object]) -> Mock:
+    scalars = Mock()
+    scalars.all.return_value = result_list
+    return scalars
+
+
+class TestWorkflowCommentService:
+    def test_validate_content_rejects_empty(self) -> None:
+        with pytest.raises(ValueError):
+            WorkflowCommentService._validate_content("   ")
+
+    def test_validate_content_rejects_too_long(self) -> None:
+        with pytest.raises(ValueError):
+            WorkflowCommentService._validate_content("a" * 1001)
+
+    def test_create_comment_creates_mentions(self, mock_session: Mock) -> None:
+        comment = Mock()
+        comment.id = "comment-1"
+        comment.created_at = "ts"
+
+        with (
+            patch.object(service_module, "WorkflowComment", return_value=comment),
+            patch.object(service_module, "WorkflowCommentMention", return_value=Mock()),
+            patch.object(service_module, "uuid_value", side_effect=[True, False]),
+        ):
+            result = WorkflowCommentService.create_comment(
+                tenant_id="tenant-1",
+                app_id="app-1",
+                created_by="user-1",
+                content="hello",
+                position_x=1.0,
+                position_y=2.0,
+                mentioned_user_ids=["user-2", "bad-id"],
+            )
+
+        assert result == {"id": "comment-1", "created_at": "ts"}
+        assert mock_session.add.call_args_list[0].args[0] is comment
+        assert mock_session.add.call_count == 2
+        mock_session.commit.assert_called_once()
+
+    def test_update_comment_raises_not_found(self, mock_session: Mock) -> None:
+        mock_session.scalar.return_value = None
+
+        with pytest.raises(NotFound):
+            WorkflowCommentService.update_comment(
+                tenant_id="tenant-1",
+                app_id="app-1",
+                comment_id="comment-1",
+                user_id="user-1",
+                content="hello",
+            )
+
+    def test_update_comment_raises_forbidden(self, mock_session: Mock) -> None:
+        comment = Mock()
+        comment.created_by = "owner"
+        mock_session.scalar.return_value = comment
+
+        with pytest.raises(Forbidden):
+            WorkflowCommentService.update_comment(
+                tenant_id="tenant-1",
+                app_id="app-1",
+                comment_id="comment-1",
+                user_id="intruder",
+                content="hello",
+            )
+
+    def test_update_comment_replaces_mentions(self, mock_session: Mock) -> None:
+        comment = Mock()
+        comment.id = "comment-1"
+        comment.created_by = "owner"
+        mock_session.scalar.return_value = comment
+
+        existing_mentions = [Mock(), Mock()]
+        mock_session.scalars.return_value = _mock_scalars(existing_mentions)
+
+        with patch.object(service_module, "uuid_value", side_effect=[True, False]):
+            result = WorkflowCommentService.update_comment(
+                tenant_id="tenant-1",
+                app_id="app-1",
+                comment_id="comment-1",
+                user_id="owner",
+                content="updated",
+                mentioned_user_ids=["user-2", "bad-id"],
+            )
+
+        assert result == {"id": "comment-1", "updated_at": comment.updated_at}
+        assert mock_session.delete.call_count == 2
+        assert mock_session.add.call_count == 1
+        mock_session.commit.assert_called_once()
+
+    def test_delete_comment_raises_forbidden(self, mock_session: Mock) -> None:
+        comment = Mock()
+        comment.created_by = "owner"
+
+        with patch.object(WorkflowCommentService, "get_comment", return_value=comment):
+            with pytest.raises(Forbidden):
+                WorkflowCommentService.delete_comment("tenant-1", "app-1", "comment-1", "intruder")
+
+    def test_delete_comment_removes_related_entities(self, mock_session: Mock) -> None:
+        comment = Mock()
+        comment.created_by = "owner"
+
+        mentions = [Mock(), Mock()]
+        replies = [Mock()]
+        mock_session.scalars.side_effect = [_mock_scalars(mentions), _mock_scalars(replies)]
+
+        with patch.object(WorkflowCommentService, "get_comment", return_value=comment):
+            WorkflowCommentService.delete_comment("tenant-1", "app-1", "comment-1", "owner")
+
+        assert mock_session.delete.call_count == 4
+        mock_session.commit.assert_called_once()
+
+    def test_resolve_comment_sets_fields(self, mock_session: Mock) -> None:
+        comment = Mock()
+        comment.resolved = False
+        comment.resolved_at = None
+        comment.resolved_by = None
+
+        with (
+            patch.object(WorkflowCommentService, "get_comment", return_value=comment),
+            patch.object(service_module, "naive_utc_now", return_value="now"),
+        ):
+            result = WorkflowCommentService.resolve_comment("tenant-1", "app-1", "comment-1", "user-1")
+
+        assert result is comment
+        assert comment.resolved is True
+        assert comment.resolved_at == "now"
+        assert comment.resolved_by == "user-1"
+        mock_session.commit.assert_called_once()
+
+    def test_resolve_comment_noop_when_already_resolved(self, mock_session: Mock) -> None:
+        comment = Mock()
+        comment.resolved = True
+
+        with patch.object(WorkflowCommentService, "get_comment", return_value=comment):
+            result = WorkflowCommentService.resolve_comment("tenant-1", "app-1", "comment-1", "user-1")
+
+        assert result is comment
+        mock_session.commit.assert_not_called()
+
+    def test_create_reply_requires_comment(self, mock_session: Mock) -> None:
+        mock_session.get.return_value = None
+
+        with pytest.raises(NotFound):
+            WorkflowCommentService.create_reply("comment-1", "hello", "user-1")
+
+    def test_create_reply_creates_mentions(self, mock_session: Mock) -> None:
+        mock_session.get.return_value = Mock()
+        reply = Mock()
+        reply.id = "reply-1"
+        reply.created_at = "ts"
+
+        with (
+            patch.object(service_module, "WorkflowCommentReply", return_value=reply),
+            patch.object(service_module, "WorkflowCommentMention", return_value=Mock()),
+            patch.object(service_module, "uuid_value", side_effect=[True, False]),
+        ):
+            result = WorkflowCommentService.create_reply(
+                comment_id="comment-1",
+                content="hello",
+                created_by="user-1",
+                mentioned_user_ids=["user-2", "bad-id"],
+            )
+
+        assert result == {"id": "reply-1", "created_at": "ts"}
+        assert mock_session.add.call_count == 2
+        mock_session.commit.assert_called_once()
+
+    def test_update_reply_raises_not_found(self, mock_session: Mock) -> None:
+        mock_session.get.return_value = None
+
+        with pytest.raises(NotFound):
+            WorkflowCommentService.update_reply("reply-1", "user-1", "hello")
+
+    def test_update_reply_raises_forbidden(self, mock_session: Mock) -> None:
+        reply = Mock()
+        reply.created_by = "owner"
+        mock_session.get.return_value = reply
+
+        with pytest.raises(Forbidden):
+            WorkflowCommentService.update_reply("reply-1", "intruder", "hello")
+
+    def test_update_reply_replaces_mentions(self, mock_session: Mock) -> None:
+        reply = Mock()
+        reply.id = "reply-1"
+        reply.comment_id = "comment-1"
+        reply.created_by = "owner"
+        reply.updated_at = "updated"
+        mock_session.get.return_value = reply
+        mock_session.scalars.return_value = _mock_scalars([Mock()])
+
+        with patch.object(service_module, "uuid_value", side_effect=[True, False]):
+            result = WorkflowCommentService.update_reply(
+                reply_id="reply-1",
+                user_id="owner",
+                content="new",
+                mentioned_user_ids=["user-2", "bad-id"],
+            )
+
+        assert result == {"id": "reply-1", "updated_at": "updated"}
+        assert mock_session.delete.call_count == 1
+        assert mock_session.add.call_count == 1
+        mock_session.commit.assert_called_once()
+        mock_session.refresh.assert_called_once_with(reply)
+
+    def test_delete_reply_raises_forbidden(self, mock_session: Mock) -> None:
+        reply = Mock()
+        reply.created_by = "owner"
+        mock_session.get.return_value = reply
+
+        with pytest.raises(Forbidden):
+            WorkflowCommentService.delete_reply("reply-1", "intruder")
+
+    def test_delete_reply_removes_mentions(self, mock_session: Mock) -> None:
+        reply = Mock()
+        reply.created_by = "owner"
+        mock_session.get.return_value = reply
+        mock_session.scalars.return_value = _mock_scalars([Mock(), Mock()])
+
+        WorkflowCommentService.delete_reply("reply-1", "owner")
+
+        assert mock_session.delete.call_count == 3
+        mock_session.commit.assert_called_once()

api/tests/unit_tests/services/test_workflow_service.py

@@ -10,7 +10,7 @@ This test suite covers:
 """
 
 import json
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock, Mock, patch
 
 import pytest
 
@@ -630,6 +630,79 @@ class TestWorkflowService:
         with pytest.raises(ValueError, match="Invalid app mode"):
             workflow_service.validate_features_structure(app, features)
 
+    # ==================== Draft Workflow Variable Update Tests ====================
+    # These tests verify updating draft workflow environment/conversation variables
+
+    def test_update_draft_workflow_environment_variables_updates_workflow(self, workflow_service, mock_db_session):
+        """Test update_draft_workflow_environment_variables updates draft fields."""
+        app = TestWorkflowAssociatedDataFactory.create_app_mock()
+        account = TestWorkflowAssociatedDataFactory.create_account_mock()
+        workflow = TestWorkflowAssociatedDataFactory.create_workflow_mock()
+        variables = [Mock()]
+
+        with (
+            patch.object(workflow_service, "get_draft_workflow", return_value=workflow),
+            patch("services.workflow_service.naive_utc_now", return_value="now"),
+        ):
+            workflow_service.update_draft_workflow_environment_variables(
+                app_model=app,
+                environment_variables=variables,
+                account=account,
+            )
+
+        assert workflow.environment_variables == variables
+        assert workflow.updated_by == account.id
+        assert workflow.updated_at == "now"
+        mock_db_session.session.commit.assert_called_once()
+
+    def test_update_draft_workflow_environment_variables_raises_when_missing(self, workflow_service):
+        """Test update_draft_workflow_environment_variables raises when draft missing."""
+        app = TestWorkflowAssociatedDataFactory.create_app_mock()
+        account = TestWorkflowAssociatedDataFactory.create_account_mock()
+
+        with patch.object(workflow_service, "get_draft_workflow", return_value=None):
+            with pytest.raises(ValueError, match="No draft workflow found."):
+                workflow_service.update_draft_workflow_environment_variables(
+                    app_model=app,
+                    environment_variables=[],
+                    account=account,
+                )
+
+    def test_update_draft_workflow_conversation_variables_updates_workflow(self, workflow_service, mock_db_session):
+        """Test update_draft_workflow_conversation_variables updates draft fields."""
+        app = TestWorkflowAssociatedDataFactory.create_app_mock()
+        account = TestWorkflowAssociatedDataFactory.create_account_mock()
+        workflow = TestWorkflowAssociatedDataFactory.create_workflow_mock()
+        variables = [Mock()]
+
+        with (
+            patch.object(workflow_service, "get_draft_workflow", return_value=workflow),
+            patch("services.workflow_service.naive_utc_now", return_value="now"),
+        ):
+            workflow_service.update_draft_workflow_conversation_variables(
+                app_model=app,
+                conversation_variables=variables,
+                account=account,
+            )
+
+        assert workflow.conversation_variables == variables
+        assert workflow.updated_by == account.id
+        assert workflow.updated_at == "now"
+        mock_db_session.session.commit.assert_called_once()
+
+    def test_update_draft_workflow_conversation_variables_raises_when_missing(self, workflow_service):
+        """Test update_draft_workflow_conversation_variables raises when draft missing."""
+        app = TestWorkflowAssociatedDataFactory.create_app_mock()
+        account = TestWorkflowAssociatedDataFactory.create_account_mock()
+
+        with patch.object(workflow_service, "get_draft_workflow", return_value=None):
+            with pytest.raises(ValueError, match="No draft workflow found."):
+                workflow_service.update_draft_workflow_conversation_variables(
+                    app_model=app,
+                    conversation_variables=[],
+                    account=account,
+                )
+
     # ==================== Publish Workflow Tests ====================
     # These tests verify creating published versions from draft workflows
 

api/uv.lock

@@ -583,6 +583,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/57/f4/a69c20ee4f660081a7dedb1ac57f29be9378e04edfcb90c526b923d4bebc/beautifulsoup4-4.12.2-py3-none-any.whl", hash = "sha256:bd2520ca0d9d7d12694a53d44ac482d181b4ec1888909b035a3dbf40d0f57d4a", size = 142979 },
 ]
 
+[[package]]
+name = "bidict"
+version = "0.23.1"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/9a/6e/026678aa5a830e07cd9498a05d3e7e650a4f56a42f267a53d22bcda1bdc9/bidict-0.23.1.tar.gz", hash = "sha256:03069d763bc387bbd20e7d49914e75fc4132a41937fa3405417e1a5a2d006d71", size = 29093, upload-time = "2024-02-18T19:09:05.748Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/99/37/e8730c3587a65eb5645d4aba2d27aae48e8003614d6aaf15dda67f702f1f/bidict-0.23.1-py3-none-any.whl", hash = "sha256:5dae8d4d79b552a71cbabc7deb25dfe8ce710b17ff41711e13010ead2abfc3e5", size = 32764, upload-time = "2024-02-18T19:09:04.156Z" },
+]
+
 [[package]]
 name = "billiard"
 version = "4.2.3"
@@ -1391,6 +1400,7 @@ dependencies = [
     { name = "flask-restx" },
     { name = "flask-sqlalchemy" },
     { name = "gevent" },
+    { name = "gevent-websocket" },
     { name = "gmpy2" },
     { name = "google-api-core" },
     { name = "google-api-python-client" },
@@ -1441,6 +1451,7 @@ dependencies = [
     { name = "pypdfium2" },
     { name = "python-docx" },
     { name = "python-dotenv" },
+    { name = "python-socketio" },
     { name = "pyyaml" },
     { name = "readabilipy" },
     { name = "redis", extra = ["hiredis"] },
@@ -1589,6 +1600,7 @@ requires-dist = [
     { name = "flask-restx", specifier = "~=1.3.0" },
     { name = "flask-sqlalchemy", specifier = "~=3.1.1" },
     { name = "gevent", specifier = "~=25.9.1" },
+    { name = "gevent-websocket", specifier = "~=0.10.1" },
     { name = "gmpy2", specifier = "~=2.2.1" },
     { name = "google-api-core", specifier = "==2.18.0" },
     { name = "google-api-python-client", specifier = "==2.90.0" },
@@ -1639,6 +1651,7 @@ requires-dist = [
     { name = "pypdfium2", specifier = "==5.2.0" },
     { name = "python-docx", specifier = "~=1.1.0" },
     { name = "python-dotenv", specifier = "==1.0.1" },
+    { name = "python-socketio", specifier = "~=5.13.0" },
     { name = "pyyaml", specifier = "~=6.0.1" },
     { name = "readabilipy", specifier = "~=0.3.0" },
     { name = "redis", extras = ["hiredis"], specifier = "~=6.1.0" },
@@ -2194,6 +2207,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/d5/98/caf06d5d22a7c129c1fb2fc1477306902a2c8ddfd399cd26bbbd4caf2141/gevent-25.9.1-cp312-cp312-win_amd64.whl", hash = "sha256:4acd6bcd5feabf22c7c5174bd3b9535ee9f088d2bbce789f740ad8d6554b18f3", size = 1682837 },
 ]
 
+[[package]]
+name = "gevent-websocket"
+version = "0.10.1"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "gevent" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/98/d2/6fa19239ff1ab072af40ebf339acd91fb97f34617c2ee625b8e34bf42393/gevent-websocket-0.10.1.tar.gz", hash = "sha256:7eaef32968290c9121f7c35b973e2cc302ffb076d018c9068d2f5ca8b2d85fb0", size = 18366, upload-time = "2017-03-12T22:46:05.68Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/7b/84/2dc373eb6493e00c884cc11e6c059ec97abae2678d42f06bf780570b0193/gevent_websocket-0.10.1-py3-none-any.whl", hash = "sha256:17b67d91282f8f4c973eba0551183fc84f56f1c90c8f6b6b30256f31f66f5242", size = 22987, upload-time = "2017-03-12T22:46:03.611Z" },
+]
+
 [[package]]
 name = "gitdb"
 version = "4.0.12"
@@ -5191,6 +5216,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/6a/3e/b68c118422ec867fa7ab88444e1274aa40681c606d59ac27de5a5588f082/python_dotenv-1.0.1-py3-none-any.whl", hash = "sha256:f7b63ef50f1b690dddf550d03497b66d609393b40b564ed0d674909a68ebf16a", size = 19863 },
 ]
 
+[[package]]
+name = "python-engineio"
+version = "4.12.3"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "simple-websocket" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/c9/d8/63e5535ab21dc4998ba1cfe13690ccf122883a38f025dca24d6e56c05eba/python_engineio-4.12.3.tar.gz", hash = "sha256:35633e55ec30915e7fc8f7e34ca8d73ee0c080cec8a8cd04faf2d7396f0a7a7a", size = 91910, upload-time = "2025-09-28T06:31:36.765Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/d8/f0/c5aa0a69fd9326f013110653543f36ece4913c17921f3e1dbd78e1b423ee/python_engineio-4.12.3-py3-none-any.whl", hash = "sha256:7c099abb2a27ea7ab429c04da86ab2d82698cdd6c52406cb73766fe454feb7e1", size = 59637, upload-time = "2025-09-28T06:31:35.354Z" },
+]
+
 [[package]]
 name = "python-http-client"
 version = "3.3.7"
@@ -5247,6 +5284,19 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/d9/4f/00be2196329ebbff56ce564aa94efb0fbc828d00de250b1980de1a34ab49/python_pptx-1.0.2-py3-none-any.whl", hash = "sha256:160838e0b8565a8b1f67947675886e9fea18aa5e795db7ae531606d68e785cba", size = 472788 },
 ]
 
+[[package]]
+name = "python-socketio"
+version = "5.13.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "bidict" },
+    { name = "python-engineio" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/21/1a/396d50ccf06ee539fa758ce5623b59a9cb27637fc4b2dc07ed08bf495e77/python_socketio-5.13.0.tar.gz", hash = "sha256:ac4e19a0302ae812e23b712ec8b6427ca0521f7c582d6abb096e36e24a263029", size = 121125, upload-time = "2025-04-12T15:46:59.933Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/3c/32/b4fb8585d1be0f68bde7e110dffbcf354915f77ad8c778563f0ad9655c02/python_socketio-5.13.0-py3-none-any.whl", hash = "sha256:51f68d6499f2df8524668c24bcec13ba1414117cfb3a90115c559b601ab10caf", size = 77800, upload-time = "2025-04-12T15:46:58.412Z" },
+]
+
 [[package]]
 name = "pytz"
 version = "2025.2"
@@ -5746,6 +5796,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/e0/f9/0595336914c5619e5f28a1fb793285925a8cd4b432c9da0a987836c7f822/shellingham-1.5.4-py2.py3-none-any.whl", hash = "sha256:7ecfff8f2fd72616f7481040475a65b2bf8af90a56c89140852d1120324e8686", size = 9755 },
 ]
 
+[[package]]
+name = "simple-websocket"
+version = "1.1.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "wsproto" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/b0/d4/bfa032f961103eba93de583b161f0e6a5b63cebb8f2c7d0c6e6efe1e3d2e/simple_websocket-1.1.0.tar.gz", hash = "sha256:7939234e7aa067c534abdab3a9ed933ec9ce4691b0713c78acb195560aa52ae4", size = 17300, upload-time = "2024-10-10T22:39:31.412Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/52/59/0782e51887ac6b07ffd1570e0364cf901ebc36345fea669969d2084baebb/simple_websocket-1.1.0-py3-none-any.whl", hash = "sha256:4af6069630a38ed6c561010f0e11a5bc0d4ca569b36306eb257cd9a192497c8c", size = 13842, upload-time = "2024-10-10T22:39:29.645Z" },
+]
+
 [[package]]
 name = "six"
 version = "1.17.0"
@@ -7166,6 +7228,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/ff/21/abdedb4cdf6ff41ebf01a74087740a709e2edb146490e4d9beea054b0b7a/wrapt-1.16.0-py3-none-any.whl", hash = "sha256:6906c4100a8fcbf2fa735f6059214bb13b97f75b1a61777fcf6432121ef12ef1", size = 23362 },
 ]
 
+[[package]]
+name = "wsproto"
+version = "1.2.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "h11" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/c9/4a/44d3c295350d776427904d73c189e10aeae66d7f555bb2feee16d1e4ba5a/wsproto-1.2.0.tar.gz", hash = "sha256:ad565f26ecb92588a3e43bc3d96164de84cd9902482b130d0ddbaa9664a85065", size = 53425, upload-time = "2022-08-23T19:58:21.447Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/78/58/e860788190eba3bcce367f74d29c4675466ce8dddfba85f7827588416f01/wsproto-1.2.0-py3-none-any.whl", hash = "sha256:b9acddd652b585d75b20477888c56642fdade28bdfd3579aa24a4d2c037dd736", size = 24226, upload-time = "2022-08-23T19:58:19.96Z" },
+]
+
 [[package]]
 name = "xinference-client"
 version = "1.2.2"

docker/.env.example

@@ -129,6 +129,10 @@ MIGRATION_ENABLED=true
 # The default value is 300 seconds.
 FILES_ACCESS_TIMEOUT=300
 
+# Collaboration mode toggle
+# To open collaboration features, you also need to set SERVER_WORKER_CLASS=geventwebsocket.gunicorn.workers.GeventWebSocketWorker
+ENABLE_COLLABORATION_MODE=false
+
 # Access token expiration time in minutes
 ACCESS_TOKEN_EXPIRE_MINUTES=60
 
@@ -164,6 +168,7 @@ SERVER_WORKER_AMOUNT=1
 # Modifying it may also decrease throughput.
 #
 # It is strongly discouraged to change this parameter.
+# If enable collaboration mode, it must be set to geventwebsocket.gunicorn.workers.GeventWebSocketWorker
 SERVER_WORKER_CLASS=gevent
 
 # Default number of worker connections, the default is 10.
@@ -401,6 +406,8 @@ CONSOLE_CORS_ALLOW_ORIGINS=*
 COOKIE_DOMAIN=
 # When the frontend and backend run on different subdomains, set NEXT_PUBLIC_COOKIE_DOMAIN=1.
 NEXT_PUBLIC_COOKIE_DOMAIN=
+# WebSocket server URL. 
+NEXT_PUBLIC_SOCKET_URL=ws://localhost
 NEXT_PUBLIC_BATCH_CONCURRENCY=5
 
 # ------------------------------

docker/docker-compose-template.yaml

@@ -139,6 +139,7 @@ services:
       APP_API_URL: ${APP_API_URL:-}
       AMPLITUDE_API_KEY: ${AMPLITUDE_API_KEY:-}
       NEXT_PUBLIC_COOKIE_DOMAIN: ${NEXT_PUBLIC_COOKIE_DOMAIN:-}
+      NEXT_PUBLIC_SOCKET_URL: ${NEXT_PUBLIC_SOCKET_URL:-ws://localhost}
       SENTRY_DSN: ${WEB_SENTRY_DSN:-}
       NEXT_TELEMETRY_DISABLED: ${NEXT_TELEMETRY_DISABLED:-0}
       TEXT_GENERATION_TIMEOUT_MS: ${TEXT_GENERATION_TIMEOUT_MS:-60000}

docker/docker-compose.yaml

@@ -33,6 +33,7 @@ x-shared-env: &shared-api-worker-env
   OPENAI_API_BASE: ${OPENAI_API_BASE:-https://api.openai.com/v1}
   MIGRATION_ENABLED: ${MIGRATION_ENABLED:-true}
   FILES_ACCESS_TIMEOUT: ${FILES_ACCESS_TIMEOUT:-300}
+  ENABLE_COLLABORATION_MODE: ${ENABLE_COLLABORATION_MODE:-false}
   ACCESS_TOKEN_EXPIRE_MINUTES: ${ACCESS_TOKEN_EXPIRE_MINUTES:-60}
   REFRESH_TOKEN_EXPIRE_DAYS: ${REFRESH_TOKEN_EXPIRE_DAYS:-30}
   APP_DEFAULT_ACTIVE_REQUESTS: ${APP_DEFAULT_ACTIVE_REQUESTS:-0}
@@ -109,6 +110,7 @@ x-shared-env: &shared-api-worker-env
   CONSOLE_CORS_ALLOW_ORIGINS: ${CONSOLE_CORS_ALLOW_ORIGINS:-*}
   COOKIE_DOMAIN: ${COOKIE_DOMAIN:-}
   NEXT_PUBLIC_COOKIE_DOMAIN: ${NEXT_PUBLIC_COOKIE_DOMAIN:-}
+  NEXT_PUBLIC_SOCKET_URL: ${NEXT_PUBLIC_SOCKET_URL:-ws://localhost}
   NEXT_PUBLIC_BATCH_CONCURRENCY: ${NEXT_PUBLIC_BATCH_CONCURRENCY:-5}
   STORAGE_TYPE: ${STORAGE_TYPE:-opendal}
   OPENDAL_SCHEME: ${OPENDAL_SCHEME:-fs}
@@ -823,6 +825,7 @@ services:
       APP_API_URL: ${APP_API_URL:-}
       AMPLITUDE_API_KEY: ${AMPLITUDE_API_KEY:-}
       NEXT_PUBLIC_COOKIE_DOMAIN: ${NEXT_PUBLIC_COOKIE_DOMAIN:-}
+      NEXT_PUBLIC_SOCKET_URL: ${NEXT_PUBLIC_SOCKET_URL:-}
       SENTRY_DSN: ${WEB_SENTRY_DSN:-}
       NEXT_TELEMETRY_DISABLED: ${NEXT_TELEMETRY_DISABLED:-0}
       TEXT_GENERATION_TIMEOUT_MS: ${TEXT_GENERATION_TIMEOUT_MS:-60000}

docker/nginx/conf.d/default.conf.template

@@ -14,6 +14,14 @@ server {
       include proxy.conf;
     }
 
+    location /socket.io/ {
+      proxy_pass http://api:5001;
+      include proxy.conf;
+      proxy_set_header Upgrade $http_upgrade;
+      proxy_set_header Connection "upgrade";
+      proxy_cache_bypass $http_upgrade;
+    }
+
     location /v1 {
       proxy_pass http://api:5001;
       include proxy.conf;

docker/nginx/proxy.conf.template

@@ -5,7 +5,7 @@ proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
 proxy_set_header X-Forwarded-Proto $scheme;
 proxy_set_header X-Forwarded-Port $server_port;
 proxy_http_version 1.1;
-proxy_set_header Connection "";
+# proxy_set_header Connection "";
 proxy_buffering off;
 proxy_read_timeout ${NGINX_PROXY_READ_TIMEOUT};
 proxy_send_timeout ${NGINX_PROXY_SEND_TIMEOUT};

web/.env.example

@@ -14,6 +14,8 @@ NEXT_PUBLIC_API_PREFIX=http://localhost:5001/console/api
 NEXT_PUBLIC_PUBLIC_API_PREFIX=http://localhost:5001/api
 # When the frontend and backend run on different subdomains, set NEXT_PUBLIC_COOKIE_DOMAIN=1.
 NEXT_PUBLIC_COOKIE_DOMAIN=
+# WebSocket server URL.
+NEXT_PUBLIC_SOCKET_URL=ws://localhost:5001
 
 # The API PREFIX for MARKETPLACE
 NEXT_PUBLIC_MARKETPLACE_API_PREFIX=https://marketplace.dify.ai/api/v1

web/README.md

@@ -43,6 +43,8 @@ NEXT_PUBLIC_EDITION=SELF_HOSTED
 # example: http://cloud.dify.ai/console/api
 NEXT_PUBLIC_API_PREFIX=http://localhost:5001/console/api
 NEXT_PUBLIC_COOKIE_DOMAIN=
+# WebSocket server URL.
+NEXT_PUBLIC_SOCKET_URL=ws://localhost:5001
 # The URL for Web APP, refers to the Web App base URL of WEB service if web app domain is different from
 # console or api domain.
 # example: http://udify.app/api

web/app/(commonLayout)/app/(appDetailLayout)/[appId]/overview/card-view.tsx

@@ -6,7 +6,7 @@ import type { UpdateAppSiteCodeResponse } from '@/models/app'
 import type { App } from '@/types/app'
 import type { I18nKeysByPrefix } from '@/types/i18n'
 import * as React from 'react'
-import { useCallback, useMemo } from 'react'
+import { useCallback, useEffect, useMemo } from 'react'
 import { useTranslation } from 'react-i18next'
 import { useContext } from 'use-context-selector'
 import AppCard from '@/app/components/app/overview/app-card'
@@ -15,6 +15,8 @@ import { useStore as useAppStore } from '@/app/components/app/store'
 import Loading from '@/app/components/base/loading'
 import { ToastContext } from '@/app/components/base/toast'
 import MCPServiceCard from '@/app/components/tools/mcp/mcp-service-card'
+import { collaborationManager } from '@/app/components/workflow/collaboration/core/collaboration-manager'
+import { webSocketClient } from '@/app/components/workflow/collaboration/core/websocket-manager'
 import { isTriggerNode } from '@/app/components/workflow/types'
 import { NEED_REFRESH_APP_LIST_KEY } from '@/config'
 import { useDocLink } from '@/context/i18n'
@@ -89,28 +91,59 @@ const CardView: FC<ICardViewProps> = ({ appId, isInPanel, className }) => {
     ? buildTriggerModeMessage(t('mcp.server.title', { ns: 'tools' }))
     : null
 
-  const updateAppDetail = async () => {
+  const updateAppDetail = useCallback(async () => {
     try {
       const res = await fetchAppDetail({ url: '/apps', id: appId })
       setAppDetail({ ...res })
     }
-    catch (error) { console.error(error) }
-  }
+    catch (error) {
+      console.error(error)
+    }
+  }, [appId, setAppDetail])
 
   const handleCallbackResult = (err: Error | null, message?: I18nKeysByPrefix<'common', 'actionMsg.'>) => {
     const type = err ? 'error' : 'success'
 
     message ||= (type === 'success' ? 'modifiedSuccessfully' : 'modifiedUnsuccessfully')
 
-    if (type === 'success')
+    if (type === 'success') {
       updateAppDetail()
 
+      // Emit collaboration event to notify other clients of app state changes
+      const socket = webSocketClient.getSocket(appId)
+      if (socket) {
+        socket.emit('collaboration_event', {
+          type: 'app_state_update',
+          data: { timestamp: Date.now() },
+          timestamp: Date.now(),
+        })
+      }
+    }
+
     notify({
       type,
       message: t(`actionMsg.${message}`, { ns: 'common' }) as string,
     })
   }
 
+  // Listen for collaborative app state updates from other clients
+  useEffect(() => {
+    if (!appId)
+      return
+
+    const unsubscribe = collaborationManager.onAppStateUpdate(async () => {
+      try {
+        // Update app detail when other clients modify app state
+        await updateAppDetail()
+      }
+      catch (error) {
+        console.error('app state update failed:', error)
+      }
+    })
+
+    return unsubscribe
+  }, [appId, updateAppDetail])
+
   const onChangeSiteStatus = async (value: boolean) => {
     const [err] = await asyncRunSafe<App>(
       updateAppSiteStatus({

web/app/components/app-sidebar/app-info.tsx

@@ -14,7 +14,7 @@ import {
 import dynamic from 'next/dynamic'
 import { useRouter } from 'next/navigation'
 import * as React from 'react'
-import { useCallback, useState } from 'react'
+import { useCallback, useEffect, useState } from 'react'
 import { useTranslation } from 'react-i18next'
 import { useContext } from 'use-context-selector'
 import CardView from '@/app/(commonLayout)/app/(appDetailLayout)/[appId]/overview/card-view'
@@ -22,10 +22,12 @@ import { useStore as useAppStore } from '@/app/components/app/store'
 import Button from '@/app/components/base/button'
 import ContentDialog from '@/app/components/base/content-dialog'
 import { ToastContext } from '@/app/components/base/toast'
+import { collaborationManager } from '@/app/components/workflow/collaboration/core/collaboration-manager'
+import { webSocketClient } from '@/app/components/workflow/collaboration/core/websocket-manager'
 import { NEED_REFRESH_APP_LIST_KEY } from '@/config'
 import { useAppContext } from '@/context/app-context'
 import { useProviderContext } from '@/context/provider-context'
-import { copyApp, deleteApp, exportAppConfig, updateAppInfo } from '@/service/apps'
+import { copyApp, deleteApp, exportAppConfig, fetchAppDetail, updateAppInfo } from '@/service/apps'
 import { useInvalidateAppList } from '@/service/use-apps'
 import { fetchWorkflowDraft } from '@/service/workflow'
 import { AppModeEnum } from '@/types/app'
@@ -77,6 +79,19 @@ const AppInfo = ({ expand, onlyShowDetail = false, openState = false, onDetailEx
   const [secretEnvList, setSecretEnvList] = useState<EnvironmentVariable[]>([])
   const [showExportWarning, setShowExportWarning] = useState(false)
 
+  const emitAppMetaUpdate = useCallback(() => {
+    if (!appDetail?.id)
+      return
+    const socket = webSocketClient.getSocket(appDetail.id)
+    if (socket) {
+      socket.emit('collaboration_event', {
+        type: 'app_meta_update',
+        data: { timestamp: Date.now() },
+        timestamp: Date.now(),
+      })
+    }
+  }, [appDetail])
+
   const onEdit: CreateAppModalProps['onConfirm'] = useCallback(async ({
     name,
     icon_type,
@@ -105,11 +120,12 @@ const AppInfo = ({ expand, onlyShowDetail = false, openState = false, onDetailEx
         message: t('editDone', { ns: 'app' }),
       })
       setAppDetail(app)
+      emitAppMetaUpdate()
     }
     catch {
       notify({ type: 'error', message: t('editFailed', { ns: 'app' }) })
     }
-  }, [appDetail, notify, setAppDetail, t])
+  }, [appDetail, notify, setAppDetail, t, emitAppMetaUpdate])
 
   const onCopy: DuplicateAppModalProps['onConfirm'] = async ({ name, icon_type, icon, icon_background }) => {
     if (!appDetail)
@@ -207,6 +223,23 @@ const AppInfo = ({ expand, onlyShowDetail = false, openState = false, onDetailEx
     setShowConfirmDelete(false)
   }, [appDetail, invalidateAppList, notify, onPlanInfoChanged, replace, setAppDetail, t])
 
+  useEffect(() => {
+    if (!appDetail?.id)
+      return
+
+    const unsubscribe = collaborationManager.onAppMetaUpdate(async () => {
+      try {
+        const res = await fetchAppDetail({ url: '/apps', id: appDetail.id })
+        setAppDetail({ ...res })
+      }
+      catch (error) {
+        console.error('failed to refresh app detail from collaboration update:', error)
+      }
+    })
+
+    return unsubscribe
+  }, [appDetail?.id, setAppDetail])
+
   const { isCurrentWorkspaceEditor } = useAppContext()
 
   if (!appDetail)

web/app/components/app/annotation/index.tsx

@@ -203,7 +203,7 @@ const Annotation: FC<Props> = (props) => {
         </Filter>
         {isLoading
           ? <Loading type="app" />
-          // eslint-disable-next-line sonarjs/no-nested-conditional
+
           : total > 0
             ? (
                 <List

web/app/components/app/app-publisher/features-wrapper.tsx

@@ -1,6 +1,7 @@
 import type { AppPublisherProps } from '@/app/components/app/app-publisher'
 import type { ModelAndParameter } from '@/app/components/app/configuration/debug/types'
 import type { FileUpload } from '@/app/components/base/features/types'
+import type { PublishWorkflowParams } from '@/types/workflow'
 import { produce } from 'immer'
 import * as React from 'react'
 import { useCallback, useState } from 'react'
@@ -13,7 +14,7 @@ import { SupportUploadFileTypes } from '@/app/components/workflow/types'
 import { Resolution } from '@/types/app'
 
 type Props = Omit<AppPublisherProps, 'onPublish'> & {
-  onPublish?: (modelAndParameter?: ModelAndParameter, features?: any) => Promise<any> | any
+  onPublish?: (params?: ModelAndParameter | PublishWorkflowParams, features?: any) => Promise<any> | any
   publishedConfig?: any
   resetAppConfig?: () => void
 }
@@ -62,8 +63,8 @@ const FeaturesWrappedAppPublisher = (props: Props) => {
     setRestoreConfirmOpen(false)
   }, [featuresStore, props])
 
-  const handlePublish = useCallback((modelAndParameter?: ModelAndParameter) => {
-    return props.onPublish?.(modelAndParameter, features)
+  const handlePublish = useCallback((params?: ModelAndParameter | PublishWorkflowParams) => {
+    return props.onPublish?.(params, features)
   }, [features, props])
 
   return (

web/app/components/app/app-publisher/index.tsx

@@ -1,5 +1,7 @@
 import type { ModelAndParameter } from '../configuration/debug/types'
+import type { CollaborationUpdate } from '@/app/components/workflow/collaboration/types/collaboration'
 import type { InputVar, Variable } from '@/app/components/workflow/types'
+import type { InstalledApp } from '@/models/explore'
 import type { I18nKeysByPrefix } from '@/types/i18n'
 import type { PublishWorkflowParams } from '@/types/workflow'
 import {
@@ -18,6 +20,7 @@ import { useKeyPress } from 'ahooks'
 import {
   memo,
   useCallback,
+  useContext,
   useEffect,
   useMemo,
   useState,
@@ -35,6 +38,9 @@ import {
 } from '@/app/components/base/portal-to-follow-elem'
 import UpgradeBtn from '@/app/components/billing/upgrade-btn'
 import WorkflowToolConfigureButton from '@/app/components/tools/workflow-tool/configure-button'
+import { collaborationManager } from '@/app/components/workflow/collaboration/core/collaboration-manager'
+import { webSocketClient } from '@/app/components/workflow/collaboration/core/websocket-manager'
+import { WorkflowContext } from '@/app/components/workflow/context'
 import { appDefaultIconBackground } from '@/config'
 import { useGlobalPublicStore } from '@/context/global-public-context'
 import { useAsyncWindowOpen } from '@/hooks/use-async-window-open'
@@ -43,6 +49,8 @@ import { AccessMode } from '@/models/access-control'
 import { useAppWhiteListSubjects, useGetUserCanAccessApp } from '@/service/access-control'
 import { fetchAppDetailDirect } from '@/service/apps'
 import { fetchInstalledAppList } from '@/service/explore'
+import { useInvalidateAppWorkflow } from '@/service/use-workflow'
+import { fetchPublishedWorkflow } from '@/service/workflow'
 import { AppModeEnum } from '@/types/app'
 import { basePath } from '@/utils/var'
 import Divider from '../../base/divider'
@@ -56,6 +64,10 @@ import SuggestedAction from './suggested-action'
 
 type AccessModeLabel = I18nKeysByPrefix<'app', 'accessControlDialog.accessItems.'>
 
+type InstalledAppsResponse = {
+  installed_apps?: InstalledApp[]
+}
+
 const ACCESS_MODE_MAP: Record<AccessMode, { label: AccessModeLabel, icon: React.ElementType }> = {
   [AccessMode.ORGANIZATION]: {
     label: 'organization',
@@ -102,8 +114,8 @@ export type AppPublisherProps = {
   debugWithMultipleModel?: boolean
   multipleModelConfigs?: ModelAndParameter[]
   /** modelAndParameter is passed when debugWithMultipleModel is true */
-  onPublish?: (params?: any) => Promise<any> | any
-  onRestore?: () => Promise<any> | any
+  onPublish?: (params?: ModelAndParameter | PublishWorkflowParams) => Promise<void> | void
+  onRestore?: () => Promise<void> | void
   onToggle?: (state: boolean) => void
   crossAxisOffset?: number
   toolPublished?: boolean
@@ -146,6 +158,7 @@ const AppPublisher = ({
   const [isAppAccessSet, setIsAppAccessSet] = useState(true)
   const [embeddingModalOpen, setEmbeddingModalOpen] = useState(false)
 
+  const workflowStore = useContext(WorkflowContext)
   const appDetail = useAppStore(state => state.appDetail)
   const setAppDetail = useAppStore(s => s.setAppDetail)
   const systemFeatures = useGlobalPublicStore(s => s.systemFeatures)
@@ -158,6 +171,7 @@ const AppPublisher = ({
 
   const { data: userCanAccessApp, isLoading: isGettingUserCanAccessApp, refetch } = useGetUserCanAccessApp({ appId: appDetail?.id, enabled: false })
   const { data: appAccessSubjects, isLoading: isGettingAppWhiteListSubjects } = useAppWhiteListSubjects(appDetail?.id, open && systemFeatures.webapp_auth.enabled && appDetail?.access_mode === AccessMode.SPECIFIC_GROUPS_MEMBERS)
+  const invalidateAppWorkflow = useInvalidateAppWorkflow()
   const openAsyncWindow = useAsyncWindowOpen()
 
   const noAccessPermission = useMemo(() => systemFeatures.webapp_auth.enabled && appDetail && appDetail.access_mode !== AccessMode.EXTERNAL_MEMBERS && !userCanAccessApp?.result, [systemFeatures, appDetail, userCanAccessApp])
@@ -193,12 +207,39 @@ const AppPublisher = ({
     try {
       await onPublish?.(params)
       setPublished(true)
+
+      const appId = appDetail?.id
+      const socket = appId ? webSocketClient.getSocket(appId) : null
+      console.warn('[app-publisher] publish success', {
+        appId,
+        hasSocket: Boolean(socket),
+      })
+      if (appId)
+        invalidateAppWorkflow(appId)
+      else
+        console.warn('[app-publisher] missing appId, skip workflow invalidate and socket emit')
+      if (socket) {
+        const timestamp = Date.now()
+        socket.emit('collaboration_event', {
+          type: 'app_publish_update',
+          data: {
+            action: 'published',
+            timestamp,
+          },
+          timestamp,
+        })
+      }
+      else if (appId) {
+        console.warn('[app-publisher] socket not ready, skip collaboration_event emit', { appId })
+      }
+
       trackEvent('app_published_time', { action_mode: 'app', app_id: appDetail?.id, app_name: appDetail?.name })
     }
-    catch {
+    catch (error) {
+      console.warn('[app-publisher] publish failed', error)
       setPublished(false)
     }
-  }, [appDetail, onPublish])
+  }, [appDetail, onPublish, invalidateAppWorkflow])
 
   const handleRestore = useCallback(async () => {
     try {
@@ -227,9 +268,10 @@ const AppPublisher = ({
     await openAsyncWindow(async () => {
       if (!appDetail?.id)
         throw new Error('App not found')
-      const { installed_apps }: any = await fetchInstalledAppList(appDetail?.id) || {}
-      if (installed_apps?.length > 0)
-        return `${basePath}/explore/installed/${installed_apps[0].id}`
+      const response = (await fetchInstalledAppList(appDetail?.id)) as InstalledAppsResponse
+      const installedApps = response?.installed_apps
+      if (installedApps?.length)
+        return `${basePath}/explore/installed/${installedApps[0].id}`
       throw new Error('No app found in Explore')
     }, {
       onError: (err) => {
@@ -257,6 +299,29 @@ const AppPublisher = ({
     handlePublish()
   }, { exactMatch: true, useCapture: true })
 
+  useEffect(() => {
+    const appId = appDetail?.id
+    if (!appId)
+      return
+
+    const unsubscribe = collaborationManager.onAppPublishUpdate((update: CollaborationUpdate) => {
+      const action = typeof update.data.action === 'string' ? update.data.action : undefined
+      if (action === 'published') {
+        invalidateAppWorkflow(appId)
+        fetchPublishedWorkflow(`/apps/${appId}/workflows/publish`)
+          .then((publishedWorkflow) => {
+            if (publishedWorkflow?.created_at)
+              workflowStore?.getState().setPublishedAt(publishedWorkflow.created_at)
+          })
+          .catch((error) => {
+            console.warn('[app-publisher] refresh published workflow failed', error)
+          })
+      }
+    })
+
+    return unsubscribe
+  }, [appDetail?.id, invalidateAppWorkflow, workflowStore])
+
   const hasPublishedVersion = !!publishedAt
   const workflowToolDisabled = !hasPublishedVersion || !workflowToolAvailable
   const workflowToolMessage = workflowToolDisabled ? t('common.workflowAsToolDisabledHint', { ns: 'workflow' }) : undefined

web/app/components/app/configuration/config/automatic/get-automatic-res.tsx

@@ -134,7 +134,6 @@ const GetAutomaticRes: FC<IGetAutomaticResProps> = ({
     },
   ] as const
 
-  // eslint-disable-next-line sonarjs/no-nested-template-literals, sonarjs/no-nested-conditional
   const [instructionFromSessionStorage, setInstruction] = useSessionStorageState<string>(`improve-instruction-${flowId}${isBasicMode ? '' : `-${nodeId}${editorId ? `-${editorId}` : ''}`}`)
   const instruction = instructionFromSessionStorage || ''
   const [ideaOutput, setIdeaOutput] = useState<string>('')

web/app/components/app/configuration/index.tsx

@@ -18,6 +18,7 @@ import type {
   TextToSpeechConfig,
 } from '@/models/debug'
 import type { ModelConfig as BackendModelConfig, UserInputFormItem, VisionSettings } from '@/types/app'
+import type { PublishWorkflowParams } from '@/types/workflow'
 import { CodeBracketIcon } from '@heroicons/react/20/solid'
 import { useBoolean, useGetState } from 'ahooks'
 import { clone } from 'es-toolkit/object'
@@ -760,7 +761,8 @@ const Configuration: FC = () => {
     else { return promptEmpty }
   })()
   const contextVarEmpty = mode === AppModeEnum.COMPLETION && dataSets.length > 0 && !hasSetContextVar
-  const onPublish = async (modelAndParameter?: ModelAndParameter, features?: FeaturesData) => {
+  const onPublish = async (params?: ModelAndParameter | PublishWorkflowParams, features?: FeaturesData) => {
+    const modelAndParameter = params && 'model' in params ? params : undefined
     const modelId = modelAndParameter?.model || modelConfig.model_id
     const promptTemplate = modelConfig.configs.prompt_template
     const promptVariables = modelConfig.configs.prompt_variables

web/app/components/app/overview/settings/index.spec.tsx

@@ -175,7 +175,7 @@ describe('SettingsModal', () => {
     renderSettingsModal()
     fireEvent.click(screen.getByText('appOverview.overview.appInfo.settings.more.entry'))
     const privacyInput = screen.getByPlaceholderText('appOverview.overview.appInfo.settings.more.privacyPolicyPlaceholder')
-    // eslint-disable-next-line sonarjs/no-clear-text-protocols
+
     fireEvent.change(privacyInput, { target: { value: 'ftp://invalid-url' } })
 
     fireEvent.click(screen.getByText('common.operation.save'))

web/app/components/apps/app-card.tsx

@@ -5,6 +5,7 @@ import type { HtmlContentProps } from '@/app/components/base/popover'
 import type { Tag } from '@/app/components/base/tag-management/constant'
 import type { CreateAppModalProps } from '@/app/components/explore/create-app-modal'
 import type { EnvironmentVariable } from '@/app/components/workflow/types'
+import type { WorkflowOnlineUser } from '@/models/app'
 import type { App } from '@/types/app'
 import { RiBuildingLine, RiGlobalLine, RiLockLine, RiMoreFill, RiVerifiedBadgeLine } from '@remixicon/react'
 import dynamic from 'next/dynamic'
@@ -20,6 +21,7 @@ import CustomPopover from '@/app/components/base/popover'
 import TagSelector from '@/app/components/base/tag-management/selector'
 import Toast, { ToastContext } from '@/app/components/base/toast'
 import Tooltip from '@/app/components/base/tooltip'
+import { UserAvatarList } from '@/app/components/base/user-avatar-list'
 import { NEED_REFRESH_APP_LIST_KEY } from '@/config'
 import { useAppContext } from '@/context/app-context'
 import { useGlobalPublicStore } from '@/context/global-public-context'
@@ -58,9 +60,10 @@ const AccessControl = dynamic(() => import('@/app/components/app/app-access-cont
 export type AppCardProps = {
   app: App
   onRefresh?: () => void
+  onlineUsers?: WorkflowOnlineUser[]
 }
 
-const AppCard = ({ app, onRefresh }: AppCardProps) => {
+const AppCard = ({ app, onRefresh, onlineUsers = [] }: AppCardProps) => {
   const { t } = useTranslation()
   const { notify } = useContext(ToastContext)
   const systemFeatures = useGlobalPublicStore(s => s.systemFeatures)
@@ -348,6 +351,19 @@ const AppCard = ({ app, onRefresh }: AppCardProps) => {
     return `${t('segment.editedAt', { ns: 'datasetDocuments' })} ${timeText}`
   }, [app.updated_at, app.created_at])
 
+  const onlineUserAvatars = useMemo(() => {
+    if (!onlineUsers.length)
+      return []
+
+    return onlineUsers
+      .map(user => ({
+        id: user.user_id || user.sid || '',
+        name: user.username || 'User',
+        avatar_url: user.avatar || undefined,
+      }))
+      .filter(user => !!user.id)
+  }, [onlineUsers])
+
   return (
     <>
       <div
@@ -400,6 +416,11 @@ const AppCard = ({ app, onRefresh }: AppCardProps) => {
               </Tooltip>
             )}
           </div>
+          <div>
+            {onlineUserAvatars.length > 0 && (
+              <UserAvatarList users={onlineUserAvatars} maxVisible={3} size={20} />
+            )}
+          </div>
         </div>
         <div className="title-wrapper h-[90px] px-[14px] text-xs leading-normal text-text-tertiary">
           <div

web/app/components/apps/list.spec.tsx

@@ -1,3 +1,4 @@
+import { QueryClient, QueryClientProvider } from '@tanstack/react-query'
 import { act, fireEvent, render, screen } from '@testing-library/react'
 import * as React from 'react'
 import { useStore as useTagStore } from '@/app/components/base/tag-management/store'
@@ -234,6 +235,21 @@ beforeAll(() => {
   } as unknown as typeof IntersectionObserver
 })
 
+const renderList = () => {
+  const queryClient = new QueryClient({
+    defaultOptions: {
+      queries: {
+        retry: false,
+      },
+    },
+  })
+  return render(
+    <QueryClientProvider client={queryClient}>
+      <List />
+    </QueryClientProvider>,
+  )
+}
+
 describe('List', () => {
   beforeEach(() => {
     vi.clearAllMocks()
@@ -260,13 +276,13 @@ describe('List', () => {
 
   describe('Rendering', () => {
     it('should render without crashing', () => {
-      render(<List />)
+      renderList()
       // Tab slider renders app type tabs
       expect(screen.getByText('app.types.all')).toBeInTheDocument()
     })
 
     it('should render tab slider with all app types', () => {
-      render(<List />)
+      renderList()
 
       expect(screen.getByText('app.types.all')).toBeInTheDocument()
       expect(screen.getByText('app.types.workflow')).toBeInTheDocument()
@@ -277,48 +293,48 @@ describe('List', () => {
     })
 
     it('should render search input', () => {
-      render(<List />)
+      renderList()
       // Input component renders a searchbox
       expect(screen.getByRole('textbox')).toBeInTheDocument()
     })
 
     it('should render tag filter', () => {
-      render(<List />)
+      renderList()
       // Tag filter renders with placeholder text
       expect(screen.getByText('common.tag.placeholder')).toBeInTheDocument()
     })
 
     it('should render created by me checkbox', () => {
-      render(<List />)
+      renderList()
       expect(screen.getByText('app.showMyCreatedAppsOnly')).toBeInTheDocument()
     })
 
     it('should render app cards when apps exist', () => {
-      render(<List />)
+      renderList()
 
       expect(screen.getByTestId('app-card-app-1')).toBeInTheDocument()
       expect(screen.getByTestId('app-card-app-2')).toBeInTheDocument()
     })
 
     it('should render new app card for editors', () => {
-      render(<List />)
+      renderList()
       expect(screen.getByTestId('new-app-card')).toBeInTheDocument()
     })
 
     it('should render footer when branding is disabled', () => {
-      render(<List />)
+      renderList()
       expect(screen.getByTestId('footer')).toBeInTheDocument()
     })
 
     it('should render drop DSL hint for editors', () => {
-      render(<List />)
+      renderList()
       expect(screen.getByText('app.newApp.dropDSLToCreateApp')).toBeInTheDocument()
     })
   })
 
   describe('Tab Navigation', () => {
     it('should call setActiveTab when tab is clicked', () => {
-      render(<List />)
+      renderList()
 
       fireEvent.click(screen.getByText('app.types.workflow'))
 
@@ -326,7 +342,7 @@ describe('List', () => {
     })
 
     it('should call setActiveTab for all tab', () => {
-      render(<List />)
+      renderList()
 
       fireEvent.click(screen.getByText('app.types.all'))
 
@@ -336,12 +352,12 @@ describe('List', () => {
 
   describe('Search Functionality', () => {
     it('should render search input field', () => {
-      render(<List />)
+      renderList()
       expect(screen.getByRole('textbox')).toBeInTheDocument()
     })
 
     it('should handle search input change', () => {
-      render(<List />)
+      renderList()
 
       const input = screen.getByRole('textbox')
       fireEvent.change(input, { target: { value: 'test search' } })
@@ -350,7 +366,7 @@ describe('List', () => {
     })
 
     it('should handle search input interaction', () => {
-      render(<List />)
+      renderList()
 
       const input = screen.getByRole('textbox')
       expect(input).toBeInTheDocument()
@@ -360,7 +376,7 @@ describe('List', () => {
       // Set initial keywords to make clear button visible
       mockQueryState.keywords = 'existing search'
 
-      render(<List />)
+      renderList()
 
       // Find and click clear button (Input component uses .group class for clear icon container)
       const clearButton = document.querySelector('.group')
@@ -375,12 +391,12 @@ describe('List', () => {
 
   describe('Tag Filter', () => {
     it('should render tag filter component', () => {
-      render(<List />)
+      renderList()
       expect(screen.getByText('common.tag.placeholder')).toBeInTheDocument()
     })
 
     it('should render tag filter with placeholder', () => {
-      render(<List />)
+      renderList()
 
       // Tag filter is rendered
       expect(screen.getByText('common.tag.placeholder')).toBeInTheDocument()
@@ -389,12 +405,12 @@ describe('List', () => {
 
   describe('Created By Me Filter', () => {
     it('should render checkbox with correct label', () => {
-      render(<List />)
+      renderList()
       expect(screen.getByText('app.showMyCreatedAppsOnly')).toBeInTheDocument()
     })
 
     it('should handle checkbox change', () => {
-      render(<List />)
+      renderList()
 
       // Checkbox component uses data-testid="checkbox-{id}"
       // CheckboxWithLabel doesn't pass testId, so id is undefined
@@ -409,7 +425,7 @@ describe('List', () => {
     it('should not render new app card for non-editors', () => {
       mockIsCurrentWorkspaceEditor.mockReturnValue(false)
 
-      render(<List />)
+      renderList()
 
       expect(screen.queryByTestId('new-app-card')).not.toBeInTheDocument()
     })
@@ -417,7 +433,7 @@ describe('List', () => {
     it('should not render drop DSL hint for non-editors', () => {
       mockIsCurrentWorkspaceEditor.mockReturnValue(false)
 
-      render(<List />)
+      renderList()
 
       expect(screen.queryByText(/drop dsl file to create app/i)).not.toBeInTheDocument()
     })
@@ -427,7 +443,7 @@ describe('List', () => {
     it('should redirect dataset operators to datasets page', () => {
       mockIsCurrentWorkspaceDatasetOperator.mockReturnValue(true)
 
-      render(<List />)
+      renderList()
 
       expect(mockReplace).toHaveBeenCalledWith('/datasets')
     })
@@ -437,7 +453,7 @@ describe('List', () => {
     it('should call refetch when refresh key is set in localStorage', () => {
       localStorage.setItem('needRefreshAppList', '1')
 
-      render(<List />)
+      renderList()
 
       expect(mockRefetch).toHaveBeenCalled()
       expect(localStorage.getItem('needRefreshAppList')).toBeNull()
@@ -446,22 +462,23 @@ describe('List', () => {
 
   describe('Edge Cases', () => {
     it('should handle multiple renders without issues', () => {
-      const { rerender } = render(<List />)
+      const { unmount } = renderList()
       expect(screen.getByText('app.types.all')).toBeInTheDocument()
 
-      rerender(<List />)
+      unmount()
+      renderList()
       expect(screen.getByText('app.types.all')).toBeInTheDocument()
     })
 
     it('should render app cards correctly', () => {
-      render(<List />)
+      renderList()
 
       expect(screen.getByText('Test App 1')).toBeInTheDocument()
       expect(screen.getByText('Test App 2')).toBeInTheDocument()
     })
 
     it('should render with all filter options visible', () => {
-      render(<List />)
+      renderList()
 
       expect(screen.getByRole('textbox')).toBeInTheDocument()
       expect(screen.getByText('common.tag.placeholder')).toBeInTheDocument()
@@ -471,14 +488,14 @@ describe('List', () => {
 
   describe('Dragging State', () => {
     it('should show drop hint when DSL feature is enabled for editors', () => {
-      render(<List />)
+      renderList()
       expect(screen.getByText('app.newApp.dropDSLToCreateApp')).toBeInTheDocument()
     })
   })
 
   describe('App Type Tabs', () => {
     it('should render all app type tabs', () => {
-      render(<List />)
+      renderList()
 
       expect(screen.getByText('app.types.all')).toBeInTheDocument()
       expect(screen.getByText('app.types.workflow')).toBeInTheDocument()
@@ -489,7 +506,7 @@ describe('List', () => {
     })
 
     it('should call setActiveTab for each app type', () => {
-      render(<List />)
+      renderList()
 
       const appTypeTexts = [
         { mode: AppModeEnum.WORKFLOW, text: 'app.types.workflow' },
@@ -508,7 +525,7 @@ describe('List', () => {
 
   describe('Search and Filter Integration', () => {
     it('should display search input with correct attributes', () => {
-      render(<List />)
+      renderList()
 
       const input = screen.getByRole('textbox')
       expect(input).toBeInTheDocument()
@@ -516,13 +533,13 @@ describe('List', () => {
     })
 
     it('should have tag filter component', () => {
-      render(<List />)
+      renderList()
 
       expect(screen.getByText('common.tag.placeholder')).toBeInTheDocument()
     })
 
     it('should display created by me label', () => {
-      render(<List />)
+      renderList()
 
       expect(screen.getByText('app.showMyCreatedAppsOnly')).toBeInTheDocument()
     })
@@ -530,14 +547,14 @@ describe('List', () => {
 
   describe('App List Display', () => {
     it('should display all app cards from data', () => {
-      render(<List />)
+      renderList()
 
       expect(screen.getByTestId('app-card-app-1')).toBeInTheDocument()
       expect(screen.getByTestId('app-card-app-2')).toBeInTheDocument()
     })
 
     it('should display app names correctly', () => {
-      render(<List />)
+      renderList()
 
       expect(screen.getByText('Test App 1')).toBeInTheDocument()
       expect(screen.getByText('Test App 2')).toBeInTheDocument()
@@ -546,7 +563,7 @@ describe('List', () => {
 
   describe('Footer Visibility', () => {
     it('should render footer when branding is disabled', () => {
-      render(<List />)
+      renderList()
 
       expect(screen.getByTestId('footer')).toBeInTheDocument()
     })
@@ -558,14 +575,14 @@ describe('List', () => {
   describe('Additional Coverage', () => {
     it('should render dragging state overlay when dragging', () => {
       mockDragging = true
-      const { container } = render(<List />)
+      const { container } = renderList()
 
       // Component should render successfully with dragging state
       expect(container).toBeInTheDocument()
     })
 
     it('should handle app mode filter in query params', () => {
-      render(<List />)
+      renderList()
 
       const workflowTab = screen.getByText('app.types.workflow')
       fireEvent.click(workflowTab)
@@ -574,7 +591,7 @@ describe('List', () => {
     })
 
     it('should render new app card for editors', () => {
-      render(<List />)
+      renderList()
 
       expect(screen.getByTestId('new-app-card')).toBeInTheDocument()
     })
@@ -582,7 +599,7 @@ describe('List', () => {
 
   describe('DSL File Drop', () => {
     it('should handle DSL file drop and show modal', () => {
-      render(<List />)
+      renderList()
 
       // Simulate DSL file drop via the callback
       const mockFile = new File(['test content'], 'test.yml', { type: 'application/yaml' })
@@ -596,7 +613,7 @@ describe('List', () => {
     })
 
     it('should close DSL modal when onClose is called', () => {
-      render(<List />)
+      renderList()
 
       // Open modal via DSL file drop
       const mockFile = new File(['test content'], 'test.yml', { type: 'application/yaml' })
@@ -614,7 +631,7 @@ describe('List', () => {
     })
 
     it('should close DSL modal and refetch when onSuccess is called', () => {
-      render(<List />)
+      renderList()
 
       // Open modal via DSL file drop
       const mockFile = new File(['test content'], 'test.yml', { type: 'application/yaml' })
@@ -637,7 +654,7 @@ describe('List', () => {
   describe('Tag Filter Change', () => {
     it('should handle tag filter value change', () => {
       vi.useFakeTimers()
-      render(<List />)
+      renderList()
 
       // TagFilter component is rendered
       expect(screen.getByTestId('tag-filter')).toBeInTheDocument()
@@ -661,7 +678,7 @@ describe('List', () => {
 
     it('should handle empty tag filter selection', () => {
       vi.useFakeTimers()
-      render(<List />)
+      renderList()
 
       // Trigger tag filter change with empty array
       act(() => {
@@ -683,7 +700,7 @@ describe('List', () => {
   describe('Infinite Scroll', () => {
     it('should call fetchNextPage when intersection observer triggers', () => {
       mockServiceState.hasNextPage = true
-      render(<List />)
+      renderList()
 
       // Simulate intersection
       if (intersectionCallback) {
@@ -700,7 +717,7 @@ describe('List', () => {
 
     it('should not call fetchNextPage when not intersecting', () => {
       mockServiceState.hasNextPage = true
-      render(<List />)
+      renderList()
 
       // Simulate non-intersection
       if (intersectionCallback) {
@@ -718,7 +735,7 @@ describe('List', () => {
     it('should not call fetchNextPage when loading', () => {
       mockServiceState.hasNextPage = true
       mockServiceState.isLoading = true
-      render(<List />)
+      renderList()
 
       if (intersectionCallback) {
         act(() => {
@@ -736,7 +753,7 @@ describe('List', () => {
   describe('Error State', () => {
     it('should handle error state in useEffect', () => {
       mockServiceState.error = new Error('Test error')
-      const { container } = render(<List />)
+      const { container } = renderList()
 
       // Component should still render
       expect(container).toBeInTheDocument()

web/app/components/apps/list.tsx

@@ -8,13 +8,14 @@ import {
   RiMessage3Line,
   RiRobot3Line,
 } from '@remixicon/react'
+import { useQuery } from '@tanstack/react-query'
 import { useDebounceFn } from 'ahooks'
 import dynamic from 'next/dynamic'
 import {
   useRouter,
 } from 'next/navigation'
 import { parseAsString, useQueryState } from 'nuqs'
-import { useCallback, useEffect, useRef, useState } from 'react'
+import { useCallback, useEffect, useMemo, useRef, useState } from 'react'
 import { useTranslation } from 'react-i18next'
 import Input from '@/app/components/base/input'
 import TabSliderNew from '@/app/components/base/tab-slider-new'
@@ -25,6 +26,7 @@ import { NEED_REFRESH_APP_LIST_KEY } from '@/config'
 import { useAppContext } from '@/context/app-context'
 import { useGlobalPublicStore } from '@/context/global-public-context'
 import { CheckModal } from '@/hooks/use-pay'
+import { fetchWorkflowOnlineUsers } from '@/service/apps'
 import { useInfiniteAppList } from '@/service/use-apps'
 import { AppModeEnum } from '@/types/app'
 import { cn } from '@/utils/classnames'
@@ -110,6 +112,37 @@ const List = () => {
     refetch,
   } = useInfiniteAppList(appListQueryParams, { enabled: !isCurrentWorkspaceDatasetOperator })
 
+  const apps = useMemo(() => data?.pages?.flatMap(page => page.data) ?? [], [data])
+
+  const workflowIds = useMemo(() => {
+    const ids = new Set<string>()
+    apps.forEach((appItem) => {
+      const workflowId = appItem.id
+      if (!workflowId)
+        return
+
+      if (appItem.mode === 'workflow' || appItem.mode === 'advanced-chat')
+        ids.add(workflowId)
+    })
+    return Array.from(ids)
+  }, [apps])
+
+  const { data: onlineUsersByWorkflow = {}, refetch: refreshOnlineUsers } = useQuery({
+    queryKey: ['apps', 'workflow-online-users', workflowIds],
+    queryFn: () => fetchWorkflowOnlineUsers({ workflowIds }),
+    enabled: workflowIds.length > 0,
+  })
+
+  useEffect(() => {
+    const timer = window.setInterval(() => {
+      refetch()
+      if (workflowIds.length)
+        refreshOnlineUsers()
+    }, 10000)
+
+    return () => window.clearInterval(timer)
+  }, [workflowIds.join(','), refetch, refreshOnlineUsers])
+
   const anchorRef = useRef<HTMLDivElement>(null)
   const options = [
     { value: 'all', text: t('types.all', { ns: 'app' }), icon: <RiApps2Line className="mr-1 h-[14px] w-[14px]" /> },
@@ -241,7 +274,7 @@ const List = () => {
 
             if (hasAnyApp) {
               return pages.flatMap(({ data: apps }) => apps).map(app => (
-                <AppCard key={app.id} app={app} onRefresh={refetch} />
+                <AppCard key={app.id} app={app} onRefresh={refetch} onlineUsers={onlineUsersByWorkflow?.[app.id] ?? []} />
               ))
             }
 

web/app/components/base/avatar/index.spec.tsx

@@ -35,12 +35,14 @@ describe('Avatar', () => {
       it.each([
         { size: undefined, expected: '30px', label: 'default (30px)' },
         { size: 50, expected: '50px', label: 'custom (50px)' },
-      ])('should apply $label size to img element', ({ size, expected }) => {
+      ])('should apply $label size to avatar container', ({ size, expected }) => {
         const props = { name: 'Test', avatar: 'https://example.com/avatar.jpg', size }
 
         render(<Avatar {...props} />)
 
-        expect(screen.getByRole('img')).toHaveStyle({
+        const img = screen.getByRole('img')
+        const wrapper = img.parentElement as HTMLElement
+        expect(wrapper).toHaveStyle({
           width: expected,
           height: expected,
           fontSize: expected,
@@ -60,7 +62,7 @@ describe('Avatar', () => {
     })
 
     describe('className prop', () => {
-      it('should merge className with default avatar classes on img', () => {
+      it('should merge className with default avatar classes on container', () => {
         const props = {
           name: 'Test',
           avatar: 'https://example.com/avatar.jpg',
@@ -70,8 +72,9 @@ describe('Avatar', () => {
         render(<Avatar {...props} />)
 
         const img = screen.getByRole('img')
-        expect(img).toHaveClass('custom-class')
-        expect(img).toHaveClass('shrink-0', 'flex', 'items-center', 'rounded-full', 'bg-primary-600')
+        const wrapper = img.parentElement as HTMLElement
+        expect(wrapper).toHaveClass('custom-class')
+        expect(wrapper).toHaveClass('shrink-0', 'flex', 'items-center', 'rounded-full', 'bg-primary-600')
       })
 
       it('should merge className with default avatar classes on fallback div', () => {
@@ -277,10 +280,11 @@ describe('Avatar', () => {
       render(<Avatar {...props} />)
 
       const img = screen.getByRole('img')
+      const wrapper = img.parentElement as HTMLElement
       expect(img).toHaveAttribute('alt', 'Test User')
       expect(img).toHaveAttribute('src', 'https://example.com/avatar.jpg')
-      expect(img).toHaveStyle({ width: '64px', height: '64px' })
-      expect(img).toHaveClass('custom-avatar')
+      expect(wrapper).toHaveStyle({ width: '64px', height: '64px' })
+      expect(wrapper).toHaveClass('custom-avatar')
 
       // Trigger load to verify onError callback
       fireEvent.load(img)

web/app/components/base/avatar/index.tsx

@@ -9,6 +9,7 @@ export type AvatarProps = {
   className?: string
   textClassName?: string
   onError?: (x: boolean) => void
+  backgroundColor?: string
 }
 const Avatar = ({
   name,
@@ -17,9 +18,18 @@ const Avatar = ({
   className,
   textClassName,
   onError,
+  backgroundColor,
 }: AvatarProps) => {
-  const avatarClassName = 'shrink-0 flex items-center rounded-full bg-primary-600'
-  const style = { width: `${size}px`, height: `${size}px`, fontSize: `${size}px`, lineHeight: `${size}px` }
+  const avatarClassName = backgroundColor
+    ? 'shrink-0 flex items-center rounded-full'
+    : 'shrink-0 flex items-center rounded-full bg-primary-600'
+  const style = {
+    width: `${size}px`,
+    height: `${size}px`,
+    fontSize: `${size}px`,
+    lineHeight: `${size}px`,
+    ...(backgroundColor && !avatar ? { backgroundColor } : {}),
+  }
   const [imgError, setImgError] = useState(false)
 
   const handleError = () => {
@@ -35,14 +45,18 @@ const Avatar = ({
 
   if (avatar && !imgError) {
     return (
-      <img
+      <span
         className={cn(avatarClassName, className)}
         style={style}
-        alt={name}
-        src={avatar}
-        onError={handleError}
-        onLoad={() => onError?.(false)}
-      />
+      >
+        <img
+          className="h-full w-full rounded-full object-cover"
+          alt={name}
+          src={avatar}
+          onError={handleError}
+          onLoad={() => onError?.(false)}
+        />
+      </span>
     )
   }
 

web/app/components/base/chat/chat/citation/popup.tsx

@@ -1,4 +1,4 @@
-import type { FC } from 'react'
+import type { FC, MouseEvent } from 'react'
 import type { Resources } from './index'
 import Link from 'next/link'
 import { Fragment, useState } from 'react'
@@ -18,6 +18,8 @@ import {
   PortalToFollowElemContent,
   PortalToFollowElemTrigger,
 } from '@/app/components/base/portal-to-follow-elem'
+import { useDocumentDownload } from '@/service/knowledge/use-document'
+import { downloadUrl } from '@/utils/download'
 import ProgressTooltip from './progress-tooltip'
 import Tooltip from './tooltip'
 
@@ -36,6 +38,30 @@ const Popup: FC<PopupProps> = ({
     ? (/\.([^.]*)$/.exec(data.documentName)?.[1] || '')
     : 'notion'
 
+  const { mutateAsync: downloadDocument, isPending: isDownloading } = useDocumentDownload()
+
+  /**
+   * Download the original uploaded file for citations whose data source is upload-file.
+   * We request a signed URL from the dataset document download endpoint, then trigger browser download.
+   */
+  const handleDownloadUploadFile = async (e: MouseEvent<HTMLElement>) => {
+    // Prevent toggling the citation popup when user clicks the download link.
+    e.preventDefault()
+    e.stopPropagation()
+
+    // Only upload-file citations can be downloaded this way (needs dataset/document ids).
+    const isUploadFile = data.dataSourceType === 'upload_file' || data.dataSourceType === 'file'
+    const datasetId = data.sources?.[0]?.dataset_id
+    const documentId = data.documentId || data.sources?.[0]?.document_id
+    if (!isUploadFile || !datasetId || !documentId || isDownloading)
+      return
+
+    // Fetch signed URL (usually points to `/files/<id>/file-preview?...&as_attachment=true`).
+    const res = await downloadDocument({ datasetId, documentId })
+    if (res?.url)
+      downloadUrl({ url: res.url, fileName: data.documentName })
+  }
+
   return (
     <PortalToFollowElem
       open={open}
@@ -49,6 +75,7 @@ const Popup: FC<PopupProps> = ({
       <PortalToFollowElemTrigger onClick={() => setOpen(v => !v)}>
         <div className="flex h-7 max-w-[240px] items-center rounded-lg bg-components-button-secondary-bg px-2">
           <FileIcon type={fileType} className="mr-1 h-4 w-4 shrink-0" />
+          {/* Keep the trigger purely for opening the popup (no download link here). */}
           <div className="truncate text-xs text-text-tertiary">{data.documentName}</div>
         </div>
       </PortalToFollowElemTrigger>
@@ -57,7 +84,21 @@ const Popup: FC<PopupProps> = ({
           <div className="px-4 pb-2 pt-3">
             <div className="flex h-[18px] items-center">
               <FileIcon type={fileType} className="mr-1 h-4 w-4 shrink-0" />
-              <div className="system-xs-medium truncate text-text-tertiary">{data.documentName}</div>
+              <div className="system-xs-medium truncate text-text-tertiary">
+                {/* If it's an upload-file reference, the title becomes a download link. */}
+                {(data.dataSourceType === 'upload_file' || data.dataSourceType === 'file') && !!data.sources?.[0]?.dataset_id
+                  ? (
+                      <button
+                        type="button"
+                        className="cursor-pointer truncate text-text-tertiary hover:underline"
+                        onClick={handleDownloadUploadFile}
+                        disabled={isDownloading}
+                      >
+                        {data.documentName}
+                      </button>
+                    )
+                  : data.documentName}
+              </div>
             </div>
           </div>
           <div className="max-h-[450px] overflow-y-auto rounded-lg bg-components-panel-bg px-4 py-0.5">

web/app/components/base/content-dialog/index.tsx

@@ -19,7 +19,7 @@ const ContentDialog = ({
     <Transition
       show={show}
       as="div"
-      className="absolute left-0 top-0 z-30 box-border h-full w-full p-2"
+      className="absolute left-0 top-0 z-[70] box-border h-full w-full p-2"
     >
       <TransitionChild>
         <div

web/app/components/base/icons/assets/public/common/enter-key.svg

@@ -0,0 +1,4 @@
+<svg width="16" height="16" viewBox="0 0 16 16" fill="none" xmlns="http://www.w3.org/2000/svg">
+<path d="M0 4C0 1.79086 1.79086 0 4 0H12C14.2091 0 16 1.79086 16 4V12C16 14.2091 14.2091 16 12 16H4C1.79086 16 0 14.2091 0 12V4Z" fill="white" fill-opacity="0.12"/>
+<path d="M3.42756 8.7358V7.62784H10.8764C11.2003 7.62784 11.4957 7.5483 11.7628 7.3892C12.0298 7.23011 12.2415 7.01705 12.3977 6.75C12.5568 6.48295 12.6364 6.1875 12.6364 5.86364C12.6364 5.53977 12.5568 5.24574 12.3977 4.98153C12.2386 4.71449 12.0256 4.50142 11.7585 4.34233C11.4943 4.18324 11.2003 4.10369 10.8764 4.10369H10.3991V3H10.8764C11.4048 3 11.8849 3.12926 12.3168 3.38778C12.7486 3.64631 13.0938 3.99148 13.3523 4.4233C13.6108 4.85511 13.7401 5.33523 13.7401 5.86364C13.7401 6.25852 13.6648 6.62926 13.5142 6.97585C13.3665 7.32244 13.1619 7.62784 12.9006 7.89205C12.6392 8.15625 12.3352 8.36364 11.9886 8.5142C11.642 8.66193 11.2713 8.7358 10.8764 8.7358H3.42756ZM6.16761 12.0554L2.29403 8.18182L6.16761 4.30824L6.9304 5.07102L3.81534 8.18182L6.9304 11.2926L6.16761 12.0554Z" fill="white"/>
+</svg>

web/app/components/base/icons/assets/public/other/comment.svg

@@ -0,0 +1,3 @@
+<svg xmlns="http://www.w3.org/2000/svg" width="14" height="12" viewBox="0 0 14 12" fill="none">
+<path d="M12.3334 4C12.3334 2.52725 11.1395 1.33333 9.66671 1.33333H4.33337C2.86062 1.33333 1.66671 2.52724 1.66671 4V10.6667H9.66671C11.1395 10.6667 12.3334 9.47274 12.3334 8V4ZM7.66671 6.66667V8H4.33337V6.66667H7.66671ZM9.66671 4V5.33333H4.33337V4H9.66671ZM13.6667 8C13.6667 10.2091 11.8758 12 9.66671 12H0.333374V4C0.333374 1.79086 2.12424 0 4.33337 0H9.66671C11.8758 0 13.6667 1.79086 13.6667 4V8Z" fill="currentColor"/>
+</svg>

web/app/components/base/icons/src/public/common/EnterKey.json

@@ -0,0 +1,36 @@
+{
+  "icon": {
+    "type": "element",
+    "isRootNode": true,
+    "name": "svg",
+    "attributes": {
+      "width": "16",
+      "height": "16",
+      "viewBox": "0 0 16 16",
+      "fill": "none",
+      "xmlns": "http://www.w3.org/2000/svg"
+    },
+    "children": [
+      {
+        "type": "element",
+        "name": "path",
+        "attributes": {
+          "d": "M0 4C0 1.79086 1.79086 0 4 0H12C14.2091 0 16 1.79086 16 4V12C16 14.2091 14.2091 16 12 16H4C1.79086 16 0 14.2091 0 12V4Z",
+          "fill": "white",
+          "fill-opacity": "0.12"
+        },
+        "children": []
+      },
+      {
+        "type": "element",
+        "name": "path",
+        "attributes": {
+          "d": "M3.42756 8.7358V7.62784H10.8764C11.2003 7.62784 11.4957 7.5483 11.7628 7.3892C12.0298 7.23011 12.2415 7.01705 12.3977 6.75C12.5568 6.48295 12.6364 6.1875 12.6364 5.86364C12.6364 5.53977 12.5568 5.24574 12.3977 4.98153C12.2386 4.71449 12.0256 4.50142 11.7585 4.34233C11.4943 4.18324 11.2003 4.10369 10.8764 4.10369H10.3991V3H10.8764C11.4048 3 11.8849 3.12926 12.3168 3.38778C12.7486 3.64631 13.0938 3.99148 13.3523 4.4233C13.6108 4.85511 13.7401 5.33523 13.7401 5.86364C13.7401 6.25852 13.6648 6.62926 13.5142 6.97585C13.3665 7.32244 13.1619 7.62784 12.9006 7.89205C12.6392 8.15625 12.3352 8.36364 11.9886 8.5142C11.642 8.66193 11.2713 8.7358 10.8764 8.7358H3.42756ZM6.16761 12.0554L2.29403 8.18182L6.16761 4.30824L6.9304 5.07102L3.81534 8.18182L6.9304 11.2926L6.16761 12.0554Z",
+          "fill": "white"
+        },
+        "children": []
+      }
+    ]
+  },
+  "name": "EnterKey"
+}

web/app/components/base/icons/src/public/common/EnterKey.tsx

@@ -0,0 +1,20 @@
+// GENERATE BY script
+// DON NOT EDIT IT MANUALLY
+
+import type { IconData } from '@/app/components/base/icons/IconBase'
+import * as React from 'react'
+import IconBase from '@/app/components/base/icons/IconBase'
+import data from './EnterKey.json'
+
+const Icon = (
+  {
+    ref,
+    ...props
+  }: React.SVGProps<SVGSVGElement> & {
+    ref?: React.RefObject<React.RefObject<HTMLOrSVGElement>>
+  },
+) => <IconBase {...props} ref={ref} data={data as IconData} />
+
+Icon.displayName = 'EnterKey'
+
+export default Icon

web/app/components/base/icons/src/public/common/index.ts

@@ -1,6 +1,7 @@
 export { default as D } from './D'
 export { default as DiagonalDividingLine } from './DiagonalDividingLine'
 export { default as Dify } from './Dify'
+export { default as EnterKey } from './EnterKey'
 export { default as Gdpr } from './Gdpr'
 export { default as Github } from './Github'
 export { default as Highlight } from './Highlight'

web/app/components/base/icons/src/public/other/Comment.json

@@ -0,0 +1,26 @@
+{
+  "icon": {
+    "type": "element",
+    "isRootNode": true,
+    "name": "svg",
+    "attributes": {
+      "xmlns": "http://www.w3.org/2000/svg",
+      "width": "14",
+      "height": "12",
+      "viewBox": "0 0 14 12",
+      "fill": "none"
+    },
+    "children": [
+      {
+        "type": "element",
+        "name": "path",
+        "attributes": {
+          "d": "M12.3334 4C12.3334 2.52725 11.1395 1.33333 9.66671 1.33333H4.33337C2.86062 1.33333 1.66671 2.52724 1.66671 4V10.6667H9.66671C11.1395 10.6667 12.3334 9.47274 12.3334 8V4ZM7.66671 6.66667V8H4.33337V6.66667H7.66671ZM9.66671 4V5.33333H4.33337V4H9.66671ZM13.6667 8C13.6667 10.2091 11.8758 12 9.66671 12H0.333374V4C0.333374 1.79086 2.12424 0 4.33337 0H9.66671C11.8758 0 13.6667 1.79086 13.6667 4V8Z",
+          "fill": "currentColor"
+        },
+        "children": []
+      }
+    ]
+  },
+  "name": "Comment"
+}

web/app/components/base/icons/src/public/other/Comment.tsx

@@ -0,0 +1,20 @@
+// GENERATE BY script
+// DON NOT EDIT IT MANUALLY
+
+import type { IconData } from '@/app/components/base/icons/IconBase'
+import * as React from 'react'
+import IconBase from '@/app/components/base/icons/IconBase'
+import data from './Comment.json'
+
+const Icon = (
+  {
+    ref,
+    ...props
+  }: React.SVGProps<SVGSVGElement> & {
+    ref?: React.RefObject<React.RefObject<HTMLOrSVGElement>>
+  },
+) => <IconBase {...props} ref={ref} data={data as IconData} />
+
+Icon.displayName = 'Comment'
+
+export default Icon

web/app/components/base/icons/src/public/other/index.ts

@@ -1,3 +1,4 @@
+export { default as Comment } from './Comment'
 export { default as DefaultToolIcon } from './DefaultToolIcon'
 export { default as Icon3Dots } from './Icon3Dots'
 export { default as Message3Fill } from './Message3Fill'

web/app/components/base/markdown-blocks/code-block.tsx

@@ -205,7 +205,7 @@ const CodeBlock: any = memo(({ inline, className, children = '', ...props }: any
       }
       catch {
         try {
-          // eslint-disable-next-line no-new-func, sonarjs/code-eval
+          // eslint-disable-next-line no-new-func
           const result = new Function(`return ${trimmedContent}`)()
           if (typeof result === 'object' && result !== null) {
             setFinalChartOption(result)
@@ -250,7 +250,7 @@ const CodeBlock: any = memo(({ inline, className, children = '', ...props }: any
       }
       catch {
         try {
-          // eslint-disable-next-line no-new-func, sonarjs/code-eval
+          // eslint-disable-next-line no-new-func
           const result = new Function(`return ${trimmedContent}`)()
           if (typeof result === 'object' && result !== null) {
             setFinalChartOption(result)

web/app/components/base/prompt-editor/index.tsx

@@ -17,6 +17,7 @@ import type {
 } from './types'
 import { CodeNode } from '@lexical/code'
 import { LexicalComposer } from '@lexical/react/LexicalComposer'
+import { useLexicalComposerContext } from '@lexical/react/LexicalComposerContext'
 import { ContentEditable } from '@lexical/react/LexicalContentEditable'
 import { LexicalErrorBoundary } from '@lexical/react/LexicalErrorBoundary'
 import { HistoryPlugin } from '@lexical/react/LexicalHistoryPlugin'
@@ -81,6 +82,29 @@ import {
 } from './plugins/workflow-variable-block'
 import { textToEditorState } from './utils'
 
+const ValueSyncPlugin: FC<{ value?: string }> = ({ value }) => {
+  const [editor] = useLexicalComposerContext()
+
+  useEffect(() => {
+    if (value === undefined)
+      return
+
+    const incomingValue = value ?? ''
+    const shouldUpdate = editor.getEditorState().read(() => {
+      const currentText = $getRoot().getChildren().map(node => node.getTextContent()).join('\n')
+      return currentText !== incomingValue
+    })
+
+    if (!shouldUpdate)
+      return
+
+    const editorState = editor.parseEditorState(textToEditorState(incomingValue))
+    editor.setEditorState(editorState)
+  }, [editor, value])
+
+  return null
+}
+
 export type PromptEditorProps = {
   instanceId?: string
   compact?: boolean
@@ -294,6 +318,7 @@ const PromptEditor: FC<PromptEditorProps> = ({
             <VariableValueBlock />
           )
         }
+        <ValueSyncPlugin value={value} />
         <OnChangePlugin onChange={handleEditorChange} />
         <OnBlurBlock onBlur={onBlur} onFocus={onFocus} />
         <UpdateBlock instanceId={instanceId} />

web/app/components/base/user-avatar-list/index.tsx

@@ -0,0 +1,79 @@
+import type { FC } from 'react'
+import { memo } from 'react'
+import Avatar from '@/app/components/base/avatar'
+import { getUserColor } from '@/app/components/workflow/collaboration/utils/user-color'
+import { useAppContext } from '@/context/app-context'
+
+type User = {
+  id: string
+  name: string
+  avatar_url?: string | null
+}
+
+type UserAvatarListProps = {
+  users: User[]
+  maxVisible?: number
+  size?: number
+  className?: string
+  showCount?: boolean
+}
+
+export const UserAvatarList: FC<UserAvatarListProps> = memo(({
+  users,
+  maxVisible = 3,
+  size = 24,
+  className = '',
+  showCount = true,
+}) => {
+  const { userProfile } = useAppContext()
+  if (!users.length)
+    return null
+
+  const shouldShowCount = showCount && users.length > maxVisible
+  const actualMaxVisible = shouldShowCount ? Math.max(1, maxVisible - 1) : maxVisible
+  const visibleUsers = users.slice(0, actualMaxVisible)
+  const remainingCount = users.length - actualMaxVisible
+
+  const currentUserId = userProfile?.id
+
+  return (
+    <div className={`flex items-center -space-x-1 ${className}`}>
+      {visibleUsers.map((user, index) => {
+        const isCurrentUser = user.id === currentUserId
+        const userColor = isCurrentUser ? undefined : getUserColor(user.id)
+        return (
+          <div
+            key={`${user.id}-${index}`}
+            className="relative"
+            style={{ zIndex: visibleUsers.length - index }}
+          >
+            <Avatar
+              name={user.name}
+              avatar={user.avatar_url || null}
+              size={size}
+              className="ring-2 ring-components-panel-bg"
+              backgroundColor={userColor}
+            />
+          </div>
+        )
+      },
+
+      )}
+      {shouldShowCount && remainingCount > 0 && (
+        <div
+          className="flex items-center justify-center rounded-full bg-gray-500 text-[10px] leading-none text-white ring-2 ring-components-panel-bg"
+          style={{
+            zIndex: 0,
+            width: size,
+            height: size,
+          }}
+        >
+          +
+          {remainingCount}
+        </div>
+      )}
+    </div>
+  )
+})
+
+UserAvatarList.displayName = 'UserAvatarList'

web/app/components/datasets/documents/components/list.tsx

@@ -30,9 +30,10 @@ import { useDatasetDetailContextWithSelector as useDatasetDetailContext } from '
 import useTimestamp from '@/hooks/use-timestamp'
 import { ChunkingMode, DataSourceType, DocumentActionType } from '@/models/datasets'
 import { DatasourceType } from '@/models/pipeline'
-import { useDocumentArchive, useDocumentBatchRetryIndex, useDocumentDelete, useDocumentDisable, useDocumentEnable } from '@/service/knowledge/use-document'
+import { useDocumentArchive, useDocumentBatchRetryIndex, useDocumentDelete, useDocumentDisable, useDocumentDownloadZip, useDocumentEnable } from '@/service/knowledge/use-document'
 import { asyncRunSafe } from '@/utils'
 import { cn } from '@/utils/classnames'
+import { downloadBlob } from '@/utils/download'
 import { formatNumber } from '@/utils/format'
 import BatchAction from '../detail/completed/common/batch-action'
 import StatusItem from '../status-item'
@@ -222,6 +223,7 @@ const DocumentList: FC<IDocumentListProps> = ({
   const { mutateAsync: disableDocument } = useDocumentDisable()
   const { mutateAsync: deleteDocument } = useDocumentDelete()
   const { mutateAsync: retryIndexDocument } = useDocumentBatchRetryIndex()
+  const { mutateAsync: requestDocumentsZip, isPending: isDownloadingZip } = useDocumentDownloadZip()
 
   const handleAction = (actionName: DocumentActionType) => {
     return async () => {
@@ -300,6 +302,39 @@ const DocumentList: FC<IDocumentListProps> = ({
     return dataSourceType === DatasourceType.onlineDrive
   }, [])
 
+  const downloadableSelectedIds = useMemo(() => {
+    const selectedSet = new Set(selectedIds)
+    return localDocs
+      .filter(doc => selectedSet.has(doc.id) && doc.data_source_type === DataSourceType.FILE)
+      .map(doc => doc.id)
+  }, [localDocs, selectedIds])
+
+  /**
+   * Generate a random ZIP filename for bulk document downloads.
+   * We intentionally avoid leaking dataset info in the exported archive name.
+   */
+  const generateDocsZipFileName = useCallback((): string => {
+    // Prefer UUID for uniqueness; fall back to time+random when unavailable.
+    const randomPart = (typeof crypto !== 'undefined' && typeof crypto.randomUUID === 'function')
+      ? crypto.randomUUID()
+      : `${Date.now().toString(36)}${Math.random().toString(36).slice(2, 10)}`
+    return `${randomPart}-docs.zip`
+  }, [])
+
+  const handleBatchDownload = useCallback(async () => {
+    if (isDownloadingZip)
+      return
+
+    // Download as a single ZIP to avoid browser caps on multiple automatic downloads.
+    const [e, blob] = await asyncRunSafe(requestDocumentsZip({ datasetId, documentIds: downloadableSelectedIds }))
+    if (e || !blob) {
+      Toast.notify({ type: 'error', message: t('actionMsg.downloadUnsuccessfully', { ns: 'common' }) })
+      return
+    }
+
+    downloadBlob({ data: blob, fileName: generateDocsZipFileName() })
+  }, [datasetId, downloadableSelectedIds, generateDocsZipFileName, isDownloadingZip, requestDocumentsZip, t])
+
   return (
     <div className="relative mt-3 flex h-full w-full flex-col">
       <div className="relative h-0 grow overflow-x-auto">
@@ -463,6 +498,7 @@ const DocumentList: FC<IDocumentListProps> = ({
           onArchive={handleAction(DocumentActionType.archive)}
           onBatchEnable={handleAction(DocumentActionType.enable)}
           onBatchDisable={handleAction(DocumentActionType.disable)}
+          onBatchDownload={downloadableSelectedIds.length > 0 ? handleBatchDownload : undefined}
           onBatchDelete={handleAction(DocumentActionType.delete)}
           onEditMetadata={showEditModal}
           onBatchReIndex={hasErrorDocumentsSelected ? handleBatchReIndex : undefined}

web/app/components/datasets/documents/components/operations.tsx

@@ -1,8 +1,10 @@
 import type { OperationName } from '../types'
 import type { CommonResponse } from '@/models/common'
+import type { DocumentDownloadResponse } from '@/service/datasets'
 import {
   RiArchive2Line,
   RiDeleteBinLine,
+  RiDownload2Line,
   RiEditLine,
   RiEqualizer2Line,
   RiLoopLeftLine,
@@ -28,6 +30,7 @@ import {
   useDocumentArchive,
   useDocumentDelete,
   useDocumentDisable,
+  useDocumentDownload,
   useDocumentEnable,
   useDocumentPause,
   useDocumentResume,
@@ -37,6 +40,7 @@ import {
 } from '@/service/knowledge/use-document'
 import { asyncRunSafe } from '@/utils'
 import { cn } from '@/utils/classnames'
+import { downloadUrl } from '@/utils/download'
 import s from '../style.module.css'
 import RenameModal from './rename-modal'
 
@@ -69,7 +73,7 @@ const Operations = ({
   scene = 'list',
   className = '',
 }: OperationsProps) => {
-  const { id, enabled = false, archived = false, data_source_type, display_status } = detail || {}
+  const { id, name, enabled = false, archived = false, data_source_type, display_status } = detail || {}
   const [showModal, setShowModal] = useState(false)
   const [deleting, setDeleting] = useState(false)
   const { notify } = useContext(ToastContext)
@@ -80,6 +84,7 @@ const Operations = ({
   const { mutateAsync: enableDocument } = useDocumentEnable()
   const { mutateAsync: disableDocument } = useDocumentDisable()
   const { mutateAsync: deleteDocument } = useDocumentDelete()
+  const { mutateAsync: downloadDocument, isPending: isDownloading } = useDocumentDownload()
   const { mutateAsync: syncDocument } = useSyncDocument()
   const { mutateAsync: syncWebsite } = useSyncWebsite()
   const { mutateAsync: pauseDocument } = useDocumentPause()
@@ -158,6 +163,24 @@ const Operations = ({
     onUpdate()
   }, [onUpdate])
 
+  const handleDownload = useCallback(async () => {
+    // Avoid repeated clicks while the signed URL request is in-flight.
+    if (isDownloading)
+      return
+
+    // Request a signed URL first (it points to `/files/<id>/file-preview?...&as_attachment=true`).
+    const [e, res] = await asyncRunSafe<DocumentDownloadResponse>(
+      downloadDocument({ datasetId, documentId: id }) as Promise<DocumentDownloadResponse>,
+    )
+    if (e || !res?.url) {
+      notify({ type: 'error', message: t('actionMsg.downloadUnsuccessfully', { ns: 'common' }) })
+      return
+    }
+
+    // Trigger download without navigating away (helps avoid duplicate downloads in some browsers).
+    downloadUrl({ url: res.url, fileName: name })
+  }, [datasetId, downloadDocument, id, isDownloading, name, notify, t])
+
   return (
     <div className="flex items-center" onClick={e => e.stopPropagation()}>
       {isListScene && !embeddingAvailable && (
@@ -214,6 +237,20 @@ const Operations = ({
                       <RiEditLine className="h-4 w-4 text-text-tertiary" />
                       <span className={s.actionName}>{t('list.table.rename', { ns: 'datasetDocuments' })}</span>
                     </div>
+                    {data_source_type === DataSourceType.FILE && (
+                      <div
+                        className={s.actionItem}
+                        onClick={(evt) => {
+                          evt.preventDefault()
+                          evt.stopPropagation()
+                          evt.nativeEvent.stopImmediatePropagation?.()
+                          handleDownload()
+                        }}
+                      >
+                        <RiDownload2Line className="h-4 w-4 text-text-tertiary" />
+                        <span className={s.actionName}>{t('list.action.download', { ns: 'datasetDocuments' })}</span>
+                      </div>
+                    )}
                     {['notion_import', DataSourceType.WEB].includes(data_source_type) && (
                       <div className={s.actionItem} onClick={() => onOperate('sync')}>
                         <RiLoopLeftLine className="h-4 w-4 text-text-tertiary" />
@@ -223,6 +260,23 @@ const Operations = ({
                     <Divider className="my-1" />
                   </>
                 )}
+                {archived && data_source_type === DataSourceType.FILE && (
+                  <>
+                    <div
+                      className={s.actionItem}
+                      onClick={(evt) => {
+                        evt.preventDefault()
+                        evt.stopPropagation()
+                        evt.nativeEvent.stopImmediatePropagation?.()
+                        handleDownload()
+                      }}
+                    >
+                      <RiDownload2Line className="h-4 w-4 text-text-tertiary" />
+                      <span className={s.actionName}>{t('list.action.download', { ns: 'datasetDocuments' })}</span>
+                    </div>
+                    <Divider className="my-1" />
+                  </>
+                )}
                 {!archived && display_status?.toLowerCase() === 'indexing' && (
                   <div className={s.actionItem} onClick={() => onOperate('pause')}>
                     <RiPauseCircleLine className="h-4 w-4 text-text-tertiary" />

web/app/components/datasets/documents/create-from-pipeline/processing/embedding-process/rule-detail.tsx

@@ -28,10 +28,10 @@ const RuleDetail = ({
       case 'mode':
         value = !sourceData?.mode
           ? value
-          // eslint-disable-next-line sonarjs/no-nested-conditional
+
           : sourceData.mode === ProcessMode.general
             ? (t('embedding.custom', { ns: 'datasetDocuments' }) as string)
-            // eslint-disable-next-line sonarjs/no-nested-conditional
+
             : `${t('embedding.hierarchical', { ns: 'datasetDocuments' })} · ${sourceData?.rules?.parent_mode === 'paragraph'
               ? t('parentMode.paragraph', { ns: 'dataset' })
               : t('parentMode.fullDoc', { ns: 'dataset' })}`
@@ -70,7 +70,7 @@ const RuleDetail = ({
             src={
               retrievalMethod === RETRIEVE_METHOD.fullText
                 ? retrievalIcon.fullText
-                // eslint-disable-next-line sonarjs/no-nested-conditional
+
                 : retrievalMethod === RETRIEVE_METHOD.hybrid
                   ? retrievalIcon.hybrid
                   : retrievalIcon.vector

web/app/components/datasets/documents/detail/completed/common/batch-action.tsx

@@ -1,5 +1,5 @@
 import type { FC } from 'react'
-import { RiArchive2Line, RiCheckboxCircleLine, RiCloseCircleLine, RiDeleteBinLine, RiDraftLine, RiRefreshLine } from '@remixicon/react'
+import { RiArchive2Line, RiCheckboxCircleLine, RiCloseCircleLine, RiDeleteBinLine, RiDownload2Line, RiDraftLine, RiRefreshLine } from '@remixicon/react'
 import { useBoolean } from 'ahooks'
 import * as React from 'react'
 import { useTranslation } from 'react-i18next'
@@ -14,6 +14,7 @@ type IBatchActionProps = {
   selectedIds: string[]
   onBatchEnable: () => void
   onBatchDisable: () => void
+  onBatchDownload?: () => void
   onBatchDelete: () => Promise<void>
   onArchive?: () => void
   onEditMetadata?: () => void
@@ -26,6 +27,7 @@ const BatchAction: FC<IBatchActionProps> = ({
   selectedIds,
   onBatchEnable,
   onBatchDisable,
+  onBatchDownload,
   onArchive,
   onBatchDelete,
   onEditMetadata,
@@ -103,6 +105,16 @@ const BatchAction: FC<IBatchActionProps> = ({
             <span className="px-0.5">{t(`${i18nPrefix}.reIndex`, { ns: 'dataset' })}</span>
           </Button>
         )}
+        {onBatchDownload && (
+          <Button
+            variant="ghost"
+            className="gap-x-0.5 px-3"
+            onClick={onBatchDownload}
+          >
+            <RiDownload2Line className="size-4" />
+            <span className="px-0.5">{t(`${i18nPrefix}.download`, { ns: 'dataset' })}</span>
+          </Button>
+        )}
         <Button
           variant="ghost"
           destructive

web/app/components/datasets/extra-info/api-access/card.tsx

@@ -0,0 +1,92 @@
+import { RiArrowRightUpLine, RiBookOpenLine } from '@remixicon/react'
+import Link from 'next/link'
+import * as React from 'react'
+import { useCallback } from 'react'
+import { useTranslation } from 'react-i18next'
+import Switch from '@/app/components/base/switch'
+import Indicator from '@/app/components/header/indicator'
+import { useSelector as useAppContextSelector } from '@/context/app-context'
+import { useDatasetDetailContextWithSelector } from '@/context/dataset-detail'
+import { useDatasetApiAccessUrl } from '@/hooks/use-api-access-url'
+import { useDisableDatasetServiceApi, useEnableDatasetServiceApi } from '@/service/knowledge/use-dataset'
+import { cn } from '@/utils/classnames'
+
+type CardProps = {
+  apiEnabled: boolean
+}
+
+const Card = ({
+  apiEnabled,
+}: CardProps) => {
+  const { t } = useTranslation()
+  const datasetId = useDatasetDetailContextWithSelector(state => state.dataset?.id)
+  const mutateDatasetRes = useDatasetDetailContextWithSelector(state => state.mutateDatasetRes)
+  const { mutateAsync: enableDatasetServiceApi } = useEnableDatasetServiceApi()
+  const { mutateAsync: disableDatasetServiceApi } = useDisableDatasetServiceApi()
+
+  const isCurrentWorkspaceManager = useAppContextSelector(state => state.isCurrentWorkspaceManager)
+
+  const apiReferenceUrl = useDatasetApiAccessUrl()
+
+  const onToggle = useCallback(async (state: boolean) => {
+    let result: 'success' | 'fail'
+    if (state)
+      result = (await enableDatasetServiceApi(datasetId ?? '')).result
+    else
+      result = (await disableDatasetServiceApi(datasetId ?? '')).result
+    if (result === 'success')
+      mutateDatasetRes?.()
+  }, [datasetId, enableDatasetServiceApi, mutateDatasetRes, disableDatasetServiceApi])
+
+  return (
+    <div className="w-[208px] rounded-xl border-[0.5px] border-components-panel-border bg-components-panel-bg-blur shadow-lg">
+      <div className="p-1">
+        <div className="p-2">
+          <div className="mb-1.5 flex justify-between">
+            <div className="flex items-center gap-1">
+              <Indicator
+                className="shrink-0"
+                color={apiEnabled ? 'green' : 'yellow'}
+              />
+              <div
+                className={cn(
+                  'system-xs-semibold-uppercase',
+                  apiEnabled ? 'text-text-success' : 'text-text-warning',
+                )}
+              >
+                {apiEnabled
+                  ? t('serviceApi.enabled', { ns: 'dataset' })
+                  : t('serviceApi.disabled', { ns: 'dataset' })}
+              </div>
+            </div>
+            <Switch
+              defaultValue={apiEnabled}
+              onChange={onToggle}
+              disabled={!isCurrentWorkspaceManager}
+            />
+          </div>
+          <div className="system-xs-regular text-text-tertiary">
+            {t('appMenus.apiAccessTip', { ns: 'common' })}
+          </div>
+        </div>
+      </div>
+      <div className="h-px bg-divider-subtle"></div>
+      <div className="p-1">
+        <Link
+          href={apiReferenceUrl}
+          target="_blank"
+          rel="noopener noreferrer"
+          className="flex h-8 items-center space-x-[7px] rounded-lg px-2 text-text-tertiary hover:bg-state-base-hover"
+        >
+          <RiBookOpenLine className="size-3.5 shrink-0" />
+          <div className="system-sm-regular grow truncate">
+            {t('overview.apiInfo.doc', { ns: 'appOverview' })}
+          </div>
+          <RiArrowRightUpLine className="size-3.5 shrink-0" />
+        </Link>
+      </div>
+    </div>
+  )
+}
+
+export default React.memo(Card)

web/app/components/datasets/extra-info/api-access/index.tsx

@@ -0,0 +1,65 @@
+import * as React from 'react'
+import { useState } from 'react'
+import { useTranslation } from 'react-i18next'
+import { ApiAggregate } from '@/app/components/base/icons/src/vender/knowledge'
+import { PortalToFollowElem, PortalToFollowElemContent, PortalToFollowElemTrigger } from '@/app/components/base/portal-to-follow-elem'
+import Indicator from '@/app/components/header/indicator'
+import { cn } from '@/utils/classnames'
+import Card from './card'
+
+type ApiAccessProps = {
+  expand: boolean
+  apiEnabled: boolean
+}
+
+const ApiAccess = ({
+  expand,
+  apiEnabled,
+}: ApiAccessProps) => {
+  const { t } = useTranslation()
+  const [open, setOpen] = useState(false)
+
+  const handleToggle = () => {
+    setOpen(!open)
+  }
+
+  return (
+    <div className="p-3 pt-2">
+      <PortalToFollowElem
+        open={open}
+        onOpenChange={setOpen}
+        placement="top-start"
+        offset={{
+          mainAxis: 4,
+          crossAxis: -4,
+        }}
+      >
+        <PortalToFollowElemTrigger
+          className="w-full"
+          onClick={handleToggle}
+        >
+          <div className={cn(
+            'relative flex h-8 cursor-pointer items-center gap-2 rounded-lg border border-components-panel-border px-3',
+            !expand && 'w-8 justify-center',
+            open ? 'bg-state-base-hover' : 'hover:bg-state-base-hover',
+          )}
+          >
+            <ApiAggregate className="size-4 shrink-0 text-text-secondary" />
+            {expand && <div className="system-sm-medium grow text-text-secondary">{t('appMenus.apiAccess', { ns: 'common' })}</div>}
+            <Indicator
+              className={cn('shrink-0', !expand && 'absolute -right-px -top-px')}
+              color={apiEnabled ? 'green' : 'yellow'}
+            />
+          </div>
+        </PortalToFollowElemTrigger>
+        <PortalToFollowElemContent className="z-[10]">
+          <Card
+            apiEnabled={apiEnabled}
+          />
+        </PortalToFollowElemContent>
+      </PortalToFollowElem>
+    </div>
+  )
+}
+
+export default React.memo(ApiAccess)

web/app/components/datasets/extra-info/index.tsx

@@ -1,8 +1,7 @@
 import type { RelatedAppResponse } from '@/models/datasets'
 import * as React from 'react'
 import { useDatasetDetailContextWithSelector } from '@/context/dataset-detail'
-import { useDatasetApiBaseUrl } from '@/service/knowledge/use-dataset'
-import ServiceApi from './service-api'
+import ApiAccess from './api-access'
 import Statistics from './statistics'
 
 type IExtraInfoProps = {
@@ -17,7 +16,6 @@ const ExtraInfo = ({
   expand,
 }: IExtraInfoProps) => {
   const apiEnabled = useDatasetDetailContextWithSelector(state => state.dataset?.enable_api)
-  const { data: apiBaseInfo } = useDatasetApiBaseUrl()
 
   return (
     <>
@@ -28,9 +26,8 @@ const ExtraInfo = ({
           relatedApps={relatedApps}
         />
       )}
-      <ServiceApi
+      <ApiAccess
         expand={expand}
-        apiBaseUrl={apiBaseInfo?.api_base_url ?? ''}
         apiEnabled={apiEnabled ?? false}
       />
     </>

web/app/components/datasets/extra-info/service-api/card.tsx

@@ -6,45 +6,22 @@ import { useTranslation } from 'react-i18next'
 import Button from '@/app/components/base/button'
 import CopyFeedback from '@/app/components/base/copy-feedback'
 import { ApiAggregate } from '@/app/components/base/icons/src/vender/knowledge'
-import Switch from '@/app/components/base/switch'
 import SecretKeyModal from '@/app/components/develop/secret-key/secret-key-modal'
 import Indicator from '@/app/components/header/indicator'
-import { useSelector as useAppContextSelector } from '@/context/app-context'
-import { useDatasetDetailContextWithSelector } from '@/context/dataset-detail'
 import { useDatasetApiAccessUrl } from '@/hooks/use-api-access-url'
-import { useDisableDatasetServiceApi, useEnableDatasetServiceApi } from '@/service/knowledge/use-dataset'
-import { cn } from '@/utils/classnames'
 
 type CardProps = {
-  apiEnabled: boolean
   apiBaseUrl: string
 }
 
 const Card = ({
-  apiEnabled,
   apiBaseUrl,
 }: CardProps) => {
   const { t } = useTranslation()
-  const datasetId = useDatasetDetailContextWithSelector(state => state.dataset?.id)
-  const mutateDatasetRes = useDatasetDetailContextWithSelector(state => state.mutateDatasetRes)
-  const { mutateAsync: enableDatasetServiceApi } = useEnableDatasetServiceApi()
-  const { mutateAsync: disableDatasetServiceApi } = useDisableDatasetServiceApi()
   const [isSecretKeyModalVisible, setIsSecretKeyModalVisible] = useState(false)
 
-  const isCurrentWorkspaceManager = useAppContextSelector(state => state.isCurrentWorkspaceManager)
-
   const apiReferenceUrl = useDatasetApiAccessUrl()
 
-  const onToggle = useCallback(async (state: boolean) => {
-    let result: 'success' | 'fail'
-    if (state)
-      result = (await enableDatasetServiceApi(datasetId ?? '')).result
-    else
-      result = (await disableDatasetServiceApi(datasetId ?? '')).result
-    if (result === 'success')
-      mutateDatasetRes?.()
-  }, [datasetId, enableDatasetServiceApi, disableDatasetServiceApi])
-
   const handleOpenSecretKeyModal = useCallback(() => {
     setIsSecretKeyModalVisible(true)
   }, [])
@@ -68,24 +45,16 @@ const Card = ({
           <div className="flex items-center gap-x-1">
             <Indicator
               className="shrink-0"
-              color={apiEnabled ? 'green' : 'yellow'}
+              color={
+                apiBaseUrl ? 'green' : 'yellow'
+              }
             />
             <div
-              className={cn(
-                'system-xs-semibold-uppercase',
-                apiEnabled ? 'text-text-success' : 'text-text-warning',
-              )}
+              className="system-xs-semibold-uppercase text-text-success"
             >
-              {apiEnabled
-                ? t('serviceApi.enabled', { ns: 'dataset' })
-                : t('serviceApi.disabled', { ns: 'dataset' })}
+              {t('serviceApi.enabled', { ns: 'dataset' })}
             </div>
           </div>
-          <Switch
-            defaultValue={apiEnabled}
-            onChange={onToggle}
-            disabled={!isCurrentWorkspaceManager}
-          />
         </div>
         <div className="flex flex-col">
           <div className="system-xs-regular leading-6 text-text-tertiary">

web/app/components/datasets/extra-info/service-api/index.tsx

@@ -1,22 +1,17 @@
 import * as React from 'react'
 import { useState } from 'react'
 import { useTranslation } from 'react-i18next'
-import { ApiAggregate } from '@/app/components/base/icons/src/vender/knowledge'
 import { PortalToFollowElem, PortalToFollowElemContent, PortalToFollowElemTrigger } from '@/app/components/base/portal-to-follow-elem'
 import Indicator from '@/app/components/header/indicator'
 import { cn } from '@/utils/classnames'
 import Card from './card'
 
 type ServiceApiProps = {
-  expand: boolean
   apiBaseUrl: string
-  apiEnabled: boolean
 }
 
 const ServiceApi = ({
-  expand,
   apiBaseUrl,
-  apiEnabled,
 }: ServiceApiProps) => {
   const { t } = useTranslation()
   const [open, setOpen] = useState(false)
@@ -26,7 +21,7 @@ const ServiceApi = ({
   }
 
   return (
-    <div className="p-3 pt-2">
+    <div>
       <PortalToFollowElem
         open={open}
         onOpenChange={setOpen}
@@ -41,22 +36,21 @@ const ServiceApi = ({
           onClick={handleToggle}
         >
           <div className={cn(
-            'relative flex h-8 cursor-pointer items-center gap-2 rounded-lg border border-components-panel-border px-3',
-            !expand && 'w-8 justify-center',
-            open ? 'bg-state-base-hover' : 'hover:bg-state-base-hover',
+            'relative flex h-8 cursor-pointer items-center gap-2 rounded-lg border-[0.5px] border-components-button-secondary-border-hover bg-components-button-secondary-bg px-3',
+            open ? 'bg-components-button-secondary-bg-hover' : 'hover:bg-components-button-secondary-bg-hover',
           )}
           >
-            <ApiAggregate className="size-4 shrink-0 text-text-secondary" />
-            {expand && <div className="system-sm-medium grow text-text-secondary">{t('serviceApi.title', { ns: 'dataset' })}</div>}
             <Indicator
-              className={cn('shrink-0', !expand && 'absolute -right-px -top-px')}
-              color={apiEnabled ? 'green' : 'yellow'}
+              className={cn('shrink-0')}
+              color={
+                apiBaseUrl ? 'green' : 'yellow'
+              }
             />
+            <div className="system-sm-medium grow text-text-secondary">{t('serviceApi.title', { ns: 'dataset' })}</div>
           </div>
         </PortalToFollowElemTrigger>
         <PortalToFollowElemContent className="z-[10]">
           <Card
-            apiEnabled={apiEnabled}
             apiBaseUrl={apiBaseUrl}
           />
         </PortalToFollowElemContent>

web/app/components/datasets/list/index.tsx

@@ -14,13 +14,14 @@ import TagFilter from '@/app/components/base/tag-management/filter'
 // Hooks
 import { useStore as useTagStore } from '@/app/components/base/tag-management/store'
 import CheckboxWithLabel from '@/app/components/datasets/create/website/base/checkbox-with-label'
-import { useAppContext } from '@/context/app-context'
+import { useAppContext, useSelector as useAppContextSelector } from '@/context/app-context'
 import { useExternalApiPanel } from '@/context/external-api-panel-context'
-
 import { useGlobalPublicStore } from '@/context/global-public-context'
 import useDocumentTitle from '@/hooks/use-document-title'
+import { useDatasetApiBaseUrl } from '@/service/knowledge/use-dataset'
 // Components
 import ExternalAPIPanel from '../external-api/external-api-panel'
+import ServiceApi from '../extra-info/service-api'
 import DatasetFooter from './dataset-footer'
 import Datasets from './datasets'
 
@@ -58,6 +59,9 @@ const List = () => {
       return router.replace('/apps')
   }, [currentWorkspace, router])
 
+  const isCurrentWorkspaceManager = useAppContextSelector(state => state.isCurrentWorkspaceManager)
+  const { data: apiBaseInfo } = useDatasetApiBaseUrl()
+
   return (
     <div className="scroll-container relative flex grow flex-col overflow-y-auto bg-background-body">
       <div className="sticky top-0 z-10 flex items-center justify-end gap-x-1 bg-background-body px-12 pb-2 pt-4">
@@ -81,6 +85,11 @@ const List = () => {
             onChange={e => handleKeywordsChange(e.target.value)}
             onClear={() => handleKeywordsChange('')}
           />
+          {
+            isCurrentWorkspaceManager && (
+              <ServiceApi apiBaseUrl={apiBaseInfo?.api_base_url ?? ''} />
+            )
+          }
           <div className="h-4 w-[1px] bg-divider-regular" />
           <Button
             className="shadows-shadow-xs gap-0.5"
@@ -96,7 +105,6 @@ const List = () => {
       {showTagManagementModal && (
         <TagManagementModal type="knowledge" show={showTagManagementModal} />
       )}
-
       {showExternalApiPanel && <ExternalAPIPanel onClose={() => setShowExternalApiPanel(false)} />}
     </div>
   )

web/app/components/datasets/settings/form/index.tsx

@@ -403,7 +403,7 @@ const Form = () => {
               </div>
             </>
           )
-        // eslint-disable-next-line sonarjs/no-nested-conditional
+
         : indexMethod
           ? (
               <>

web/app/components/header/account-setting/model-provider-page/system-model-selector/index.tsx

@@ -144,7 +144,7 @@ const SystemModel: FC<SystemModelSelectorProps> = ({
           {t('modelProvider.systemModelSettings', { ns: 'common' })}
         </Button>
       </PortalToFollowElemTrigger>
-      <PortalToFollowElemContent className="z-[60]">
+      <PortalToFollowElemContent className="z-[75]">
         <div className="w-[360px] rounded-xl border-[0.5px] border-components-panel-border bg-components-panel-bg pt-4 shadow-xl">
           <div className="px-6 py-1">
             <div className="flex h-8 items-center text-[13px] font-medium text-text-primary">

web/app/components/plugins/plugin-detail-panel/datasource-action-list.tsx

@@ -76,7 +76,7 @@ const ActionList = ({
             className='w-full'
             onClick={() => setShowSettingAuth(true)}
             disabled={!isCurrentWorkspaceManager}
-          >{t('workflow.nodes.tool.authorize')}</Button>
+          >{t('nodes.tool.authorize', { ns: 'workflow' })}</Button>
         )} */}
       </div>
       {/* <div className='flex flex-col gap-2'>

web/app/components/rag-pipeline/hooks/use-available-nodes-meta-data.ts

@@ -23,7 +23,7 @@ export const useAvailableNodesMetaData = () => {
     },
     knowledgeBaseDefault,
     dataSourceEmptyDefault,
-  ], [])
+  ] as AvailableNodesMetaData['nodes'], [])
 
   const helpLinkUri = useMemo(() => {
     if (language === 'zh_Hans')
@@ -52,7 +52,7 @@ export const useAvailableNodesMetaData = () => {
         title,
       },
     }
-  }), [mergedNodesMetaData, t])
+  }) as AvailableNodesMetaData['nodes'], [mergedNodesMetaData, t])
 
   const availableNodesMetaDataMap = useMemo(() => availableNodesMetaData.reduce((acc, node) => {
     acc![node.metaData.type] = node

web/app/components/rag-pipeline/hooks/use-get-run-and-trace-url.ts

@@ -3,8 +3,14 @@ import { useWorkflowStore } from '@/app/components/workflow/store'
 
 export const useGetRunAndTraceUrl = () => {
   const workflowStore = useWorkflowStore()
-  const getWorkflowRunAndTraceUrl = useCallback((runId: string) => {
+  const getWorkflowRunAndTraceUrl = useCallback((runId?: string) => {
     const { pipelineId } = workflowStore.getState()
+    if (!pipelineId || !runId) {
+      return {
+        runUrl: '',
+        traceUrl: '',
+      }
+    }
 
     return {
       runUrl: `/rag/pipelines/${pipelineId}/workflow-runs/${runId}`,

web/app/components/tools/edit-custom-collection-modal/get-schema.spec.tsx

@@ -23,7 +23,7 @@ describe('GetSchema', () => {
   it('shows an error when the URL is not http', () => {
     fireEvent.click(screen.getByText('tools.createTool.importFromUrl'))
     const input = screen.getByPlaceholderText('tools.createTool.importFromUrlPlaceHolder')
-    // eslint-disable-next-line sonarjs/no-clear-text-protocols
+
     fireEvent.change(input, { target: { value: 'ftp://invalid' } })
     fireEvent.click(screen.getByText('common.operation.ok'))
 

web/app/components/tools/mcp/mcp-server-modal.tsx

@@ -10,6 +10,7 @@ import Divider from '@/app/components/base/divider'
 import Modal from '@/app/components/base/modal'
 import Textarea from '@/app/components/base/textarea'
 import MCPServerParamItem from '@/app/components/tools/mcp/mcp-server-param-item'
+import { webSocketClient } from '@/app/components/workflow/collaboration/core/websocket-manager'
 import {
   useCreateMCPServer,
   useInvalidateMCPServerDetail,
@@ -59,6 +60,22 @@ const MCPServerModal = ({
     return res
   }
 
+  const emitMcpServerUpdate = (action: 'created' | 'updated') => {
+    const socket = webSocketClient.getSocket(appID)
+    if (!socket)
+      return
+
+    const timestamp = Date.now()
+    socket.emit('collaboration_event', {
+      type: 'mcp_server_update',
+      data: {
+        action,
+        timestamp,
+      },
+      timestamp,
+    })
+  }
+
   const submit = async () => {
     if (!data) {
       const payload: any = {
@@ -71,6 +88,7 @@ const MCPServerModal = ({
 
       await createMCPServer(payload)
       invalidateMCPServerDetail(appID)
+      emitMcpServerUpdate('created')
       onHide()
     }
     else {
@@ -83,6 +101,7 @@ const MCPServerModal = ({
       payload.description = description
       await updateMCPServer(payload)
       invalidateMCPServerDetail(appID)
+      emitMcpServerUpdate('updated')
       onHide()
     }
   }
@@ -92,6 +111,7 @@ const MCPServerModal = ({
       isShow={show}
       onClose={onHide}
       className={cn('relative !max-w-[520px] !p-0')}
+      highPriority
     >
       <div className="absolute right-5 top-5 z-10 cursor-pointer p-1.5" onClick={onHide}>
         <RiCloseLine className="h-5 w-5 text-text-tertiary" />

web/app/components/tools/mcp/mcp-service-card.tsx

@@ -1,6 +1,8 @@
 'use client'
+import type { CollaborationUpdate } from '@/app/components/workflow/collaboration/types/collaboration'
+import type { InputVar } from '@/app/components/workflow/types'
 import type { AppDetailResponse } from '@/models/app'
-import type { AppSSO } from '@/types/app'
+import type { AppSSO, ModelConfig, UserInputFormItem } from '@/types/app'
 import { RiEditLine, RiLoopLeftLine } from '@remixicon/react'
 import * as React from 'react'
 import { useEffect, useMemo, useState } from 'react'
@@ -16,6 +18,8 @@ import Switch from '@/app/components/base/switch'
 import Tooltip from '@/app/components/base/tooltip'
 import Indicator from '@/app/components/header/indicator'
 import MCPServerModal from '@/app/components/tools/mcp/mcp-server-modal'
+import { collaborationManager } from '@/app/components/workflow/collaboration/core/collaboration-manager'
+import { webSocketClient } from '@/app/components/workflow/collaboration/core/websocket-manager'
 import { BlockEnum } from '@/app/components/workflow/types'
 import { useAppContext } from '@/context/app-context'
 import { useDocLink } from '@/context/i18n'
@@ -36,6 +40,16 @@ export type IAppCardProps = {
   triggerModeMessage?: React.ReactNode // display-only message explaining the trigger restriction
 }
 
+type BasicAppConfig = Partial<ModelConfig> & {
+  updated_at?: number
+}
+
+type McpServerParam = {
+  label: string
+  variable: string
+  type: string
+}
+
 function MCPServiceCard({
   appInfo,
   triggerModeDisabled = false,
@@ -54,16 +68,16 @@ function MCPServiceCard({
   const isAdvancedApp = appInfo?.mode === AppModeEnum.ADVANCED_CHAT || appInfo?.mode === AppModeEnum.WORKFLOW
   const isBasicApp = !isAdvancedApp
   const { data: currentWorkflow } = useAppWorkflow(isAdvancedApp ? appId : '')
-  const [basicAppConfig, setBasicAppConfig] = useState<any>({})
-  const basicAppInputForm = useMemo(() => {
-    if (!isBasicApp || !basicAppConfig?.user_input_form)
+  const [basicAppConfig, setBasicAppConfig] = useState<BasicAppConfig>({})
+  const basicAppInputForm = useMemo<McpServerParam[]>(() => {
+    if (!isBasicApp || !basicAppConfig.user_input_form)
       return []
-    return basicAppConfig.user_input_form.map((item: any) => {
-      const type = Object.keys(item)[0]
-      return {
-        ...item[type],
-        type: type || 'text-input',
-      }
+    return basicAppConfig.user_input_form.map((item: UserInputFormItem) => {
+      if ('text-input' in item)
+        return { label: item['text-input'].label, variable: item['text-input'].variable, type: 'text-input' }
+      if ('select' in item)
+        return { label: item.select.label, variable: item.select.variable, type: 'select' }
+      return { label: item.paragraph.label, variable: item.paragraph.variable, type: 'paragraph' }
     })
   }, [basicAppConfig.user_input_form, isBasicApp])
   useEffect(() => {
@@ -90,12 +104,22 @@ function MCPServiceCard({
 
   const [activated, setActivated] = useState(serverActivated)
 
-  const latestParams = useMemo(() => {
+  const latestParams = useMemo<McpServerParam[]>(() => {
     if (isAdvancedApp) {
       if (!currentWorkflow?.graph)
         return []
-      const startNode = currentWorkflow?.graph.nodes.find(node => node.data.type === BlockEnum.Start) as any
-      return startNode?.data.variables as any[] || []
+      const startNode = currentWorkflow?.graph.nodes.find(node => node.data.type === BlockEnum.Start)
+      const variables = (startNode?.data as { variables?: InputVar[] } | undefined)?.variables || []
+      return variables.map((variable) => {
+        const label = typeof variable.label === 'string'
+          ? variable.label
+          : (variable.label.variable || variable.label.nodeName)
+        return {
+          label,
+          variable: variable.variable,
+          type: variable.type,
+        }
+      })
     }
     return basicAppInputForm
   }, [currentWorkflow, basicAppInputForm, isAdvancedApp])
@@ -103,6 +127,19 @@ function MCPServiceCard({
   const onGenCode = async () => {
     await refreshMCPServerCode(detail?.id || '')
     invalidateMCPServerDetail(appId)
+
+    // Emit collaboration event to notify other clients of MCP server changes
+    const socket = webSocketClient.getSocket(appId)
+    if (socket) {
+      socket.emit('collaboration_event', {
+        type: 'mcp_server_update',
+        data: {
+          action: 'codeRegenerated',
+          timestamp: Date.now(),
+        },
+        timestamp: Date.now(),
+      })
+    }
   }
 
   const onChangeStatus = async (state: boolean) => {
@@ -132,6 +169,20 @@ function MCPServiceCard({
       })
       invalidateMCPServerDetail(appId)
     }
+
+    // Emit collaboration event to notify other clients of MCP server status change
+    const socket = webSocketClient.getSocket(appId)
+    if (socket) {
+      socket.emit('collaboration_event', {
+        type: 'mcp_server_update',
+        data: {
+          action: 'statusChanged',
+          status: state ? 'active' : 'inactive',
+          timestamp: Date.now(),
+        },
+        timestamp: Date.now(),
+      })
+    }
   }
 
   const handleServerModalHide = () => {
@@ -144,6 +195,23 @@ function MCPServiceCard({
     setActivated(serverActivated)
   }, [serverActivated])
 
+  // Listen for collaborative MCP server updates from other clients
+  useEffect(() => {
+    if (!appId)
+      return
+
+    const unsubscribe = collaborationManager.onMcpServerUpdate(async (_update: CollaborationUpdate) => {
+      try {
+        invalidateMCPServerDetail(appId)
+      }
+      catch (error) {
+        console.error('MCP server update failed:', error)
+      }
+    })
+
+    return unsubscribe
+  }, [appId, invalidateMCPServerDetail])
+
   if (!currentWorkflow && isAdvancedApp)
     return null
 

web/app/components/workflow-app/components/workflow-header/features-trigger.spec.tsx

@@ -108,7 +108,7 @@ vi.mock('@/app/components/app/app-publisher', () => ({
         <button type="button" onClick={() => { Promise.resolve(props.onPublish?.()).catch(() => undefined) }}>
           publisher-publish
         </button>
-        <button type="button" onClick={() => { Promise.resolve(props.onPublish?.({ title: 'Test title', releaseNotes: 'Test notes' })).catch(() => undefined) }}>
+        <button type="button" onClick={() => { Promise.resolve(props.onPublish?.({ url: '/apps/app-id/workflows/publish', title: 'Test title', releaseNotes: 'Test notes' })).catch(() => undefined) }}>
           publisher-publish-with-params
         </button>
       </div>

web/app/components/workflow-app/components/workflow-header/features-trigger.tsx

@@ -1,3 +1,4 @@
+import type { ModelAndParameter } from '@/app/components/app/configuration/debug/types'
 import type { EndNodeType } from '@/app/components/workflow/nodes/end/types'
 import type { StartNodeType } from '@/app/components/workflow/nodes/start/types'
 import type {
@@ -140,24 +141,38 @@ const FeaturesTrigger = () => {
   const needWarningNodes = useChecklist(nodes, edges)
 
   const updatePublishedWorkflow = useInvalidateAppWorkflow()
-  const onPublish = useCallback(async (params?: PublishWorkflowParams) => {
+  const onPublish = useCallback(async (params?: ModelAndParameter | PublishWorkflowParams) => {
+    const publishParams = params && 'title' in params ? params : undefined
     // First check if there are any items in the checklist
     // if (!validateBeforeRun())
     //   throw new Error('Checklist has unresolved items')
 
     if (needWarningNodes.length > 0) {
+      console.warn('[workflow-header] publish blocked by checklist', {
+        appId: appID,
+        warningCount: needWarningNodes.length,
+      })
       notify({ type: 'error', message: t('panel.checklistTip', { ns: 'workflow' }) })
       throw new Error('Checklist has unresolved items')
     }
 
     // Then perform the detailed validation
     if (await handleCheckBeforePublish()) {
+      console.warn('[workflow-header] publish start', {
+        appId: appID,
+        title: publishParams?.title ?? '',
+      })
       const res = await publishWorkflow({
         url: `/apps/${appID}/workflows/publish`,
-        title: params?.title || '',
-        releaseNotes: params?.releaseNotes || '',
+        title: publishParams?.title || '',
+        releaseNotes: publishParams?.releaseNotes || '',
       })
 
+      console.warn('[workflow-header] publish response', {
+        appId: appID,
+        hasResponse: Boolean(res),
+        createdAt: res?.created_at,
+      })
       if (res) {
         notify({ type: 'success', message: t('api.actionSuccess', { ns: 'common' }) })
         updatePublishedWorkflow(appID!)