Merge branch 'feat/support-remove-first-and-remove-last-in-variable-assigner' into deploy/dev

.editorconfig

@@ -5,18 +5,35 @@ root = true
 
 # Unix-style newlines with a newline ending every file
 [*]
+charset = utf-8
 end_of_line = lf
 insert_final_newline = true
+trim_trailing_whitespace = true
+
+[*.py]
+indent_size = 4
+indent_style = space
+
+[*.{yml,yaml}]
+indent_style = space
+indent_size = 2
+
+[*.toml]
+indent_size = 4
+indent_style = space
+
+# Markdown and MDX are whitespace sensitive languages.
+# Do not remove trailing spaces.
+[*.{md,mdx}]
+trim_trailing_whitespace = false
 
 # Matches multiple files with brace expansion notation
 # Set default charset
 [*.{js,tsx}]
-charset = utf-8
 indent_style = space
 indent_size = 2
 
-
+# Matches the exact files package.json
-# Matches the exact files either package.json or .travis.yml
+[package.json]
-[{package.json,.travis.yml}]
 indent_style = space
 indent_size = 2

.github/linters/editorconfig-checker.json

@@ -0,0 +1,22 @@
+{
+    "Verbose": false,
+    "Debug": false,
+    "IgnoreDefaults": false,
+    "SpacesAfterTabs": false,
+    "NoColor": false,
+    "Exclude": [
+        "^web/public/vs/",
+        "^web/public/pdf.worker.min.mjs$",
+        "web/app/components/base/icons/src/vender/"
+    ],
+    "AllowedContentTypes": [],
+    "PassedFiles": [],
+    "Disable": {
+        "EndOfLine": false,
+        "Indentation": false,
+        "IndentSize": true,
+        "InsertFinalNewline": false,
+        "TrimTrailingWhitespace": false,
+        "MaxLineLength": false
+    }
+}

.github/workflows/style.yml

@@ -9,6 +9,12 @@ concurrency:
   group: style-${{ github.head_ref || github.run_id }}
   cancel-in-progress: true
 
+permissions:
+  checks: write
+  statuses: write
+  contents: read
+
+
 jobs:
   python-style:
     name: Python Style
@@ -163,3 +169,14 @@ jobs:
           VALIDATE_DOCKERFILE_HADOLINT: true
           VALIDATE_XML: true
           VALIDATE_YAML: true
+
+      - name: EditorConfig checks
+        uses: super-linter/super-linter/slim@v7
+        env:
+          DEFAULT_BRANCH: main
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          IGNORE_GENERATED_FILES: true
+          IGNORE_GITIGNORED_FILES: true
+          # EditorConfig validation
+          VALIDATE_EDITORCONFIG: true
+          EDITORCONFIG_FILE_NAME: editorconfig-checker.json

api/app_factory.py

@@ -52,7 +52,6 @@ def initialize_extensions(app: DifyApp):
         ext_mail,
         ext_migrate,
         ext_otel,
-        ext_otel_patch,
         ext_proxy_fix,
         ext_redis,
         ext_repositories,
@@ -85,7 +84,6 @@ def initialize_extensions(app: DifyApp):
         ext_proxy_fix,
         ext_blueprints,
         ext_commands,
-        ext_otel_patch,  # Apply patch before initializing OpenTelemetry
         ext_otel,
     ]
     for ext in extensions:

api/commands.py

@@ -17,6 +17,7 @@ from core.rag.models.document import Document
 from events.app_event import app_was_created
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
+from extensions.ext_storage import storage
 from libs.helper import email as email_validate
 from libs.password import hash_password, password_pattern, valid_password
 from libs.rsa import generate_key_pair
