chore: Bump version references to 1.11.1 (#29568)

Co-authored-by: 01393547 <nieronghua@sf-express.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

api/controllers/console/app/workflow_trigger.py

@@ -114,7 +114,7 @@ class AppTriggersApi(Resource):
 
 @console_ns.route("/apps/<uuid:app_id>/trigger-enable")
 class AppTriggerEnableApi(Resource):
-    @console_ns.expect(console_ns.models[ParserEnable.__name__], validate=True)
+    @console_ns.expect(console_ns.models[ParserEnable.__name__])
     @setup_required
     @login_required
     @account_initialization_required

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

@@ -26,7 +26,7 @@ console_ns.schema_model(Parser.__name__, Parser.model_json_schema(ref_template=D
 
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/published/datasource/nodes/<string:node_id>/preview")
 class DataSourceContentPreviewApi(Resource):
-    @console_ns.expect(console_ns.models[Parser.__name__], validate=True)
+    @console_ns.expect(console_ns.models[Parser.__name__])
     @setup_required
     @login_required
     @account_initialization_required

api/controllers/console/workspace/models.py

@@ -282,9 +282,10 @@ class ModelProviderModelCredentialApi(Resource):
                 tenant_id=tenant_id, provider_name=provider
             )
         else:
-            model_type = args.model_type
+            # Normalize model_type to the origin value stored in DB (e.g., "text-generation" for LLM)
+            normalized_model_type = args.model_type.to_origin_model_type()
             available_credentials = model_provider_service.provider_manager.get_provider_model_available_credentials(
-                tenant_id=tenant_id, provider_name=provider, model_type=model_type, model_name=args.model
+                tenant_id=tenant_id, provider_name=provider, model_type=normalized_model_type, model_name=args.model
             )
 
         return jsonable_encoder(

api/core/entities/mcp_provider.py

@@ -213,12 +213,23 @@ class MCPProviderEntity(BaseModel):
         return None
 
     def retrieve_tokens(self) -> OAuthTokens | None:
-        """OAuth tokens if available"""
+        """Retrieve OAuth tokens if authentication is complete.
+
+        Returns:
+            OAuthTokens if the provider has been authenticated, None otherwise.
+        """
         if not self.credentials:
             return None
         credentials = self.decrypt_credentials()
+        access_token = credentials.get("access_token", "")
+        # Return None if access_token is empty to avoid generating invalid "Authorization: Bearer " header.
+        # Note: We don't check for whitespace-only strings here because:
+        # 1. OAuth servers don't return whitespace-only access tokens in practice
+        # 2. Even if they did, the server would return 401, triggering the OAuth flow correctly
+        if not access_token:
+            return None
         return OAuthTokens(
-            access_token=credentials.get("access_token", ""),
+            access_token=access_token,
             token_type=credentials.get("token_type", DEFAULT_TOKEN_TYPE),
             expires_in=int(credentials.get("expires_in", str(DEFAULT_EXPIRES_IN)) or DEFAULT_EXPIRES_IN),
             refresh_token=credentials.get("refresh_token", ""),

api/core/rag/extractor/excel_extractor.py

@@ -1,7 +1,7 @@
 """Abstract interface for document loader implementations."""
 
 import os
-from typing import cast
+from typing import TypedDict
 
 import pandas as pd
 from openpyxl import load_workbook
@@ -10,6 +10,12 @@ from core.rag.extractor.extractor_base import BaseExtractor
 from core.rag.models.document import Document
 
 
+class Candidate(TypedDict):
+    idx: int
+    count: int
+    map: dict[int, str]
+
+
 class ExcelExtractor(BaseExtractor):
     """Load Excel files.
 
@@ -30,32 +36,38 @@ class ExcelExtractor(BaseExtractor):
         file_extension = os.path.splitext(self._file_path)[-1].lower()
 
         if file_extension == ".xlsx":
-            wb = load_workbook(self._file_path, data_only=True)
-            for sheet_name in wb.sheetnames:
-                sheet = wb[sheet_name]
-                data = sheet.values
-                cols = next(data, None)
-                if cols is None:
-                    continue
-                df = pd.DataFrame(data, columns=cols)
-
-                df.dropna(how="all", inplace=True)
-
-                for index, row in df.iterrows():
-                    page_content = []
-                    for col_index, (k, v) in enumerate(row.items()):
-                        if pd.notna(v):
-                            cell = sheet.cell(
-                                row=cast(int, index) + 2, column=col_index + 1
-                            )  # +2 to account for header and 1-based index
-                            if cell.hyperlink:
-                                value = f"[{v}]({cell.hyperlink.target})"
-                                page_content.append(f'"{k}":"{value}"')
-                            else:
-                                page_content.append(f'"{k}":"{v}"')
-                    documents.append(
-                        Document(page_content=";".join(page_content), metadata={"source": self._file_path})
-                    )
+            wb = load_workbook(self._file_path, read_only=True, data_only=True)
+            try:
+                for sheet_name in wb.sheetnames:
+                    sheet = wb[sheet_name]
+                    header_row_idx, column_map, max_col_idx = self._find_header_and_columns(sheet)
+                    if not column_map:
+                        continue
+                    start_row = header_row_idx + 1
+                    for row in sheet.iter_rows(min_row=start_row, max_col=max_col_idx, values_only=False):
+                        if all(cell.value is None for cell in row):
+                            continue
+                        page_content = []
+                        for col_idx, cell in enumerate(row):
+                            value = cell.value
+                            if col_idx in column_map:
+                                col_name = column_map[col_idx]
+                                if hasattr(cell, "hyperlink") and cell.hyperlink:
+                                    target = getattr(cell.hyperlink, "target", None)
+                                    if target:
+                                        value = f"[{value}]({target})"
+                                if value is None:
+                                    value = ""
+                                elif not isinstance(value, str):
+                                    value = str(value)
+                                value = value.strip().replace('"', '\\"')
+                                page_content.append(f'"{col_name}":"{value}"')
+                        if page_content:
+                            documents.append(
+                                Document(page_content=";".join(page_content), metadata={"source": self._file_path})
+                            )
+            finally:
+                wb.close()
 
         elif file_extension == ".xls":
             excel_file = pd.ExcelFile(self._file_path, engine="xlrd")
@@ -63,9 +75,9 @@ class ExcelExtractor(BaseExtractor):
                 df = excel_file.parse(sheet_name=excel_sheet_name)
                 df.dropna(how="all", inplace=True)
 
-                for _, row in df.iterrows():
+                for _, series_row in df.iterrows():
                     page_content = []
-                    for k, v in row.items():
+                    for k, v in series_row.items():
                         if pd.notna(v):
                             page_content.append(f'"{k}":"{v}"')
                     documents.append(
@@ -75,3 +87,61 @@ class ExcelExtractor(BaseExtractor):
             raise ValueError(f"Unsupported file extension: {file_extension}")
 
         return documents
+
+    def _find_header_and_columns(self, sheet, scan_rows=10) -> tuple[int, dict[int, str], int]:
+        """
+        Scan first N rows to find the most likely header row.
+        Returns:
+            header_row_idx: 1-based index of the header row
+            column_map: Dict mapping 0-based column index to column name
+            max_col_idx: 1-based index of the last valid column (for iter_rows boundary)
+        """
+        # Store potential candidates: (row_index, non_empty_count, column_map)
+        candidates: list[Candidate] = []
+
+        # Limit scan to avoid performance issues on huge files
+        # We iterate manually to control the read scope
+        for current_row_idx, row in enumerate(sheet.iter_rows(min_row=1, max_row=scan_rows, values_only=True), start=1):
+            # Filter out empty cells and build a temp map for this row
+            # col_idx is 0-based
+            row_map = {}
+            for col_idx, cell_value in enumerate(row):
+                if cell_value is not None and str(cell_value).strip():
+                    row_map[col_idx] = str(cell_value).strip().replace('"', '\\"')
+
+            if not row_map:
+                continue
+
+            non_empty_count = len(row_map)
+
+            # Header selection heuristic (implemented):
+            # - Prefer the first row with at least 2 non-empty columns.
+            # - Fallback: choose the row with the most non-empty columns
+            #   (tie-breaker: smaller row index).
+            candidates.append({"idx": current_row_idx, "count": non_empty_count, "map": row_map})
+
+        if not candidates:
+            return 0, {}, 0
+
+        # Choose the best candidate header row.
+
+        best_candidate: Candidate | None = None
+
+        # Strategy: prefer the first row with >= 2 non-empty columns; otherwise fallback.
+
+        for cand in candidates:
+            if cand["count"] >= 2:
+                best_candidate = cand
+                break
+
+        # Fallback: if no row has >= 2 columns, or all have 1, just take the one with max columns
+        if not best_candidate:
+            # Sort by count desc, then index asc
+            candidates.sort(key=lambda x: (-x["count"], x["idx"]))
+            best_candidate = candidates[0]
+
+        # Determine max_col_idx (1-based for openpyxl)
+        # It is the index of the last valid column in our map + 1
+        max_col_idx = max(best_candidate["map"].keys()) + 1
+
+        return best_candidate["idx"], best_candidate["map"], max_col_idx

api/core/rag/extractor/word_extractor.py

@@ -84,22 +84,46 @@ class WordExtractor(BaseExtractor):
         image_count = 0
         image_map = {}
 
-        for rel in doc.part.rels.values():
+        for rId, rel in doc.part.rels.items():
             if "image" in rel.target_ref:
                 image_count += 1
                 if rel.is_external:
                     url = rel.target_ref
-                    response = ssrf_proxy.get(url)
+                    if not self._is_valid_url(url):
+                        continue
+                    try:
+                        response = ssrf_proxy.get(url)
+                    except Exception as e:
+                        logger.warning("Failed to download image from URL: %s: %s", url, str(e))
+                        continue
                     if response.status_code == 200:
-                        image_ext = mimetypes.guess_extension(response.headers["Content-Type"])
+                        image_ext = mimetypes.guess_extension(response.headers.get("Content-Type", ""))
                         if image_ext is None:
                             continue
                         file_uuid = str(uuid.uuid4())
-                        file_key = "image_files/" + self.tenant_id + "/" + file_uuid + "." + image_ext
+                        file_key = "image_files/" + self.tenant_id + "/" + file_uuid + image_ext
                         mime_type, _ = mimetypes.guess_type(file_key)
                         storage.save(file_key, response.content)
-                    else:
-                        continue
+                        # save file to db
+                        upload_file = UploadFile(
+                            tenant_id=self.tenant_id,
+                            storage_type=dify_config.STORAGE_TYPE,
+                            key=file_key,
+                            name=file_key,
+                            size=0,
+                            extension=str(image_ext),
+                            mime_type=mime_type or "",
+                            created_by=self.user_id,
+                            created_by_role=CreatorUserRole.ACCOUNT,
+                            created_at=naive_utc_now(),
+                            used=True,
+                            used_by=self.user_id,
+                            used_at=naive_utc_now(),
+                        )
+                        db.session.add(upload_file)
+                        db.session.commit()
+                        # Use rId as key for external images since target_part is undefined
+                        image_map[rId] = f"![image]({dify_config.FILES_URL}/files/{upload_file.id}/file-preview)"
                 else:
                     image_ext = rel.target_ref.split(".")[-1]
                     if image_ext is None:
@@ -110,26 +134,28 @@ class WordExtractor(BaseExtractor):
                     mime_type, _ = mimetypes.guess_type(file_key)
 
                     storage.save(file_key, rel.target_part.blob)
-                # save file to db
-                upload_file = UploadFile(
-                    tenant_id=self.tenant_id,
-                    storage_type=dify_config.STORAGE_TYPE,
-                    key=file_key,
-                    name=file_key,
-                    size=0,
-                    extension=str(image_ext),
-                    mime_type=mime_type or "",
-                    created_by=self.user_id,
-                    created_by_role=CreatorUserRole.ACCOUNT,
-                    created_at=naive_utc_now(),
-                    used=True,
-                    used_by=self.user_id,
-                    used_at=naive_utc_now(),
-                )
-
-                db.session.add(upload_file)
-                db.session.commit()
-                image_map[rel.target_part] = f"![image]({dify_config.FILES_URL}/files/{upload_file.id}/file-preview)"
+                    # save file to db
+                    upload_file = UploadFile(
+                        tenant_id=self.tenant_id,
+                        storage_type=dify_config.STORAGE_TYPE,
+                        key=file_key,
+                        name=file_key,
+                        size=0,
+                        extension=str(image_ext),
+                        mime_type=mime_type or "",
+                        created_by=self.user_id,
+                        created_by_role=CreatorUserRole.ACCOUNT,
+                        created_at=naive_utc_now(),
+                        used=True,
+                        used_by=self.user_id,
+                        used_at=naive_utc_now(),
+                    )
+                    db.session.add(upload_file)
+                    db.session.commit()
+                    # Use target_part as key for internal images
+                    image_map[rel.target_part] = (
+                        f"![image]({dify_config.FILES_URL}/files/{upload_file.id}/file-preview)"
+                    )
 
         return image_map
 
@@ -186,11 +212,17 @@ class WordExtractor(BaseExtractor):
                     image_id = blip.get("{http://schemas.openxmlformats.org/officeDocument/2006/relationships}embed")
                     if not image_id:
                         continue
-                    image_part = paragraph.part.rels[image_id].target_part
-
-                    if image_part in image_map:
-                        image_link = image_map[image_part]
-                        paragraph_content.append(image_link)
+                    rel = paragraph.part.rels.get(image_id)
+                    if rel is None:
+                        continue
+                    # For external images, use image_id as key; for internal, use target_part
+                    if rel.is_external:
+                        if image_id in image_map:
+                            paragraph_content.append(image_map[image_id])
+                    else:
+                        image_part = rel.target_part
+                        if image_part in image_map:
+                            paragraph_content.append(image_map[image_part])
             else:
                 paragraph_content.append(run.text)
         return "".join(paragraph_content).strip()
@@ -227,6 +259,18 @@ class WordExtractor(BaseExtractor):
 
         def parse_paragraph(paragraph):
             paragraph_content = []
+
+            def append_image_link(image_id, has_drawing):
+                """Helper to append image link from image_map based on relationship type."""
+                rel = doc.part.rels[image_id]
+                if rel.is_external:
+                    if image_id in image_map and not has_drawing:
+                        paragraph_content.append(image_map[image_id])
+                else:
+                    image_part = rel.target_part
+                    if image_part in image_map and not has_drawing:
+                        paragraph_content.append(image_map[image_part])
+
             for run in paragraph.runs:
                 if hasattr(run.element, "tag") and isinstance(run.element.tag, str) and run.element.tag.endswith("r"):
                     # Process drawing type images
@@ -243,10 +287,18 @@ class WordExtractor(BaseExtractor):
                                 "{http://schemas.openxmlformats.org/officeDocument/2006/relationships}embed"
                             )
                             if embed_id:
-                                image_part = doc.part.related_parts.get(embed_id)
-                                if image_part in image_map:
-                                    has_drawing = True
-                                    paragraph_content.append(image_map[image_part])
+                                rel = doc.part.rels.get(embed_id)
+                                if rel is not None and rel.is_external:
+                                    # External image: use embed_id as key
+                                    if embed_id in image_map:
+                                        has_drawing = True
+                                        paragraph_content.append(image_map[embed_id])
+                                else:
+                                    # Internal image: use target_part as key
+                                    image_part = doc.part.related_parts.get(embed_id)
+                                    if image_part in image_map:
+                                        has_drawing = True
+                                        paragraph_content.append(image_map[image_part])
                     # Process pict type images
                     shape_elements = run.element.findall(
                         ".//{http://schemas.openxmlformats.org/wordprocessingml/2006/main}pict"
@@ -261,9 +313,7 @@ class WordExtractor(BaseExtractor):
                                 "{http://schemas.openxmlformats.org/officeDocument/2006/relationships}id"
                             )
                             if image_id and image_id in doc.part.rels:
-                                image_part = doc.part.rels[image_id].target_part
-                                if image_part in image_map and not has_drawing:
-                                    paragraph_content.append(image_map[image_part])
+                                append_image_link(image_id, has_drawing)
                         # Find imagedata element in VML
                         image_data = shape.find(".//{urn:schemas-microsoft-com:vml}imagedata")
                         if image_data is not None:
@@ -271,9 +321,7 @@ class WordExtractor(BaseExtractor):
                                 "{http://schemas.openxmlformats.org/officeDocument/2006/relationships}id"
                             )
                             if image_id and image_id in doc.part.rels:
-                                image_part = doc.part.rels[image_id].target_part
-                                if image_part in image_map and not has_drawing:
-                                    paragraph_content.append(image_map[image_part])
+                                append_image_link(image_id, has_drawing)
                 if run.text.strip():
                     paragraph_content.append(run.text.strip())
             return "".join(paragraph_content) if paragraph_content else ""

api/libs/helper.py

@@ -215,7 +215,11 @@ def generate_text_hash(text: str) -> str:
 
 def compact_generate_response(response: Union[Mapping, Generator, RateLimitGenerator]) -> Response:
     if isinstance(response, dict):
-        return Response(response=json.dumps(jsonable_encoder(response)), status=200, mimetype="application/json")
+        return Response(
+            response=json.dumps(jsonable_encoder(response)),
+            status=200,
+            content_type="application/json; charset=utf-8",
+        )
     else:
 
         def generate() -> Generator:

api/models/model.py

@@ -111,7 +111,11 @@ class App(Base):
         else:
             app_model_config = self.app_model_config
             if app_model_config:
-                return app_model_config.pre_prompt
+                pre_prompt = app_model_config.pre_prompt or ""
+                # Truncate to 200 characters with ellipsis if using prompt as description
+                if len(pre_prompt) > 200:
+                    return pre_prompt[:200] + "..."
+                return pre_prompt
             else:
                 return ""
 

api/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "dify-api"
-version = "1.11.0"
+version = "1.11.1"
 requires-python = ">=3.11,<3.13"
 
 dependencies = [

api/services/dataset_service.py

@@ -1636,6 +1636,20 @@ class DocumentService:
                         return [], ""
                     db.session.add(dataset_process_rule)
                     db.session.flush()
+                else:
+                    # Fallback when no process_rule provided in knowledge_config:
+                    # 1) reuse dataset.latest_process_rule if present
+                    # 2) otherwise create an automatic rule
+                    dataset_process_rule = getattr(dataset, "latest_process_rule", None)
+                    if not dataset_process_rule:
+                        dataset_process_rule = DatasetProcessRule(
+                            dataset_id=dataset.id,
+                            mode="automatic",
+                            rules=json.dumps(DatasetProcessRule.AUTOMATIC_RULES),
+                            created_by=account.id,
+                        )
+                        db.session.add(dataset_process_rule)
+                        db.session.flush()
             lock_name = f"add_document_lock_dataset_id_{dataset.id}"
             try:
                 with redis_client.lock(lock_name, timeout=600):
@@ -1647,65 +1661,67 @@ class DocumentService:
                         if not knowledge_config.data_source.info_list.file_info_list:
                             raise ValueError("File source info is required")
                         upload_file_list = knowledge_config.data_source.info_list.file_info_list.file_ids
-                        for file_id in upload_file_list:
-                            file = (
-                                db.session.query(UploadFile)
-                                .where(UploadFile.tenant_id == dataset.tenant_id, UploadFile.id == file_id)
-                                .first()
+                        files = (
+                            db.session.query(UploadFile)
+                            .where(
+                                UploadFile.tenant_id == dataset.tenant_id,
+                                UploadFile.id.in_(upload_file_list),
                             )
+                            .all()
+                        )
+                        if len(files) != len(set(upload_file_list)):
+                            raise FileNotExistsError("One or more files not found.")
 
-                            # raise error if file not found
-                            if not file:
-                                raise FileNotExistsError()
-
-                            file_name = file.name
+                        file_names = [file.name for file in files]
+                        db_documents = (
+                            db.session.query(Document)
+                            .where(
+                                Document.dataset_id == dataset.id,
+                                Document.tenant_id == current_user.current_tenant_id,
+                                Document.data_source_type == "upload_file",
+                                Document.enabled == True,
+                                Document.name.in_(file_names),
+                            )
+                            .all()
+                        )
+                        documents_map = {document.name: document for document in db_documents}
+                        for file in files:
                             data_source_info: dict[str, str | bool] = {
-                                "upload_file_id": file_id,
+                                "upload_file_id": file.id,
                             }
-                            # check duplicate
-                            if knowledge_config.duplicate:
-                                document = (
-                                    db.session.query(Document)
-                                    .filter_by(
-                                        dataset_id=dataset.id,
-                                        tenant_id=current_user.current_tenant_id,
-                                        data_source_type="upload_file",
-                                        enabled=True,
-                                        name=file_name,
-                                    )
-                                    .first()
+                            document = documents_map.get(file.name)
+                            if knowledge_config.duplicate and document:
+                                document.dataset_process_rule_id = dataset_process_rule.id
+                                document.updated_at = naive_utc_now()
+                                document.created_from = created_from
+                                document.doc_form = knowledge_config.doc_form
+                                document.doc_language = knowledge_config.doc_language
+                                document.data_source_info = json.dumps(data_source_info)
+                                document.batch = batch
+                                document.indexing_status = "waiting"
+                                db.session.add(document)
+                                documents.append(document)
+                                duplicate_document_ids.append(document.id)
+                                continue
+                            else:
+                                document = DocumentService.build_document(
+                                    dataset,
+                                    dataset_process_rule.id,
+                                    knowledge_config.data_source.info_list.data_source_type,
+                                    knowledge_config.doc_form,
+                                    knowledge_config.doc_language,
+                                    data_source_info,
+                                    created_from,
+                                    position,
+                                    account,
+                                    file.name,
+                                    batch,
                                 )
-                                if document:
-                                    document.dataset_process_rule_id = dataset_process_rule.id
-                                    document.updated_at = naive_utc_now()
-                                    document.created_from = created_from
-                                    document.doc_form = knowledge_config.doc_form
-                                    document.doc_language = knowledge_config.doc_language
-                                    document.data_source_info = json.dumps(data_source_info)
-                                    document.batch = batch
-                                    document.indexing_status = "waiting"
-                                    db.session.add(document)
-                                    documents.append(document)
-                                    duplicate_document_ids.append(document.id)
-                                    continue
-                            document = DocumentService.build_document(
-                                dataset,
-                                dataset_process_rule.id,
-                                knowledge_config.data_source.info_list.data_source_type,
-                                knowledge_config.doc_form,
-                                knowledge_config.doc_language,
-                                data_source_info,
-                                created_from,
-                                position,
-                                account,
-                                file_name,
-                                batch,
-                            )
-                            db.session.add(document)
-                            db.session.flush()
-                            document_ids.append(document.id)
-                            documents.append(document)
-                            position += 1
+                                db.session.add(document)
+                                db.session.flush()
+                                document_ids.append(document.id)
+                                documents.append(document)
+                                position += 1
                     elif knowledge_config.data_source.info_list.data_source_type == "notion_import":
                         notion_info_list = knowledge_config.data_source.info_list.notion_info_list  # type: ignore
                         if not notion_info_list:

api/services/hit_testing_service.py

@@ -178,8 +178,8 @@ class HitTestingService:
 
     @classmethod
     def hit_testing_args_check(cls, args):
-        query = args["query"]
-        attachment_ids = args["attachment_ids"]
+        query = args.get("query")
+        attachment_ids = args.get("attachment_ids")
 
         if not attachment_ids and not query:
             raise ValueError("Query or attachment_ids is required")

api/tests/test_containers_integration_tests/trigger/__init__.py

@@ -0,0 +1 @@
+

api/tests/test_containers_integration_tests/trigger/conftest.py

@@ -0,0 +1,182 @@
+"""
+Fixtures for trigger integration tests.
+
+This module provides fixtures for creating test data (tenant, account, app)
+and mock objects used across trigger-related tests.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Generator
+from typing import Any
+
+import pytest
+from sqlalchemy.orm import Session
+
+from models.account import Account, Tenant, TenantAccountJoin, TenantAccountRole
+from models.model import App
+
+
+@pytest.fixture
+def tenant_and_account(db_session_with_containers: Session) -> Generator[tuple[Tenant, Account], None, None]:
+    """
+    Create a tenant and account for testing.
+
+    This fixture creates a tenant, account, and their association,
+    then cleans up after the test completes.
+
+    Yields:
+        tuple[Tenant, Account]: The created tenant and account
+    """
+    tenant = Tenant(name="trigger-e2e")
+    account = Account(name="tester", email="tester@example.com", interface_language="en-US")
+    db_session_with_containers.add_all([tenant, account])
+    db_session_with_containers.commit()
+
+    join = TenantAccountJoin(tenant_id=tenant.id, account_id=account.id, role=TenantAccountRole.OWNER.value)
+    db_session_with_containers.add(join)
+    db_session_with_containers.commit()
+
+    yield tenant, account
+
+    # Cleanup
+    db_session_with_containers.query(TenantAccountJoin).filter_by(tenant_id=tenant.id).delete()
+    db_session_with_containers.query(Account).filter_by(id=account.id).delete()
+    db_session_with_containers.query(Tenant).filter_by(id=tenant.id).delete()
+    db_session_with_containers.commit()
+
+
+@pytest.fixture
+def app_model(
+    db_session_with_containers: Session, tenant_and_account: tuple[Tenant, Account]
+) -> Generator[App, None, None]:
+    """
+    Create an app for testing.
+
+    This fixture creates a workflow app associated with the tenant and account,
+    then cleans up after the test completes.
+
+    Yields:
+        App: The created app
+    """
+    tenant, account = tenant_and_account
+    app = App(
+        tenant_id=tenant.id,
+        name="trigger-app",
+        description="trigger e2e",
+        mode="workflow",
+        icon_type="emoji",
+        icon="robot",
+        icon_background="#FFEAD5",
+        enable_site=True,
+        enable_api=True,
+        api_rpm=100,
+        api_rph=1000,
+        is_demo=False,
+        is_public=False,
+        is_universal=False,
+        created_by=account.id,
+    )
+    db_session_with_containers.add(app)
+    db_session_with_containers.commit()
+
+    yield app
+
+    # Cleanup - delete related records first
+    from models.trigger import (
+        AppTrigger,
+        TriggerSubscription,
+        WorkflowPluginTrigger,
+        WorkflowSchedulePlan,
+        WorkflowTriggerLog,
+        WorkflowWebhookTrigger,
+    )
+    from models.workflow import Workflow
+
+    db_session_with_containers.query(WorkflowTriggerLog).filter_by(app_id=app.id).delete()
+    db_session_with_containers.query(WorkflowSchedulePlan).filter_by(app_id=app.id).delete()
+    db_session_with_containers.query(WorkflowWebhookTrigger).filter_by(app_id=app.id).delete()
+    db_session_with_containers.query(WorkflowPluginTrigger).filter_by(app_id=app.id).delete()
+    db_session_with_containers.query(AppTrigger).filter_by(app_id=app.id).delete()
+    db_session_with_containers.query(TriggerSubscription).filter_by(tenant_id=tenant.id).delete()
+    db_session_with_containers.query(Workflow).filter_by(app_id=app.id).delete()
+    db_session_with_containers.query(App).filter_by(id=app.id).delete()
+    db_session_with_containers.commit()
+
+
+class MockCeleryGroup:
+    """Mock for celery group() function that collects dispatched tasks."""
+
+    def __init__(self) -> None:
+        self.collected: list[dict[str, Any]] = []
+        self._applied = False
+
+    def __call__(self, items: Any) -> MockCeleryGroup:
+        self.collected = list(items)
+        return self
+
+    def apply_async(self) -> None:
+        self._applied = True
+
+    @property
+    def applied(self) -> bool:
+        return self._applied
+
+
+class MockCelerySignature:
+    """Mock for celery task signature that returns task info dict."""
+
+    def s(self, schedule_id: str) -> dict[str, str]:
+        return {"schedule_id": schedule_id}
+
+
+@pytest.fixture
+def mock_celery_group() -> MockCeleryGroup:
+    """
+    Provide a mock celery group for testing task dispatch.
+
+    Returns:
+        MockCeleryGroup: Mock group that collects dispatched tasks
+    """
+    return MockCeleryGroup()
+
+
+@pytest.fixture
+def mock_celery_signature() -> MockCelerySignature:
+    """
+    Provide a mock celery signature for testing task dispatch.
+
+    Returns:
+        MockCelerySignature: Mock signature generator
+    """
+    return MockCelerySignature()
+
+
+class MockPluginSubscription:
+    """Mock plugin subscription for testing plugin triggers."""
+
+    def __init__(
+        self,
+        subscription_id: str = "sub-1",
+        tenant_id: str = "tenant-1",
+        provider_id: str = "provider-1",
+    ) -> None:
+        self.id = subscription_id
+        self.tenant_id = tenant_id
+        self.provider_id = provider_id
+        self.credentials: dict[str, str] = {"token": "secret"}
+        self.credential_type = "api-key"
+
+    def to_entity(self) -> MockPluginSubscription:
+        return self
+
+
+@pytest.fixture
+def mock_plugin_subscription() -> MockPluginSubscription:
+    """
+    Provide a mock plugin subscription for testing.
+
+    Returns:
+        MockPluginSubscription: Mock subscription instance
+    """
+    return MockPluginSubscription()

api/tests/test_containers_integration_tests/trigger/test_trigger_e2e.py

@@ -0,0 +1,911 @@
+from __future__ import annotations
+
+import importlib
+import json
+import time
+from datetime import timedelta
+from types import SimpleNamespace
+from typing import Any
+
+import pytest
+from flask import Flask, Response
+from flask.testing import FlaskClient
+from sqlalchemy.orm import Session
+
+from configs import dify_config
+from core.plugin.entities.request import TriggerInvokeEventResponse
+from core.trigger.debug import event_selectors
+from core.trigger.debug.event_bus import TriggerDebugEventBus
+from core.trigger.debug.event_selectors import PluginTriggerDebugEventPoller, WebhookTriggerDebugEventPoller
+from core.trigger.debug.events import PluginTriggerDebugEvent, build_plugin_pool_key
+from core.workflow.enums import NodeType
+from libs.datetime_utils import naive_utc_now
+from models.account import Account, Tenant
+from models.enums import AppTriggerStatus, AppTriggerType, CreatorUserRole, WorkflowTriggerStatus
+from models.model import App
+from models.trigger import (
+    AppTrigger,
+    TriggerSubscription,
+    WorkflowPluginTrigger,
+    WorkflowSchedulePlan,
+    WorkflowTriggerLog,
+    WorkflowWebhookTrigger,
+)
+from models.workflow import Workflow
+from schedule import workflow_schedule_task
+from schedule.workflow_schedule_task import poll_workflow_schedules
+from services import feature_service as feature_service_module
+from services.trigger import webhook_service
+from services.trigger.schedule_service import ScheduleService
+from services.workflow_service import WorkflowService
+from tasks import trigger_processing_tasks
+
+from .conftest import MockCeleryGroup, MockCelerySignature, MockPluginSubscription
+
+# Test constants
+WEBHOOK_ID_PRODUCTION = "wh1234567890123456789012"
+WEBHOOK_ID_DEBUG = "whdebug1234567890123456"
+TEST_TRIGGER_URL = "https://trigger.example.com/base"
+
+
+def _build_workflow_graph(root_node_id: str, trigger_type: NodeType) -> str:
+    """Build a minimal workflow graph JSON for testing."""
+    node_data: dict[str, Any] = {"type": trigger_type.value, "title": "trigger"}
+    if trigger_type == NodeType.TRIGGER_WEBHOOK:
+        node_data.update(
+            {
+                "method": "POST",
+                "content_type": "application/json",
+                "headers": [],
+                "params": [],
+                "body": [],
+            }
+        )
+    graph = {
+        "nodes": [
+            {"id": root_node_id, "data": node_data},
+            {"id": "answer-1", "data": {"type": NodeType.ANSWER.value, "title": "answer"}},
+        ],
+        "edges": [{"source": root_node_id, "target": "answer-1", "sourceHandle": "success"}],
+    }
+    return json.dumps(graph)
+
+
+def test_publish_blocks_start_and_trigger_coexistence(
+    db_session_with_containers: Session,
+    tenant_and_account: tuple[Tenant, Account],
+    app_model: App,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Publishing should fail when both start and trigger nodes coexist."""
+    tenant, account = tenant_and_account
+
+    graph = {
+        "nodes": [
+            {"id": "start", "data": {"type": NodeType.START.value}},
+            {"id": "trig", "data": {"type": NodeType.TRIGGER_WEBHOOK.value}},
+        ],
+        "edges": [],
+    }
+    draft_workflow = Workflow.new(
+        tenant_id=tenant.id,
+        app_id=app_model.id,
+        type="workflow",
+        version=Workflow.VERSION_DRAFT,
+        graph=json.dumps(graph),
+        features=json.dumps({}),
+        created_by=account.id,
+        environment_variables=[],
+        conversation_variables=[],
+        rag_pipeline_variables=[],
+    )
+    db_session_with_containers.add(draft_workflow)
+    db_session_with_containers.commit()
+
+    workflow_service = WorkflowService()
+
+    monkeypatch.setattr(
+        feature_service_module.FeatureService,
+        "get_system_features",
+        classmethod(lambda _cls: SimpleNamespace(plugin_manager=SimpleNamespace(enabled=False))),
+    )
+    monkeypatch.setattr("services.workflow_service.dify_config", SimpleNamespace(BILLING_ENABLED=False))
+
+    with pytest.raises(ValueError, match="Start node and trigger nodes cannot coexist"):
+        workflow_service.publish_workflow(session=db_session_with_containers, app_model=app_model, account=account)
+
+
+def test_trigger_url_uses_config_base(monkeypatch: pytest.MonkeyPatch) -> None:
+    """TRIGGER_URL config should be reflected in generated webhook and plugin endpoints."""
+    original_url = getattr(dify_config, "TRIGGER_URL", None)
+
+    try:
+        monkeypatch.setattr(dify_config, "TRIGGER_URL", TEST_TRIGGER_URL)
+        endpoint_module = importlib.reload(importlib.import_module("core.trigger.utils.endpoint"))
+
+        assert (
+            endpoint_module.generate_webhook_trigger_endpoint(WEBHOOK_ID_PRODUCTION)
+            == f"{TEST_TRIGGER_URL}/triggers/webhook/{WEBHOOK_ID_PRODUCTION}"
+        )
+        assert (
+            endpoint_module.generate_webhook_trigger_endpoint(WEBHOOK_ID_PRODUCTION, True)
+            == f"{TEST_TRIGGER_URL}/triggers/webhook-debug/{WEBHOOK_ID_PRODUCTION}"
+        )
+        assert (
+            endpoint_module.generate_plugin_trigger_endpoint_url("end-1") == f"{TEST_TRIGGER_URL}/triggers/plugin/end-1"
+        )
+    finally:
+        # Restore original config and reload module
+        if original_url is not None:
+            monkeypatch.setattr(dify_config, "TRIGGER_URL", original_url)
+        importlib.reload(importlib.import_module("core.trigger.utils.endpoint"))
+
+
+def test_webhook_trigger_creates_trigger_log(
+    test_client_with_containers: FlaskClient,
+    db_session_with_containers: Session,
+    tenant_and_account: tuple[Tenant, Account],
+    app_model: App,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Production webhook trigger should create a trigger log in the database."""
+    tenant, account = tenant_and_account
+
+    webhook_node_id = "webhook-node"
+    graph_json = _build_workflow_graph(webhook_node_id, NodeType.TRIGGER_WEBHOOK)
+    published_workflow = Workflow.new(
+        tenant_id=tenant.id,
+        app_id=app_model.id,
+        type="workflow",
+        version=Workflow.version_from_datetime(naive_utc_now()),
+        graph=graph_json,
+        features=json.dumps({}),
+        created_by=account.id,
+        environment_variables=[],
+        conversation_variables=[],
+        rag_pipeline_variables=[],
+    )
+    db_session_with_containers.add(published_workflow)
+    app_model.workflow_id = published_workflow.id
+    db_session_with_containers.commit()
+
+    webhook_trigger = WorkflowWebhookTrigger(
+        app_id=app_model.id,
+        node_id=webhook_node_id,
+        tenant_id=tenant.id,
+        webhook_id=WEBHOOK_ID_PRODUCTION,
+        created_by=account.id,
+    )
+    app_trigger = AppTrigger(
+        tenant_id=tenant.id,
+        app_id=app_model.id,
+        node_id=webhook_node_id,
+        trigger_type=AppTriggerType.TRIGGER_WEBHOOK,
+        status=AppTriggerStatus.ENABLED,
+        title="webhook",
+    )
+
+    db_session_with_containers.add_all([webhook_trigger, app_trigger])
+    db_session_with_containers.commit()
+
+    def _fake_trigger_workflow_async(session: Session, user: Any, trigger_data: Any) -> SimpleNamespace:
+        log = WorkflowTriggerLog(
+            tenant_id=trigger_data.tenant_id,
+            app_id=trigger_data.app_id,
+            workflow_id=trigger_data.workflow_id,
+            root_node_id=trigger_data.root_node_id,
+            trigger_metadata=trigger_data.trigger_metadata.model_dump_json() if trigger_data.trigger_metadata else "{}",
+            trigger_type=trigger_data.trigger_type,
+            workflow_run_id=None,
+            outputs=None,
+            trigger_data=trigger_data.model_dump_json(),
+            inputs=json.dumps(dict(trigger_data.inputs)),
+            status=WorkflowTriggerStatus.SUCCEEDED,
+            error="",
+            queue_name="triggered_workflow_dispatcher",
+            celery_task_id="celery-test",
+            created_by_role=CreatorUserRole.ACCOUNT,
+            created_by=account.id,
+        )
+        session.add(log)
+        session.commit()
+        return SimpleNamespace(workflow_trigger_log_id=log.id, task_id=None, status="queued", queue="test")
+
+    monkeypatch.setattr(
+        webhook_service.AsyncWorkflowService,
+        "trigger_workflow_async",
+        _fake_trigger_workflow_async,
+    )
+
+    response = test_client_with_containers.post(f"/triggers/webhook/{webhook_trigger.webhook_id}", json={"foo": "bar"})
+
+    assert response.status_code == 200
+
+    db_session_with_containers.expire_all()
+    logs = db_session_with_containers.query(WorkflowTriggerLog).filter_by(app_id=app_model.id).all()
+    assert logs, "Webhook trigger should create trigger log"
+
+
+@pytest.mark.parametrize("schedule_type", ["visual", "cron"])
+def test_schedule_poll_dispatches_due_plan(
+    db_session_with_containers: Session,
+    tenant_and_account: tuple[Tenant, Account],
+    app_model: App,
+    mock_celery_group: MockCeleryGroup,
+    mock_celery_signature: MockCelerySignature,
+    monkeypatch: pytest.MonkeyPatch,
+    schedule_type: str,
+) -> None:
+    """Schedule plans (both visual and cron) should be polled and dispatched when due."""
+    tenant, _ = tenant_and_account
+
+    app_trigger = AppTrigger(
+        tenant_id=tenant.id,
+        app_id=app_model.id,
+        node_id=f"schedule-{schedule_type}",
+        trigger_type=AppTriggerType.TRIGGER_SCHEDULE,
+        status=AppTriggerStatus.ENABLED,
+        title=f"schedule-{schedule_type}",
+    )
+    plan = WorkflowSchedulePlan(
+        app_id=app_model.id,
+        node_id=f"schedule-{schedule_type}",
+        tenant_id=tenant.id,
+        cron_expression="* * * * *",
+        timezone="UTC",
+        next_run_at=naive_utc_now() - timedelta(minutes=1),
+    )
+    db_session_with_containers.add_all([app_trigger, plan])
+    db_session_with_containers.commit()
+
+    next_time = naive_utc_now() + timedelta(hours=1)
+    monkeypatch.setattr(workflow_schedule_task, "calculate_next_run_at", lambda *_args, **_kwargs: next_time)
+    monkeypatch.setattr(workflow_schedule_task, "group", mock_celery_group)
+    monkeypatch.setattr(workflow_schedule_task, "run_schedule_trigger", mock_celery_signature)
+
+    poll_workflow_schedules()
+
+    assert mock_celery_group.collected, f"Should dispatch signatures for due {schedule_type} schedules"
+    scheduled_ids = {sig["schedule_id"] for sig in mock_celery_group.collected}
+    assert plan.id in scheduled_ids
+
+
+def test_schedule_visual_debug_poll_generates_event(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Visual mode schedule node should generate event in single-step debug."""
+    base_now = naive_utc_now()
+    monkeypatch.setattr(event_selectors, "naive_utc_now", lambda: base_now)
+    monkeypatch.setattr(
+        event_selectors,
+        "calculate_next_run_at",
+        lambda *_args, **_kwargs: base_now - timedelta(minutes=1),
+    )
+    node_config = {
+        "id": "schedule-visual",
+        "data": {
+            "type": NodeType.TRIGGER_SCHEDULE.value,
+            "mode": "visual",
+            "frequency": "daily",
+            "visual_config": {"time": "3:00 PM"},
+            "timezone": "UTC",
+        },
+    }
+    poller = event_selectors.ScheduleTriggerDebugEventPoller(
+        tenant_id="tenant",
+        user_id="user",
+        app_id="app",
+        node_config=node_config,
+        node_id="schedule-visual",
+    )
+    event = poller.poll()
+    assert event is not None
+    assert event.workflow_args["inputs"] == {}
+
+
+def test_plugin_trigger_dispatches_and_debug_events(
+    test_client_with_containers: FlaskClient,
+    mock_plugin_subscription: MockPluginSubscription,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Plugin trigger endpoint should dispatch events and generate debug events."""
+    endpoint_id = "1cc7fa12-3f7b-4f6a-9c8d-1234567890ab"
+
+    debug_events: list[dict[str, Any]] = []
+    dispatched_payloads: list[dict[str, Any]] = []
+
+    def _fake_process_endpoint(_endpoint_id: str, _request: Any) -> Response:
+        dispatch_data = {
+            "user_id": "end-user",
+            "tenant_id": mock_plugin_subscription.tenant_id,
+            "endpoint_id": _endpoint_id,
+            "provider_id": mock_plugin_subscription.provider_id,
+            "subscription_id": mock_plugin_subscription.id,
+            "timestamp": int(time.time()),
+            "events": ["created", "updated"],
+            "request_id": f"req-{_endpoint_id}",
+        }
+        trigger_processing_tasks.dispatch_triggered_workflows_async.delay(dispatch_data)
+        return Response("ok", status=202)
+
+    monkeypatch.setattr(
+        "services.trigger.trigger_service.TriggerService.process_endpoint",
+        staticmethod(_fake_process_endpoint),
+    )
+
+    monkeypatch.setattr(
+        trigger_processing_tasks.TriggerDebugEventBus,
+        "dispatch",
+        staticmethod(lambda **kwargs: debug_events.append(kwargs) or 1),
+    )
+
+    def _fake_delay(dispatch_data: dict[str, Any]) -> None:
+        dispatched_payloads.append(dispatch_data)
+        trigger_processing_tasks.dispatch_trigger_debug_event(
+            events=dispatch_data["events"],
+            user_id=dispatch_data["user_id"],
+            timestamp=dispatch_data["timestamp"],
+            request_id=dispatch_data["request_id"],
+            subscription=mock_plugin_subscription,
+        )
+
+    monkeypatch.setattr(
+        trigger_processing_tasks.dispatch_triggered_workflows_async,
+        "delay",
+        staticmethod(_fake_delay),
+    )
+
+    response = test_client_with_containers.post(f"/triggers/plugin/{endpoint_id}", json={"hello": "world"})
+
+    assert response.status_code == 202
+    assert dispatched_payloads, "Plugin trigger should enqueue workflow dispatch payload"
+    assert debug_events, "Plugin trigger should dispatch debug events"
+    dispatched_event_names = {event["event"].name for event in debug_events}
+    assert dispatched_event_names == {"created", "updated"}
+
+
+def test_webhook_debug_dispatches_event(
+    test_client_with_containers: FlaskClient,
+    db_session_with_containers: Session,
+    tenant_and_account: tuple[Tenant, Account],
+    app_model: App,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Webhook single-step debug should dispatch debug event and be pollable."""
+    tenant, account = tenant_and_account
+    webhook_node_id = "webhook-debug-node"
+    graph_json = _build_workflow_graph(webhook_node_id, NodeType.TRIGGER_WEBHOOK)
+    draft_workflow = Workflow.new(
+        tenant_id=tenant.id,
+        app_id=app_model.id,
+        type="workflow",
+        version=Workflow.VERSION_DRAFT,
+        graph=graph_json,
+        features=json.dumps({}),
+        created_by=account.id,
+        environment_variables=[],
+        conversation_variables=[],
+        rag_pipeline_variables=[],
+    )
+    db_session_with_containers.add(draft_workflow)
+    db_session_with_containers.commit()
+
+    webhook_trigger = WorkflowWebhookTrigger(
+        app_id=app_model.id,
+        node_id=webhook_node_id,
+        tenant_id=tenant.id,
+        webhook_id=WEBHOOK_ID_DEBUG,
+        created_by=account.id,
+    )
+    db_session_with_containers.add(webhook_trigger)
+    db_session_with_containers.commit()
+
+    debug_events: list[dict[str, Any]] = []
+    original_dispatch = TriggerDebugEventBus.dispatch
+    monkeypatch.setattr(
+        "controllers.trigger.webhook.TriggerDebugEventBus.dispatch",
+        lambda **kwargs: (debug_events.append(kwargs), original_dispatch(**kwargs))[1],
+    )
+
+    # Listener polls first to enter waiting pool
+    poller = WebhookTriggerDebugEventPoller(
+        tenant_id=tenant.id,
+        user_id=account.id,
+        app_id=app_model.id,
+        node_config=draft_workflow.get_node_config_by_id(webhook_node_id),
+        node_id=webhook_node_id,
+    )
+    assert poller.poll() is None
+
+    response = test_client_with_containers.post(
+        f"/triggers/webhook-debug/{webhook_trigger.webhook_id}",
+        json={"foo": "bar"},
+        headers={"Content-Type": "application/json"},
+    )
+
+    assert response.status_code == 200
+    assert debug_events, "Debug event should be sent to event bus"
+    # Second poll should get the event
+    event = poller.poll()
+    assert event is not None
+    assert event.workflow_args["inputs"]["webhook_body"]["foo"] == "bar"
+    assert debug_events[0]["pool_key"].endswith(f":{app_model.id}:{webhook_node_id}")
+
+
+def test_plugin_single_step_debug_flow(
+    flask_app_with_containers: Flask,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Plugin single-step debug: listen -> dispatch event -> poller receives and returns variables."""
+    tenant_id = "tenant-1"
+    app_id = "app-1"
+    user_id = "user-1"
+    node_id = "plugin-node"
+    provider_id = "langgenius/provider-1/provider-1"
+    node_config = {
+        "id": node_id,
+        "data": {
+            "type": NodeType.TRIGGER_PLUGIN.value,
+            "title": "plugin",
+            "plugin_id": "plugin-1",
+            "plugin_unique_identifier": "plugin-1",
+            "provider_id": provider_id,
+            "event_name": "created",
+            "subscription_id": "sub-1",
+            "parameters": {},
+        },
+    }
+    # Start listening
+    poller = PluginTriggerDebugEventPoller(
+        tenant_id=tenant_id,
+        user_id=user_id,
+        app_id=app_id,
+        node_config=node_config,
+        node_id=node_id,
+    )
+    assert poller.poll() is None
+
+    from core.trigger.debug.events import build_plugin_pool_key
+
+    pool_key = build_plugin_pool_key(
+        tenant_id=tenant_id,
+        provider_id=provider_id,
+        subscription_id="sub-1",
+        name="created",
+    )
+    TriggerDebugEventBus.dispatch(
+        tenant_id=tenant_id,
+        event=PluginTriggerDebugEvent(
+            timestamp=int(time.time()),
+            user_id=user_id,
+            name="created",
+            request_id="req-1",
+            subscription_id="sub-1",
+            provider_id="provider-1",
+        ),
+        pool_key=pool_key,
+    )
+
+    from core.plugin.entities.request import TriggerInvokeEventResponse
+
+    monkeypatch.setattr(
+        "services.trigger.trigger_service.TriggerService.invoke_trigger_event",
+        staticmethod(
+            lambda **_kwargs: TriggerInvokeEventResponse(
+                variables={"echo": "pong"},
+                cancelled=False,
+            )
+        ),
+    )
+
+    event = poller.poll()
+    assert event is not None
+    assert event.workflow_args["inputs"]["echo"] == "pong"
+
+
+def test_schedule_trigger_creates_trigger_log(
+    db_session_with_containers: Session,
+    tenant_and_account: tuple[Tenant, Account],
+    app_model: App,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Schedule trigger execution should create WorkflowTriggerLog in database."""
+    from tasks import workflow_schedule_tasks
+
+    tenant, account = tenant_and_account
+
+    # Create published workflow with schedule trigger node
+    schedule_node_id = "schedule-node"
+    graph = {
+        "nodes": [
+            {
+                "id": schedule_node_id,
+                "data": {
+                    "type": NodeType.TRIGGER_SCHEDULE.value,
+                    "title": "schedule",
+                    "mode": "cron",
+                    "cron_expression": "0 9 * * *",
+                    "timezone": "UTC",
+                },
+            },
+            {"id": "answer-1", "data": {"type": NodeType.ANSWER.value, "title": "answer"}},
+        ],
+        "edges": [{"source": schedule_node_id, "target": "answer-1", "sourceHandle": "success"}],
+    }
+    published_workflow = Workflow.new(
+        tenant_id=tenant.id,
+        app_id=app_model.id,
+        type="workflow",
+        version=Workflow.version_from_datetime(naive_utc_now()),
+        graph=json.dumps(graph),
+        features=json.dumps({}),
+        created_by=account.id,
+        environment_variables=[],
+        conversation_variables=[],
+        rag_pipeline_variables=[],
+    )
+    db_session_with_containers.add(published_workflow)
+    app_model.workflow_id = published_workflow.id
+    db_session_with_containers.commit()
+
+    # Create schedule plan
+    plan = WorkflowSchedulePlan(
+        app_id=app_model.id,
+        node_id=schedule_node_id,
+        tenant_id=tenant.id,
+        cron_expression="0 9 * * *",
+        timezone="UTC",
+        next_run_at=naive_utc_now() - timedelta(minutes=1),
+    )
+    app_trigger = AppTrigger(
+        tenant_id=tenant.id,
+        app_id=app_model.id,
+        node_id=schedule_node_id,
+        trigger_type=AppTriggerType.TRIGGER_SCHEDULE,
+        status=AppTriggerStatus.ENABLED,
+        title="schedule",
+    )
+    db_session_with_containers.add_all([plan, app_trigger])
+    db_session_with_containers.commit()
+
+    # Mock AsyncWorkflowService to create WorkflowTriggerLog
+    def _fake_trigger_workflow_async(session: Session, user: Any, trigger_data: Any) -> SimpleNamespace:
+        log = WorkflowTriggerLog(
+            tenant_id=trigger_data.tenant_id,
+            app_id=trigger_data.app_id,
+            workflow_id=published_workflow.id,
+            root_node_id=trigger_data.root_node_id,
+            trigger_metadata="{}",
+            trigger_type=AppTriggerType.TRIGGER_SCHEDULE,
+            workflow_run_id=None,
+            outputs=None,
+            trigger_data=trigger_data.model_dump_json(),
+            inputs=json.dumps(dict(trigger_data.inputs)),
+            status=WorkflowTriggerStatus.SUCCEEDED,
+            error="",
+            queue_name="schedule_executor",
+            celery_task_id="celery-schedule-test",
+            created_by_role=CreatorUserRole.ACCOUNT,
+            created_by=account.id,
+        )
+        session.add(log)
+        session.commit()
+        return SimpleNamespace(workflow_trigger_log_id=log.id, task_id=None, status="queued", queue="test")
+
+    monkeypatch.setattr(
+        workflow_schedule_tasks.AsyncWorkflowService,
+        "trigger_workflow_async",
+        _fake_trigger_workflow_async,
+    )
+
+    # Mock quota to avoid rate limiting
+    from enums import quota_type
+
+    monkeypatch.setattr(quota_type.QuotaType.TRIGGER, "consume", lambda _tenant_id: quota_type.unlimited())
+
+    # Execute schedule trigger
+    workflow_schedule_tasks.run_schedule_trigger(plan.id)
+
+    # Verify WorkflowTriggerLog was created
+    db_session_with_containers.expire_all()
+    logs = db_session_with_containers.query(WorkflowTriggerLog).filter_by(app_id=app_model.id).all()
+    assert logs, "Schedule trigger should create WorkflowTriggerLog"
+    assert logs[0].trigger_type == AppTriggerType.TRIGGER_SCHEDULE
+    assert logs[0].root_node_id == schedule_node_id
+
+
+@pytest.mark.parametrize(
+    ("mode", "frequency", "visual_config", "cron_expression", "expected_cron"),
+    [
+        # Visual mode: hourly
+        ("visual", "hourly", {"on_minute": 30}, None, "30 * * * *"),
+        # Visual mode: daily
+        ("visual", "daily", {"time": "3:00 PM"}, None, "0 15 * * *"),
+        # Visual mode: weekly
+        ("visual", "weekly", {"time": "9:00 AM", "weekdays": ["mon", "wed", "fri"]}, None, "0 9 * * 1,3,5"),
+        # Visual mode: monthly
+        ("visual", "monthly", {"time": "10:30 AM", "monthly_days": [1, 15]}, None, "30 10 1,15 * *"),
+        # Cron mode: direct expression
+        ("cron", None, None, "*/5 * * * *", "*/5 * * * *"),
+    ],
+)
+def test_schedule_visual_cron_conversion(
+    mode: str,
+    frequency: str | None,
+    visual_config: dict[str, Any] | None,
+    cron_expression: str | None,
+    expected_cron: str,
+) -> None:
+    """Schedule visual config should correctly convert to cron expression."""
+
+    node_config: dict[str, Any] = {
+        "id": "schedule-node",
+        "data": {
+            "type": NodeType.TRIGGER_SCHEDULE.value,
+            "mode": mode,
+            "timezone": "UTC",
+        },
+    }
+
+    if mode == "visual":
+        node_config["data"]["frequency"] = frequency
+        node_config["data"]["visual_config"] = visual_config
+    else:
+        node_config["data"]["cron_expression"] = cron_expression
+
+    config = ScheduleService.to_schedule_config(node_config)
+
+    assert config.cron_expression == expected_cron, f"Expected {expected_cron}, got {config.cron_expression}"
+    assert config.timezone == "UTC"
+    assert config.node_id == "schedule-node"
+
+
+def test_plugin_trigger_full_chain_with_db_verification(
+    test_client_with_containers: FlaskClient,
+    db_session_with_containers: Session,
+    tenant_and_account: tuple[Tenant, Account],
+    app_model: App,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Plugin trigger should create WorkflowTriggerLog and WorkflowPluginTrigger records."""
+
+    tenant, account = tenant_and_account
+
+    # Create published workflow with plugin trigger node
+    plugin_node_id = "plugin-trigger-node"
+    provider_id = "langgenius/test-provider/test-provider"
+    subscription_id = "sub-plugin-test"
+    endpoint_id = "2cc7fa12-3f7b-4f6a-9c8d-1234567890ab"
+
+    graph = {
+        "nodes": [
+            {
+                "id": plugin_node_id,
+                "data": {
+                    "type": NodeType.TRIGGER_PLUGIN.value,
+                    "title": "plugin",
+                    "plugin_id": "test-plugin",
+                    "plugin_unique_identifier": "test-plugin",
+                    "provider_id": provider_id,
+                    "event_name": "test_event",
+                    "subscription_id": subscription_id,
+                    "parameters": {},
+                },
+            },
+            {"id": "answer-1", "data": {"type": NodeType.ANSWER.value, "title": "answer"}},
+        ],
+        "edges": [{"source": plugin_node_id, "target": "answer-1", "sourceHandle": "success"}],
+    }
+    published_workflow = Workflow.new(
+        tenant_id=tenant.id,
+        app_id=app_model.id,
+        type="workflow",
+        version=Workflow.version_from_datetime(naive_utc_now()),
+        graph=json.dumps(graph),
+        features=json.dumps({}),
+        created_by=account.id,
+        environment_variables=[],
+        conversation_variables=[],
+        rag_pipeline_variables=[],
+    )
+    db_session_with_containers.add(published_workflow)
+    app_model.workflow_id = published_workflow.id
+    db_session_with_containers.commit()
+
+    # Create trigger subscription
+    subscription = TriggerSubscription(
+        name="test-subscription",
+        tenant_id=tenant.id,
+        user_id=account.id,
+        provider_id=provider_id,
+        endpoint_id=endpoint_id,
+        parameters={},
+        properties={},
+        credentials={"token": "test-secret"},
+        credential_type="api-key",
+    )
+    db_session_with_containers.add(subscription)
+    db_session_with_containers.commit()
+
+    # Update subscription_id to match the created subscription
+    graph["nodes"][0]["data"]["subscription_id"] = subscription.id
+    published_workflow.graph = json.dumps(graph)
+    db_session_with_containers.commit()
+
+    # Create WorkflowPluginTrigger
+    plugin_trigger = WorkflowPluginTrigger(
+        app_id=app_model.id,
+        tenant_id=tenant.id,
+        node_id=plugin_node_id,
+        provider_id=provider_id,
+        event_name="test_event",
+        subscription_id=subscription.id,
+    )
+    app_trigger = AppTrigger(
+        tenant_id=tenant.id,
+        app_id=app_model.id,
+        node_id=plugin_node_id,
+        trigger_type=AppTriggerType.TRIGGER_PLUGIN,
+        status=AppTriggerStatus.ENABLED,
+        title="plugin",
+    )
+    db_session_with_containers.add_all([plugin_trigger, app_trigger])
+    db_session_with_containers.commit()
+
+    # Track dispatched data
+    dispatched_data: list[dict[str, Any]] = []
+
+    def _fake_process_endpoint(_endpoint_id: str, _request: Any) -> Response:
+        dispatch_data = {
+            "user_id": "end-user",
+            "tenant_id": tenant.id,
+            "endpoint_id": _endpoint_id,
+            "provider_id": provider_id,
+            "subscription_id": subscription.id,
+            "timestamp": int(time.time()),
+            "events": ["test_event"],
+            "request_id": f"req-{_endpoint_id}",
+        }
+        dispatched_data.append(dispatch_data)
+        return Response("ok", status=202)
+
+    monkeypatch.setattr(
+        "services.trigger.trigger_service.TriggerService.process_endpoint",
+        staticmethod(_fake_process_endpoint),
+    )
+
+    response = test_client_with_containers.post(f"/triggers/plugin/{endpoint_id}", json={"test": "data"})
+
+    assert response.status_code == 202
+    assert dispatched_data, "Plugin trigger should dispatch event data"
+    assert dispatched_data[0]["subscription_id"] == subscription.id
+    assert dispatched_data[0]["events"] == ["test_event"]
+
+    # Verify database records exist
+    db_session_with_containers.expire_all()
+    plugin_triggers = (
+        db_session_with_containers.query(WorkflowPluginTrigger)
+        .filter_by(app_id=app_model.id, node_id=plugin_node_id)
+        .all()
+    )
+    assert plugin_triggers, "WorkflowPluginTrigger record should exist"
+    assert plugin_triggers[0].provider_id == provider_id
+    assert plugin_triggers[0].event_name == "test_event"
+
+
+def test_plugin_debug_via_http_endpoint(
+    test_client_with_containers: FlaskClient,
+    db_session_with_containers: Session,
+    tenant_and_account: tuple[Tenant, Account],
+    app_model: App,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Plugin single-step debug via HTTP endpoint should dispatch debug event and be pollable."""
+
+    tenant, account = tenant_and_account
+
+    provider_id = "langgenius/debug-provider/debug-provider"
+    endpoint_id = "3cc7fa12-3f7b-4f6a-9c8d-1234567890ab"
+    event_name = "debug_event"
+
+    # Create subscription
+    subscription = TriggerSubscription(
+        name="debug-subscription",
+        tenant_id=tenant.id,
+        user_id=account.id,
+        provider_id=provider_id,
+        endpoint_id=endpoint_id,
+        parameters={},
+        properties={},
+        credentials={"token": "debug-secret"},
+        credential_type="api-key",
+    )
+    db_session_with_containers.add(subscription)
+    db_session_with_containers.commit()
+
+    # Create plugin trigger node config
+    node_id = "plugin-debug-node"
+    node_config = {
+        "id": node_id,
+        "data": {
+            "type": NodeType.TRIGGER_PLUGIN.value,
+            "title": "plugin-debug",
+            "plugin_id": "debug-plugin",
+            "plugin_unique_identifier": "debug-plugin",
+            "provider_id": provider_id,
+            "event_name": event_name,
+            "subscription_id": subscription.id,
+            "parameters": {},
+        },
+    }
+
+    # Start listening with poller
+
+    poller = PluginTriggerDebugEventPoller(
+        tenant_id=tenant.id,
+        user_id=account.id,
+        app_id=app_model.id,
+        node_config=node_config,
+        node_id=node_id,
+    )
+    assert poller.poll() is None, "First poll should return None (waiting)"
+
+    # Track debug events dispatched
+    debug_events: list[dict[str, Any]] = []
+    original_dispatch = TriggerDebugEventBus.dispatch
+
+    def _tracking_dispatch(**kwargs: Any) -> int:
+        debug_events.append(kwargs)
+        return original_dispatch(**kwargs)
+
+    monkeypatch.setattr(TriggerDebugEventBus, "dispatch", staticmethod(_tracking_dispatch))
+
+    # Mock process_endpoint to trigger debug event dispatch
+    def _fake_process_endpoint(_endpoint_id: str, _request: Any) -> Response:
+        # Simulate what happens inside process_endpoint + dispatch_triggered_workflows_async
+        pool_key = build_plugin_pool_key(
+            tenant_id=tenant.id,
+            provider_id=provider_id,
+            subscription_id=subscription.id,
+            name=event_name,
+        )
+        TriggerDebugEventBus.dispatch(
+            tenant_id=tenant.id,
+            event=PluginTriggerDebugEvent(
+                timestamp=int(time.time()),
+                user_id="end-user",
+                name=event_name,
+                request_id=f"req-{_endpoint_id}",
+                subscription_id=subscription.id,
+                provider_id=provider_id,
+            ),
+            pool_key=pool_key,
+        )
+        return Response("ok", status=202)
+
+    monkeypatch.setattr(
+        "services.trigger.trigger_service.TriggerService.process_endpoint",
+        staticmethod(_fake_process_endpoint),
+    )
+
+    # Call HTTP endpoint
+    response = test_client_with_containers.post(f"/triggers/plugin/{endpoint_id}", json={"debug": "payload"})
+
+    assert response.status_code == 202
+    assert debug_events, "Debug event should be dispatched via HTTP endpoint"
+    assert debug_events[0]["event"].name == event_name
+
+    # Mock invoke_trigger_event for poller
+
+    monkeypatch.setattr(
+        "services.trigger.trigger_service.TriggerService.invoke_trigger_event",
+        staticmethod(
+            lambda **_kwargs: TriggerInvokeEventResponse(
+                variables={"http_debug": "success"},
+                cancelled=False,
+            )
+        ),
+    )
+
+    # Second poll should receive the event
+    event = poller.poll()
+    assert event is not None, "Poller should receive debug event after HTTP trigger"
+    assert event.workflow_args["inputs"]["http_debug"] == "success"

api/uv.lock

@@ -1337,7 +1337,7 @@ wheels = [
 
 [[package]]
 name = "dify-api"
-version = "1.11.0"
+version = "1.11.1"
 source = { virtual = "." }
 dependencies = [
     { name = "apscheduler" },

docker/docker-compose-template.yaml

@@ -21,7 +21,7 @@ services:
 
   # API service
   api:
-    image: langgenius/dify-api:1.11.0
+    image: langgenius/dify-api:1.11.1
     restart: always
     environment:
       # Use the shared environment variables.
@@ -62,7 +62,7 @@ services:
   # worker service
   # The Celery worker for processing all queues (dataset, workflow, mail, etc.)
   worker:
-    image: langgenius/dify-api:1.11.0
+    image: langgenius/dify-api:1.11.1
     restart: always
     environment:
       # Use the shared environment variables.
@@ -101,7 +101,7 @@ services:
   # worker_beat service
   # Celery beat for scheduling periodic tasks.
   worker_beat:
-    image: langgenius/dify-api:1.11.0
+    image: langgenius/dify-api:1.11.1
     restart: always
     environment:
       # Use the shared environment variables.
@@ -131,7 +131,7 @@ services:
 
   # Frontend web application.
   web:
-    image: langgenius/dify-web:1.11.0
+    image: langgenius/dify-web:1.11.1
     restart: always
     environment:
       CONSOLE_API_URL: ${CONSOLE_API_URL:-}

docker/docker-compose.yaml

@@ -659,7 +659,7 @@ services:
 
   # API service
   api:
-    image: langgenius/dify-api:1.11.0
+    image: langgenius/dify-api:1.11.1
     restart: always
     environment:
       # Use the shared environment variables.
@@ -700,7 +700,7 @@ services:
   # worker service
   # The Celery worker for processing all queues (dataset, workflow, mail, etc.)
   worker:
-    image: langgenius/dify-api:1.11.0
+    image: langgenius/dify-api:1.11.1
     restart: always
     environment:
       # Use the shared environment variables.
@@ -739,7 +739,7 @@ services:
   # worker_beat service
   # Celery beat for scheduling periodic tasks.
   worker_beat:
-    image: langgenius/dify-api:1.11.0
+    image: langgenius/dify-api:1.11.1
     restart: always
     environment:
       # Use the shared environment variables.
@@ -769,7 +769,7 @@ services:
 
   # Frontend web application.
   web:
-    image: langgenius/dify-web:1.11.0
+    image: langgenius/dify-web:1.11.1
     restart: always
     environment:
       CONSOLE_API_URL: ${CONSOLE_API_URL:-}

web/app/components/base/amplitude/AmplitudeProvider.tsx

@@ -4,21 +4,18 @@ import type { FC } from 'react'
 import React, { useEffect } from 'react'
 import * as amplitude from '@amplitude/analytics-browser'
 import { sessionReplayPlugin } from '@amplitude/plugin-session-replay-browser'
-import { IS_CLOUD_EDITION } from '@/config'
+import { AMPLITUDE_API_KEY, IS_CLOUD_EDITION } from '@/config'
 
 export type IAmplitudeProps = {
-  apiKey?: string
   sessionReplaySampleRate?: number
 }
 
 // Check if Amplitude should be enabled
 export const isAmplitudeEnabled = () => {
-  const apiKey = process.env.NEXT_PUBLIC_AMPLITUDE_API_KEY
-  return IS_CLOUD_EDITION && !!apiKey
+  return IS_CLOUD_EDITION && !!AMPLITUDE_API_KEY
 }
 
 const AmplitudeProvider: FC<IAmplitudeProps> = ({
-  apiKey = process.env.NEXT_PUBLIC_AMPLITUDE_API_KEY ?? '',
   sessionReplaySampleRate = 1,
 }) => {
   useEffect(() => {
@@ -27,7 +24,7 @@ const AmplitudeProvider: FC<IAmplitudeProps> = ({
       return
 
     // Initialize Amplitude
-    amplitude.init(apiKey, {
+    amplitude.init(AMPLITUDE_API_KEY, {
       defaultTracking: {
         sessions: true,
         pageViews: true,

web/app/layout.tsx

@@ -42,6 +42,7 @@ const LocaleLayout = async ({
     [DatasetAttr.DATA_MARKETPLACE_API_PREFIX]: process.env.NEXT_PUBLIC_MARKETPLACE_API_PREFIX,
     [DatasetAttr.DATA_MARKETPLACE_URL_PREFIX]: process.env.NEXT_PUBLIC_MARKETPLACE_URL_PREFIX,
     [DatasetAttr.DATA_PUBLIC_EDITION]: process.env.NEXT_PUBLIC_EDITION,
+    [DatasetAttr.DATA_PUBLIC_AMPLITUDE_API_KEY]: process.env.NEXT_PUBLIC_AMPLITUDE_API_KEY,
     [DatasetAttr.DATA_PUBLIC_COOKIE_DOMAIN]: process.env.NEXT_PUBLIC_COOKIE_DOMAIN,
     [DatasetAttr.DATA_PUBLIC_SUPPORT_MAIL_LOGIN]: process.env.NEXT_PUBLIC_SUPPORT_MAIL_LOGIN,
     [DatasetAttr.DATA_PUBLIC_SENTRY_DSN]: process.env.NEXT_PUBLIC_SENTRY_DSN,

web/config/index.ts

@@ -77,6 +77,12 @@ const EDITION = getStringConfig(
 export const IS_CE_EDITION = EDITION === 'SELF_HOSTED'
 export const IS_CLOUD_EDITION = EDITION === 'CLOUD'
 
+export const AMPLITUDE_API_KEY = getStringConfig(
+  process.env.NEXT_PUBLIC_AMPLITUDE_API_KEY,
+  DatasetAttr.DATA_PUBLIC_AMPLITUDE_API_KEY,
+  '',
+)
+
 export const IS_DEV = process.env.NODE_ENV === 'development'
 export const IS_PROD = process.env.NODE_ENV === 'production'
 

web/package.json

@@ -1,6 +1,6 @@
 {
   "name": "dify-web",
-  "version": "1.11.0",
+  "version": "1.11.1",
   "private": true,
   "packageManager": "pnpm@10.25.0+sha512.5e82639027af37cf832061bcc6d639c219634488e0f2baebe785028a793de7b525ffcd3f7ff574f5e9860654e098fe852ba8ac5dd5cefe1767d23a020a92f501",
   "engines": {
@@ -106,15 +106,15 @@
     "mime": "^4.1.0",
     "mitt": "^3.0.1",
     "negotiator": "^1.0.0",
-    "next": "~15.5.7",
+    "next": "~15.5.9",
     "next-pwa": "^5.6.0",
     "next-themes": "^0.4.6",
     "pinyin-pro": "^3.27.0",
     "qrcode.react": "^4.2.0",
     "qs": "^6.14.0",
-    "react": "19.2.1",
+    "react": "19.2.3",
     "react-18-input-autosize": "^3.0.0",
-    "react-dom": "19.2.1",
+    "react-dom": "19.2.3",
     "react-easy-crop": "^5.5.3",
     "react-hook-form": "^7.65.0",
     "react-hotkeys-hook": "^4.6.2",
@@ -155,9 +155,9 @@
     "@happy-dom/jest-environment": "^20.0.8",
     "@mdx-js/loader": "^3.1.1",
     "@mdx-js/react": "^3.1.1",
-    "@next/bundle-analyzer": "15.5.7",
-    "@next/eslint-plugin-next": "15.5.7",
-    "@next/mdx": "15.5.7",
+    "@next/bundle-analyzer": "15.5.9",
+    "@next/eslint-plugin-next": "15.5.9",
+    "@next/mdx": "15.5.9",
     "@rgrove/parse-xml": "^4.2.0",
     "@storybook/addon-docs": "9.1.13",
     "@storybook/addon-links": "9.1.13",

web/types/feature.ts

@@ -106,6 +106,7 @@ export enum DatasetAttr {
   DATA_MARKETPLACE_API_PREFIX = 'data-marketplace-api-prefix',
   DATA_MARKETPLACE_URL_PREFIX = 'data-marketplace-url-prefix',
   DATA_PUBLIC_EDITION = 'data-public-edition',
+  DATA_PUBLIC_AMPLITUDE_API_KEY = 'data-public-amplitude-api-key',
   DATA_PUBLIC_COOKIE_DOMAIN = 'data-public-cookie-domain',
   DATA_PUBLIC_SUPPORT_MAIL_LOGIN = 'data-public-support-mail-login',
   DATA_PUBLIC_SENTRY_DSN = 'data-public-sentry-dsn',