@@ -815,3 +816,331 @@ def clear_free_plan_tenant_expired_logs(days: int, batch: int, tenant_ids: list[
     ClearFreePlanTenantExpiredLogs.process(days, batch, tenant_ids)
 
     click.echo(click.style("Clear free plan tenant expired logs completed.", fg="green"))
+
+
+@click.option("-f", "--force", is_flag=True, help="Skip user confirmation and force the command to execute.")
+@click.command("clear-orphaned-file-records", help="Clear orphaned file records.")
+def clear_orphaned_file_records(force: bool):
+    """
+    Clear orphaned file records in the database.
+    """
+
+    # define tables and columns to process
+    files_tables = [
+        {"table": "upload_files", "id_column": "id", "key_column": "key"},
+        {"table": "tool_files", "id_column": "id", "key_column": "file_key"},
+    ]
+    ids_tables = [
+        {"type": "uuid", "table": "message_files", "column": "upload_file_id"},
+        {"type": "text", "table": "documents", "column": "data_source_info"},
+        {"type": "text", "table": "document_segments", "column": "content"},
+        {"type": "text", "table": "messages", "column": "answer"},
+        {"type": "text", "table": "workflow_node_executions", "column": "inputs"},
+        {"type": "text", "table": "workflow_node_executions", "column": "process_data"},
+        {"type": "text", "table": "workflow_node_executions", "column": "outputs"},
+        {"type": "text", "table": "conversations", "column": "introduction"},
+        {"type": "text", "table": "conversations", "column": "system_instruction"},
+        {"type": "json", "table": "messages", "column": "inputs"},
+        {"type": "json", "table": "messages", "column": "message"},
+    ]
+
+    # notify user and ask for confirmation
+    click.echo(
+        click.style(
+            "This command will first find and delete orphaned file records from the message_files table,", fg="yellow"
+        )
+    )
+    click.echo(
+        click.style(
+            "and then it will find and delete orphaned file records in the following tables:",
+            fg="yellow",
+        )
+    )
+    for files_table in files_tables:
+        click.echo(click.style(f"- {files_table['table']}", fg="yellow"))
+    click.echo(
+        click.style("The following tables and columns will be scanned to find orphaned file records:", fg="yellow")
+    )
+    for ids_table in ids_tables:
+        click.echo(click.style(f"- {ids_table['table']} ({ids_table['column']})", fg="yellow"))
+    click.echo("")
+
+    click.echo(click.style("!!! USE WITH CAUTION !!!", fg="red"))
+    click.echo(
+        click.style(
+            (
+                "Since not all patterns have been fully tested, "
+                "please note that this command may delete unintended file records."
+            ),
+            fg="yellow",
+        )
+    )
+    click.echo(
+        click.style("This cannot be undone. Please make sure to back up your database before proceeding.", fg="yellow")
+    )
+    click.echo(
+        click.style(
+            (
+                "It is also recommended to run this during the maintenance window, "
+                "as this may cause high load on your instance."
+            ),
+            fg="yellow",
+        )
+    )
+    if not force:
+        click.confirm("Do you want to proceed?", abort=True)
+
+    # start the cleanup process
+    click.echo(click.style("Starting orphaned file records cleanup.", fg="white"))
+
+    # clean up the orphaned records in the message_files table where message_id doesn't exist in messages table
+    try:
+        click.echo(
+            click.style("- Listing message_files records where message_id doesn't exist in messages table", fg="white")
+        )
+        query = (
+            "SELECT mf.id, mf.message_id "
+            "FROM message_files mf LEFT JOIN messages m ON mf.message_id = m.id "
+            "WHERE m.id IS NULL"
+        )
+        orphaned_message_files = []
+        with db.engine.begin() as conn:
+            rs = conn.execute(db.text(query))
+            for i in rs:
+                orphaned_message_files.append({"id": str(i[0]), "message_id": str(i[1])})
+
+        if orphaned_message_files:
+            click.echo(click.style(f"Found {len(orphaned_message_files)} orphaned message_files records:", fg="white"))
+            for record in orphaned_message_files:
+                click.echo(click.style(f"  - id: {record['id']}, message_id: {record['message_id']}", fg="black"))
+
+            if not force:
+                click.confirm(
+                    (
+                        f"Do you want to proceed "
+                        f"to delete all {len(orphaned_message_files)} orphaned message_files records?"
+                    ),
+                    abort=True,
+                )
+
+            click.echo(click.style("- Deleting orphaned message_files records", fg="white"))
+            query = "DELETE FROM message_files WHERE id IN :ids"
+            with db.engine.begin() as conn:
+                conn.execute(db.text(query), {"ids": tuple([record["id"] for record in orphaned_message_files])})
+            click.echo(
+                click.style(f"Removed {len(orphaned_message_files)} orphaned message_files records.", fg="green")
+            )
+        else:
+            click.echo(click.style("No orphaned message_files records found. There is nothing to delete.", fg="green"))
+    except Exception as e:
+        click.echo(click.style(f"Error deleting orphaned message_files records: {str(e)}", fg="red"))
+
+    # clean up the orphaned records in the rest of the *_files tables
+    try:
+        # fetch file id and keys from each table
+        all_files_in_tables = []
+        for files_table in files_tables:
+            click.echo(click.style(f"- Listing file records in table {files_table['table']}", fg="white"))
+            query = f"SELECT {files_table['id_column']}, {files_table['key_column']} FROM {files_table['table']}"
+            with db.engine.begin() as conn:
+                rs = conn.execute(db.text(query))
+            for i in rs:
+                all_files_in_tables.append({"table": files_table["table"], "id": str(i[0]), "key": i[1]})
+        click.echo(click.style(f"Found {len(all_files_in_tables)} files in tables.", fg="white"))
+
+        # fetch referred table and columns
+        guid_regexp = "[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}"
+        all_ids_in_tables = []
+        for ids_table in ids_tables:
+            query = ""
+            if ids_table["type"] == "uuid":
+                click.echo(
+                    click.style(
+                        f"- Listing file ids in column {ids_table['column']} in table {ids_table['table']}", fg="white"
+                    )
+                )
+                query = (
+                    f"SELECT {ids_table['column']} FROM {ids_table['table']} WHERE {ids_table['column']} IS NOT NULL"
+                )
+                with db.engine.begin() as conn:
+                    rs = conn.execute(db.text(query))
+                for i in rs:
+                    all_ids_in_tables.append({"table": ids_table["table"], "id": str(i[0])})
+            elif ids_table["type"] == "text":
+                click.echo(
+                    click.style(
+                        f"- Listing file-id-like strings in column {ids_table['column']} in table {ids_table['table']}",
+                        fg="white",
+                    )
+                )
+                query = (
+                    f"SELECT regexp_matches({ids_table['column']}, '{guid_regexp}', 'g') AS extracted_id "
+                    f"FROM {ids_table['table']}"
+                )
+                with db.engine.begin() as conn:
+                    rs = conn.execute(db.text(query))
+                for i in rs:
+                    for j in i[0]:
+                        all_ids_in_tables.append({"table": ids_table["table"], "id": j})
+            elif ids_table["type"] == "json":
+                click.echo(
+                    click.style(
+                        (
+                            f"- Listing file-id-like JSON string in column {ids_table['column']} "
+                            f"in table {ids_table['table']}"
+                        ),
+                        fg="white",
+                    )
+                )
+                query = (
+                    f"SELECT regexp_matches({ids_table['column']}::text, '{guid_regexp}', 'g') AS extracted_id "
+                    f"FROM {ids_table['table']}"
+                )
+                with db.engine.begin() as conn:
+                    rs = conn.execute(db.text(query))
+                for i in rs:
+                    for j in i[0]:
+                        all_ids_in_tables.append({"table": ids_table["table"], "id": j})
+        click.echo(click.style(f"Found {len(all_ids_in_tables)} file ids in tables.", fg="white"))
+
+    except Exception as e:
+        click.echo(click.style(f"Error fetching keys: {str(e)}", fg="red"))
+        return
+
+    # find orphaned files
+    all_files = [file["id"] for file in all_files_in_tables]
+    all_ids = [file["id"] for file in all_ids_in_tables]
+    orphaned_files = list(set(all_files) - set(all_ids))
+    if not orphaned_files:
+        click.echo(click.style("No orphaned file records found. There is nothing to delete.", fg="green"))
+        return
+    click.echo(click.style(f"Found {len(orphaned_files)} orphaned file records.", fg="white"))
+    for file in orphaned_files:
+        click.echo(click.style(f"- orphaned file id: {file}", fg="black"))
+    if not force:
+        click.confirm(f"Do you want to proceed to delete all {len(orphaned_files)} orphaned file records?", abort=True)
+
+    # delete orphaned records for each file
+    try:
+        for files_table in files_tables:
+            click.echo(click.style(f"- Deleting orphaned file records in table {files_table['table']}", fg="white"))
+            query = f"DELETE FROM {files_table['table']} WHERE {files_table['id_column']} IN :ids"
+            with db.engine.begin() as conn:
+                conn.execute(db.text(query), {"ids": tuple(orphaned_files)})
+    except Exception as e:
+        click.echo(click.style(f"Error deleting orphaned file records: {str(e)}", fg="red"))
+        return
+    click.echo(click.style(f"Removed {len(orphaned_files)} orphaned file records.", fg="green"))
+
+
+@click.option("-f", "--force", is_flag=True, help="Skip user confirmation and force the command to execute.")
+@click.command("remove-orphaned-files-on-storage", help="Remove orphaned files on the storage.")
+def remove_orphaned_files_on_storage(force: bool):
+    """
+    Remove orphaned files on the storage.
+    """
+
+    # define tables and columns to process
+    files_tables = [
+        {"table": "upload_files", "key_column": "key"},
+        {"table": "tool_files", "key_column": "file_key"},
+    ]
+    storage_paths = ["image_files", "tools", "upload_files"]
+
+    # notify user and ask for confirmation
+    click.echo(click.style("This command will find and remove orphaned files on the storage,", fg="yellow"))
+    click.echo(
+        click.style("by comparing the files on the storage with the records in the following tables:", fg="yellow")
+    )
+    for files_table in files_tables:
+        click.echo(click.style(f"- {files_table['table']}", fg="yellow"))
+    click.echo(click.style("The following paths on the storage will be scanned to find orphaned files:", fg="yellow"))
+    for storage_path in storage_paths:
+        click.echo(click.style(f"- {storage_path}", fg="yellow"))
+    click.echo("")
+
+    click.echo(click.style("!!! USE WITH CAUTION !!!", fg="red"))
+    click.echo(
+        click.style(
+            "Currently, this command will work only for opendal based storage (STORAGE_TYPE=opendal).", fg="yellow"
+        )
+    )
+    click.echo(
+        click.style(
+            "Since not all patterns have been fully tested, please note that this command may delete unintended files.",
+            fg="yellow",
+        )
+    )
+    click.echo(
+        click.style("This cannot be undone. Please make sure to back up your storage before proceeding.", fg="yellow")
+    )
+    click.echo(
+        click.style(
+            (
+                "It is also recommended to run this during the maintenance window, "
+                "as this may cause high load on your instance."
+            ),
+            fg="yellow",
+        )
+    )
+    if not force:
+        click.confirm("Do you want to proceed?", abort=True)
+
+    # start the cleanup process
+    click.echo(click.style("Starting orphaned files cleanup.", fg="white"))
+
+    # fetch file id and keys from each table
+    all_files_in_tables = []
+    try:
+        for files_table in files_tables:
+            click.echo(click.style(f"- Listing files from table {files_table['table']}", fg="white"))
+            query = f"SELECT {files_table['key_column']} FROM {files_table['table']}"
+            with db.engine.begin() as conn:
+                rs = conn.execute(db.text(query))
+            for i in rs:
+                all_files_in_tables.append(str(i[0]))
+        click.echo(click.style(f"Found {len(all_files_in_tables)} files in tables.", fg="white"))
+    except Exception as e:
+        click.echo(click.style(f"Error fetching keys: {str(e)}", fg="red"))
+
+    all_files_on_storage = []
+    for storage_path in storage_paths:
+        try:
+            click.echo(click.style(f"- Scanning files on storage path {storage_path}", fg="white"))
+            files = storage.scan(path=storage_path, files=True, directories=False)
+            all_files_on_storage.extend(files)
+        except FileNotFoundError as e:
+            click.echo(click.style(f"  -> Skipping path {storage_path} as it does not exist.", fg="yellow"))
+            continue
+        except Exception as e:
+            click.echo(click.style(f"  -> Error scanning files on storage path {storage_path}: {str(e)}", fg="red"))
+            continue
+    click.echo(click.style(f"Found {len(all_files_on_storage)} files on storage.", fg="white"))
+
+    # find orphaned files
+    orphaned_files = list(set(all_files_on_storage) - set(all_files_in_tables))
+    if not orphaned_files:
+        click.echo(click.style("No orphaned files found. There is nothing to remove.", fg="green"))
+        return
+    click.echo(click.style(f"Found {len(orphaned_files)} orphaned files.", fg="white"))
+    for file in orphaned_files:
+        click.echo(click.style(f"- orphaned file: {file}", fg="black"))
+    if not force:
+        click.confirm(f"Do you want to proceed to remove all {len(orphaned_files)} orphaned files?", abort=True)
+
+    # delete orphaned files
+    removed_files = 0
+    error_files = 0
+    for file in orphaned_files:
+        try:
+            storage.delete(file)
+            removed_files += 1
+            click.echo(click.style(f"- Removing orphaned file: {file}", fg="white"))
+        except Exception as e:
+            error_files += 1
+            click.echo(click.style(f"- Error deleting orphaned file {file}: {str(e)}", fg="red"))
+            continue
+    if error_files == 0:
+        click.echo(click.style(f"Removed {removed_files} orphaned files without errors.", fg="green"))
+    else:
+        click.echo(click.style(f"Removed {removed_files} orphaned files, with {error_files} errors.", fg="yellow"))

api/configs/middleware/vdb/opensearch_config.py

@@ -1,4 +1,5 @@
-from typing import Optional
+import enum
+from typing import Literal, Optional
 
 from pydantic import Field, PositiveInt
 from pydantic_settings import BaseSettings
@@ -9,6 +10,14 @@ class OpenSearchConfig(BaseSettings):
     Configuration settings for OpenSearch
     """
 
+    class AuthMethod(enum.StrEnum):
+        """
+        Authentication method for OpenSearch
+        """
+
+        BASIC = "basic"
+        AWS_MANAGED_IAM = "aws_managed_iam"
+
     OPENSEARCH_HOST: Optional[str] = Field(
         description="Hostname or IP address of the OpenSearch server (e.g., 'localhost' or 'opensearch.example.com')",
         default=None,
@@ -19,6 +28,16 @@ class OpenSearchConfig(BaseSettings):
         default=9200,
     )
 
+    OPENSEARCH_SECURE: bool = Field(
+        description="Whether to use SSL/TLS encrypted connection for OpenSearch (True for HTTPS, False for HTTP)",
+        default=False,
+    )
+
+    OPENSEARCH_AUTH_METHOD: AuthMethod = Field(
+        description="Authentication method for OpenSearch connection (default is 'basic')",
+        default=AuthMethod.BASIC,
+    )
+
     OPENSEARCH_USER: Optional[str] = Field(
         description="Username for authenticating with OpenSearch",
         default=None,
@@ -29,7 +48,11 @@ class OpenSearchConfig(BaseSettings):
         default=None,
     )
 
-    OPENSEARCH_SECURE: bool = Field(
+    OPENSEARCH_AWS_REGION: Optional[str] = Field(
-        description="Whether to use SSL/TLS encrypted connection for OpenSearch (True for HTTPS, False for HTTP)",
+        description="AWS region for OpenSearch (e.g. 'us-west-2')",
-        default=False,
+        default=None,
+    )
+
+    OPENSEARCH_AWS_SERVICE: Optional[Literal["es", "aoss"]] = Field(
+        description="AWS service for OpenSearch (e.g. 'aoss' for OpenSearch Serverless)", default=None
     )

api/configs/packaging/__init__.py

@@ -9,7 +9,7 @@ class PackagingInfo(BaseSettings):
 
     CURRENT_VERSION: str = Field(
         description="Dify version",
-        default="1.3.0",
+        default="1.3.1",
     )
 
     COMMIT_SHA: str = Field(

api/constants/__init__.py

@@ -16,11 +16,25 @@ AUDIO_EXTENSIONS.extend([ext.upper() for ext in AUDIO_EXTENSIONS])
 
 
 if dify_config.ETL_TYPE == "Unstructured":
-    DOCUMENT_EXTENSIONS = ["txt", "markdown", "md", "mdx", "pdf", "html", "htm", "xlsx", "xls"]
+    DOCUMENT_EXTENSIONS = ["txt", "markdown", "md", "mdx", "pdf", "html", "htm", "xlsx", "xls", "vtt", "properties"]
     DOCUMENT_EXTENSIONS.extend(("doc", "docx", "csv", "eml", "msg", "pptx", "xml", "epub"))
     if dify_config.UNSTRUCTURED_API_URL:
         DOCUMENT_EXTENSIONS.append("ppt")
     DOCUMENT_EXTENSIONS.extend([ext.upper() for ext in DOCUMENT_EXTENSIONS])
 else:
-    DOCUMENT_EXTENSIONS = ["txt", "markdown", "md", "mdx", "pdf", "html", "htm", "xlsx", "xls", "docx", "csv"]
+    DOCUMENT_EXTENSIONS = [
+        "txt",
+        "markdown",
+        "md",
+        "mdx",
+        "pdf",
+        "html",
+        "htm",
+        "xlsx",
+        "xls",
+        "docx",
+        "csv",
+        "vtt",
+        "properties",
+    ]
     DOCUMENT_EXTENSIONS.extend([ext.upper() for ext in DOCUMENT_EXTENSIONS])

api/controllers/console/app/annotation.py

@@ -186,7 +186,7 @@ class AnnotationUpdateDeleteApi(Resource):
         app_id = str(app_id)
         annotation_id = str(annotation_id)
         AppAnnotationService.delete_app_annotation(app_id, annotation_id)
-        return {"result": "success"}, 200
+        return {"result": "success"}, 204
 
 
 class AnnotationBatchImportApi(Resource):

api/controllers/console/app/ops_trace.py

@@ -84,7 +84,7 @@ class TraceAppConfigApi(Resource):
             result = OpsService.delete_tracing_app_config(app_id=app_id, tracing_provider=args["tracing_provider"])
             if not result:
                 raise TracingConfigNotExist()
-            return {"result": "success"}
+            return {"result": "success"}, 204
         except Exception as e:
             raise BadRequest(str(e))
 

api/controllers/console/auth/data_source_bearer_auth.py

@@ -65,7 +65,7 @@ class ApiKeyAuthDataSourceBindingDelete(Resource):
 
         ApiKeyAuthService.delete_provider_auth(current_user.current_tenant_id, binding_id)
 
-        return {"result": "success"}, 200
+        return {"result": "success"}, 204
 
 
 api.add_resource(ApiKeyAuthDataSource, "/api-key-auth/data-source")

api/controllers/console/datasets/datasets_document.py

@@ -40,7 +40,7 @@ from core.indexing_runner import IndexingRunner
 from core.model_manager import ModelManager
 from core.model_runtime.entities.model_entities import ModelType
 from core.model_runtime.errors.invoke import InvokeAuthorizationError
-from core.plugin.manager.exc import PluginDaemonClientSideError
+from core.plugin.impl.exc import PluginDaemonClientSideError
 from core.rag.extractor.entity.extract_setting import ExtractSetting
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client

api/controllers/console/datasets/datasets_segments.py

@@ -131,7 +131,7 @@ class DatasetDocumentSegmentListApi(Resource):
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
         SegmentService.delete_segments(segment_ids, document, dataset)
-        return {"result": "success"}, 200
+        return {"result": "success"}, 204
 
 
 class DatasetDocumentSegmentApi(Resource):
@@ -333,7 +333,7 @@ class DatasetDocumentSegmentUpdateApi(Resource):
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
         SegmentService.delete_segment(segment, document, dataset)
-        return {"result": "success"}, 200
+        return {"result": "success"}, 204
 
 
 class DatasetDocumentSegmentBatchImportApi(Resource):
@@ -590,7 +590,7 @@ class ChildChunkUpdateApi(Resource):
             SegmentService.delete_child_chunk(child_chunk, dataset)
         except ChildChunkDeleteIndexServiceError as e:
             raise ChildChunkDeleteIndexError(str(e))
-        return {"result": "success"}, 200
+        return {"result": "success"}, 204
 
     @setup_required
     @login_required

api/controllers/console/datasets/external.py

@@ -135,7 +135,7 @@ class ExternalApiTemplateApi(Resource):
             raise Forbidden()
 
         ExternalDatasetService.delete_external_knowledge_api(current_user.current_tenant_id, external_knowledge_api_id)
-        return {"result": "success"}, 200
+        return {"result": "success"}, 204
 
 
 class ExternalApiUseCheckApi(Resource):

api/controllers/console/datasets/metadata.py

@@ -82,7 +82,7 @@ class DatasetMetadataApi(Resource):
         DatasetService.check_dataset_permission(dataset, current_user)
 
         MetadataService.delete_metadata(dataset_id_str, metadata_id_str)
-        return 200
+        return {"result": "success"}, 204
 
 
 class DatasetMetadataBuiltInFieldApi(Resource):

api/controllers/console/explore/installed_app.py

@@ -113,7 +113,7 @@ class InstalledAppApi(InstalledAppResource):
         db.session.delete(installed_app)
         db.session.commit()
 
-        return {"result": "success", "message": "App uninstalled successfully"}
+        return {"result": "success", "message": "App uninstalled successfully"}, 204
 
     def patch(self, installed_app):
         parser = reqparse.RequestParser()

api/controllers/console/explore/saved_message.py

@@ -72,7 +72,7 @@ class SavedMessageApi(InstalledAppResource):
 
         SavedMessageService.delete(app_model, current_user, message_id)
 
-        return {"result": "success"}
+        return {"result": "success"}, 204
 
 
 api.add_resource(

api/controllers/console/extension.py

@@ -99,7 +99,7 @@ class APIBasedExtensionDetailAPI(Resource):
 
         APIBasedExtensionService.delete(extension_data_from_db)
 
-        return {"result": "success"}
+        return {"result": "success"}, 204
 
 
 api.add_resource(CodeBasedExtensionAPI, "/code-based-extension")

api/controllers/console/tag/tags.py

@@ -86,7 +86,7 @@ class TagUpdateDeleteApi(Resource):
 
         TagService.delete_tag(tag_id)
 
-        return 200
+        return 204
 
 
 class TagBindingCreateApi(Resource):

api/controllers/console/workspace/endpoint.py

@@ -5,7 +5,7 @@ from werkzeug.exceptions import Forbidden
 from controllers.console import api
 from controllers.console.wraps import account_initialization_required, setup_required
 from core.model_runtime.utils.encoders import jsonable_encoder
-from core.plugin.manager.exc import PluginPermissionDeniedError
+from core.plugin.impl.exc import PluginPermissionDeniedError
 from libs.login import login_required
 from services.plugin.endpoint_service import EndpointService
 

api/controllers/console/workspace/plugin.py

@@ -10,7 +10,7 @@ from controllers.console import api
 from controllers.console.workspace import plugin_permission_required
 from controllers.console.wraps import account_initialization_required, setup_required
 from core.model_runtime.utils.encoders import jsonable_encoder
-from core.plugin.manager.exc import PluginDaemonClientSideError
+from core.plugin.impl.exc import PluginDaemonClientSideError
 from libs.login import login_required
 from models.account import TenantPluginPermission
 from services.plugin.plugin_permission_service import PluginPermissionService

api/controllers/files/image_preview.py

@@ -70,6 +70,20 @@ class FilePreviewApi(Resource):
             direct_passthrough=True,
             headers={},
         )
+        # add Accept-Ranges header for audio/video files
+        if upload_file.mime_type in [
+            "audio/mpeg",
+            "audio/wav",
+            "audio/mp4",
+            "audio/ogg",
+            "audio/flac",
+            "audio/aac",
+            "video/mp4",
+            "video/webm",
+            "video/quicktime",
+            "audio/x-m4a",
+        ]:
+            response.headers["Accept-Ranges"] = "bytes"
         if upload_file.size > 0:
             response.headers["Content-Length"] = str(upload_file.size)
         if args["as_attachment"]:

api/controllers/service_api/app/annotation.py

@@ -79,7 +79,7 @@ class AnnotationListApi(Resource):
 class AnnotationUpdateDeleteApi(Resource):
     @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON))
     @marshal_with(annotation_fields)
-    def post(self, app_model: App, end_user: EndUser, annotation_id):
+    def put(self, app_model: App, end_user: EndUser, annotation_id):
         if not current_user.is_editor:
             raise Forbidden()
 
@@ -98,7 +98,7 @@ class AnnotationUpdateDeleteApi(Resource):
 
         annotation_id = str(annotation_id)
         AppAnnotationService.delete_app_annotation(app_model.id, annotation_id)
-        return {"result": "success"}, 200
+        return {"result": "success"}, 204
 
 
 api.add_resource(AnnotationReplyActionApi, "/apps/annotation-reply/<string:action>")

api/controllers/service_api/app/conversation.py

@@ -72,7 +72,7 @@ class ConversationDetailApi(Resource):
             ConversationService.delete(app_model, conversation_id, end_user)
         except services.errors.conversation.ConversationNotExistsError:
             raise NotFound("Conversation Not Exists.")
-        return {"result": "success"}, 200
+        return {"result": "success"}, 204
 
 
 class ConversationRenameApi(Resource):

api/controllers/service_api/dataset/document.py

@@ -323,7 +323,7 @@ class DocumentDeleteApi(DatasetApiResource):
         except services.errors.document.DocumentIndexingError:
             raise DocumentIndexingError("Cannot delete document during indexing.")
 
-        return {"result": "success"}, 200
+        return {"result": "success"}, 204
 
 
 class DocumentListApi(DatasetApiResource):

api/controllers/service_api/dataset/metadata.py

@@ -63,7 +63,7 @@ class DatasetMetadataServiceApi(DatasetApiResource):
         DatasetService.check_dataset_permission(dataset, current_user)
 
         MetadataService.delete_metadata(dataset_id_str, metadata_id_str)
-        return 200
+        return 204
 
 
 class DatasetMetadataBuiltInFieldServiceApi(DatasetApiResource):

api/controllers/service_api/dataset/segment.py

@@ -159,7 +159,7 @@ class DatasetSegmentApi(DatasetApiResource):
         if not segment:
             raise NotFound("Segment not found.")
         SegmentService.delete_segment(segment, document, dataset)
-        return {"result": "success"}, 200
+        return {"result": "success"}, 204
 
     @cloud_edition_billing_resource_check("vector_space", "dataset")
     def post(self, tenant_id, dataset_id, document_id, segment_id):
@@ -344,7 +344,7 @@ class DatasetChildChunkApi(DatasetApiResource):
         except ChildChunkDeleteIndexServiceError as e:
             raise ChildChunkDeleteIndexError(str(e))
 
-        return {"result": "success"}, 200
+        return {"result": "success"}, 204
 
     @cloud_edition_billing_resource_check("vector_space", "dataset")
     @cloud_edition_billing_knowledge_limit_check("add_segment", "dataset")

api/controllers/web/saved_message.py

@@ -67,7 +67,7 @@ class SavedMessageApi(WebApiResource):
 
         SavedMessageService.delete(app_model, end_user, message_id)
 
-        return {"result": "success"}
+        return {"result": "success"}, 204
 
 
 api.add_resource(SavedMessageListApi, "/saved-messages")

api/core/agent/cot_agent_runner.py

@@ -69,6 +69,13 @@ class CotAgentRunner(BaseAgentRunner, ABC):
         tool_instances, prompt_messages_tools = self._init_prompt_tools()
         self._prompt_messages_tools = prompt_messages_tools
 
+        # fix metadata filter not work
+        if app_config.dataset is not None:
+            metadata_filtering_conditions = app_config.dataset.retrieve_config.metadata_filtering_conditions
+            for key, dataset_retriever_tool in tool_instances.items():
+                if hasattr(dataset_retriever_tool, "retrieval_tool"):
+                    dataset_retriever_tool.retrieval_tool.metadata_filtering_conditions = metadata_filtering_conditions
+
         function_call_state = True
         llm_usage: dict[str, Optional[LLMUsage]] = {"usage": None}
         final_answer = ""

api/core/agent/fc_agent_runner.py

@@ -45,6 +45,13 @@ class FunctionCallAgentRunner(BaseAgentRunner):
         # convert tools into ModelRuntime Tool format
         tool_instances, prompt_messages_tools = self._init_prompt_tools()
 
+        # fix metadata filter not work
+        if app_config.dataset is not None:
+            metadata_filtering_conditions = app_config.dataset.retrieve_config.metadata_filtering_conditions
+            for key, dataset_retriever_tool in tool_instances.items():
+                if hasattr(dataset_retriever_tool, "retrieval_tool"):
+                    dataset_retriever_tool.retrieval_tool.metadata_filtering_conditions = metadata_filtering_conditions
+
         assert app_config.agent
 
         iteration_step = 1

api/core/agent/strategy/plugin.py

@@ -4,7 +4,7 @@ from typing import Any, Optional
 from core.agent.entities import AgentInvokeMessage
 from core.agent.plugin_entities import AgentStrategyEntity, AgentStrategyParameter
 from core.agent.strategy.base import BaseAgentStrategy
-from core.plugin.manager.agent import PluginAgentManager
+from core.plugin.impl.agent import PluginAgentClient
 from core.plugin.utils.converter import convert_parameters_to_plugin_format
 
 
@@ -42,7 +42,7 @@ class PluginAgentStrategy(BaseAgentStrategy):
         """
         Invoke the agent strategy.
         """
-        manager = PluginAgentManager()
+        manager = PluginAgentClient()
 
         initialized_params = self.initialize_parameters(params)
         params = convert_parameters_to_plugin_format(initialized_params)

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

@@ -25,8 +25,8 @@ from core.app.entities.task_entities import ChatbotAppBlockingResponse, ChatbotA
 from core.model_runtime.errors.invoke import InvokeAuthorizationError
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.prompt.utils.get_thread_messages_length import get_thread_messages_length
-from core.repository import RepositoryFactory
+from core.workflow.repository import RepositoryFactory
-from core.repository.workflow_node_execution_repository import WorkflowNodeExecutionRepository
+from core.workflow.repository.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 from extensions.ext_database import db
 from factories import file_factory
 from models.account import Account

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

@@ -62,10 +62,10 @@ from core.app.task_pipeline.workflow_cycle_manage import WorkflowCycleManage
 from core.model_runtime.entities.llm_entities import LLMUsage
 from core.model_runtime.utils.encoders import jsonable_encoder
 from core.ops.ops_trace_manager import TraceQueueManager
-from core.repository.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 from core.workflow.enums import SystemVariableKey
 from core.workflow.graph_engine.entities.graph_runtime_state import GraphRuntimeState
 from core.workflow.nodes import NodeType
+from core.workflow.repository.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 from events.message_event import message_was_created
 from extensions.ext_database import db
 from models import Conversation, EndUser, Message, MessageFile

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

@@ -23,8 +23,8 @@ from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerat
 from core.app.entities.task_entities import WorkflowAppBlockingResponse, WorkflowAppStreamResponse
 from core.model_runtime.errors.invoke import InvokeAuthorizationError
 from core.ops.ops_trace_manager import TraceQueueManager
-from core.repository import RepositoryFactory
+from core.workflow.repository import RepositoryFactory
-from core.repository.workflow_node_execution_repository import WorkflowNodeExecutionRepository
+from core.workflow.repository.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 from extensions.ext_database import db
 from factories import file_factory
 from models import Account, App, EndUser, Workflow

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

@@ -54,8 +54,8 @@ from core.app.entities.task_entities import (
 from core.app.task_pipeline.based_generate_task_pipeline import BasedGenerateTaskPipeline
 from core.app.task_pipeline.workflow_cycle_manage import WorkflowCycleManage
 from core.ops.ops_trace_manager import TraceQueueManager
-from core.repository.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 from core.workflow.enums import SystemVariableKey
+from core.workflow.repository.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 from extensions.ext_database import db
 from models.account import Account
 from models.enums import CreatedByRole

api/core/app/task_pipeline/workflow_cycle_manage.py

@@ -49,12 +49,12 @@ from core.file import FILE_MODEL_IDENTITY, File
 from core.model_runtime.utils.encoders import jsonable_encoder
 from core.ops.entities.trace_entity import TraceTaskName
 from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
-from core.repository.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 from core.tools.tool_manager import ToolManager
 from core.workflow.entities.node_entities import NodeRunMetadataKey
 from core.workflow.enums import SystemVariableKey
 from core.workflow.nodes import NodeType
 from core.workflow.nodes.tool.entities import ToolNodeData
+from core.workflow.repository.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 from core.workflow.workflow_entry import WorkflowEntry
 from models.account import Account
 from models.enums import CreatedByRole, WorkflowRunTriggeredFrom
@@ -381,6 +381,8 @@ class WorkflowCycleManage:
         workflow_node_execution.elapsed_time = elapsed_time
         workflow_node_execution.execution_metadata = execution_metadata
 
+        self._workflow_node_execution_repository.update(workflow_node_execution)
+
         return workflow_node_execution
 
     def _handle_workflow_node_execution_retried(

api/core/llm_generator/llm_generator.py

@@ -3,6 +3,8 @@ import logging
 import re
 from typing import Optional, cast
 
+import json_repair
+
 from core.llm_generator.output_parser.rule_config_generator import RuleConfigGeneratorOutputParser
 from core.llm_generator.output_parser.suggested_questions_after_answer import SuggestedQuestionsAfterAnswerOutputParser
 from core.llm_generator.prompts import (
@@ -366,7 +368,20 @@ class LLMGenerator:
                 ),
             )
 
-            generated_json_schema = cast(str, response.message.content)
+            raw_content = response.message.content
+
+            if not isinstance(raw_content, str):
+                raise ValueError(f"LLM response content must be a string, got: {type(raw_content)}")
+
+            try:
+                parsed_content = json.loads(raw_content)
+            except json.JSONDecodeError:
+                parsed_content = json_repair.loads(raw_content)
+
+            if not isinstance(parsed_content, dict | list):
+                raise ValueError(f"Failed to parse structured output from llm: {raw_content}")
+
+            generated_json_schema = json.dumps(parsed_content, indent=2, ensure_ascii=False)
             return {"output": generated_json_schema, "error": ""}
 
         except InvokeError as e:

api/core/model_runtime/model_providers/__base/ai_model.py

@@ -26,7 +26,7 @@ from core.model_runtime.errors.invoke import (
 )
 from core.model_runtime.model_providers.__base.tokenizers.gpt2_tokenzier import GPT2Tokenizer
 from core.plugin.entities.plugin_daemon import PluginDaemonInnerError, PluginModelProviderEntity
-from core.plugin.manager.model import PluginModelManager
+from core.plugin.impl.model import PluginModelClient
 
 
 class AIModel(BaseModel):
@@ -141,7 +141,7 @@ class AIModel(BaseModel):
         :param credentials: model credentials
         :return: model schema
         """
-        plugin_model_manager = PluginModelManager()
+        plugin_model_manager = PluginModelClient()
         cache_key = f"{self.tenant_id}:{self.plugin_id}:{self.provider_name}:{self.model_type.value}:{model}"
         # sort credentials
         sorted_credentials = sorted(credentials.items()) if credentials else []

api/core/model_runtime/model_providers/__base/large_language_model.py

@@ -2,7 +2,7 @@ import logging
 import time
 import uuid
 from collections.abc import Generator, Sequence
-from typing import Optional, Union
+from typing import Optional, Union, cast
 
 from pydantic import ConfigDict
 
@@ -20,7 +20,8 @@ from core.model_runtime.entities.model_entities import (
     PriceType,
 )
 from core.model_runtime.model_providers.__base.ai_model import AIModel
-from core.plugin.manager.model import PluginModelManager
+from core.model_runtime.utils.helper import convert_llm_result_chunk_to_str
+from core.plugin.impl.model import PluginModelClient
 
 logger = logging.getLogger(__name__)
 
@@ -140,7 +141,7 @@ class LargeLanguageModel(AIModel):
         result: Union[LLMResult, Generator[LLMResultChunk, None, None]]
 
         try:
-            plugin_model_manager = PluginModelManager()
+            plugin_model_manager = PluginModelClient()
             result = plugin_model_manager.invoke_llm(
                 tenant_id=self.tenant_id,
                 user_id=user or "unknown",
@@ -280,7 +281,9 @@ class LargeLanguageModel(AIModel):
                     callbacks=callbacks,
                 )
 
-                assistant_message.content += chunk.delta.message.content
+                text = convert_llm_result_chunk_to_str(chunk.delta.message.content)
+                current_content = cast(str, assistant_message.content)
+                assistant_message.content = current_content + text
                 real_model = chunk.model
                 if chunk.delta.usage:
                     usage = chunk.delta.usage
@@ -326,7 +329,7 @@ class LargeLanguageModel(AIModel):
         :return:
         """
         if dify_config.PLUGIN_BASED_TOKEN_COUNTING_ENABLED:
-            plugin_model_manager = PluginModelManager()
+            plugin_model_manager = PluginModelClient()
             return plugin_model_manager.get_llm_num_tokens(
                 tenant_id=self.tenant_id,
                 user_id="unknown",

api/core/model_runtime/model_providers/__base/moderation_model.py

@@ -5,7 +5,7 @@ from pydantic import ConfigDict
 
 from core.model_runtime.entities.model_entities import ModelType
 from core.model_runtime.model_providers.__base.ai_model import AIModel
-from core.plugin.manager.model import PluginModelManager
+from core.plugin.impl.model import PluginModelClient
 
 
 class ModerationModel(AIModel):
@@ -31,7 +31,7 @@ class ModerationModel(AIModel):
         self.started_at = time.perf_counter()
 
         try:
-            plugin_model_manager = PluginModelManager()
+            plugin_model_manager = PluginModelClient()
             return plugin_model_manager.invoke_moderation(
                 tenant_id=self.tenant_id,
                 user_id=user or "unknown",

api/core/model_runtime/model_providers/__base/rerank_model.py

@@ -3,7 +3,7 @@ from typing import Optional
 from core.model_runtime.entities.model_entities import ModelType
 from core.model_runtime.entities.rerank_entities import RerankResult
 from core.model_runtime.model_providers.__base.ai_model import AIModel
-from core.plugin.manager.model import PluginModelManager
+from core.plugin.impl.model import PluginModelClient
 
 
 class RerankModel(AIModel):
@@ -36,7 +36,7 @@ class RerankModel(AIModel):
         :return: rerank result
         """
         try:
-            plugin_model_manager = PluginModelManager()
+            plugin_model_manager = PluginModelClient()
             return plugin_model_manager.invoke_rerank(
                 tenant_id=self.tenant_id,
                 user_id=user or "unknown",

api/core/model_runtime/model_providers/__base/speech2text_model.py

@@ -4,7 +4,7 @@ from pydantic import ConfigDict
 
 from core.model_runtime.entities.model_entities import ModelType
 from core.model_runtime.model_providers.__base.ai_model import AIModel
-from core.plugin.manager.model import PluginModelManager
+from core.plugin.impl.model import PluginModelClient
 
 
 class Speech2TextModel(AIModel):
@@ -28,7 +28,7 @@ class Speech2TextModel(AIModel):
         :return: text for given audio file
         """
         try:
-            plugin_model_manager = PluginModelManager()
+            plugin_model_manager = PluginModelClient()
             return plugin_model_manager.invoke_speech_to_text(
                 tenant_id=self.tenant_id,
                 user_id=user or "unknown",

api/core/model_runtime/model_providers/__base/text_embedding_model.py

@@ -6,7 +6,7 @@ from core.entities.embedding_type import EmbeddingInputType
 from core.model_runtime.entities.model_entities import ModelPropertyKey, ModelType
 from core.model_runtime.entities.text_embedding_entities import TextEmbeddingResult
 from core.model_runtime.model_providers.__base.ai_model import AIModel
-from core.plugin.manager.model import PluginModelManager
+from core.plugin.impl.model import PluginModelClient
 
 
 class TextEmbeddingModel(AIModel):
@@ -38,7 +38,7 @@ class TextEmbeddingModel(AIModel):
         :return: embeddings result
         """
         try:
-            plugin_model_manager = PluginModelManager()
+            plugin_model_manager = PluginModelClient()
             return plugin_model_manager.invoke_text_embedding(
                 tenant_id=self.tenant_id,
                 user_id=user or "unknown",
@@ -61,7 +61,7 @@ class TextEmbeddingModel(AIModel):
         :param texts: texts to embed
         :return:
         """
-        plugin_model_manager = PluginModelManager()
+        plugin_model_manager = PluginModelClient()
         return plugin_model_manager.get_text_embedding_num_tokens(
             tenant_id=self.tenant_id,
             user_id="unknown",

api/core/model_runtime/model_providers/__base/tts_model.py

@@ -6,7 +6,7 @@ from pydantic import ConfigDict
 
 from core.model_runtime.entities.model_entities import ModelType
 from core.model_runtime.model_providers.__base.ai_model import AIModel
-from core.plugin.manager.model import PluginModelManager
+from core.plugin.impl.model import PluginModelClient
 
 logger = logging.getLogger(__name__)
 
@@ -42,7 +42,7 @@ class TTSModel(AIModel):
         :return: translated audio file
         """
         try:
-            plugin_model_manager = PluginModelManager()
+            plugin_model_manager = PluginModelClient()
             return plugin_model_manager.invoke_tts(
                 tenant_id=self.tenant_id,
                 user_id=user or "unknown",
@@ -65,7 +65,7 @@ class TTSModel(AIModel):
         :param credentials: The credentials required to access the TTS model.
         :return: A list of voices supported by the TTS model.
         """
-        plugin_model_manager = PluginModelManager()
+        plugin_model_manager = PluginModelClient()
         return plugin_model_manager.get_tts_model_voices(
             tenant_id=self.tenant_id,
             user_id="unknown",

api/core/model_runtime/model_providers/model_provider_factory.py

@@ -22,8 +22,8 @@ from core.model_runtime.schema_validators.model_credential_schema_validator impo
 from core.model_runtime.schema_validators.provider_credential_schema_validator import ProviderCredentialSchemaValidator
 from core.plugin.entities.plugin import ModelProviderID
 from core.plugin.entities.plugin_daemon import PluginModelProviderEntity
-from core.plugin.manager.asset import PluginAssetManager
+from core.plugin.impl.asset import PluginAssetManager
-from core.plugin.manager.model import PluginModelManager
+from core.plugin.impl.model import PluginModelClient
 
 logger = logging.getLogger(__name__)
 
@@ -40,7 +40,7 @@ class ModelProviderFactory:
         self.provider_position_map = {}
 
         self.tenant_id = tenant_id
-        self.plugin_model_manager = PluginModelManager()
+        self.plugin_model_manager = PluginModelClient()
 
         if not self.provider_position_map:
             # get the path of current classes

api/core/model_runtime/utils/helper.py

@@ -1,6 +1,8 @@
 import pydantic
 from pydantic import BaseModel
 
+from core.model_runtime.entities.message_entities import PromptMessageContentUnionTypes
+
 
 def dump_model(model: BaseModel) -> dict:
     if hasattr(pydantic, "model_dump"):
@@ -8,3 +10,18 @@ def dump_model(model: BaseModel) -> dict:
         return pydantic.model_dump(model)  # type: ignore
     else:
         return model.model_dump()
+
+
+def convert_llm_result_chunk_to_str(content: None | str | list[PromptMessageContentUnionTypes]) -> str:
+    if content is None:
+        message_text = ""
+    elif isinstance(content, str):
+        message_text = content
+    elif isinstance(content, list):
+        # Assuming the list contains PromptMessageContent objects with a "data" attribute
+        message_text = "".join(
+            item.data if hasattr(item, "data") and isinstance(item.data, str) else str(item) for item in content
+        )
+    else:
+        message_text = str(content)
+    return message_text

api/core/ops/langfuse_trace/langfuse_trace.py

@@ -29,7 +29,7 @@ from core.ops.langfuse_trace.entities.langfuse_trace_entity import (
     UnitEnum,
 )
 from core.ops.utils import filter_none_values
-from core.repository.repository_factory import RepositoryFactory
+from core.workflow.repository.repository_factory import RepositoryFactory
 from extensions.ext_database import db
 from models.model import EndUser
 

api/core/ops/langsmith_trace/langsmith_trace.py

@@ -28,7 +28,7 @@ from core.ops.langsmith_trace.entities.langsmith_trace_entity import (
     LangSmithRunUpdateModel,
 )
 from core.ops.utils import filter_none_values, generate_dotted_order
-from core.repository.repository_factory import RepositoryFactory
+from core.workflow.repository.repository_factory import RepositoryFactory
 from extensions.ext_database import db
 from models.model import EndUser, MessageFile
 

api/core/ops/opik_trace/opik_trace.py

@@ -22,7 +22,7 @@ from core.ops.entities.trace_entity import (
     TraceTaskName,
     WorkflowTraceInfo,
 )
-from core.repository.repository_factory import RepositoryFactory
+from core.workflow.repository.repository_factory import RepositoryFactory
 from extensions.ext_database import db
 from models.model import EndUser, MessageFile
 

api/core/plugin/backwards_invocation/app.py

@@ -72,7 +72,7 @@ class PluginAppBackwardsInvocation(BaseBackwardsInvocation):
                 raise ValueError("missing query")
 
             return cls.invoke_chat_app(app, user, conversation_id, query, stream, inputs, files)
-        elif app.mode == AppMode.WORKFLOW.value:
+        elif app.mode == AppMode.WORKFLOW:
             return cls.invoke_workflow_app(app, user, stream, inputs, files)
         elif app.mode == AppMode.COMPLETION:
             return cls.invoke_completion_app(app, user, stream, inputs, files)

api/core/plugin/entities/plugin_daemon.py

@@ -1,6 +1,7 @@
+from collections.abc import Mapping
 from datetime import datetime
 from enum import StrEnum
-from typing import Generic, Optional, TypeVar
+from typing import Any, Generic, Optional, TypeVar
 
 from pydantic import BaseModel, ConfigDict, Field
 
@@ -158,3 +159,11 @@ class PluginInstallTaskStartResponse(BaseModel):
 class PluginUploadResponse(BaseModel):
     unique_identifier: str = Field(description="The unique identifier of the plugin.")
     manifest: PluginDeclaration
+
+
+class PluginOAuthAuthorizationUrlResponse(BaseModel):
+    authorization_url: str = Field(description="The URL of the authorization.")
+
+
+class PluginOAuthCredentialsResponse(BaseModel):
+    credentials: Mapping[str, Any] = Field(description="The credentials of the OAuth.")

api/core/plugin/impl/agent.py

@@ -6,10 +6,10 @@ from core.plugin.entities.plugin import GenericProviderID
 from core.plugin.entities.plugin_daemon import (
     PluginAgentProviderEntity,
 )
-from core.plugin.manager.base import BasePluginManager
+from core.plugin.impl.base import BasePluginClient
 
 
-class PluginAgentManager(BasePluginManager):
+class PluginAgentClient(BasePluginClient):
     def fetch_agent_strategy_providers(self, tenant_id: str) -> list[PluginAgentProviderEntity]:
         """
         Fetch agent providers for the given tenant.

api/core/plugin/impl/asset.py

@@ -1,7 +1,7 @@
-from core.plugin.manager.base import BasePluginManager
+from core.plugin.impl.base import BasePluginClient
 
 
-class PluginAssetManager(BasePluginManager):
+class PluginAssetManager(BasePluginClient):
     def fetch_asset(self, tenant_id: str, id: str) -> bytes:
         """
         Fetch an asset by id.

api/core/plugin/impl/base.py

@@ -18,7 +18,7 @@ from core.model_runtime.errors.invoke import (
 )
 from core.model_runtime.errors.validate import CredentialsValidateFailedError
 from core.plugin.entities.plugin_daemon import PluginDaemonBasicResponse, PluginDaemonError, PluginDaemonInnerError
-from core.plugin.manager.exc import (
+from core.plugin.impl.exc import (
     PluginDaemonBadRequestError,
     PluginDaemonInternalServerError,
     PluginDaemonNotFoundError,
@@ -37,7 +37,7 @@ T = TypeVar("T", bound=(BaseModel | dict | list | bool | str))
 logger = logging.getLogger(__name__)
 
 
-class BasePluginManager:
+class BasePluginClient:
     def _request(
         self,
         method: str,

api/core/plugin/impl/debugging.py

@@ -1,9 +1,9 @@
 from pydantic import BaseModel
 
-from core.plugin.manager.base import BasePluginManager
+from core.plugin.impl.base import BasePluginClient
 
 
-class PluginDebuggingManager(BasePluginManager):
+class PluginDebuggingClient(BasePluginClient):
     def get_debugging_key(self, tenant_id: str) -> str:
         """
         Get the debugging key for the given tenant.

api/core/plugin/impl/endpoint.py

@@ -1,8 +1,8 @@
 from core.plugin.entities.endpoint import EndpointEntityWithInstance
-from core.plugin.manager.base import BasePluginManager
+from core.plugin.impl.base import BasePluginClient
 
 
-class PluginEndpointManager(BasePluginManager):
+class PluginEndpointClient(BasePluginClient):
     def create_endpoint(
         self, tenant_id: str, user_id: str, plugin_unique_identifier: str, name: str, settings: dict
     ) -> bool:

api/core/plugin/impl/model.py

@@ -18,10 +18,10 @@ from core.plugin.entities.plugin_daemon import (
     PluginTextEmbeddingNumTokensResponse,
     PluginVoicesResponse,
 )
-from core.plugin.manager.base import BasePluginManager
+from core.plugin.impl.base import BasePluginClient
 
 
-class PluginModelManager(BasePluginManager):
+class PluginModelClient(BasePluginClient):
     def fetch_model_providers(self, tenant_id: str) -> Sequence[PluginModelProviderEntity]:
         """
         Fetch model providers for the given tenant.

api/core/plugin/impl/oauth.py

@@ -0,0 +1,98 @@
+from collections.abc import Mapping
+from typing import Any
+
+from werkzeug import Request
+
+from core.plugin.entities.plugin_daemon import PluginOAuthAuthorizationUrlResponse, PluginOAuthCredentialsResponse
+from core.plugin.impl.base import BasePluginClient
+
+
+class OAuthHandler(BasePluginClient):
+    def get_authorization_url(
+        self,
+        tenant_id: str,
+        user_id: str,
+        plugin_id: str,
+        provider: str,
+        system_credentials: Mapping[str, Any],
+    ) -> PluginOAuthAuthorizationUrlResponse:
+        return self._request_with_plugin_daemon_response(
+            "POST",
+            f"plugin/{tenant_id}/dispatch/oauth/get_authorization_url",
+            PluginOAuthAuthorizationUrlResponse,
+            data={
+                "user_id": user_id,
+                "data": {
+                    "provider": provider,
+                    "system_credentials": system_credentials,
+                },
+            },
+            headers={
+                "X-Plugin-ID": plugin_id,
+                "Content-Type": "application/json",
+            },
+        )
+
+    def get_credentials(
+        self,
+        tenant_id: str,
+        user_id: str,
+        plugin_id: str,
+        provider: str,
+        system_credentials: Mapping[str, Any],
+        request: Request,
+    ) -> PluginOAuthCredentialsResponse:
+        """
+        Get credentials from the given request.
+        """
+
+        # encode request to raw http request
+        raw_request_bytes = self._convert_request_to_raw_data(request)
+
+        return self._request_with_plugin_daemon_response(
+            "POST",
+            f"plugin/{tenant_id}/dispatch/oauth/get_credentials",
+            PluginOAuthCredentialsResponse,
+            data={
+                "user_id": user_id,
+                "data": {
+                    "provider": provider,
+                    "system_credentials": system_credentials,
+                    "raw_request_bytes": raw_request_bytes,
+                },
+            },
+            headers={
+                "X-Plugin-ID": plugin_id,
+                "Content-Type": "application/json",
+            },
+        )
+
+    def _convert_request_to_raw_data(self, request: Request) -> bytes:
+        """
+        Convert a Request object to raw HTTP data.
+
+        Args:
+            request: The Request object to convert.
+
+        Returns:
+            The raw HTTP data as bytes.
+        """
+        # Start with the request line
+        method = request.method
+        path = request.path
+        protocol = request.headers.get("HTTP_VERSION", "HTTP/1.1")
+        raw_data = f"{method} {path} {protocol}\r\n".encode()
+
+        # Add headers
+        for header_name, header_value in request.headers.items():
+            raw_data += f"{header_name}: {header_value}\r\n".encode()
+
+        # Add empty line to separate headers from body
+        raw_data += b"\r\n"
+
+        # Add body if exists
+        body = request.get_data(as_text=False)
+        if body:
+            raw_data += body
+
+        return raw_data

api/core/plugin/impl/plugin.py

@@ -10,10 +10,10 @@ from core.plugin.entities.plugin import (
     PluginInstallationSource,
 )
 from core.plugin.entities.plugin_daemon import PluginInstallTask, PluginInstallTaskStartResponse, PluginUploadResponse
-from core.plugin.manager.base import BasePluginManager
+from core.plugin.impl.base import BasePluginClient
 
 
-class PluginInstallationManager(BasePluginManager):
+class PluginInstaller(BasePluginClient):
     def fetch_plugin_by_identifier(
         self,
         tenant_id: str,

api/core/plugin/impl/tool.py

@@ -5,11 +5,11 @@ from pydantic import BaseModel
 
 from core.plugin.entities.plugin import GenericProviderID, ToolProviderID
 from core.plugin.entities.plugin_daemon import PluginBasicBooleanResponse, PluginToolProviderEntity
-from core.plugin.manager.base import BasePluginManager
+from core.plugin.impl.base import BasePluginClient
 from core.tools.entities.tool_entities import ToolInvokeMessage, ToolParameter
 
 
-class PluginToolManager(BasePluginManager):
+class PluginToolManager(BasePluginClient):
     def fetch_tool_providers(self, tenant_id: str) -> list[PluginToolProviderEntity]:
         """
         Fetch tool providers for the given tenant.

api/core/rag/datasource/vdb/milvus/milvus_vector.py

@@ -27,8 +27,8 @@ class MilvusConfig(BaseModel):
 
     uri: str  # Milvus server URI
     token: Optional[str] = None  # Optional token for authentication
-    user: str  # Username for authentication
+    user: Optional[str] = None  # Username for authentication
-    password: str  # Password for authentication
+    password: Optional[str] = None  # Password for authentication
     batch_size: int = 100  # Batch size for operations
     database: str = "default"  # Database name
     enable_hybrid_search: bool = False  # Flag to enable hybrid search
@@ -43,6 +43,7 @@ class MilvusConfig(BaseModel):
         """
         if not values.get("uri"):
             raise ValueError("config MILVUS_URI is required")
+        if not values.get("token"):
             if not values.get("user"):
                 raise ValueError("config MILVUS_USER is required")
             if not values.get("password"):
@@ -356,10 +357,13 @@ class MilvusVector(BaseVector):
                 )
             redis_client.set(collection_exist_cache_key, 1, ex=3600)
 
-    def _init_client(self, config) -> MilvusClient:
+    def _init_client(self, config: MilvusConfig) -> MilvusClient:
         """
         Initialize and return a Milvus client.
         """
+        if config.token:
+            client = MilvusClient(uri=config.uri, token=config.token, db_name=config.database)
+        else:
             client = MilvusClient(uri=config.uri, user=config.user, password=config.password, db_name=config.database)
         return client
 

api/core/rag/datasource/vdb/opensearch/opensearch_vector.py

@@ -1,10 +1,9 @@
 import json
 import logging
-import ssl
+from typing import Any, Literal, Optional
-from typing import Any, Optional
 from uuid import uuid4
 
-from opensearchpy import OpenSearch, helpers
+from opensearchpy import OpenSearch, Urllib3AWSV4SignerAuth, Urllib3HttpConnection, helpers
 from opensearchpy.helpers import BulkIndexError
 from pydantic import BaseModel, model_validator
 
@@ -24,9 +23,12 @@ logger = logging.getLogger(__name__)
 class OpenSearchConfig(BaseModel):
     host: str
     port: int
+    secure: bool = False
+    auth_method: Literal["basic", "aws_managed_iam"] = "basic"
     user: Optional[str] = None
     password: Optional[str] = None
-    secure: bool = False
+    aws_region: Optional[str] = None
+    aws_service: Optional[str] = None
 
     @model_validator(mode="before")
     @classmethod
@@ -35,24 +37,40 @@ class OpenSearchConfig(BaseModel):
             raise ValueError("config OPENSEARCH_HOST is required")
         if not values.get("port"):
             raise ValueError("config OPENSEARCH_PORT is required")
+        if values.get("auth_method") == "aws_managed_iam":
+            if not values.get("aws_region"):
+                raise ValueError("config OPENSEARCH_AWS_REGION is required for AWS_MANAGED_IAM auth method")
+            if not values.get("aws_service"):
+                raise ValueError("config OPENSEARCH_AWS_SERVICE is required for AWS_MANAGED_IAM auth method")
         return values
 
-    def create_ssl_context(self) -> ssl.SSLContext:
+    def create_aws_managed_iam_auth(self) -> Urllib3AWSV4SignerAuth:
-        ssl_context = ssl.create_default_context()
+        import boto3  # type: ignore
-        ssl_context.check_hostname = False
+
-        ssl_context.verify_mode = ssl.CERT_NONE  # Disable Certificate Validation
+        return Urllib3AWSV4SignerAuth(
-        return ssl_context
+            credentials=boto3.Session().get_credentials(),
+            region=self.aws_region,
+            service=self.aws_service,  # type: ignore[arg-type]
+        )
 
     def to_opensearch_params(self) -> dict[str, Any]:
         params = {
             "hosts": [{"host": self.host, "port": self.port}],
             "use_ssl": self.secure,
             "verify_certs": self.secure,
+            "connection_class": Urllib3HttpConnection,
+            "pool_maxsize": 20,
         }
-        if self.user and self.password:
+
+        if self.auth_method == "basic":
+            logger.info("Using basic authentication for OpenSearch Vector DB")
+
             params["http_auth"] = (self.user, self.password)
-        if self.secure:
+        elif self.auth_method == "aws_managed_iam":
-            params["ssl_context"] = self.create_ssl_context()
+            logger.info("Using AWS managed IAM role for OpenSearch Vector DB")
+
+            params["http_auth"] = self.create_aws_managed_iam_auth()
+
         return params
 
 
@@ -76,16 +94,23 @@ class OpenSearchVector(BaseVector):
             action = {
                 "_op_type": "index",
                 "_index": self._collection_name.lower(),
-                "_id": uuid4().hex,
                 "_source": {
                     Field.CONTENT_KEY.value: documents[i].page_content,
                     Field.VECTOR.value: embeddings[i],  # Make sure you pass an array here
                     Field.METADATA_KEY.value: documents[i].metadata,
                 },
             }
+            # See https://github.com/langchain-ai/langchainjs/issues/4346#issuecomment-1935123377
+            if self._client_config.aws_service not in ["aoss"]:
+                action["_id"] = uuid4().hex
             actions.append(action)
 
-        helpers.bulk(self._client, actions)
+        helpers.bulk(
+            client=self._client,
+            actions=actions,
+            timeout=30,
+            max_retries=3,
+        )
 
     def get_ids_by_metadata_field(self, key: str, value: str):
         query = {"query": {"term": {f"{Field.METADATA_KEY.value}.{key}": value}}}
@@ -234,6 +259,7 @@ class OpenSearchVector(BaseVector):
                     },
                 }
 
+                logger.info(f"Creating OpenSearch index {self._collection_name.lower()}")
                 self._client.indices.create(index=self._collection_name.lower(), body=index_body)
 
             redis_client.set(collection_exist_cache_key, 1, ex=3600)
@@ -252,9 +278,12 @@ class OpenSearchVectorFactory(AbstractVectorFactory):
         open_search_config = OpenSearchConfig(
             host=dify_config.OPENSEARCH_HOST or "localhost",
             port=dify_config.OPENSEARCH_PORT,
+            secure=dify_config.OPENSEARCH_SECURE,
+            auth_method=dify_config.OPENSEARCH_AUTH_METHOD.value,
             user=dify_config.OPENSEARCH_USER,
             password=dify_config.OPENSEARCH_PASSWORD,
-            secure=dify_config.OPENSEARCH_SECURE,
+            aws_region=dify_config.OPENSEARCH_AWS_REGION,
+            aws_service=dify_config.OPENSEARCH_AWS_SERVICE,
         )
 
         return OpenSearchVector(collection_name=collection_name, config=open_search_config)

api/core/rag/extractor/watercrawl/provider.py

@@ -20,7 +20,7 @@ class WaterCrawlProvider:
         }
         if options.get("crawl_sub_pages", True):
             spider_options["page_limit"] = options.get("limit", 1)
-            spider_options["max_depth"] = options.get("depth", 1)
+            spider_options["max_depth"] = options.get("max_depth", 1)
             spider_options["include_paths"] = options.get("includes", "").split(",") if options.get("includes") else []
             spider_options["exclude_paths"] = options.get("excludes", "").split(",") if options.get("excludes") else []
 

api/core/rag/rerank/rerank_model.py

@@ -52,6 +52,7 @@ class RerankModelRunner(BaseRerankRunner):
         rerank_documents = []
 
         for result in rerank_result.docs:
+            if score_threshold is None or result.score >= score_threshold:
                 # format document
                 rerank_document = Document(
                     page_content=result.text,
@@ -62,4 +63,5 @@ class RerankModelRunner(BaseRerankRunner):
                     rerank_document.metadata["score"] = result.score
                     rerank_documents.append(rerank_document)
 
-        return rerank_documents
+        rerank_documents.sort(key=lambda x: x.metadata.get("score", 0.0), reverse=True)
+        return rerank_documents[:top_n] if top_n else rerank_documents

api/core/repositories/__init__.py

@@ -2,5 +2,5 @@
 Repository implementations for data access.
 
 This package contains concrete implementations of the repository interfaces
-defined in the core.repository package.
+defined in the core.workflow.repository package.
 """

api/core/repositories/repository_registry.py

@@ -11,9 +11,9 @@ from typing import Any
 from sqlalchemy.orm import sessionmaker
 
 from configs import dify_config
-from core.repository.repository_factory import RepositoryFactory
+from core.repositories.workflow_node_execution import SQLAlchemyWorkflowNodeExecutionRepository
+from core.workflow.repository.repository_factory import RepositoryFactory
 from extensions.ext_database import db
-from repositories.workflow_node_execution import SQLAlchemyWorkflowNodeExecutionRepository
 
 logger = logging.getLogger(__name__)
 

api/core/repositories/workflow_node_execution/__init__.py

@@ -2,7 +2,7 @@
 WorkflowNodeExecution repository implementations.
 """
 
-from repositories.workflow_node_execution.sqlalchemy_repository import SQLAlchemyWorkflowNodeExecutionRepository
+from core.repositories.workflow_node_execution.sqlalchemy_repository import SQLAlchemyWorkflowNodeExecutionRepository
 
 __all__ = [
     "SQLAlchemyWorkflowNodeExecutionRepository",

api/core/repositories/workflow_node_execution/sqlalchemy_repository.py

@@ -10,7 +10,7 @@ from sqlalchemy import UnaryExpression, asc, delete, desc, select
 from sqlalchemy.engine import Engine
 from sqlalchemy.orm import sessionmaker
 
-from core.repository.workflow_node_execution_repository import OrderConfig
+from core.workflow.repository.workflow_node_execution_repository import OrderConfig
 from models.workflow import WorkflowNodeExecution, WorkflowNodeExecutionStatus, WorkflowNodeExecutionTriggeredFrom
 
 logger = logging.getLogger(__name__)

api/core/tools/plugin_tool/provider.py

@@ -1,6 +1,6 @@
 from typing import Any
 
-from core.plugin.manager.tool import PluginToolManager
+from core.plugin.impl.tool import PluginToolManager
 from core.tools.__base.tool_runtime import ToolRuntime
 from core.tools.builtin_tool.provider import BuiltinToolProviderController
 from core.tools.entities.tool_entities import ToolProviderEntityWithPlugin, ToolProviderType

api/core/tools/plugin_tool/tool.py

@@ -1,7 +1,7 @@
 from collections.abc import Generator
 from typing import Any, Optional
 
-from core.plugin.manager.tool import PluginToolManager
+from core.plugin.impl.tool import PluginToolManager
 from core.plugin.utils.converter import convert_parameters_to_plugin_format
 from core.tools.__base.tool import Tool
 from core.tools.__base.tool_runtime import ToolRuntime

api/core/tools/tool_engine.py

@@ -246,7 +246,7 @@ class ToolEngine:
                     + "you do not need to create it, just tell the user to check it now."
                 )
             elif response.type == ToolInvokeMessage.MessageType.JSON:
-                result = json.dumps(
+                result += json.dumps(
                     cast(ToolInvokeMessage.JsonMessage, response.message).json_object, ensure_ascii=False
                 )
             else:

api/core/tools/tool_manager.py

@@ -10,7 +10,7 @@ from yarl import URL
 
 import contexts
 from core.plugin.entities.plugin import ToolProviderID
-from core.plugin.manager.tool import PluginToolManager
+from core.plugin.impl.tool import PluginToolManager
 from core.tools.__base.tool_provider import ToolProviderController
 from core.tools.__base.tool_runtime import ToolRuntime
 from core.tools.plugin_tool.provider import PluginToolProviderController

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

@@ -4,6 +4,7 @@ from pydantic import BaseModel, Field
 
 from core.rag.datasource.retrieval_service import RetrievalService
 from core.rag.entities.context_entities import DocumentContext
+from core.rag.entities.metadata_entities import MetadataCondition
 from core.rag.models.document import Document as RetrievalDocument
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
 from core.tools.utils.dataset_retriever.dataset_retriever_base_tool import DatasetRetrieverBaseTool
@@ -33,6 +34,7 @@ class DatasetRetrieverTool(DatasetRetrieverBaseTool):
     args_schema: type[BaseModel] = DatasetRetrieverToolInput
     description: str = "use this to retrieve a dataset. "
     dataset_id: str
+    metadata_filtering_conditions: MetadataCondition
 
     @classmethod
     def from_dataset(cls, dataset: Dataset, **kwargs):
@@ -46,6 +48,7 @@ class DatasetRetrieverTool(DatasetRetrieverBaseTool):
             tenant_id=dataset.tenant_id,
             dataset_id=dataset.id,
             description=description,
+            metadata_filtering_conditions=MetadataCondition(),
             **kwargs,
         )
 
@@ -65,6 +68,7 @@ class DatasetRetrieverTool(DatasetRetrieverBaseTool):
                 dataset_id=dataset.id,
                 query=query,
                 external_retrieval_parameters=dataset.retrieval_model,
+                metadata_condition=self.metadata_filtering_conditions,
             )
             for external_document in external_documents:
                 document = RetrievalDocument(

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

@@ -7,8 +7,8 @@ from core.agent.plugin_entities import AgentStrategyParameter
 from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance, ModelManager
 from core.model_runtime.entities.model_entities import AIModelEntity, ModelType
-from core.plugin.manager.exc import PluginDaemonClientSideError
+from core.plugin.impl.exc import PluginDaemonClientSideError
-from core.plugin.manager.plugin import PluginInstallationManager
+from core.plugin.impl.plugin import PluginInstaller
 from core.provider_manager import ProviderManager
 from core.tools.entities.tool_entities import ToolParameter, ToolProviderType
 from core.tools.tool_manager import ToolManager
@@ -297,7 +297,7 @@ class AgentNode(ToolNode):
         Get agent strategy icon
         :return:
         """
-        manager = PluginInstallationManager()
+        manager = PluginInstaller()
         plugins = manager.list_plugins(self.tenant_id)
         try:
             current_plugin = next(

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

@@ -11,6 +11,7 @@ import docx
 import pandas as pd
 import pypandoc  # type: ignore
 import pypdfium2  # type: ignore
+import webvtt  # type: ignore
 import yaml  # type: ignore
 from docx.document import Document
 from docx.oxml.table import CT_Tbl
@@ -132,6 +133,10 @@ def _extract_text_by_mime_type(*, file_content: bytes, mime_type: str) -> str:
             return _extract_text_from_json(file_content)
         case "application/x-yaml" | "text/yaml":
             return _extract_text_from_yaml(file_content)
+        case "text/vtt":
+            return _extract_text_from_vtt(file_content)
+        case "text/properties":
+            return _extract_text_from_properties(file_content)
         case _:
             raise UnsupportedFileTypeError(f"Unsupported MIME type: {mime_type}")
 
@@ -139,7 +144,7 @@ def _extract_text_by_mime_type(*, file_content: bytes, mime_type: str) -> str:
 def _extract_text_by_file_extension(*, file_content: bytes, file_extension: str) -> str:
     """Extract text from a file based on its file extension."""
     match file_extension:
-        case ".txt" | ".markdown" | ".md" | ".html" | ".htm" | ".xml" | ".vtt":
+        case ".txt" | ".markdown" | ".md" | ".html" | ".htm" | ".xml":
             return _extract_text_from_plain_text(file_content)
         case ".json":
             return _extract_text_from_json(file_content)
@@ -165,6 +170,10 @@ def _extract_text_by_file_extension(*, file_content: bytes, file_extension: str)
             return _extract_text_from_eml(file_content)
         case ".msg":
             return _extract_text_from_msg(file_content)
+        case ".vtt":
+            return _extract_text_from_vtt(file_content)
+        case ".properties":
+            return _extract_text_from_properties(file_content)
         case _:
             raise UnsupportedFileTypeError(f"Unsupported Extension Type: {file_extension}")
 
@@ -214,8 +223,8 @@ def _extract_text_from_doc(file_content: bytes) -> str:
     """
     from unstructured.partition.api import partition_via_api
 
-    if not (dify_config.UNSTRUCTURED_API_URL and dify_config.UNSTRUCTURED_API_KEY):
+    if not dify_config.UNSTRUCTURED_API_URL:
-        raise TextExtractionError("UNSTRUCTURED_API_URL and UNSTRUCTURED_API_KEY must be set")
+        raise TextExtractionError("UNSTRUCTURED_API_URL must be set")
 
     try:
         with tempfile.NamedTemporaryFile(suffix=".doc", delete=False) as temp_file:
@@ -226,7 +235,7 @@ def _extract_text_from_doc(file_content: bytes) -> str:
                     file=file,
                     metadata_filename=temp_file.name,
                     api_url=dify_config.UNSTRUCTURED_API_URL,
-                    api_key=dify_config.UNSTRUCTURED_API_KEY,
+                    api_key=dify_config.UNSTRUCTURED_API_KEY,  # type: ignore
                 )
             os.unlink(temp_file.name)
         return "\n".join([getattr(element, "text", "") for element in elements])
@@ -462,3 +471,68 @@ def _extract_text_from_msg(file_content: bytes) -> str:
         return "\n".join([str(element) for element in elements])
     except Exception as e:
         raise TextExtractionError(f"Failed to extract text from MSG: {str(e)}") from e
+
+
+def _extract_text_from_vtt(vtt_bytes: bytes) -> str:
+    text = _extract_text_from_plain_text(vtt_bytes)
+
+    # remove bom
+    text = text.lstrip("\ufeff")
+
+    raw_results = []
+    for caption in webvtt.from_string(text):
+        raw_results.append((caption.voice, caption.text))
+
+    # Merge consecutive utterances by the same speaker
+    merged_results = []
+    if raw_results:
+        current_speaker, current_text = raw_results[0]
+
+        for i in range(1, len(raw_results)):
+            spk, txt = raw_results[i]
+            if spk == None:
+                merged_results.append((None, current_text))
+                continue
+
+            if spk == current_speaker:
+                # If it is the same speaker, merge the utterances (joined by space)
+                current_text += " " + txt
+            else:
+                # If the speaker changes, register the utterance so far and move on
+                merged_results.append((current_speaker, current_text))
+                current_speaker, current_text = spk, txt
+
+        # Add the last element
+        merged_results.append((current_speaker, current_text))
+    else:
+        merged_results = raw_results
+
+    # Return the result in the specified format: Speaker "text" style
+    formatted = [f'{spk or ""} "{txt}"' for spk, txt in merged_results]
+    return "\n".join(formatted)
+
+
+def _extract_text_from_properties(file_content: bytes) -> str:
+    try:
+        text = _extract_text_from_plain_text(file_content)
+        lines = text.splitlines()
+        result = []
+        for line in lines:
+            line = line.strip()
+            # Preserve comments and empty lines
+            if not line or line.startswith("#") or line.startswith("!"):
+                result.append(line)
+                continue
+
+            if "=" in line:
+                key, value = line.split("=", 1)
+            elif ":" in line:
+                key, value = line.split(":", 1)
+            else:
+                key, value = line, ""
+
+            result.append(f"{key.strip()}: {value.strip()}")
+
+        return "\n".join(result)
+    except Exception as e:
+        raise TextExtractionError(f"Failed to extract text from properties file: {str(e)}") from e

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

@@ -38,6 +38,7 @@ from core.model_runtime.entities.model_entities import (
 )
 from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
 from core.model_runtime.utils.encoders import jsonable_encoder
+from core.model_runtime.utils.helper import convert_llm_result_chunk_to_str
 from core.plugin.entities.plugin import ModelProviderID
 from core.prompt.entities.advanced_prompt_entities import CompletionModelPromptTemplate, MemoryConfig
 from core.prompt.utils.prompt_message_util import PromptMessageUtil
@@ -269,18 +270,7 @@ class LLMNode(BaseNode[LLMNodeData]):
 
     def _handle_invoke_result(self, invoke_result: LLMResult | Generator) -> Generator[NodeEvent, None, None]:
         if isinstance(invoke_result, LLMResult):
-            content = invoke_result.message.content
+            message_text = convert_llm_result_chunk_to_str(invoke_result.message.content)
-            if content is None:
-                message_text = ""
-            elif isinstance(content, str):
-                message_text = content
-            elif isinstance(content, list):
-                # Assuming the list contains PromptMessageContent objects with a "data" attribute
-                message_text = "".join(
-                    item.data if hasattr(item, "data") and isinstance(item.data, str) else str(item) for item in content
-                )
-            else:
-                message_text = str(content)
 
             yield ModelInvokeCompletedEvent(
                 text=message_text,
@@ -295,7 +285,7 @@ class LLMNode(BaseNode[LLMNodeData]):
         usage = None
         finish_reason = None
         for result in invoke_result:
-            text = result.delta.message.content
+            text = convert_llm_result_chunk_to_str(result.delta.message.content)
             full_text += text
 
             yield RunStreamChunkEvent(chunk_content=text, from_variable_selector=[self.node_id, "text"])

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

@@ -6,8 +6,8 @@ from sqlalchemy.orm import Session
 
 from core.callback_handler.workflow_tool_callback_handler import DifyWorkflowCallbackHandler
 from core.file import File, FileTransferMethod
-from core.plugin.manager.exc import PluginDaemonClientSideError
+from core.plugin.impl.exc import PluginDaemonClientSideError
-from core.plugin.manager.plugin import PluginInstallationManager
+from core.plugin.impl.plugin import PluginInstaller
 from core.tools.entities.tool_entities import ToolInvokeMessage, ToolParameter
 from core.tools.errors import ToolInvokeError
 from core.tools.tool_engine import ToolEngine
@@ -307,7 +307,7 @@ class ToolNode(BaseNode[ToolNodeData]):
                     icon = tool_info.get("icon", "")
                     dict_metadata = dict(message.message.metadata)
                     if dict_metadata.get("provider"):
-                        manager = PluginInstallationManager()
+                        manager = PluginInstaller()
                         plugins = manager.list_plugins(self.tenant_id)
                         try:
                             current_plugin = next(

api/core/workflow/nodes/variable_assigner/v2/enums.py

@@ -11,6 +11,8 @@ class Operation(StrEnum):
     SUBTRACT = "-="
     MULTIPLY = "*="
     DIVIDE = "/="
+    REMOVE_FIRST = "remove-first"
+    REMOVE_LAST = "remove-last"
 
 
 class InputType(StrEnum):

api/core/workflow/nodes/variable_assigner/v2/helpers.py

@@ -23,6 +23,15 @@ def is_operation_supported(*, variable_type: SegmentType, operation: Operation):
                 SegmentType.ARRAY_NUMBER,
                 SegmentType.ARRAY_FILE,
             }
+        case Operation.REMOVE_FIRST | Operation.REMOVE_LAST:
+            # Only array variable can have elements removed
+            return variable_type in {
+                SegmentType.ARRAY_ANY,
+                SegmentType.ARRAY_OBJECT,
+                SegmentType.ARRAY_STRING,
+                SegmentType.ARRAY_NUMBER,
+                SegmentType.ARRAY_FILE,
+            }
         case _:
             return False
 
@@ -51,7 +60,7 @@ def is_constant_input_supported(*, variable_type: SegmentType, operation: Operat
 
 
 def is_input_value_valid(*, variable_type: SegmentType, operation: Operation, value: Any):
-    if operation == Operation.CLEAR:
+    if operation in {Operation.CLEAR, Operation.REMOVE_FIRST, Operation.REMOVE_LAST}:
         return True
     match variable_type:
         case SegmentType.STRING:

api/core/workflow/nodes/variable_assigner/v2/node.py

@@ -64,7 +64,7 @@ class VariableAssignerNode(BaseNode[VariableAssignerNodeData]):
                 # Get value from variable pool
                 if (
                     item.input_type == InputType.VARIABLE
-                    and item.operation != Operation.CLEAR
+                    and item.operation not in {Operation.CLEAR, Operation.REMOVE_FIRST, Operation.REMOVE_LAST}
                     and item.value is not None
                 ):
                     value = self.graph_runtime_state.variable_pool.get(item.value)
@@ -165,5 +165,15 @@ class VariableAssignerNode(BaseNode[VariableAssignerNodeData]):
                 return variable.value * value
             case Operation.DIVIDE:
                 return variable.value / value
+            case Operation.REMOVE_FIRST:
+                # If array is empty, do nothing
+                if not variable.value:
+                    return variable.value
+                return variable.value[1:]
+            case Operation.REMOVE_LAST:
+                # If array is empty, do nothing
+                if not variable.value:
+                    return variable.value
+                return variable.value[:-1]
             case _:
                 raise OperationNotSupportedError(operation=operation, variable_type=variable.value_type)

api/core/workflow/repository/__init__.py

@@ -6,8 +6,8 @@ for accessing and manipulating data, regardless of the underlying
 storage mechanism.
 """
 
-from core.repository.repository_factory import RepositoryFactory
+from core.workflow.repository.repository_factory import RepositoryFactory
-from core.repository.workflow_node_execution_repository import WorkflowNodeExecutionRepository
+from core.workflow.repository.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 
 __all__ = [
     "RepositoryFactory",

api/core/workflow/repository/repository_factory.py

@@ -8,7 +8,7 @@ It does not contain any implementation details or dependencies on specific repos
 from collections.abc import Callable, Mapping
 from typing import Any, Literal, Optional, cast
 
-from core.repository.workflow_node_execution_repository import WorkflowNodeExecutionRepository
+from core.workflow.repository.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 
 # Type for factory functions - takes a dict of parameters and returns any repository type
 RepositoryFactoryFunc = Callable[[Mapping[str, Any]], Any]

api/core/workflow/workflow_entry.py

@@ -9,6 +9,7 @@ from core.app.apps.base_app_queue_manager import GenerateTaskStoppedError
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.file.models import File
 from core.workflow.callbacks import WorkflowCallback
+from core.workflow.constants import ENVIRONMENT_VARIABLE_NODE_ID
 from core.workflow.entities.variable_pool import VariablePool
 from core.workflow.errors import WorkflowNodeRunFailedError
 from core.workflow.graph_engine.entities.event import GraphEngineEvent, GraphRunFailedEvent, InNodeEvent
@@ -364,4 +365,5 @@ class WorkflowEntry:
                 input_value = file_factory.build_from_mappings(mappings=input_value, tenant_id=tenant_id)
 
             # append variable and value to variable pool
+            if variable_node_id != ENVIRONMENT_VARIABLE_NODE_ID:
                 variable_pool.add([variable_node_id] + variable_key_list, input_value)

api/docker/entrypoint.sh

@@ -20,7 +20,8 @@ if [[ "${MODE}" == "worker" ]]; then
     CONCURRENCY_OPTION="-c ${CELERY_WORKER_AMOUNT:-1}"
   fi
 
-  exec celery -A app.celery worker -P ${CELERY_WORKER_CLASS:-gevent} $CONCURRENCY_OPTION --loglevel ${LOG_LEVEL:-INFO} \
+  exec celery -A app.celery worker -P ${CELERY_WORKER_CLASS:-gevent} $CONCURRENCY_OPTION \
+    --max-tasks-per-child ${MAX_TASK_PRE_CHILD:-50} --loglevel ${LOG_LEVEL:-INFO} \
     -Q ${CELERY_QUEUES:-dataset,mail,ops_trace,app_deletion}
 
 elif [[ "${MODE}" == "beat" ]]; then

api/extensions/ext_commands.py

@@ -5,6 +5,7 @@ def init_app(app: DifyApp):
     from commands import (
         add_qdrant_index,
         clear_free_plan_tenant_expired_logs,
+        clear_orphaned_file_records,
         convert_to_agent_apps,
         create_tenant,
         extract_plugins,
@@ -13,6 +14,7 @@ def init_app(app: DifyApp):
         install_plugins,
         migrate_data_for_plugin,
         old_metadata_migration,
+        remove_orphaned_files_on_storage,
         reset_email,
         reset_encrypt_key_pair,
         reset_password,
@@ -36,6 +38,8 @@ def init_app(app: DifyApp):
         install_plugins,
         old_metadata_migration,
         clear_free_plan_tenant_expired_logs,
+        clear_orphaned_file_records,
+        remove_orphaned_files_on_storage,
     ]
     for cmd in cmds_to_register:
         app.cli.add_command(cmd)

api/extensions/ext_otel.py

@@ -8,6 +8,100 @@ from typing import Union
 
 from celery.signals import worker_init  # type: ignore
 from flask_login import user_loaded_from_request, user_logged_in  # type: ignore
+
+from configs import dify_config
+from dify_app import DifyApp
+
+
+@user_logged_in.connect
+@user_loaded_from_request.connect
+def on_user_loaded(_sender, user):
+    if dify_config.ENABLE_OTEL:
+        from opentelemetry.trace import get_current_span
+
+        if user:
+            current_span = get_current_span()
+            if current_span:
+                current_span.set_attribute("service.tenant.id", user.current_tenant_id)
+                current_span.set_attribute("service.user.id", user.id)
+
+
+def init_app(app: DifyApp):
+    def is_celery_worker():
+        return "celery" in sys.argv[0].lower()
+
+    def instrument_exception_logging():
+        exception_handler = ExceptionLoggingHandler()
+        logging.getLogger().addHandler(exception_handler)
+
+    def init_flask_instrumentor(app: DifyApp):
+        meter = get_meter("http_metrics", version=dify_config.CURRENT_VERSION)
+        _http_response_counter = meter.create_counter(
+            "http.server.response.count", description="Total number of HTTP responses by status code", unit="{response}"
+        )
+
+        def response_hook(span: Span, status: str, response_headers: list):
+            if span and span.is_recording():
+                if status.startswith("2"):
+                    span.set_status(StatusCode.OK)
+                else:
+                    span.set_status(StatusCode.ERROR, status)
+
+                status = status.split(" ")[0]
+                status_code = int(status)
+                status_class = f"{status_code // 100}xx"
+                _http_response_counter.add(1, {"status_code": status_code, "status_class": status_class})
+
+        instrumentor = FlaskInstrumentor()
+        if dify_config.DEBUG:
+            logging.info("Initializing Flask instrumentor")
+        instrumentor.instrument_app(app, response_hook=response_hook)
+
+    def init_sqlalchemy_instrumentor(app: DifyApp):
+        with app.app_context():
+            engines = list(app.extensions["sqlalchemy"].engines.values())
+            SQLAlchemyInstrumentor().instrument(enable_commenter=True, engines=engines)
+
+    def setup_context_propagation():
+        # Configure propagators
+        set_global_textmap(
+            CompositePropagator(
+                [
+                    TraceContextTextMapPropagator(),  # W3C trace context
+                    B3Format(),  # B3 propagation (used by many systems)
+                ]
+            )
+        )
+
+    def shutdown_tracer():
+        provider = trace.get_tracer_provider()
+        if hasattr(provider, "force_flush"):
+            provider.force_flush()
+
+    class ExceptionLoggingHandler(logging.Handler):
+        """Custom logging handler that creates spans for logging.exception() calls"""
+
+        def emit(self, record):
+            try:
+                if record.exc_info:
+                    tracer = get_tracer_provider().get_tracer("dify.exception.logging")
+                    with tracer.start_as_current_span(
+                        "log.exception",
+                        attributes={
+                            "log.level": record.levelname,
+                            "log.message": record.getMessage(),
+                            "log.logger": record.name,
+                            "log.file.path": record.pathname,
+                            "log.file.line": record.lineno,
+                        },
+                    ) as span:
+                        span.set_status(StatusCode.ERROR)
+                        span.record_exception(record.exc_info[1])
+                        span.set_attribute("exception.type", record.exc_info[0].__name__)
+                        span.set_attribute("exception.message", str(record.exc_info[1]))
+            except Exception:
+                pass
+
     from opentelemetry import trace
     from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter
     from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
@@ -28,26 +122,10 @@ from opentelemetry.sdk.trace.export import (
     )
     from opentelemetry.sdk.trace.sampling import ParentBasedTraceIdRatio
     from opentelemetry.semconv.resource import ResourceAttributes
-from opentelemetry.trace import Span, get_current_span, get_tracer_provider, set_tracer_provider
+    from opentelemetry.trace import Span, get_tracer_provider, set_tracer_provider
     from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
     from opentelemetry.trace.status import StatusCode
 
-from configs import dify_config
-from dify_app import DifyApp
-
-
-@user_logged_in.connect
-@user_loaded_from_request.connect
-def on_user_loaded(_sender, user):
-    if user:
-        current_span = get_current_span()
-        if current_span:
-            current_span.set_attribute("service.tenant.id", user.current_tenant_id)
-            current_span.set_attribute("service.user.id", user.id)
-
-
-def init_app(app: DifyApp):
-    if dify_config.ENABLE_OTEL:
     setup_context_propagation()
     # Initialize OpenTelemetry
     # Follow Semantic Convertions 1.32.0 to define resource attributes
@@ -103,66 +181,24 @@ def init_app(app: DifyApp):
     if not is_celery_worker():
         init_flask_instrumentor(app)
         CeleryInstrumentor(tracer_provider=get_tracer_provider(), meter_provider=get_meter_provider()).instrument()
+    instrument_exception_logging()
     init_sqlalchemy_instrumentor(app)
     atexit.register(shutdown_tracer)
 
 
-def is_celery_worker():
+def is_enabled():
-    return "celery" in sys.argv[0].lower()
+    return dify_config.ENABLE_OTEL
-
-
-def init_flask_instrumentor(app: DifyApp):
-    meter = get_meter("http_metrics", version=dify_config.CURRENT_VERSION)
-    _http_response_counter = meter.create_counter(
-        "http.server.response.count", description="Total number of HTTP responses by status code", unit="{response}"
-    )
-
-    def response_hook(span: Span, status: str, response_headers: list):
-        if span and span.is_recording():
-            if status.startswith("2"):
-                span.set_status(StatusCode.OK)
-            else:
-                span.set_status(StatusCode.ERROR, status)
-
-            status = status.split(" ")[0]
-            status_code = int(status)
-            status_class = f"{status_code // 100}xx"
-            _http_response_counter.add(1, {"status_code": status_code, "status_class": status_class})
-
-    instrumentor = FlaskInstrumentor()
-    if dify_config.DEBUG:
-        logging.info("Initializing Flask instrumentor")
-    instrumentor.instrument_app(app, response_hook=response_hook)
-
-
-def init_sqlalchemy_instrumentor(app: DifyApp):
-    with app.app_context():
-        engines = list(app.extensions["sqlalchemy"].engines.values())
-        SQLAlchemyInstrumentor().instrument(enable_commenter=True, engines=engines)
-
-
-def setup_context_propagation():
-    # Configure propagators
-    set_global_textmap(
-        CompositePropagator(
-            [
-                TraceContextTextMapPropagator(),  # W3C trace context
-                B3Format(),  # B3 propagation (used by many systems)
-            ]
-        )
-    )
 
 
 @worker_init.connect(weak=False)
 def init_celery_worker(*args, **kwargs):
+    if dify_config.ENABLE_OTEL:
+        from opentelemetry.instrumentation.celery import CeleryInstrumentor
+        from opentelemetry.metrics import get_meter_provider
+        from opentelemetry.trace import get_tracer_provider
+
         tracer_provider = get_tracer_provider()
         metric_provider = get_meter_provider()
         if dify_config.DEBUG:
             logging.info("Initializing OpenTelemetry for Celery worker")
         CeleryInstrumentor(tracer_provider=tracer_provider, meter_provider=metric_provider).instrument()
-
-
-def shutdown_tracer():
-    provider = trace.get_tracer_provider()
-    if hasattr(provider, "force_flush"):
-        provider.force_flush()

api/extensions/ext_otel_patch.py

@@ -1,63 +0,0 @@
-"""
-Patch for OpenTelemetry context detach method to handle None tokens gracefully.
-
-This patch addresses the issue where OpenTelemetry's context.detach() method raises a TypeError
-when called with a None token. The error occurs in the contextvars_context.py file where it tries
-to call reset() on a None token.
-
-Related GitHub issue: https://github.com/langgenius/dify/issues/18496
-
-Error being fixed:
-```
-Traceback (most recent call last):
-  File "opentelemetry/context/__init__.py", line 154, in detach
-    _RUNTIME_CONTEXT.detach(token)
-  File "opentelemetry/context/contextvars_context.py", line 50, in detach
-    self._current_context.reset(token)  # type: ignore
-    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
-TypeError: expected an instance of Token, got None
-```
-
-Instead of modifying the third-party package directly, this patch monkey-patches the
-context.detach method to gracefully handle None tokens.
-"""
-
-import logging
-from functools import wraps
-
-from opentelemetry import context
-
-logger = logging.getLogger(__name__)
-
-# Store the original detach method
-original_detach = context.detach
-
-
-# Create a patched version that handles None tokens
-@wraps(original_detach)
-def patched_detach(token):
-    """
-    A patched version of context.detach that handles None tokens gracefully.
-    """
-    if token is None:
-        logger.debug("Attempted to detach a None token, skipping")
-        return
-
-    return original_detach(token)
-
-
-def is_enabled():
-    """
-    Check if the extension is enabled.
-    Always enable this patch to prevent errors even when OpenTelemetry is disabled.
-    """
-    return True
-
-
-def init_app(app):
-    """
-    Initialize the OpenTelemetry context patch.
-    """
-    # Replace the original detach method with our patched version
-    context.detach = patched_detach
-    logger.info("OpenTelemetry context.detach patched to handle None tokens")

api/extensions/ext_repositories.py

@@ -4,8 +4,8 @@ Extension for initializing repositories.
 This extension registers repository implementations with the RepositoryFactory.
 """
 
+from core.repositories.repository_registry import register_repositories
 from dify_app import DifyApp
-from repositories.repository_registry import register_repositories
 
 
 def init_app(_app: DifyApp) -> None:

api/extensions/ext_storage.py

@@ -102,6 +102,9 @@ class Storage:
     def delete(self, filename):
         return self.storage_runner.delete(filename)
 
+    def scan(self, path: str, files: bool = True, directories: bool = False) -> list[str]:
+        return self.storage_runner.scan(path, files=files, directories=directories)
+
 
 storage = Storage()
 

api/extensions/storage/base_storage.py

@@ -30,3 +30,11 @@ class BaseStorage(ABC):
     @abstractmethod
     def delete(self, filename):
         raise NotImplementedError
+
+    def scan(self, path, files=True, directories=False) -> list[str]:
+        """
+        Scan files and directories in the given path.
+        This method is implemented only in some storage backends.
+        If a storage backend doesn't support scanning, it will raise NotImplementedError.
+        """
+        raise NotImplementedError("This storage backend doesn't support scanning")

api/extensions/storage/opendal_storage.py

@@ -80,3 +80,20 @@ class OpenDALStorage(BaseStorage):
             logger.debug(f"file {filename} deleted")
             return
         logger.debug(f"file {filename} not found, skip delete")
+
+    def scan(self, path: str, files: bool = True, directories: bool = False) -> list[str]:
+        if not self.exists(path):
+            raise FileNotFoundError("Path not found")
+
+        all_files = self.op.scan(path=path)
+        if files and directories:
+            logger.debug(f"files and directories on {path} scanned")
+            return [f.path for f in all_files]
+        if files:
+            logger.debug(f"files on {path} scanned")
+            return [f.path for f in all_files if not f.path.endswith("/")]
+        elif directories:
+            logger.debug(f"directories on {path} scanned")
+            return [f.path for f in all_files if f.path.endswith("/")]
+        else:
+            raise ValueError("At least one of files or directories must be True")

api/factories/agent_factory.py

@@ -1,12 +1,12 @@
 from core.agent.strategy.plugin import PluginAgentStrategy
-from core.plugin.manager.agent import PluginAgentManager
+from core.plugin.impl.agent import PluginAgentClient
 
 
 def get_plugin_agent_strategy(
     tenant_id: str, agent_strategy_provider_name: str, agent_strategy_name: str
 ) -> PluginAgentStrategy:
     # TODO: use contexts to cache the agent provider
-    manager = PluginAgentManager()
+    manager = PluginAgentClient()
     agent_provider = manager.fetch_agent_strategy_provider(tenant_id, agent_strategy_provider_name)
     for agent_strategy in agent_provider.declaration.strategies:
         if agent_strategy.identity.name == agent_strategy_name:

api/models/model.py

@@ -1012,7 +1012,9 @@ class Message(db.Model):  # type: ignore[name-defined]
                 sign_url = file_helpers.get_signed_file_url(upload_file_id)
             else:
                 continue
-
+            # if as_attachment is in the url, add it to the sign_url.
+            if "as_attachment" in url:
+                sign_url += "&as_attachment=true"
             re_sign_file_url_answer = re_sign_file_url_answer.replace(url, sign_url)
 
         return re_sign_file_url_answer

api/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "dify-api"
-version = "1.3.0"
+dynamic = ["version"]
 requires-python = ">=3.11,<3.13"
 
 dependencies = [
@@ -81,15 +81,19 @@ dependencies = [
     "tokenizers~=0.15.0",
     "transformers~=4.35.0",
     "unstructured[docx,epub,md,ppt,pptx]~=0.16.1",
-    "validators==0.21.0",
     "weave~=0.51.34",
     "yarl~=1.18.3",
+    "webvtt-py~=0.5.1",
 ]
 # Before adding new dependency, consider place it in
 # alphabet order (a-z) and suitable group.
 
+[tool.setuptools]
+packages = []
+
 [tool.uv]
 default-groups = ["storage", "tools", "vdb"]
+package = false
 
 [dependency-groups]
 
@@ -191,6 +195,6 @@ vdb = [
     "tidb-vector==0.0.9",
     "upstash-vector==0.6.0",
     "volcengine-compat~=1.0.156",
-    "weaviate-client~=3.21.0",
+    "weaviate-client~=3.24.0",
     "xinference-client~=1.2.2",
 ]

api/services/agent_service.py

@@ -6,8 +6,8 @@ from flask_login import current_user  # type: ignore
 
 import contexts
 from core.app.app_config.easy_ui_based_app.agent.manager import AgentConfigManager
-from core.plugin.manager.agent import PluginAgentManager
+from core.plugin.impl.agent import PluginAgentClient
-from core.plugin.manager.exc import PluginDaemonClientSideError
+from core.plugin.impl.exc import PluginDaemonClientSideError
 from core.tools.tool_manager import ToolManager
 from extensions.ext_database import db
 from models.account import Account
@@ -161,7 +161,7 @@ class AgentService:
         """
         List agent providers
         """
-        manager = PluginAgentManager()
+        manager = PluginAgentClient()
         return manager.fetch_agent_strategy_providers(tenant_id)
 
     @classmethod
@@ -169,7 +169,7 @@ class AgentService:
         """
         Get agent provider
         """
-        manager = PluginAgentManager()
+        manager = PluginAgentClient()
         try:
             return manager.fetch_agent_strategy_provider(tenant_id, provider_name)
         except PluginDaemonClientSideError as e:

api/services/external_knowledge_service.py

@@ -2,9 +2,9 @@ import json
 from copy import deepcopy
 from datetime import UTC, datetime
 from typing import Any, Optional, Union, cast
+from urllib.parse import urlparse
 
 import httpx
-import validators
 
 from constants import HIDDEN_VALUE
 from core.helper import ssrf_proxy
@@ -72,7 +72,9 @@ class ExternalDatasetService:
 
         endpoint = f"{settings['endpoint']}/retrieval"
         api_key = settings["api_key"]
-        if not validators.url(endpoint, simple_host=True):
+
+        parsed_url = urlparse(endpoint)
+        if not all([parsed_url.scheme, parsed_url.netloc]):
             if not endpoint.startswith("http://") and not endpoint.startswith("https://"):
                 raise ValueError(f"invalid endpoint: {endpoint} must start with http:// or https://")
             else:

api/services/plugin/dependencies_analysis.py

@@ -1,7 +1,7 @@
 from configs import dify_config
 from core.helper import marketplace
 from core.plugin.entities.plugin import ModelProviderID, PluginDependency, PluginInstallationSource, ToolProviderID
-from core.plugin.manager.plugin import PluginInstallationManager
+from core.plugin.impl.plugin import PluginInstaller
 
 
 class DependenciesAnalysisService:
@@ -38,7 +38,7 @@ class DependenciesAnalysisService:
         for dependency in dependencies:
             required_plugin_unique_identifiers.append(dependency.value.plugin_unique_identifier)
 
-        manager = PluginInstallationManager()
+        manager = PluginInstaller()
 
         # get leaked dependencies
         missing_plugins = manager.fetch_missing_dependencies(tenant_id, required_plugin_unique_identifiers)
@@ -64,7 +64,7 @@ class DependenciesAnalysisService:
         Generate dependencies through the list of plugin ids
         """
         dependencies = list(set(dependencies))
-        manager = PluginInstallationManager()
+        manager = PluginInstaller()
         plugins = manager.fetch_plugin_installation_by_ids(tenant_id, dependencies)
         result = []
         for plugin in plugins:

api/services/plugin/endpoint_service.py

@@ -1,10 +1,10 @@
-from core.plugin.manager.endpoint import PluginEndpointManager
+from core.plugin.impl.endpoint import PluginEndpointClient
 
 
 class EndpointService:
     @classmethod
     def create_endpoint(cls, tenant_id: str, user_id: str, plugin_unique_identifier: str, name: str, settings: dict):
-        return PluginEndpointManager().create_endpoint(
+        return PluginEndpointClient().create_endpoint(
             tenant_id=tenant_id,
             user_id=user_id,
             plugin_unique_identifier=plugin_unique_identifier,
@@ -14,7 +14,7 @@ class EndpointService:
 
     @classmethod
     def list_endpoints(cls, tenant_id: str, user_id: str, page: int, page_size: int):
-        return PluginEndpointManager().list_endpoints(
+        return PluginEndpointClient().list_endpoints(
             tenant_id=tenant_id,
             user_id=user_id,
             page=page,
@@ -23,7 +23,7 @@ class EndpointService:
 
     @classmethod
     def list_endpoints_for_single_plugin(cls, tenant_id: str, user_id: str, plugin_id: str, page: int, page_size: int):
-        return PluginEndpointManager().list_endpoints_for_single_plugin(
+        return PluginEndpointClient().list_endpoints_for_single_plugin(
             tenant_id=tenant_id,
             user_id=user_id,
             plugin_id=plugin_id,
@@ -33,7 +33,7 @@ class EndpointService:
 
     @classmethod
     def update_endpoint(cls, tenant_id: str, user_id: str, endpoint_id: str, name: str, settings: dict):
-        return PluginEndpointManager().update_endpoint(
+        return PluginEndpointClient().update_endpoint(
             tenant_id=tenant_id,
             user_id=user_id,
             endpoint_id=endpoint_id,
@@ -43,7 +43,7 @@ class EndpointService:
 
     @classmethod
     def delete_endpoint(cls, tenant_id: str, user_id: str, endpoint_id: str):
-        return PluginEndpointManager().delete_endpoint(
+        return PluginEndpointClient().delete_endpoint(
             tenant_id=tenant_id,
             user_id=user_id,
             endpoint_id=endpoint_id,
@@ -51,7 +51,7 @@ class EndpointService:
 
     @classmethod
     def enable_endpoint(cls, tenant_id: str, user_id: str, endpoint_id: str):
-        return PluginEndpointManager().enable_endpoint(
+        return PluginEndpointClient().enable_endpoint(
             tenant_id=tenant_id,
             user_id=user_id,
             endpoint_id=endpoint_id,
@@ -59,7 +59,7 @@ class EndpointService:
 
     @classmethod
     def disable_endpoint(cls, tenant_id: str, user_id: str, endpoint_id: str):
-        return PluginEndpointManager().disable_endpoint(
+        return PluginEndpointClient().disable_endpoint(
             tenant_id=tenant_id,
             user_id=user_id,
             endpoint_id=endpoint_id,