try

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

.github/dependabot.yml

@@ -1,12 +1,25 @@
 version: 2
+
+multi-ecosystem-groups:
+  python:
+    schedule:
+      interval: "weekly"  # or whatever schedule you want
+
 updates:
+  - package-ecosystem: "pip"
+    directory: "/api"
+    open-pull-requests-limit: 2
+    patterns: ["*"]
+    schedule:
+      interval: "weekly"
+  - package-ecosystem: "uv"
+    directory: "/api"
+    open-pull-requests-limit: 2
+    patterns: ["*"]
+    schedule:
+      interval: "weekly"
   - package-ecosystem: "npm"
     directory: "/web"
     schedule:
       interval: "weekly"
     open-pull-requests-limit: 2
-  - package-ecosystem: "uv"
-    directory: "/api"
-    schedule:
-      interval: "weekly"
-    open-pull-requests-limit: 2

.github/workflows/pyrefly-diff-comment.yml

@@ -0,0 +1,88 @@
+name: Comment with Pyrefly Diff
+
+on:
+  workflow_run:
+    workflows:
+      - Pyrefly Diff Check
+    types:
+      - completed
+
+permissions: {}
+
+jobs:
+  comment:
+    name: Comment PR with pyrefly diff
+    runs-on: ubuntu-latest
+    permissions:
+      actions: read
+      contents: read
+      issues: write
+      pull-requests: write
+    if: ${{ github.event.workflow_run.conclusion == 'success' && github.event.workflow_run.pull_requests[0].head.repo.full_name != github.repository }}
+    steps:
+      - name: Download pyrefly diff artifact
+        uses: actions/github-script@v8
+        with:
+          github-token: ${{ secrets.GITHUB_TOKEN }}
+          script: |
+            const fs = require('fs');
+            const artifacts = await github.rest.actions.listWorkflowRunArtifacts({
+              owner: context.repo.owner,
+              repo: context.repo.repo,
+              run_id: ${{ github.event.workflow_run.id }},
+            });
+            const match = artifacts.data.artifacts.find((artifact) =>
+              artifact.name === 'pyrefly_diff'
+            );
+            if (!match) {
+              throw new Error('pyrefly_diff artifact not found');
+            }
+            const download = await github.rest.actions.downloadArtifact({
+              owner: context.repo.owner,
+              repo: context.repo.repo,
+              artifact_id: match.id,
+              archive_format: 'zip',
+            });
+            fs.writeFileSync('pyrefly_diff.zip', Buffer.from(download.data));
+
+      - name: Unzip artifact
+        run: unzip -o pyrefly_diff.zip
+
+      - name: Post comment
+        uses: actions/github-script@v8
+        with:
+          github-token: ${{ secrets.GITHUB_TOKEN }}
+          script: |
+            const fs = require('fs');
+            let diff = fs.readFileSync('pyrefly_diff.txt', { encoding: 'utf8' });
+            let prNumber = null;
+            try {
+              prNumber = parseInt(fs.readFileSync('pr_number.txt', { encoding: 'utf8' }), 10);
+            } catch (err) {
+              // Fallback to workflow_run payload if artifact is missing or incomplete.
+              const prs = context.payload.workflow_run.pull_requests || [];
+              if (prs.length > 0 && prs[0].number) {
+                prNumber = prs[0].number;
+              }
+            }
+            if (!prNumber) {
+              throw new Error('PR number not found in artifact or workflow_run payload');
+            }
+
+            const MAX_CHARS = 65000;
+            if (diff.length > MAX_CHARS) {
+              diff = diff.slice(0, MAX_CHARS);
+              diff = diff.slice(0, diff.lastIndexOf('\\n'));
+              diff += '\\n\\n... (truncated) ...';
+            }
+
+            const body = diff.trim()
+              ? `### Pyrefly Diff (base → PR)\\n\\`\\`\\`diff\\n${diff}\\n\\`\\`\\``
+              : '### Pyrefly Diff\\nNo changes detected.';
+
+            await github.rest.issues.createComment({
+              issue_number: prNumber,
+              owner: context.repo.owner,
+              repo: context.repo.repo,
+              body,
+            });

.github/workflows/pyrefly-diff.yml

@@ -0,0 +1,85 @@
+name: Pyrefly Diff Check
+
+on:
+  pull_request:
+    paths:
+      - 'api/**/*.py'
+
+permissions:
+  contents: read
+
+jobs:
+  pyrefly-diff:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      issues: write
+      pull-requests: write
+    steps:
+      - name: Checkout PR branch
+        uses: actions/checkout@v6
+        with:
+          fetch-depth: 0
+
+      - name: Setup Python & UV
+        uses: astral-sh/setup-uv@v5
+        with:
+          enable-cache: true
+
+      - name: Install dependencies
+        run: uv sync --project api --dev
+
+      - name: Run pyrefly on PR branch
+        run: |
+          uv run --directory api pyrefly check > /tmp/pyrefly_pr.txt 2>&1 || true
+
+      - name: Checkout base branch
+        run: git checkout ${{ github.base_ref }}
+
+      - name: Run pyrefly on base branch
+        run: |
+          uv run --directory api pyrefly check > /tmp/pyrefly_base.txt 2>&1 || true
+
+      - name: Compute diff
+        run: |
+          diff /tmp/pyrefly_base.txt /tmp/pyrefly_pr.txt > pyrefly_diff.txt || true
+
+      - name: Save PR number
+        run: |
+          echo ${{ github.event.pull_request.number }} > pr_number.txt
+
+      - name: Upload pyrefly diff
+        uses: actions/upload-artifact@v4
+        with:
+          name: pyrefly_diff
+          path: |
+            pyrefly_diff.txt
+            pr_number.txt
+
+      - name: Comment PR with pyrefly diff
+        if: ${{ github.event.pull_request.head.repo.full_name == github.repository }}
+        uses: actions/github-script@v8
+        with:
+          github-token: ${{ secrets.GITHUB_TOKEN }}
+          script: |
+            const fs = require('fs');
+            let diff = fs.readFileSync('pyrefly_diff.txt', { encoding: 'utf8' });
+            const prNumber = context.payload.pull_request.number;
+
+            const MAX_CHARS = 65000;
+            if (diff.length > MAX_CHARS) {
+              diff = diff.slice(0, MAX_CHARS);
+              diff = diff.slice(0, diff.lastIndexOf('\n'));
+              diff += '\n\n... (truncated) ...';
+            }
+
+            const body = diff.trim()
+              ? `### Pyrefly Diff (base → PR)\n\`\`\`diff\n${diff}\n\`\`\``
+              : '### Pyrefly Diff\nNo changes detected.';
+
+            await github.rest.issues.createComment({
+              issue_number: prNumber,
+              owner: context.repo.owner,
+              repo: context.repo.repo,
+              body,
+            });

api/.env.example

@@ -553,6 +553,8 @@ WORKFLOW_LOG_CLEANUP_ENABLED=false
 WORKFLOW_LOG_RETENTION_DAYS=30
 # Batch size for workflow log cleanup operations (default: 100)
 WORKFLOW_LOG_CLEANUP_BATCH_SIZE=100
+# Comma-separated list of workflow IDs to clean logs for
+WORKFLOW_LOG_CLEANUP_SPECIFIC_WORKFLOW_IDS=
 
 # App configuration
 APP_MAX_EXECUTION_TIME=1200
@@ -715,6 +717,7 @@ ANNOTATION_IMPORT_MAX_CONCURRENT=5
 # Sandbox expired records clean configuration
 SANDBOX_EXPIRED_RECORDS_CLEAN_GRACEFUL_PERIOD=21
 SANDBOX_EXPIRED_RECORDS_CLEAN_BATCH_SIZE=1000
+SANDBOX_EXPIRED_RECORDS_CLEAN_BATCH_MAX_INTERVAL=200
 SANDBOX_EXPIRED_RECORDS_RETENTION_DAYS=30
 SANDBOX_EXPIRED_RECORDS_CLEAN_TASK_LOCK_TTL=90000
 

api/.importlinter

@@ -114,19 +114,12 @@ ignore_imports =
     core.workflow.nodes.datasource.datasource_node -> models.model
     core.workflow.nodes.datasource.datasource_node -> models.tools
     core.workflow.nodes.datasource.datasource_node -> services.datasource_provider_service
-    core.workflow.nodes.document_extractor.node -> configs
-    core.workflow.nodes.document_extractor.node -> core.file.file_manager
     core.workflow.nodes.document_extractor.node -> core.helper.ssrf_proxy
-    core.workflow.nodes.http_request.entities -> configs
-    core.workflow.nodes.http_request.executor -> configs
-    core.workflow.nodes.http_request.executor -> core.file.file_manager
-    core.workflow.nodes.http_request.node -> configs
     core.workflow.nodes.http_request.node -> core.tools.tool_file_manager
     core.workflow.nodes.iteration.iteration_node -> core.app.workflow.node_factory
     core.workflow.nodes.knowledge_index.knowledge_index_node -> core.rag.index_processor.index_processor_factory
     core.workflow.nodes.llm.llm_utils -> configs
     core.workflow.nodes.llm.llm_utils -> core.app.entities.app_invoke_entities
-    core.workflow.nodes.llm.llm_utils -> core.file.models
     core.workflow.nodes.llm.llm_utils -> core.model_manager
     core.workflow.nodes.llm.llm_utils -> core.model_runtime.model_providers.__base.large_language_model
     core.workflow.nodes.llm.llm_utils -> models.model
@@ -162,36 +155,10 @@ ignore_imports =
     core.workflow.nodes.llm.llm_utils -> core.entities.provider_entities
     core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.model_manager
     core.workflow.nodes.question_classifier.question_classifier_node -> core.model_manager
-    core.workflow.node_events.node -> core.file
-    core.workflow.nodes.agent.agent_node -> core.file
-    core.workflow.nodes.datasource.datasource_node -> core.file
-    core.workflow.nodes.datasource.datasource_node -> core.file.enums
-    core.workflow.nodes.document_extractor.node -> core.file
-    core.workflow.nodes.http_request.executor -> core.file.enums
-    core.workflow.nodes.http_request.node -> core.file
-    core.workflow.nodes.http_request.node -> core.file.file_manager
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.file.models
-    core.workflow.nodes.list_operator.node -> core.file
-    core.workflow.nodes.llm.file_saver -> core.file
     core.workflow.nodes.llm.llm_utils -> core.variables.segments
-    core.workflow.nodes.llm.node -> core.file
-    core.workflow.nodes.llm.node -> core.file.file_manager
-    core.workflow.nodes.llm.node -> core.file.models
     core.workflow.nodes.loop.entities -> core.variables.types
-    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.file
-    core.workflow.nodes.protocols -> core.file
-    core.workflow.nodes.question_classifier.question_classifier_node -> core.file.models
-    core.workflow.nodes.tool.tool_node -> core.file
     core.workflow.nodes.tool.tool_node -> core.tools.utils.message_transformer
     core.workflow.nodes.tool.tool_node -> models
-    core.workflow.nodes.trigger_webhook.node -> core.file
-    core.workflow.runtime.variable_pool -> core.file
-    core.workflow.runtime.variable_pool -> core.file.file_manager
-    core.workflow.system_variable -> core.file.models
-    core.workflow.utils.condition.processor -> core.file
-    core.workflow.utils.condition.processor -> core.file.file_manager
-    core.workflow.workflow_entry -> core.file.models
-    core.workflow.workflow_type_encoder -> core.file.models
     core.workflow.nodes.agent.agent_node -> models.model
     core.workflow.nodes.code.code_node -> core.helper.code_executor.code_node_provider
     core.workflow.nodes.code.code_node -> core.helper.code_executor.javascript.javascript_code_provider

api/.vscode/launch.json.example

@@ -54,7 +54,7 @@
                 "--loglevel",
                 "DEBUG",
                 "-Q",
-                "dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor"
+                "dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,workflow_based_app_execution,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor"
             ]
         }
     ]

api/commands.py

@@ -30,6 +30,7 @@ from extensions.ext_redis import redis_client
 from extensions.ext_storage import storage
 from extensions.storage.opendal_storage import OpenDALStorage
 from extensions.storage.storage_type import StorageType
+from libs.db_migration_lock import DbMigrationAutoRenewLock
 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
@@ -54,6 +55,8 @@ from tasks.remove_app_and_related_data_task import delete_draft_variables_batch
 
 logger = logging.getLogger(__name__)
 
+DB_UPGRADE_LOCK_TTL_SECONDS = 60
+
 
 @click.command("reset-password", help="Reset the account password.")
 @click.option("--email", prompt=True, help="Account email to reset password for")
@@ -727,8 +730,15 @@ def create_tenant(email: str, language: str | None = None, name: str | None = No
 @click.command("upgrade-db", help="Upgrade the database")
 def upgrade_db():
     click.echo("Preparing database migration...")
-    lock = redis_client.lock(name="db_upgrade_lock", timeout=60)
+    lock = DbMigrationAutoRenewLock(
+        redis_client=redis_client,
+        name="db_upgrade_lock",
+        ttl_seconds=DB_UPGRADE_LOCK_TTL_SECONDS,
+        logger=logger,
+        log_context="db_migration",
+    )
     if lock.acquire(blocking=False):
+        migration_succeeded = False
         try:
             click.echo(click.style("Starting database migration.", fg="green"))
 
@@ -737,6 +747,7 @@ def upgrade_db():
 
             flask_migrate.upgrade()
 
+            migration_succeeded = True
             click.echo(click.style("Database migration successful!", fg="green"))
 
         except Exception as e:
@@ -744,7 +755,8 @@ def upgrade_db():
             click.echo(click.style(f"Database migration failed: {e}", fg="red"))
             raise SystemExit(1)
         finally:
-            lock.release()
+            status = "successful" if migration_succeeded else "failed"
+            lock.release_safely(status=status)
     else:
         click.echo("Database migration skipped")
 

api/configs/feature/__init__.py

@@ -265,6 +265,11 @@ class PluginConfig(BaseSettings):
         default=60 * 60,
     )
 
+    PLUGIN_MAX_FILE_SIZE: PositiveInt = Field(
+        description="Maximum allowed size (bytes) for plugin-generated files",
+        default=50 * 1024 * 1024,
+    )
+
 
 class MarketplaceConfig(BaseSettings):
     """
@@ -1314,6 +1319,9 @@ class WorkflowLogConfig(BaseSettings):
     WORKFLOW_LOG_CLEANUP_BATCH_SIZE: int = Field(
         default=100, description="Batch size for workflow run log cleanup operations"
     )
+    WORKFLOW_LOG_CLEANUP_SPECIFIC_WORKFLOW_IDS: str = Field(
+        default="", description="Comma-separated list of workflow IDs to clean logs for"
+    )
 
 
 class SwaggerUIConfig(BaseSettings):
@@ -1344,6 +1352,10 @@ class SandboxExpiredRecordsCleanConfig(BaseSettings):
         description="Maximum number of records to process in each batch",
         default=1000,
     )
+    SANDBOX_EXPIRED_RECORDS_CLEAN_BATCH_MAX_INTERVAL: PositiveInt = Field(
+        description="Maximum interval in milliseconds between batches",
+        default=200,
+    )
     SANDBOX_EXPIRED_RECORDS_RETENTION_DAYS: PositiveInt = Field(
         description="Retention days for sandbox expired workflow_run records and message records",
         default=30,

api/configs/middleware/__init__.py

@@ -259,11 +259,20 @@ class CeleryConfig(DatabaseConfig):
         description="Password of the Redis Sentinel master.",
         default=None,
     )
+
     CELERY_SENTINEL_SOCKET_TIMEOUT: PositiveFloat | None = Field(
         description="Timeout for Redis Sentinel socket operations in seconds.",
         default=0.1,
     )
 
+    CELERY_TASK_ANNOTATIONS: dict[str, Any] | None = Field(
+        description=(
+            "Annotations for Celery tasks as a JSON mapping of task name -> options "
+            "(for example, rate limits or other task-specific settings)."
+        ),
+        default=None,
+    )
+
     @computed_field
     def CELERY_RESULT_BACKEND(self) -> str | None:
         if self.CELERY_BACKEND in ("database", "rabbitmq"):

api/configs/middleware/vdb/oceanbase_config.py

@@ -1,3 +1,5 @@
+from typing import Literal
+
 from pydantic import Field, PositiveInt
 from pydantic_settings import BaseSettings
 
@@ -49,3 +51,43 @@ class OceanBaseVectorConfig(BaseSettings):
         ),
         default="ik",
     )
+
+    OCEANBASE_VECTOR_BATCH_SIZE: PositiveInt = Field(
+        description="Number of documents to insert per batch",
+        default=100,
+    )
+
+    OCEANBASE_VECTOR_METRIC_TYPE: Literal["l2", "cosine", "inner_product"] = Field(
+        description="Distance metric type for vector index: l2, cosine, or inner_product",
+        default="l2",
+    )
+
+    OCEANBASE_HNSW_M: PositiveInt = Field(
+        description="HNSW M parameter (max number of connections per node)",
+        default=16,
+    )
+
+    OCEANBASE_HNSW_EF_CONSTRUCTION: PositiveInt = Field(
+        description="HNSW efConstruction parameter (index build-time search width)",
+        default=256,
+    )
+
+    OCEANBASE_HNSW_EF_SEARCH: int = Field(
+        description="HNSW efSearch parameter (query-time search width, -1 uses server default)",
+        default=-1,
+    )
+
+    OCEANBASE_VECTOR_POOL_SIZE: PositiveInt = Field(
+        description="SQLAlchemy connection pool size",
+        default=5,
+    )
+
+    OCEANBASE_VECTOR_MAX_OVERFLOW: int = Field(
+        description="SQLAlchemy connection pool max overflow connections",
+        default=10,
+    )
+
+    OCEANBASE_HNSW_REFRESH_THRESHOLD: int = Field(
+        description="Minimum number of inserted documents to trigger an automatic HNSW index refresh (0 to disable)",
+        default=1000,
+    )

api/constants/languages.py

@@ -21,6 +21,7 @@ language_timezone_mapping = {
     "th-TH": "Asia/Bangkok",
     "id-ID": "Asia/Jakarta",
     "ar-TN": "Africa/Tunis",
+    "nl-NL": "Europe/Amsterdam",
 }
 
 languages = list(language_timezone_mapping.keys())

api/controllers/common/fields.py

@@ -4,7 +4,7 @@ from typing import Any, TypeAlias
 
 from pydantic import BaseModel, ConfigDict, computed_field
 
-from core.file import helpers as file_helpers
+from core.workflow.file import helpers as file_helpers
 from models.model import IconType
 
 JSONValue: TypeAlias = str | int | float | bool | None | dict[str, Any] | list[Any]

api/controllers/console/app/app.py

@@ -23,10 +23,10 @@ from controllers.console.wraps import (
     is_admin_or_owner_required,
     setup_required,
 )
-from core.file import helpers as file_helpers
 from core.ops.ops_trace_manager import OpsTraceManager
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
 from core.workflow.enums import NodeType, WorkflowExecutionStatus
+from core.workflow.file import helpers as file_helpers
 from extensions.ext_database import db
 from libs.login import current_account_with_tenant, login_required
 from models import App, DatasetPermissionEnum, Workflow
@@ -660,6 +660,19 @@ class AppCopyApi(Resource):
             )
             session.commit()
 
+            # Inherit web app permission from original app
+            if result.app_id and FeatureService.get_system_features().webapp_auth.enabled:
+                try:
+                    # Get the original app's access mode
+                    original_settings = EnterpriseService.WebAppAuth.get_app_access_mode_by_id(app_model.id)
+                    access_mode = original_settings.access_mode
+                except Exception:
+                    # If original app has no settings (old app), default to public to match fallback behavior
+                    access_mode = "public"
+
+                # Apply the same access mode to the copied app
+                EnterpriseService.WebAppAuth.update_app_access_mode(result.app_id, access_mode)
+
             stmt = select(App).where(App.id == result.app_id)
             app = session.scalar(stmt)
 

api/controllers/console/app/conversation.py

@@ -599,7 +599,12 @@ def _get_conversation(app_model, conversation_id):
     db.session.execute(
         sa.update(Conversation)
         .where(Conversation.id == conversation_id, Conversation.read_at.is_(None))
-        .values(read_at=naive_utc_now(), read_account_id=current_user.id)
+        # Keep updated_at unchanged when only marking a conversation as read.
+        .values(
+            read_at=naive_utc_now(),
+            read_account_id=current_user.id,
+            updated_at=Conversation.updated_at,
+        )
     )
     db.session.commit()
     db.session.refresh(conversation)

api/controllers/console/app/workflow.py

@@ -20,7 +20,6 @@ from core.app.app_config.features.file_upload.manager import FileUploadConfigMan
 from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.apps.workflow.app_generator import SKIP_PREPARE_USER_INPUTS_KEY
 from core.app.entities.app_invoke_entities import InvokeFrom
-from core.file.models import File
 from core.helper.trace_id_helper import get_external_trace_id
 from core.model_runtime.utils.encoders import jsonable_encoder
 from core.plugin.impl.exc import PluginInvokeError
@@ -31,6 +30,7 @@ from core.trigger.debug.event_selectors import (
     select_trigger_debug_events,
 )
 from core.workflow.enums import NodeType
+from core.workflow.file.models import File
 from core.workflow.graph_engine.manager import GraphEngineManager
 from extensions.ext_database import db
 from factories import file_factory, variable_factory

api/controllers/console/app/workflow_draft_variable.py

@@ -15,11 +15,11 @@ from controllers.console.app.error import (
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
 from controllers.web.error import InvalidArgumentError, NotFoundError
-from core.file import helpers as file_helpers
 from core.variables.segment_group import SegmentGroup
 from core.variables.segments import ArrayFileSegment, FileSegment, Segment
 from core.variables.types import SegmentType
 from core.workflow.constants import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
+from core.workflow.file import helpers as file_helpers
 from extensions.ext_database import db
 from factories.file_factory import build_from_mapping, build_from_mappings
 from factories.variable_factory import build_segment_with_type

api/controllers/console/explore/trial.py

@@ -10,7 +10,7 @@ import services
 from controllers.common.fields import Parameters as ParametersResponse
 from controllers.common.fields import Site as SiteResponse
 from controllers.common.schema import get_or_create_model
-from controllers.console import api, console_ns
+from controllers.console import console_ns
 from controllers.console.app.error import (
     AppUnavailableError,
     AudioTooLargeError,
@@ -469,7 +469,7 @@ class TrialSitApi(Resource):
     """Resource for trial app sites."""
 
     @trial_feature_enable
-    @get_app_model_with_trial
+    @get_app_model_with_trial(None)
     def get(self, app_model):
         """Retrieve app site info.
 
@@ -491,7 +491,7 @@ class TrialAppParameterApi(Resource):
     """Resource for app variables."""
 
     @trial_feature_enable
-    @get_app_model_with_trial
+    @get_app_model_with_trial(None)
     def get(self, app_model):
         """Retrieve app parameters."""
 
@@ -520,7 +520,7 @@ class TrialAppParameterApi(Resource):
 
 class AppApi(Resource):
     @trial_feature_enable
-    @get_app_model_with_trial
+    @get_app_model_with_trial(None)
     @marshal_with(app_detail_with_site_model)
     def get(self, app_model):
         """Get app detail"""
@@ -533,7 +533,7 @@ class AppApi(Resource):
 
 class AppWorkflowApi(Resource):
     @trial_feature_enable
-    @get_app_model_with_trial
+    @get_app_model_with_trial(None)
     @marshal_with(workflow_model)
     def get(self, app_model):
         """Get workflow detail"""
@@ -552,7 +552,7 @@ class AppWorkflowApi(Resource):
 
 class DatasetListApi(Resource):
     @trial_feature_enable
-    @get_app_model_with_trial
+    @get_app_model_with_trial(None)
     def get(self, app_model):
         page = request.args.get("page", default=1, type=int)
         limit = request.args.get("limit", default=20, type=int)
@@ -570,27 +570,31 @@ class DatasetListApi(Resource):
         return response
 
 
-api.add_resource(TrialChatApi, "/trial-apps/<uuid:app_id>/chat-messages", endpoint="trial_app_chat_completion")
+console_ns.add_resource(TrialChatApi, "/trial-apps/<uuid:app_id>/chat-messages", endpoint="trial_app_chat_completion")
 
-api.add_resource(
+console_ns.add_resource(
     TrialMessageSuggestedQuestionApi,
     "/trial-apps/<uuid:app_id>/messages/<uuid:message_id>/suggested-questions",
     endpoint="trial_app_suggested_question",
 )
 
-api.add_resource(TrialChatAudioApi, "/trial-apps/<uuid:app_id>/audio-to-text", endpoint="trial_app_audio")
-api.add_resource(TrialChatTextApi, "/trial-apps/<uuid:app_id>/text-to-audio", endpoint="trial_app_text")
+console_ns.add_resource(TrialChatAudioApi, "/trial-apps/<uuid:app_id>/audio-to-text", endpoint="trial_app_audio")
+console_ns.add_resource(TrialChatTextApi, "/trial-apps/<uuid:app_id>/text-to-audio", endpoint="trial_app_text")
 
-api.add_resource(TrialCompletionApi, "/trial-apps/<uuid:app_id>/completion-messages", endpoint="trial_app_completion")
+console_ns.add_resource(
+    TrialCompletionApi, "/trial-apps/<uuid:app_id>/completion-messages", endpoint="trial_app_completion"
+)
 
-api.add_resource(TrialSitApi, "/trial-apps/<uuid:app_id>/site")
+console_ns.add_resource(TrialSitApi, "/trial-apps/<uuid:app_id>/site")
 
-api.add_resource(TrialAppParameterApi, "/trial-apps/<uuid:app_id>/parameters", endpoint="trial_app_parameters")
+console_ns.add_resource(TrialAppParameterApi, "/trial-apps/<uuid:app_id>/parameters", endpoint="trial_app_parameters")
 
-api.add_resource(AppApi, "/trial-apps/<uuid:app_id>", endpoint="trial_app")
+console_ns.add_resource(AppApi, "/trial-apps/<uuid:app_id>", endpoint="trial_app")
 
-api.add_resource(TrialAppWorkflowRunApi, "/trial-apps/<uuid:app_id>/workflows/run", endpoint="trial_app_workflow_run")
-api.add_resource(TrialAppWorkflowTaskStopApi, "/trial-apps/<uuid:app_id>/workflows/tasks/<string:task_id>/stop")
+console_ns.add_resource(
+    TrialAppWorkflowRunApi, "/trial-apps/<uuid:app_id>/workflows/run", endpoint="trial_app_workflow_run"
+)
+console_ns.add_resource(TrialAppWorkflowTaskStopApi, "/trial-apps/<uuid:app_id>/workflows/tasks/<string:task_id>/stop")
 
-api.add_resource(AppWorkflowApi, "/trial-apps/<uuid:app_id>/workflows", endpoint="trial_app_workflow")
-api.add_resource(DatasetListApi, "/trial-apps/<uuid:app_id>/datasets", endpoint="trial_app_datasets")
+console_ns.add_resource(AppWorkflowApi, "/trial-apps/<uuid:app_id>/workflows", endpoint="trial_app_workflow")
+console_ns.add_resource(DatasetListApi, "/trial-apps/<uuid:app_id>/datasets", endpoint="trial_app_datasets")

api/controllers/console/explore/wraps.py

@@ -105,9 +105,9 @@ def trial_app_required(view: Callable[Concatenate[App, P], R] | None = None):
     return decorator
 
 
-def trial_feature_enable(view: Callable[..., R]) -> Callable[..., R]:
+def trial_feature_enable(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         features = FeatureService.get_system_features()
         if not features.enable_trial_app:
             abort(403, "Trial app feature is not enabled.")
@@ -116,9 +116,9 @@ def trial_feature_enable(view: Callable[..., R]) -> Callable[..., R]:
     return decorated
 
 
-def explore_banner_enabled(view: Callable[..., R]) -> Callable[..., R]:
+def explore_banner_enabled(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         features = FeatureService.get_system_features()
         if not features.enable_explore_banner:
             abort(403, "Explore banner feature is not enabled.")

api/controllers/console/remote_files.py

@@ -1,6 +1,7 @@
 import urllib.parse
 
 import httpx
+from flask_restx import Resource
 from pydantic import BaseModel, Field
 
 import services
@@ -10,12 +11,12 @@ from controllers.common.errors import (
     RemoteFileUploadError,
     UnsupportedFileTypeError,
 )
-from controllers.fastopenapi import console_router
-from core.file import helpers as file_helpers
+from controllers.console import console_ns
 from core.helper import ssrf_proxy
+from core.workflow.file import helpers as file_helpers
 from extensions.ext_database import db
 from fields.file_fields import FileWithSignedUrl, RemoteFileInfo
-from libs.login import current_account_with_tenant
+from libs.login import current_account_with_tenant, login_required
 from services.file_service import FileService
 
 
@@ -23,69 +24,73 @@ class RemoteFileUploadPayload(BaseModel):
     url: str = Field(..., description="URL to fetch")
 
 
-@console_router.get(
-    "/remote-files/<path:url>",
-    response_model=RemoteFileInfo,
-    tags=["console"],
-)
-def get_remote_file_info(url: str) -> RemoteFileInfo:
-    decoded_url = urllib.parse.unquote(url)
-    resp = ssrf_proxy.head(decoded_url)
-    if resp.status_code != httpx.codes.OK:
-        resp = ssrf_proxy.get(decoded_url, timeout=3)
-    resp.raise_for_status()
-    return RemoteFileInfo(
-        file_type=resp.headers.get("Content-Type", "application/octet-stream"),
-        file_length=int(resp.headers.get("Content-Length", 0)),
-    )
-
-
-@console_router.post(
-    "/remote-files/upload",
-    response_model=FileWithSignedUrl,
-    tags=["console"],
-    status_code=201,
-)
-def upload_remote_file(payload: RemoteFileUploadPayload) -> FileWithSignedUrl:
-    url = payload.url
-
-    try:
-        resp = ssrf_proxy.head(url=url)
+@console_ns.route("/remote-files/<path:url>")
+class GetRemoteFileInfo(Resource):
+    @login_required
+    def get(self, url: str):
+        decoded_url = urllib.parse.unquote(url)
+        resp = ssrf_proxy.head(decoded_url)
         if resp.status_code != httpx.codes.OK:
-            resp = ssrf_proxy.get(url=url, timeout=3, follow_redirects=True)
-        if resp.status_code != httpx.codes.OK:
-            raise RemoteFileUploadError(f"Failed to fetch file from {url}: {resp.text}")
-    except httpx.RequestError as e:
-        raise RemoteFileUploadError(f"Failed to fetch file from {url}: {str(e)}")
+            resp = ssrf_proxy.get(decoded_url, timeout=3)
+        resp.raise_for_status()
+        return RemoteFileInfo(
+            file_type=resp.headers.get("Content-Type", "application/octet-stream"),
+            file_length=int(resp.headers.get("Content-Length", 0)),
+        ).model_dump(mode="json")
 
-    file_info = helpers.guess_file_info_from_response(resp)
 
-    if not FileService.is_file_size_within_limit(extension=file_info.extension, file_size=file_info.size):
-        raise FileTooLargeError
+@console_ns.route("/remote-files/upload")
+class RemoteFileUpload(Resource):
+    @login_required
+    def post(self):
+        payload = RemoteFileUploadPayload.model_validate(console_ns.payload)
+        url = payload.url
 
-    content = resp.content if resp.request.method == "GET" else ssrf_proxy.get(url).content
+        # Try to fetch remote file metadata/content first
+        try:
+            resp = ssrf_proxy.head(url=url)
+            if resp.status_code != httpx.codes.OK:
+                resp = ssrf_proxy.get(url=url, timeout=3, follow_redirects=True)
+            if resp.status_code != httpx.codes.OK:
+                # Normalize into a user-friendly error message expected by tests
+                raise RemoteFileUploadError(f"Failed to fetch file from {url}: {resp.text}")
+        except httpx.RequestError as e:
+            raise RemoteFileUploadError(f"Failed to fetch file from {url}: {str(e)}")
 
-    try:
-        user, _ = current_account_with_tenant()
-        upload_file = FileService(db.engine).upload_file(
-            filename=file_info.filename,
-            content=content,
-            mimetype=file_info.mimetype,
-            user=user,
-            source_url=url,
+        file_info = helpers.guess_file_info_from_response(resp)
+
+        # Enforce file size limit with 400 (Bad Request) per tests' expectation
+        if not FileService.is_file_size_within_limit(extension=file_info.extension, file_size=file_info.size):
+            raise FileTooLargeError()
+
+        # Load content if needed
+        content = resp.content if resp.request.method == "GET" else ssrf_proxy.get(url).content
+
+        try:
+            user, _ = current_account_with_tenant()
+            upload_file = FileService(db.engine).upload_file(
+                filename=file_info.filename,
+                content=content,
+                mimetype=file_info.mimetype,
+                user=user,
+                source_url=url,
+            )
+        except services.errors.file.FileTooLargeError as file_too_large_error:
+            raise FileTooLargeError(file_too_large_error.description)
+        except services.errors.file.UnsupportedFileTypeError:
+            raise UnsupportedFileTypeError()
+
+        # Success: return created resource with 201 status
+        return (
+            FileWithSignedUrl(
+                id=upload_file.id,
+                name=upload_file.name,
+                size=upload_file.size,
+                extension=upload_file.extension,
+                url=file_helpers.get_signed_file_url(upload_file_id=upload_file.id),
+                mime_type=upload_file.mime_type,
+                created_by=upload_file.created_by,
+                created_at=int(upload_file.created_at.timestamp()),
+            ).model_dump(mode="json"),
+            201,
         )
-    except services.errors.file.FileTooLargeError as file_too_large_error:
-        raise FileTooLargeError(file_too_large_error.description)
-    except services.errors.file.UnsupportedFileTypeError:
-        raise UnsupportedFileTypeError()
-
-    return FileWithSignedUrl(
-        id=upload_file.id,
-        name=upload_file.name,
-        size=upload_file.size,
-        extension=upload_file.extension,
-        url=file_helpers.get_signed_file_url(upload_file_id=upload_file.id),
-        mime_type=upload_file.mime_type,
-        created_by=upload_file.created_by,
-        created_at=int(upload_file.created_at.timestamp()),
-    )

api/controllers/console/setup.py

@@ -42,7 +42,15 @@ class SetupResponse(BaseModel):
     tags=["console"],
 )
 def get_setup_status_api() -> SetupStatusResponse:
-    """Get system setup status."""
+    """Get system setup status.
+
+    NOTE: This endpoint is unauthenticated by design.
+
+    During first-time bootstrap there is no admin account yet, so frontend initialization must be
+    able to query setup progress before any login flow exists.
+
+    Only bootstrap-safe status information should be returned by this endpoint.
+    """
     if dify_config.EDITION == "SELF_HOSTED":
         setup_status = get_setup_status()
         if setup_status and not isinstance(setup_status, bool):
@@ -61,7 +69,12 @@ def get_setup_status_api() -> SetupStatusResponse:
 )
 @only_edition_self_hosted
 def setup_system(payload: SetupRequestPayload) -> SetupResponse:
-    """Initialize system setup with admin account."""
+    """Initialize system setup with admin account.
+
+    NOTE: This endpoint is unauthenticated by design for first-time bootstrap.
+    Access is restricted by deployment mode (`SELF_HOSTED`), one-time setup guards,
+    and init-password validation rather than user session authentication.
+    """
     if get_setup_status():
         raise AlreadySetupError()
 

api/controllers/files/tool_files.py

@@ -64,6 +64,10 @@ class ToolFileApi(Resource):
 
             if not stream or not tool_file:
                 raise NotFound("file is not found")
+
+        except NotFound:
+            raise
+
         except Exception:
             raise UnsupportedFileTypeError()
 

api/controllers/files/upload.py

@@ -7,8 +7,8 @@ from pydantic import BaseModel, Field
 from werkzeug.exceptions import Forbidden
 
 import services
-from core.file.helpers import verify_plugin_file_signature
 from core.tools.tool_file_manager import ToolFileManager
+from core.workflow.file.helpers import verify_plugin_file_signature
 from fields.file_fields import FileResponse
 
 from ..common.errors import (

api/controllers/inner_api/plugin/plugin.py

@@ -4,7 +4,6 @@ from controllers.console.wraps import setup_required
 from controllers.inner_api import inner_api_ns
 from controllers.inner_api.plugin.wraps import get_user_tenant, plugin_data
 from controllers.inner_api.wraps import plugin_inner_api_only
-from core.file.helpers import get_signed_file_url_for_plugin
 from core.model_runtime.utils.encoders import jsonable_encoder
 from core.plugin.backwards_invocation.app import PluginAppBackwardsInvocation
 from core.plugin.backwards_invocation.base import BaseBackwardsInvocationResponse
@@ -30,6 +29,7 @@ from core.plugin.entities.request import (
     RequestRequestUploadFile,
 )
 from core.tools.entities.tool_entities import ToolProviderType
+from core.workflow.file.helpers import get_signed_file_url_for_plugin
 from libs.helper import length_prefixed_response
 from models import Account, Tenant
 from models.model import EndUser

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

@@ -3,7 +3,8 @@ from typing import Any
 
 from flask import request
 from pydantic import BaseModel
-from werkzeug.exceptions import Forbidden
+from sqlalchemy import select
+from werkzeug.exceptions import Forbidden, NotFound
 
 import services
 from controllers.common.errors import FilenameNotExistsError, NoFileUploadedError, TooManyFilesError
@@ -17,7 +18,7 @@ from core.app.entities.app_invoke_entities import InvokeFrom
 from libs import helper
 from libs.login import current_user
 from models import Account
-from models.dataset import Pipeline
+from models.dataset import Dataset, Pipeline
 from models.engine import db
 from services.errors.file import FileTooLargeError, UnsupportedFileTypeError
 from services.file_service import FileService
@@ -65,6 +66,12 @@ class DatasourcePluginsApi(DatasetApiResource):
     )
     def get(self, tenant_id: str, dataset_id: str):
         """Resource for getting datasource plugins."""
+        # Verify dataset ownership
+        stmt = select(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id)
+        dataset = db.session.scalar(stmt)
+        if not dataset:
+            raise NotFound("Dataset not found.")
+
         # Get query parameter to determine published or draft
         is_published: bool = request.args.get("is_published", default=True, type=bool)
 
@@ -104,6 +111,12 @@ class DatasourceNodeRunApi(DatasetApiResource):
     @service_api_ns.expect(service_api_ns.models[DatasourceNodeRunPayload.__name__])
     def post(self, tenant_id: str, dataset_id: str, node_id: str):
         """Resource for getting datasource plugins."""
+        # Verify dataset ownership
+        stmt = select(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id)
+        dataset = db.session.scalar(stmt)
+        if not dataset:
+            raise NotFound("Dataset not found.")
+
         payload = DatasourceNodeRunPayload.model_validate(service_api_ns.payload or {})
         assert isinstance(current_user, Account)
         rag_pipeline_service: RagPipelineService = RagPipelineService()
@@ -161,6 +174,12 @@ class PipelineRunApi(DatasetApiResource):
     @service_api_ns.expect(service_api_ns.models[PipelineRunApiEntity.__name__])
     def post(self, tenant_id: str, dataset_id: str):
         """Resource for running a rag pipeline."""
+        # Verify dataset ownership
+        stmt = select(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id)
+        dataset = db.session.scalar(stmt)
+        if not dataset:
+            raise NotFound("Dataset not found.")
+
         payload = PipelineRunApiEntity.model_validate(service_api_ns.payload or {})
 
         if not isinstance(current_user, Account):

api/controllers/web/remote_files.py

@@ -10,8 +10,8 @@ from controllers.common.errors import (
     RemoteFileUploadError,
     UnsupportedFileTypeError,
 )
-from core.file import helpers as file_helpers
 from core.helper import ssrf_proxy
+from core.workflow.file import helpers as file_helpers
 from extensions.ext_database import db
 from fields.file_fields import FileWithSignedUrl, RemoteFileInfo
 from services.file_service import FileService

api/core/agent/base_agent_runner.py

@@ -17,7 +17,6 @@ from core.app.entities.app_invoke_entities import (
 )
 from core.callback_handler.agent_tool_callback_handler import DifyAgentCallbackHandler
 from core.callback_handler.index_tool_callback_handler import DatasetIndexToolCallbackHandler
-from core.file import file_manager
 from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance
 from core.model_runtime.entities import (
@@ -40,6 +39,7 @@ from core.tools.entities.tool_entities import (
 )
 from core.tools.tool_manager import ToolManager
 from core.tools.utils.dataset_retriever_tool import DatasetRetrieverTool
+from core.workflow.file import file_manager
 from extensions.ext_database import db
 from factories import file_factory
 from models.enums import CreatorUserRole

api/core/agent/cot_chat_agent_runner.py

@@ -1,7 +1,6 @@
 import json
 
 from core.agent.cot_agent_runner import CotAgentRunner
-from core.file import file_manager
 from core.model_runtime.entities import (
     AssistantPromptMessage,
     PromptMessage,
@@ -11,6 +10,7 @@ from core.model_runtime.entities import (
 )
 from core.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
 from core.model_runtime.utils.encoders import jsonable_encoder
+from core.workflow.file import file_manager
 
 
 class CotChatAgentRunner(CotAgentRunner):

api/core/agent/fc_agent_runner.py

@@ -7,7 +7,6 @@ from typing import Any, Union
 from core.agent.base_agent_runner import BaseAgentRunner
 from core.app.apps.base_app_queue_manager import PublishFrom
 from core.app.entities.queue_entities import QueueAgentThoughtEvent, QueueMessageEndEvent, QueueMessageFileEvent
-from core.file import file_manager
 from core.model_runtime.entities import (
     AssistantPromptMessage,
     LLMResult,
@@ -25,6 +24,7 @@ from core.model_runtime.entities.message_entities import ImagePromptMessageConte
 from core.prompt.agent_history_prompt_transform import AgentHistoryPromptTransform
 from core.tools.entities.tool_entities import ToolInvokeMeta
 from core.tools.tool_engine import ToolEngine
+from core.workflow.file import file_manager
 from core.workflow.nodes.agent.exc import AgentMaxIterationError
 from models.model import Message
 

api/core/app/app_config/entities.py

@@ -5,9 +5,9 @@ from typing import Any, Literal
 from jsonschema import Draft7Validator, SchemaError
 from pydantic import BaseModel, Field, field_validator
 
-from core.file import FileTransferMethod, FileType, FileUploadConfig
 from core.model_runtime.entities.llm_entities import LLMMode
 from core.model_runtime.entities.message_entities import PromptMessageRole
+from core.workflow.file import FileTransferMethod, FileType, FileUploadConfig
 from models.model import AppMode
 
 

api/core/app/app_config/features/file_upload/manager.py

@@ -2,7 +2,7 @@ from collections.abc import Mapping
 from typing import Any
 
 from constants import DEFAULT_FILE_NUMBER_LIMITS
-from core.file import FileUploadConfig
+from core.workflow.file import FileUploadConfig
 
 
 class FileUploadConfigManager:

api/core/app/apps/base_app_generator.py

@@ -5,8 +5,8 @@ from sqlalchemy.orm import Session
 
 from core.app.app_config.entities import VariableEntityType
 from core.app.entities.app_invoke_entities import InvokeFrom
-from core.file import File, FileUploadConfig
 from core.workflow.enums import NodeType
+from core.workflow.file import File, FileUploadConfig
 from core.workflow.repositories.draft_variable_repository import (
     DraftVariableSaver,
     DraftVariableSaverFactory,

api/core/app/apps/base_app_queue_manager.py

@@ -2,7 +2,7 @@ import logging
 import queue
 import threading
 import time
-from abc import abstractmethod
+from abc import ABC, abstractmethod
 from enum import IntEnum, auto
 from typing import Any
 
@@ -31,7 +31,7 @@ class PublishFrom(IntEnum):
     TASK_PIPELINE = auto()
 
 
-class AppQueueManager:
+class AppQueueManager(ABC):
     def __init__(self, task_id: str, user_id: str, invoke_from: InvokeFrom):
         if not user_id:
             raise ValueError("user is required")
@@ -133,7 +133,7 @@ class AppQueueManager:
         self._publish(event, pub_from)
 
     @abstractmethod
-    def _publish(self, event: AppQueueEvent, pub_from: PublishFrom):
+    def _publish(self, event: AppQueueEvent, pub_from: PublishFrom) -> None:
         """
         Publish event to queue
         :param event:

api/core/app/apps/base_app_runner.py

@@ -22,7 +22,6 @@ from core.app.entities.queue_entities import (
 from core.app.features.annotation_reply.annotation_reply import AnnotationReplyFeature
 from core.app.features.hosting_moderation.hosting_moderation import HostingModerationFeature
 from core.external_data_tool.external_data_fetch import ExternalDataFetch
-from core.file.enums import FileTransferMethod, FileType
 from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance
 from core.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk, LLMResultChunkDelta, LLMUsage
@@ -39,12 +38,13 @@ from core.prompt.advanced_prompt_transform import AdvancedPromptTransform
 from core.prompt.entities.advanced_prompt_entities import ChatModelMessage, CompletionModelPromptTemplate, MemoryConfig
 from core.prompt.simple_prompt_transform import ModelMode, SimplePromptTransform
 from core.tools.tool_file_manager import ToolFileManager
+from core.workflow.file.enums import FileTransferMethod, FileType
 from extensions.ext_database import db
 from models.enums import CreatorUserRole
 from models.model import App, AppMode, Message, MessageAnnotation, MessageFile
 
 if TYPE_CHECKING:
-    from core.file.models import File
+    from core.workflow.file.models import File
 
 _logger = logging.getLogger(__name__)
 

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

@@ -11,12 +11,12 @@ from core.app.entities.app_invoke_entities import (
 )
 from core.app.entities.queue_entities import QueueAnnotationReplyEvent
 from core.callback_handler.index_tool_callback_handler import DatasetIndexToolCallbackHandler
-from core.file import File
 from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance
 from core.model_runtime.entities.message_entities import ImagePromptMessageContent
 from core.moderation.base import ModerationError
 from core.rag.retrieval.dataset_retrieval import DatasetRetrieval
+from core.workflow.file import File
 from extensions.ext_database import db
 from models.model import App, Conversation, Message
 

api/core/app/apps/common/workflow_response_converter.py

@@ -45,7 +45,6 @@ from core.app.entities.task_entities import (
     WorkflowPauseStreamResponse,
     WorkflowStartStreamResponse,
 )
-from core.file import FILE_MODEL_IDENTITY, File
 from core.plugin.impl.datasource import PluginDatasourceManager
 from core.tools.entities.tool_entities import ToolProviderType
 from core.tools.tool_manager import ToolManager
@@ -60,6 +59,7 @@ from core.workflow.enums import (
     WorkflowNodeExecutionMetadataKey,
     WorkflowNodeExecutionStatus,
 )
+from core.workflow.file import FILE_MODEL_IDENTITY, File
 from core.workflow.runtime import GraphRuntimeState
 from core.workflow.system_variable import SystemVariable
 from core.workflow.workflow_entry import WorkflowEntry

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

@@ -10,11 +10,11 @@ from core.app.entities.app_invoke_entities import (
     CompletionAppGenerateEntity,
 )
 from core.callback_handler.index_tool_callback_handler import DatasetIndexToolCallbackHandler
-from core.file import File
 from core.model_manager import ModelInstance
 from core.model_runtime.entities.message_entities import ImagePromptMessageContent
 from core.moderation.base import ModerationError
 from core.rag.retrieval.dataset_retrieval import DatasetRetrieval
+from core.workflow.file import File
 from extensions.ext_database import db
 from models.model import App, Message
 

api/core/app/apps/streaming_utils.py

@@ -34,7 +34,7 @@ def stream_topic_events(
             on_subscribe()
         while True:
             try:
-                msg = sub.receive(timeout=0.1)
+                msg = sub.receive(timeout=1)
             except SubscriptionClosedError:
                 return
             if msg is None:

api/core/app/entities/app_invoke_entities.py

@@ -7,8 +7,8 @@ from pydantic import BaseModel, ConfigDict, Field, ValidationInfo, field_validat
 from constants import UUID_NIL
 from core.app.app_config.entities import EasyUIBasedAppConfig, WorkflowUIBasedAppConfig
 from core.entities.provider_configuration import ProviderModelBundle
-from core.file import File, FileUploadConfig
 from core.model_runtime.entities.model_entities import AIModelEntity
+from core.workflow.file import File, FileUploadConfig
 
 if TYPE_CHECKING:
     from core.ops.ops_trace_manager import TraceQueueManager

api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py

@@ -56,10 +56,13 @@ from core.ops.entities.trace_entity import TraceTaskName
 from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
 from core.prompt.utils.prompt_message_util import PromptMessageUtil
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
+from core.tools.signature import sign_tool_file
+from core.workflow.file import helpers as file_helpers
+from core.workflow.file.enums import FileTransferMethod
 from events.message_event import message_was_created
 from extensions.ext_database import db
 from libs.datetime_utils import naive_utc_now
-from models.model import AppMode, Conversation, Message, MessageAgentThought
+from models.model import AppMode, Conversation, Message, MessageAgentThought, MessageFile, UploadFile
 
 logger = logging.getLogger(__name__)
 
@@ -463,6 +466,85 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
             metadata=metadata_dict,
         )
 
+    def _record_files(self):
+        with Session(db.engine, expire_on_commit=False) as session:
+            message_files = session.scalars(select(MessageFile).where(MessageFile.message_id == self._message_id)).all()
+            if not message_files:
+                return None
+
+            files_list = []
+            upload_file_ids = [
+                mf.upload_file_id
+                for mf in message_files
+                if mf.transfer_method == FileTransferMethod.LOCAL_FILE and mf.upload_file_id
+            ]
+            upload_files_map = {}
+            if upload_file_ids:
+                upload_files = session.scalars(select(UploadFile).where(UploadFile.id.in_(upload_file_ids))).all()
+                upload_files_map = {uf.id: uf for uf in upload_files}
+
+            for message_file in message_files:
+                upload_file = None
+                if message_file.transfer_method == FileTransferMethod.LOCAL_FILE and message_file.upload_file_id:
+                    upload_file = upload_files_map.get(message_file.upload_file_id)
+
+                url = None
+                filename = "file"
+                mime_type = "application/octet-stream"
+                size = 0
+                extension = ""
+
+                if message_file.transfer_method == FileTransferMethod.REMOTE_URL:
+                    url = message_file.url
+                    if message_file.url:
+                        filename = message_file.url.split("/")[-1].split("?")[0]  # Remove query params
+                elif message_file.transfer_method == FileTransferMethod.LOCAL_FILE:
+                    if upload_file:
+                        url = file_helpers.get_signed_file_url(upload_file_id=str(upload_file.id))
+                        filename = upload_file.name
+                        mime_type = upload_file.mime_type or "application/octet-stream"
+                        size = upload_file.size or 0
+                        extension = f".{upload_file.extension}" if upload_file.extension else ""
+                    elif message_file.upload_file_id:
+                        # Fallback: generate URL even if upload_file not found
+                        url = file_helpers.get_signed_file_url(upload_file_id=str(message_file.upload_file_id))
+                elif message_file.transfer_method == FileTransferMethod.TOOL_FILE and message_file.url:
+                    # For tool files, use URL directly if it's HTTP, otherwise sign it
+                    if message_file.url.startswith("http"):
+                        url = message_file.url
+                        filename = message_file.url.split("/")[-1].split("?")[0]
+                    else:
+                        # Extract tool file id and extension from URL
+                        url_parts = message_file.url.split("/")
+                        if url_parts:
+                            file_part = url_parts[-1].split("?")[0]  # Remove query params first
+                            # Use rsplit to correctly handle filenames with multiple dots
+                            if "." in file_part:
+                                tool_file_id, ext = file_part.rsplit(".", 1)
+                                extension = f".{ext}"
+                            else:
+                                tool_file_id = file_part
+                                extension = ".bin"
+                            url = sign_tool_file(tool_file_id=tool_file_id, extension=extension)
+                            filename = file_part
+
+                transfer_method_value = message_file.transfer_method
+                remote_url = message_file.url if message_file.transfer_method == FileTransferMethod.REMOTE_URL else ""
+                file_dict = {
+                    "related_id": message_file.id,
+                    "extension": extension,
+                    "filename": filename,
+                    "size": size,
+                    "mime_type": mime_type,
+                    "transfer_method": transfer_method_value,
+                    "type": message_file.type,
+                    "url": url or "",
+                    "upload_file_id": message_file.upload_file_id or message_file.id,
+                    "remote_url": remote_url,
+                }
+                files_list.append(file_dict)
+            return files_list or None
+
     def _agent_message_to_stream_response(self, answer: str, message_id: str) -> AgentMessageStreamResponse:
         """
         Agent message to stream response.

api/core/app/task_pipeline/message_cycle_manager.py

@@ -64,7 +64,13 @@ class MessageCycleManager:
 
         # Use SQLAlchemy 2.x style session.scalar(select(...))
         with session_factory.create_session() as session:
-            message_file = session.scalar(select(MessageFile).where(MessageFile.message_id == message_id))
+            message_file = session.scalar(
+                select(MessageFile)
+                .where(
+                    MessageFile.message_id == message_id,
+                )
+                .where(MessageFile.belongs_to == "assistant")
+            )
 
         if message_file:
             self._message_has_file.add(message_id)

api/core/app/workflow/file_runtime.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+from collections.abc import Generator
+
+from configs import dify_config
+from core.helper.ssrf_proxy import ssrf_proxy
+from core.tools.signature import sign_tool_file
+from core.workflow.file.protocols import HttpResponseProtocol, WorkflowFileRuntimeProtocol
+from core.workflow.file.runtime import set_workflow_file_runtime
+from extensions.ext_storage import storage
+
+
+class DifyWorkflowFileRuntime(WorkflowFileRuntimeProtocol):
+    """Production runtime wiring for ``core.workflow.file``."""
+
+    @property
+    def files_url(self) -> str:
+        return dify_config.FILES_URL
+
+    @property
+    def internal_files_url(self) -> str | None:
+        return dify_config.INTERNAL_FILES_URL
+
+    @property
+    def secret_key(self) -> str:
+        return dify_config.SECRET_KEY
+
+    @property
+    def files_access_timeout(self) -> int:
+        return dify_config.FILES_ACCESS_TIMEOUT
+
+    @property
+    def multimodal_send_format(self) -> str:
+        return dify_config.MULTIMODAL_SEND_FORMAT
+
+    def http_get(self, url: str, *, follow_redirects: bool = True) -> HttpResponseProtocol:
+        return ssrf_proxy.get(url, follow_redirects=follow_redirects)
+
+    def storage_load(self, path: str, *, stream: bool = False) -> bytes | Generator:
+        return storage.load(path, stream=stream)
+
+    def sign_tool_file(self, *, tool_file_id: str, extension: str, for_external: bool = True) -> str:
+        return sign_tool_file(tool_file_id=tool_file_id, extension=extension, for_external=for_external)
+
+
+def bind_dify_workflow_file_runtime() -> None:
+    set_workflow_file_runtime(DifyWorkflowFileRuntime())

api/core/app/workflow/node_factory.py

@@ -1,10 +1,8 @@
-from collections.abc import Callable, Sequence
 from typing import TYPE_CHECKING, final
 
 from typing_extensions import override
 
 from configs import dify_config
-from core.file.file_manager import file_manager
 from core.helper.code_executor.code_executor import CodeExecutor
 from core.helper.code_executor.code_node_provider import CodeNodeProvider
 from core.helper.ssrf_proxy import ssrf_proxy
@@ -12,18 +10,16 @@ from core.rag.retrieval.dataset_retrieval import DatasetRetrieval
 from core.tools.tool_file_manager import ToolFileManager
 from core.workflow.entities.graph_config import NodeConfigDict
 from core.workflow.enums import NodeType
+from core.workflow.file.file_manager import file_manager
 from core.workflow.graph.graph import NodeFactory
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.code.code_node import CodeNode
 from core.workflow.nodes.code.limits import CodeNodeLimits
-from core.workflow.nodes.http_request.node import HttpRequestNode
+from core.workflow.nodes.document_extractor import DocumentExtractorNode, UnstructuredApiConfig
+from core.workflow.nodes.http_request import HttpRequestNode, build_http_request_config
 from core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node import KnowledgeRetrievalNode
 from core.workflow.nodes.node_mapping import LATEST_VERSION, NODE_TYPE_CLASSES_MAPPING
-from core.workflow.nodes.protocols import FileManagerProtocol, HttpClientProtocol
-from core.workflow.nodes.template_transform.template_renderer import (
-    CodeExecutorJinja2TemplateRenderer,
-    Jinja2TemplateRenderer,
-)
+from core.workflow.nodes.template_transform.template_renderer import CodeExecutorJinja2TemplateRenderer
 from core.workflow.nodes.template_transform.template_transform_node import TemplateTransformNode
 
 if TYPE_CHECKING:
@@ -44,23 +40,12 @@ class DifyNodeFactory(NodeFactory):
         self,
         graph_init_params: "GraphInitParams",
         graph_runtime_state: "GraphRuntimeState",
-        *,
-        code_executor: type[CodeExecutor] | None = None,
-        code_providers: Sequence[type[CodeNodeProvider]] | None = None,
-        code_limits: CodeNodeLimits | None = None,
-        template_renderer: Jinja2TemplateRenderer | None = None,
-        template_transform_max_output_length: int | None = None,
-        http_request_http_client: HttpClientProtocol | None = None,
-        http_request_tool_file_manager_factory: Callable[[], ToolFileManager] = ToolFileManager,
-        http_request_file_manager: FileManagerProtocol | None = None,
     ) -> None:
         self.graph_init_params = graph_init_params
         self.graph_runtime_state = graph_runtime_state
-        self._code_executor: type[CodeExecutor] = code_executor or CodeExecutor
-        self._code_providers: tuple[type[CodeNodeProvider], ...] = (
-            tuple(code_providers) if code_providers else CodeNode.default_code_providers()
-        )
-        self._code_limits = code_limits or CodeNodeLimits(
+        self._code_executor: type[CodeExecutor] = CodeExecutor
+        self._code_providers: tuple[type[CodeNodeProvider], ...] = CodeNode.default_code_providers()
+        self._code_limits = CodeNodeLimits(
             max_string_length=dify_config.CODE_MAX_STRING_LENGTH,
             max_number=dify_config.CODE_MAX_NUMBER,
             min_number=dify_config.CODE_MIN_NUMBER,
@@ -70,14 +55,25 @@ class DifyNodeFactory(NodeFactory):
             max_string_array_length=dify_config.CODE_MAX_STRING_ARRAY_LENGTH,
             max_object_array_length=dify_config.CODE_MAX_OBJECT_ARRAY_LENGTH,
         )
-        self._template_renderer = template_renderer or CodeExecutorJinja2TemplateRenderer()
-        self._template_transform_max_output_length = (
-            template_transform_max_output_length or dify_config.TEMPLATE_TRANSFORM_MAX_LENGTH
-        )
-        self._http_request_http_client = http_request_http_client or ssrf_proxy
-        self._http_request_tool_file_manager_factory = http_request_tool_file_manager_factory
-        self._http_request_file_manager = http_request_file_manager or file_manager
+        self._template_renderer = CodeExecutorJinja2TemplateRenderer()
+        self._template_transform_max_output_length = dify_config.TEMPLATE_TRANSFORM_MAX_LENGTH
+        self._http_request_http_client = ssrf_proxy
+        self._http_request_tool_file_manager_factory = ToolFileManager
+        self._http_request_file_manager = file_manager
         self._rag_retrieval = DatasetRetrieval()
+        self._document_extractor_unstructured_api_config = UnstructuredApiConfig(
+            api_url=dify_config.UNSTRUCTURED_API_URL,
+            api_key=dify_config.UNSTRUCTURED_API_KEY or "",
+        )
+        self._http_request_config = build_http_request_config(
+            max_connect_timeout=dify_config.HTTP_REQUEST_MAX_CONNECT_TIMEOUT,
+            max_read_timeout=dify_config.HTTP_REQUEST_MAX_READ_TIMEOUT,
+            max_write_timeout=dify_config.HTTP_REQUEST_MAX_WRITE_TIMEOUT,
+            max_binary_size=dify_config.HTTP_REQUEST_NODE_MAX_BINARY_SIZE,
+            max_text_size=dify_config.HTTP_REQUEST_NODE_MAX_TEXT_SIZE,
+            ssl_verify=dify_config.HTTP_REQUEST_NODE_SSL_VERIFY,
+            ssrf_default_max_retries=dify_config.SSRF_DEFAULT_MAX_RETRIES,
+        )
 
     @override
     def create_node(self, node_config: NodeConfigDict) -> Node:
@@ -138,6 +134,7 @@ class DifyNodeFactory(NodeFactory):
                 config=node_config,
                 graph_init_params=self.graph_init_params,
                 graph_runtime_state=self.graph_runtime_state,
+                http_request_config=self._http_request_config,
                 http_client=self._http_request_http_client,
                 tool_file_manager_factory=self._http_request_tool_file_manager_factory,
                 file_manager=self._http_request_file_manager,
@@ -152,6 +149,15 @@ class DifyNodeFactory(NodeFactory):
                 rag_retrieval=self._rag_retrieval,
             )
 
+        if node_type == NodeType.DOCUMENT_EXTRACTOR:
+            return DocumentExtractorNode(
+                id=node_id,
+                config=node_config,
+                graph_init_params=self.graph_init_params,
+                graph_runtime_state=self.graph_runtime_state,
+                unstructured_api_config=self._document_extractor_unstructured_api_config,
+            )
+
         return node_class(
             id=node_id,
             config=node_config,

api/core/datasource/datasource_file_manager.py

@@ -213,6 +213,6 @@ class DatasourceFileManager:
 
 
 # init tool_file_parser
-# from core.file.datasource_file_parser import datasource_file_manager
+# from core.workflow.file.datasource_file_parser import datasource_file_manager
 #
 # datasource_file_manager["manager"] = DatasourceFileManager

api/core/datasource/utils/message_transformer.py

@@ -3,8 +3,8 @@ from collections.abc import Generator
 from mimetypes import guess_extension, guess_type
 
 from core.datasource.entities.datasource_entities import DatasourceMessage
-from core.file import File, FileTransferMethod, FileType
 from core.tools.tool_file_manager import ToolFileManager
+from core.workflow.file import File, FileTransferMethod, FileType
 from models.tools import ToolFile
 
 logger = logging.getLogger(__name__)

api/core/entities/mcp_provider.py

@@ -10,12 +10,12 @@ from pydantic import BaseModel
 
 from configs import dify_config
 from core.entities.provider_entities import BasicProviderConfig
-from core.file import helpers as file_helpers
 from core.helper import encrypter
 from core.helper.provider_cache import NoOpProviderCredentialCache
 from core.mcp.types import OAuthClientInformation, OAuthClientMetadata, OAuthTokens
 from core.tools.entities.common_entities import I18nObject
 from core.tools.entities.tool_entities import ToolProviderType
+from core.workflow.file import helpers as file_helpers
 
 if TYPE_CHECKING:
     from models.tools import MCPToolProvider

api/core/file/tool_file_parser.py

@@ -1,12 +0,0 @@
-from collections.abc import Callable
-from typing import TYPE_CHECKING
-
-if TYPE_CHECKING:
-    from core.tools.tool_file_manager import ToolFileManager
-
-_tool_file_manager_factory: Callable[[], "ToolFileManager"] | None = None
-
-
-def set_tool_file_manager_factory(factory: Callable[[], "ToolFileManager"]):
-    global _tool_file_manager_factory
-    _tool_file_manager_factory = factory

api/core/memory/token_buffer_memory.py

@@ -4,7 +4,6 @@ from sqlalchemy import select
 from sqlalchemy.orm import sessionmaker
 
 from core.app.app_config.features.file_upload.manager import FileUploadConfigManager
-from core.file import file_manager
 from core.model_manager import ModelInstance
 from core.model_runtime.entities import (
     AssistantPromptMessage,
@@ -16,6 +15,7 @@ from core.model_runtime.entities import (
 )
 from core.model_runtime.entities.message_entities import PromptMessageContentUnionTypes
 from core.prompt.utils.extract_thread_messages import extract_thread_messages
+from core.workflow.file import file_manager
 from extensions.ext_database import db
 from factories import file_factory
 from models.model import AppMode, Conversation, Message, MessageFile

api/core/moderation/base.py

@@ -39,7 +39,7 @@ class Moderation(Extensible, ABC):
 
     @classmethod
     @abstractmethod
-    def validate_config(cls, tenant_id: str, config: dict):
+    def validate_config(cls, tenant_id: str, config: dict) -> None:
         """
         Validate the incoming form config data.
 

api/core/ops/aliyun_trace/aliyun_trace.py

@@ -14,6 +14,7 @@ from core.ops.aliyun_trace.data_exporter.traceclient import (
 )
 from core.ops.aliyun_trace.entities.aliyun_trace_entity import SpanData, TraceMetadata
 from core.ops.aliyun_trace.entities.semconv import (
+    DIFY_APP_ID,
     GEN_AI_COMPLETION,
     GEN_AI_INPUT_MESSAGE,
     GEN_AI_OUTPUT_MESSAGE,
@@ -99,6 +100,16 @@ class AliyunDataTrace(BaseTraceInstance):
             logger.info("Aliyun get project url failed: %s", str(e), exc_info=True)
             raise ValueError(f"Aliyun get project url failed: {str(e)}")
 
+    def _extract_app_id(self, trace_info: BaseTraceInfo) -> str:
+        """Extract app_id from trace_info, trying metadata first then message_data."""
+        app_id = trace_info.metadata.get("app_id")
+        if app_id:
+            return str(app_id)
+        message_data = getattr(trace_info, "message_data", None)
+        if message_data is not None:
+            return str(getattr(message_data, "app_id", ""))
+        return ""
+
     def workflow_trace(self, trace_info: WorkflowTraceInfo):
         trace_metadata = TraceMetadata(
             trace_id=convert_to_trace_id(trace_info.workflow_run_id),
@@ -143,13 +154,16 @@ class AliyunDataTrace(BaseTraceInstance):
             name="message",
             start_time=convert_datetime_to_nanoseconds(trace_info.start_time),
             end_time=convert_datetime_to_nanoseconds(trace_info.end_time),
-            attributes=create_common_span_attributes(
-                session_id=trace_metadata.session_id,
-                user_id=trace_metadata.user_id,
-                span_kind=GenAISpanKind.CHAIN,
-                inputs=inputs_json,
-                outputs=outputs_str,
-            ),
+            attributes={
+                **create_common_span_attributes(
+                    session_id=trace_metadata.session_id,
+                    user_id=trace_metadata.user_id,
+                    span_kind=GenAISpanKind.CHAIN,
+                    inputs=inputs_json,
+                    outputs=outputs_str,
+                ),
+                DIFY_APP_ID: self._extract_app_id(trace_info),
+            },
             status=status,
             links=trace_metadata.links,
             span_kind=SpanKind.SERVER,
@@ -441,6 +455,8 @@ class AliyunDataTrace(BaseTraceInstance):
         inputs_json = serialize_json_data(trace_info.workflow_run_inputs)
         outputs_json = serialize_json_data(trace_info.workflow_run_outputs)
 
+        app_id = self._extract_app_id(trace_info)
+
         if message_span_id:
             message_span = SpanData(
                 trace_id=trace_metadata.trace_id,
@@ -449,13 +465,16 @@ class AliyunDataTrace(BaseTraceInstance):
                 name="message",
                 start_time=convert_datetime_to_nanoseconds(trace_info.start_time),
                 end_time=convert_datetime_to_nanoseconds(trace_info.end_time),
-                attributes=create_common_span_attributes(
-                    session_id=trace_metadata.session_id,
-                    user_id=trace_metadata.user_id,
-                    span_kind=GenAISpanKind.CHAIN,
-                    inputs=trace_info.workflow_run_inputs.get("sys.query") or "",
-                    outputs=outputs_json,
-                ),
+                attributes={
+                    **create_common_span_attributes(
+                        session_id=trace_metadata.session_id,
+                        user_id=trace_metadata.user_id,
+                        span_kind=GenAISpanKind.CHAIN,
+                        inputs=trace_info.workflow_run_inputs.get("sys.query") or "",
+                        outputs=outputs_json,
+                    ),
+                    DIFY_APP_ID: app_id,
+                },
                 status=status,
                 links=trace_metadata.links,
                 span_kind=SpanKind.SERVER,
@@ -469,13 +488,16 @@ class AliyunDataTrace(BaseTraceInstance):
             name="workflow",
             start_time=convert_datetime_to_nanoseconds(trace_info.start_time),
             end_time=convert_datetime_to_nanoseconds(trace_info.end_time),
-            attributes=create_common_span_attributes(
-                session_id=trace_metadata.session_id,
-                user_id=trace_metadata.user_id,
-                span_kind=GenAISpanKind.CHAIN,
-                inputs=inputs_json,
-                outputs=outputs_json,
-            ),
+            attributes={
+                **create_common_span_attributes(
+                    session_id=trace_metadata.session_id,
+                    user_id=trace_metadata.user_id,
+                    span_kind=GenAISpanKind.CHAIN,
+                    inputs=inputs_json,
+                    outputs=outputs_json,
+                ),
+                **({DIFY_APP_ID: app_id} if message_span_id is None else {}),
+            },
             status=status,
             links=trace_metadata.links,
             span_kind=SpanKind.SERVER if message_span_id is None else SpanKind.INTERNAL,

api/core/ops/aliyun_trace/entities/semconv.py

@@ -3,6 +3,9 @@ from typing import Final
 
 ACS_ARMS_SERVICE_FEATURE: Final[str] = "acs.arms.service.feature"
 
+# Dify-specific attributes
+DIFY_APP_ID: Final[str] = "dify.app_id"
+
 # Public attributes
 GEN_AI_SESSION_ID: Final[str] = "gen_ai.session.id"
 GEN_AI_USER_ID: Final[str] = "gen_ai.user.id"

api/core/ops/langfuse_trace/entities/langfuse_trace_entity.py

@@ -129,11 +129,11 @@ class LangfuseSpan(BaseModel):
         default=None,
         description="The id of the user that triggered the execution. Used to provide user-level analytics.",
     )
-    start_time: datetime | str | None = Field(
+    start_time: datetime | None = Field(
         default_factory=datetime.now,
         description="The time at which the span started, defaults to the current time.",
     )
-    end_time: datetime | str | None = Field(
+    end_time: datetime | None = Field(
         default=None,
         description="The time at which the span ended. Automatically set by span.end().",
     )
@@ -146,7 +146,7 @@ class LangfuseSpan(BaseModel):
         description="Additional metadata of the span. Can be any JSON object. Metadata is merged when being updated "
         "via the API.",
     )
-    level: str | None = Field(
+    level: LevelEnum | None = Field(
         default=None,
         description="The level of the span. Can be DEBUG, DEFAULT, WARNING or ERROR. Used for sorting/filtering of "
         "traces with elevated error levels and for highlighting in the UI.",
@@ -222,16 +222,16 @@ class LangfuseGeneration(BaseModel):
         default=None,
         description="Identifier of the generation. Useful for sorting/filtering in the UI.",
     )
-    start_time: datetime | str | None = Field(
+    start_time: datetime | None = Field(
         default_factory=datetime.now,
         description="The time at which the generation started, defaults to the current time.",
     )
-    completion_start_time: datetime | str | None = Field(
+    completion_start_time: datetime | None = Field(
         default=None,
         description="The time at which the completion started (streaming). Set it to get latency analytics broken "
         "down into time until completion started and completion duration.",
     )
-    end_time: datetime | str | None = Field(
+    end_time: datetime | None = Field(
         default=None,
         description="The time at which the generation ended. Automatically set by generation.end().",
     )

api/core/ops/tencent_trace/client.py

@@ -18,8 +18,7 @@ except ImportError:
     from importlib_metadata import version  # type: ignore[import-not-found]
 
 if TYPE_CHECKING:
-    from opentelemetry.metrics import Meter
-    from opentelemetry.metrics._internal.instrument import Histogram
+    from opentelemetry.metrics import Histogram, Meter
     from opentelemetry.sdk.metrics.export import MetricReader
 
 from opentelemetry import trace as trace_api

api/core/plugin/impl/tool.py

@@ -3,6 +3,8 @@ from typing import Any
 
 from pydantic import BaseModel
 
+from configs import dify_config
+
 # from core.plugin.entities.plugin import GenericProviderID, ToolProviderID
 from core.plugin.entities.plugin_daemon import CredentialType, PluginBasicBooleanResponse, PluginToolProviderEntity
 from core.plugin.impl.base import BasePluginClient
@@ -122,7 +124,7 @@ class PluginToolManager(BasePluginClient):
             },
         )
 
-        return merge_blob_chunks(response)
+        return merge_blob_chunks(response, max_file_size=dify_config.PLUGIN_MAX_FILE_SIZE)
 
     def validate_provider_credentials(
         self, tenant_id: str, user_id: str, provider: str, credentials: dict[str, Any]

api/core/plugin/utils/converter.py

@@ -1,7 +1,7 @@
 from typing import Any
 
-from core.file.models import File
 from core.tools.entities.tool_entities import ToolSelector
+from core.workflow.file.models import File
 
 
 def convert_parameters_to_plugin_format(parameters: dict[str, Any]) -> dict[str, Any]:

api/core/prompt/advanced_prompt_transform.py

@@ -2,8 +2,6 @@ from collections.abc import Mapping, Sequence
 from typing import cast
 
 from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
-from core.file import file_manager
-from core.file.models import File
 from core.helper.code_executor.jinja2.jinja2_formatter import Jinja2Formatter
 from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_runtime.entities import (
@@ -18,6 +16,8 @@ from core.model_runtime.entities.message_entities import ImagePromptMessageConte
 from core.prompt.entities.advanced_prompt_entities import ChatModelMessage, CompletionModelPromptTemplate, MemoryConfig
 from core.prompt.prompt_transform import PromptTransform
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
+from core.workflow.file import file_manager
+from core.workflow.file.models import File
 from core.workflow.runtime import VariablePool
 
 

api/core/prompt/simple_prompt_transform.py

@@ -6,7 +6,6 @@ from typing import TYPE_CHECKING, Any, cast
 
 from core.app.app_config.entities import PromptTemplateEntity
 from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
-from core.file import file_manager
 from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_runtime.entities.message_entities import (
     ImagePromptMessageContent,
@@ -19,10 +18,11 @@ from core.model_runtime.entities.message_entities import (
 from core.prompt.entities.advanced_prompt_entities import MemoryConfig
 from core.prompt.prompt_transform import PromptTransform
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
+from core.workflow.file import file_manager
 from models.model import AppMode
 
 if TYPE_CHECKING:
-    from core.file.models import File
+    from core.workflow.file.models import File
 
 
 class ModelMode(StrEnum):

api/core/rag/datasource/vdb/oceanbase/oceanbase_vector.py

@@ -1,12 +1,13 @@
 import json
 import logging
-import math
-from typing import Any
+import re
+from typing import Any, Literal
 
 from pydantic import BaseModel, model_validator
-from pyobvector import VECTOR, ObVecClient, l2_distance  # type: ignore
+from pyobvector import VECTOR, ObVecClient, cosine_distance, inner_product, l2_distance  # type: ignore
 from sqlalchemy import JSON, Column, String
 from sqlalchemy.dialects.mysql import LONGTEXT
+from sqlalchemy.exc import SQLAlchemyError
 
 from configs import dify_config
 from core.rag.datasource.vdb.vector_base import BaseVector
@@ -19,10 +20,14 @@ from models.dataset import Dataset
 
 logger = logging.getLogger(__name__)
 
-DEFAULT_OCEANBASE_HNSW_BUILD_PARAM = {"M": 16, "efConstruction": 256}
-DEFAULT_OCEANBASE_HNSW_SEARCH_PARAM = {"efSearch": 64}
 OCEANBASE_SUPPORTED_VECTOR_INDEX_TYPE = "HNSW"
-DEFAULT_OCEANBASE_VECTOR_METRIC_TYPE = "l2"
+_VALID_TABLE_NAME_RE = re.compile(r"^[a-zA-Z0-9_]+$")
+
+_DISTANCE_FUNC_MAP = {
+    "l2": l2_distance,
+    "cosine": cosine_distance,
+    "inner_product": inner_product,
+}
 
 
 class OceanBaseVectorConfig(BaseModel):
@@ -32,6 +37,14 @@ class OceanBaseVectorConfig(BaseModel):
     password: str
     database: str
     enable_hybrid_search: bool = False
+    batch_size: int = 100
+    metric_type: Literal["l2", "cosine", "inner_product"] = "l2"
+    hnsw_m: int = 16
+    hnsw_ef_construction: int = 256
+    hnsw_ef_search: int = -1
+    pool_size: int = 5
+    max_overflow: int = 10
+    hnsw_refresh_threshold: int = 1000
 
     @model_validator(mode="before")
     @classmethod
@@ -49,14 +62,23 @@ class OceanBaseVectorConfig(BaseModel):
 
 class OceanBaseVector(BaseVector):
     def __init__(self, collection_name: str, config: OceanBaseVectorConfig):
+        if not _VALID_TABLE_NAME_RE.match(collection_name):
+            raise ValueError(
+                f"Invalid collection name '{collection_name}': "
+                "only alphanumeric characters and underscores are allowed."
+            )
         super().__init__(collection_name)
         self._config = config
-        self._hnsw_ef_search = -1
+        self._hnsw_ef_search = self._config.hnsw_ef_search
         self._client = ObVecClient(
             uri=f"{self._config.host}:{self._config.port}",
             user=self._config.user,
             password=self._config.password,
             db_name=self._config.database,
+            pool_size=self._config.pool_size,
+            max_overflow=self._config.max_overflow,
+            pool_recycle=3600,
+            pool_pre_ping=True,
         )
         self._fields: list[str] = []  # List of fields in the collection
         if self._client.check_table_exists(collection_name):
@@ -136,8 +158,8 @@ class OceanBaseVector(BaseVector):
                 field_name="vector",
                 index_type=OCEANBASE_SUPPORTED_VECTOR_INDEX_TYPE,
                 index_name="vector_index",
-                metric_type=DEFAULT_OCEANBASE_VECTOR_METRIC_TYPE,
-                params=DEFAULT_OCEANBASE_HNSW_BUILD_PARAM,
+                metric_type=self._config.metric_type,
+                params={"M": self._config.hnsw_m, "efConstruction": self._config.hnsw_ef_construction},
             )
 
             self._client.create_table_with_index_params(
@@ -178,6 +200,17 @@ class OceanBaseVector(BaseVector):
             else:
                 logger.debug("DEBUG: Hybrid search is NOT enabled for '%s'", self._collection_name)
 
+            try:
+                self._client.perform_raw_text_sql(
+                    f"CREATE INDEX IF NOT EXISTS idx_metadata_doc_id ON `{self._collection_name}` "
+                    f"((CAST(metadata->>'$.document_id' AS CHAR(64))))"
+                )
+            except SQLAlchemyError:
+                logger.warning(
+                    "Failed to create metadata functional index on '%s'; metadata queries may be slow without it.",
+                    self._collection_name,
+                )
+
             self._client.refresh_metadata([self._collection_name])
             self._load_collection_fields()
             redis_client.set(collection_exist_cache_key, 1, ex=3600)
@@ -205,24 +238,49 @@ class OceanBaseVector(BaseVector):
 
     def add_texts(self, documents: list[Document], embeddings: list[list[float]], **kwargs):
         ids = self._get_uuids(documents)
-        for id, doc, emb in zip(ids, documents, embeddings):
+        batch_size = self._config.batch_size
+        total = len(documents)
+
+        all_data = [
+            {
+                "id": doc_id,
+                "vector": emb,
+                "text": doc.page_content,
+                "metadata": doc.metadata,
+            }
+            for doc_id, doc, emb in zip(ids, documents, embeddings)
+        ]
+
+        for start in range(0, total, batch_size):
+            batch = all_data[start : start + batch_size]
             try:
                 self._client.insert(
                     table_name=self._collection_name,
-                    data={
-                        "id": id,
-                        "vector": emb,
-                        "text": doc.page_content,
-                        "metadata": doc.metadata,
-                    },
+                    data=batch,
                 )
             except Exception as e:
                 logger.exception(
-                    "Failed to insert document with id '%s' in collection '%s'",
-                    id,
+                    "Failed to insert batch [%d:%d] into collection '%s'",
+                    start,
+                    start + len(batch),
+                    self._collection_name,
+                )
+                raise Exception(
+                    f"Failed to insert batch [{start}:{start + len(batch)}] into collection '{self._collection_name}'"
+                ) from e
+
+        if self._config.hnsw_refresh_threshold > 0 and total >= self._config.hnsw_refresh_threshold:
+            try:
+                self._client.refresh_index(
+                    table_name=self._collection_name,
+                    index_name="vector_index",
+                )
+            except SQLAlchemyError:
+                logger.warning(
+                    "Failed to refresh HNSW index after inserting %d documents into '%s'",
+                    total,
                     self._collection_name,
                 )
-                raise Exception(f"Failed to insert document with id '{id}'") from e
 
     def text_exists(self, id: str) -> bool:
         try:
@@ -412,7 +470,7 @@ class OceanBaseVector(BaseVector):
                 vec_column_name="vector",
                 vec_data=query_vector,
                 topk=topk,
-                distance_func=l2_distance,
+                distance_func=self._get_distance_func(),
                 output_column_names=["text", "metadata"],
                 with_dist=True,
                 where_clause=_where_clause,
@@ -424,14 +482,31 @@ class OceanBaseVector(BaseVector):
             )
             raise Exception(f"Vector search failed for collection '{self._collection_name}'") from e
 
-        # Convert distance to score and prepare results for processing
         results = []
         for _text, metadata_str, distance in cur:
-            score = 1 - distance / math.sqrt(2)
+            score = self._distance_to_score(distance)
             results.append((_text, metadata_str, score))
 
         return self._process_search_results(results, score_threshold=score_threshold)
 
+    def _get_distance_func(self):
+        func = _DISTANCE_FUNC_MAP.get(self._config.metric_type)
+        if func is None:
+            raise ValueError(
+                f"Unsupported metric_type '{self._config.metric_type}'. Supported: {', '.join(_DISTANCE_FUNC_MAP)}"
+            )
+        return func
+
+    def _distance_to_score(self, distance: float) -> float:
+        metric = self._config.metric_type
+        if metric == "l2":
+            return 1.0 / (1.0 + distance)
+        elif metric == "cosine":
+            return 1.0 - distance
+        elif metric == "inner_product":
+            return -distance
+        raise ValueError(f"Unsupported metric_type '{metric}'")
+
     def delete(self):
         try:
             self._client.drop_table_if_exist(self._collection_name)
@@ -464,5 +539,13 @@ class OceanBaseVectorFactory(AbstractVectorFactory):
                 password=(dify_config.OCEANBASE_VECTOR_PASSWORD or ""),
                 database=dify_config.OCEANBASE_VECTOR_DATABASE or "",
                 enable_hybrid_search=dify_config.OCEANBASE_ENABLE_HYBRID_SEARCH or False,
+                batch_size=dify_config.OCEANBASE_VECTOR_BATCH_SIZE,
+                metric_type=dify_config.OCEANBASE_VECTOR_METRIC_TYPE,
+                hnsw_m=dify_config.OCEANBASE_HNSW_M,
+                hnsw_ef_construction=dify_config.OCEANBASE_HNSW_EF_CONSTRUCTION,
+                hnsw_ef_search=dify_config.OCEANBASE_HNSW_EF_SEARCH,
+                pool_size=dify_config.OCEANBASE_VECTOR_POOL_SIZE,
+                max_overflow=dify_config.OCEANBASE_VECTOR_MAX_OVERFLOW,
+                hnsw_refresh_threshold=dify_config.OCEANBASE_HNSW_REFRESH_THRESHOLD,
             ),
         )

api/core/rag/datasource/vdb/vector_base.py

@@ -15,11 +15,11 @@ class BaseVector(ABC):
         raise NotImplementedError
 
     @abstractmethod
-    def create(self, texts: list[Document], embeddings: list[list[float]], **kwargs):
+    def create(self, texts: list[Document], embeddings: list[list[float]], **kwargs) -> list[str] | None:
         raise NotImplementedError
 
     @abstractmethod
-    def add_texts(self, documents: list[Document], embeddings: list[list[float]], **kwargs):
+    def add_texts(self, documents: list[Document], embeddings: list[list[float]], **kwargs) -> list[str]:
         raise NotImplementedError
 
     @abstractmethod
@@ -27,14 +27,14 @@ class BaseVector(ABC):
         raise NotImplementedError
 
     @abstractmethod
-    def delete_by_ids(self, ids: list[str]):
+    def delete_by_ids(self, ids: list[str]) -> None:
         raise NotImplementedError
 
     def get_ids_by_metadata_field(self, key: str, value: str):
         raise NotImplementedError
 
     @abstractmethod
-    def delete_by_metadata_field(self, key: str, value: str):
+    def delete_by_metadata_field(self, key: str, value: str) -> None:
         raise NotImplementedError
 
     @abstractmethod
@@ -46,7 +46,7 @@ class BaseVector(ABC):
         raise NotImplementedError
 
     @abstractmethod
-    def delete(self):
+    def delete(self) -> None:
         raise NotImplementedError
 
     def _filter_duplicate_texts(self, texts: list[Document]) -> list[Document]:

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

@@ -9,7 +9,6 @@ from typing import Any, cast
 logger = logging.getLogger(__name__)
 
 from core.entities.knowledge_entities import PreviewDetail
-from core.file import File, FileTransferMethod, FileType, file_manager
 from core.llm_generator.prompts import DEFAULT_GENERATOR_SUMMARY_PROMPT
 from core.model_manager import ModelInstance
 from core.model_runtime.entities.llm_entities import LLMResult, LLMUsage
@@ -35,6 +34,7 @@ from core.rag.index_processor.index_processor_base import BaseIndexProcessor
 from core.rag.models.document import AttachmentDocument, Document, MultimodalGeneralStructureChunk
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
 from core.tools.utils.text_processing_utils import remove_leading_symbols
+from core.workflow.file import File, FileTransferMethod, FileType, file_manager
 from core.workflow.nodes.llm import llm_utils
 from extensions.ext_database import db
 from factories.file_factory import build_from_mapping

api/core/rag/models/document.py

@@ -4,7 +4,7 @@ from typing import Any
 
 from pydantic import BaseModel, Field
 
-from core.file import File
+from core.workflow.file import File
 
 
 class ChildDocument(BaseModel):

api/core/rag/retrieval/dataset_retrieval.py

@@ -23,7 +23,6 @@ from core.callback_handler.index_tool_callback_handler import DatasetIndexToolCa
 from core.db.session_factory import session_factory
 from core.entities.agent_entities import PlanningStrategy
 from core.entities.model_entities import ModelStatus
-from core.file import File, FileTransferMethod, FileType
 from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance, ModelManager
 from core.model_runtime.entities.llm_entities import LLMResult, LLMUsage
@@ -61,6 +60,7 @@ from core.rag.retrieval.template_prompts import (
 )
 from core.tools.signature import sign_upload_file
 from core.tools.utils.dataset_retriever.dataset_retriever_base_tool import DatasetRetrieverBaseTool
+from core.workflow.file import File, FileTransferMethod, FileType
 from core.workflow.nodes.knowledge_retrieval import exc
 from core.workflow.repositories.rag_retrieval_protocol import (
     KnowledgeRetrievalRequest,

api/core/tools/__base/tool.py

@@ -5,7 +5,7 @@ from collections.abc import Generator
 from copy import deepcopy
 from typing import TYPE_CHECKING, Any
 
-if TYPE_CHECKING:
+if TYPE_CHECKING:  # pragma: no cover
     from models.model import File
 
 from core.tools.__base.tool_runtime import ToolRuntime
@@ -171,7 +171,7 @@ class Tool(ABC):
     def create_file_message(self, file: File) -> ToolInvokeMessage:
         return ToolInvokeMessage(
             type=ToolInvokeMessage.MessageType.FILE,
-            message=ToolInvokeMessage.FileMessage(),
+            message=ToolInvokeMessage.FileMessage(file_marker="file_marker"),
             meta={"file": file},
         )
 

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

@@ -2,14 +2,14 @@ import io
 from collections.abc import Generator
 from typing import Any
 
-from core.file.enums import FileType
-from core.file.file_manager import download
 from core.model_manager import ModelManager
 from core.model_runtime.entities.model_entities import ModelType
 from core.plugin.entities.parameters import PluginParameterOption
 from core.tools.builtin_tool.tool import BuiltinTool
 from core.tools.entities.common_entities import I18nObject
 from core.tools.entities.tool_entities import ToolInvokeMessage, ToolParameter
+from core.workflow.file.enums import FileType
+from core.workflow.file.file_manager import download
 from services.model_provider_service import ModelProviderService
 
 

api/core/tools/custom_tool/tool.py

@@ -7,13 +7,13 @@ from urllib.parse import urlencode
 
 import httpx
 
-from core.file.file_manager import download
 from core.helper import ssrf_proxy
 from core.tools.__base.tool import Tool
 from core.tools.__base.tool_runtime import ToolRuntime
 from core.tools.entities.tool_bundle import ApiToolBundle
 from core.tools.entities.tool_entities import ToolEntity, ToolInvokeMessage, ToolProviderType
 from core.tools.errors import ToolInvokeError, ToolParameterValidationError, ToolProviderCredentialValidationError
+from core.workflow.file.file_manager import download
 
 API_TOOL_DEFAULT_TIMEOUT = (
     int(getenv("API_TOOL_DEFAULT_CONNECT_TIMEOUT", "10")),

api/core/tools/tool_engine.py

@@ -12,8 +12,6 @@ from yarl import URL
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.callback_handler.agent_tool_callback_handler import DifyAgentCallbackHandler
 from core.callback_handler.workflow_tool_callback_handler import DifyWorkflowCallbackHandler
-from core.file import FileType
-from core.file.models import FileTransferMethod
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.tools.__base.tool import Tool
 from core.tools.entities.tool_entities import (
@@ -33,6 +31,8 @@ from core.tools.errors import (
 )
 from core.tools.utils.message_transformer import ToolFileMessageTransformer, safe_json_value
 from core.tools.workflow_as_tool.tool import WorkflowTool
+from core.workflow.file import FileType
+from core.workflow.file.models import FileTransferMethod
 from extensions.ext_database import db
 from models.enums import CreatorUserRole
 from models.model import Message, MessageFile

api/core/tools/tool_file_manager.py

@@ -243,7 +243,7 @@ class ToolFileManager:
 
 
 # init tool_file_parser
-from core.file.tool_file_parser import set_tool_file_manager_factory
+from core.workflow.file.tool_file_parser import set_tool_file_manager_factory
 
 
 def _factory() -> ToolFileManager:

api/core/tools/utils/message_transformer.py

@@ -8,9 +8,9 @@ from uuid import UUID
 import numpy as np
 import pytz
 
-from core.file import File, FileTransferMethod, FileType
 from core.tools.entities.tool_entities import ToolInvokeMessage
 from core.tools.tool_file_manager import ToolFileManager
+from core.workflow.file import File, FileTransferMethod, FileType
 from libs.login import current_user
 from models import Account
 

api/core/tools/utils/parser.py

@@ -2,7 +2,7 @@ import re
 from json import dumps as json_dumps
 from json import loads as json_loads
 from json.decoder import JSONDecodeError
-from typing import Any
+from typing import Any, TypedDict
 
 import httpx
 from flask import request
@@ -14,6 +14,12 @@ from core.tools.entities.tool_entities import ApiProviderSchemaType, ToolParamet
 from core.tools.errors import ToolApiSchemaError, ToolNotSupportedError, ToolProviderNotFoundError
 
 
+class InterfaceDict(TypedDict):
+    path: str
+    method: str
+    operation: dict[str, Any]
+
+
 class ApiBasedToolSchemaParser:
     @staticmethod
     def parse_openapi_to_tool_bundle(
@@ -35,7 +41,7 @@ class ApiBasedToolSchemaParser:
             server_url = matched_servers[0] if matched_servers else server_url
 
         # list all interfaces
-        interfaces = []
+        interfaces: list[InterfaceDict] = []
         for path, path_item in openapi["paths"].items():
             methods = ["get", "post", "put", "delete", "patch", "head", "options", "trace"]
             for method in methods:

api/core/tools/workflow_as_tool/tool.py

@@ -8,7 +8,6 @@ from typing import Any, cast
 from sqlalchemy import select
 
 from core.db.session_factory import session_factory
-from core.file import FILE_MODEL_IDENTITY, File, FileTransferMethod
 from core.model_runtime.entities.llm_entities import LLMUsage, LLMUsageMetadata
 from core.tools.__base.tool import Tool
 from core.tools.__base.tool_runtime import ToolRuntime
@@ -19,6 +18,7 @@ from core.tools.entities.tool_entities import (
     ToolProviderType,
 )
 from core.tools.errors import ToolInvokeError
+from core.workflow.file import FILE_MODEL_IDENTITY, File, FileTransferMethod
 from factories.file_factory import build_from_mapping
 from models import Account, Tenant
 from models.model import App, EndUser

api/core/variables/segments.py

@@ -5,7 +5,7 @@ from typing import Annotated, Any, TypeAlias
 
 from pydantic import BaseModel, ConfigDict, Discriminator, Tag, field_validator
 
-from core.file import File
+from core.workflow.file import File
 
 from .types import SegmentType
 

api/core/variables/types.py

@@ -4,7 +4,7 @@ from collections.abc import Mapping
 from enum import StrEnum
 from typing import TYPE_CHECKING, Any
 
-from core.file.models import File
+from core.workflow.file.models import File
 
 if TYPE_CHECKING:
     pass

api/core/workflow/file/file_manager.py

@@ -1,8 +1,8 @@
+from __future__ import annotations
+
 import base64
 from collections.abc import Mapping
 
-from configs import dify_config
-from core.helper import ssrf_proxy
 from core.model_runtime.entities import (
     AudioPromptMessageContent,
     DocumentPromptMessageContent,
@@ -11,12 +11,11 @@ from core.model_runtime.entities import (
     VideoPromptMessageContent,
 )
 from core.model_runtime.entities.message_entities import PromptMessageContentUnionTypes
-from core.tools.signature import sign_tool_file
-from extensions.ext_storage import storage
 
 from . import helpers
 from .enums import FileAttribute
 from .models import File, FileTransferMethod, FileType
+from .runtime import get_workflow_file_runtime
 
 
 def get_attr(*, file: File, attr: FileAttribute):
@@ -45,26 +44,7 @@ def to_prompt_message_content(
     *,
     image_detail_config: ImagePromptMessageContent.DETAIL | None = None,
 ) -> PromptMessageContentUnionTypes:
-    """
-    Convert a file to prompt message content.
-
-    This function converts files to their appropriate prompt message content types.
-    For supported file types (IMAGE, AUDIO, VIDEO, DOCUMENT), it creates the
-    corresponding message content with proper encoding/URL.
-
-    For unsupported file types, instead of raising an error, it returns a
-    TextPromptMessageContent with a descriptive message about the file.
-
-    Args:
-        f: The file to convert
-        image_detail_config: Optional detail configuration for image files
-
-    Returns:
-        PromptMessageContentUnionTypes: The appropriate message content type
-
-    Raises:
-        ValueError: If file extension or mime_type is missing
-    """
+    """Convert a file to prompt message content."""
     if f.extension is None:
         raise ValueError("Missing file extension")
     if f.mime_type is None:
@@ -77,15 +57,13 @@ def to_prompt_message_content(
         FileType.DOCUMENT: DocumentPromptMessageContent,
     }
 
-    # Check if file type is supported
     if f.type not in prompt_class_map:
-        # For unsupported file types, return a text description
         return TextPromptMessageContent(data=f"[Unsupported file type: {f.filename} ({f.type.value})]")
 
-    # Process supported file types
+    send_format = get_workflow_file_runtime().multimodal_send_format
     params = {
-        "base64_data": _get_encoded_string(f) if dify_config.MULTIMODAL_SEND_FORMAT == "base64" else "",
-        "url": _to_url(f) if dify_config.MULTIMODAL_SEND_FORMAT == "url" else "",
+        "base64_data": _get_encoded_string(f) if send_format == "base64" else "",
+        "url": _to_url(f) if send_format == "url" else "",
         "format": f.extension.removeprefix("."),
         "mime_type": f.mime_type,
         "filename": f.filename or "",
@@ -96,7 +74,7 @@ def to_prompt_message_content(
     return prompt_class_map[f.type].model_validate(params)
 
 
-def download(f: File, /):
+def download(f: File, /) -> bytes:
     if f.transfer_method in (
         FileTransferMethod.TOOL_FILE,
         FileTransferMethod.LOCAL_FILE,
@@ -106,39 +84,26 @@ def download(f: File, /):
     elif f.transfer_method == FileTransferMethod.REMOTE_URL:
         if f.remote_url is None:
             raise ValueError("Missing file remote_url")
-        response = ssrf_proxy.get(f.remote_url, follow_redirects=True)
+        response = get_workflow_file_runtime().http_get(f.remote_url, follow_redirects=True)
         response.raise_for_status()
         return response.content
     raise ValueError(f"unsupported transfer method: {f.transfer_method}")
 
 
-def _download_file_content(path: str, /):
-    """
-    Download and return the contents of a file as bytes.
-
-    This function loads the file from storage and ensures it's in bytes format.
-
-    Args:
-        path (str): The path to the file in storage.
-
-    Returns:
-        bytes: The contents of the file as a bytes object.
-
-    Raises:
-        ValueError: If the loaded file is not a bytes object.
-    """
-    data = storage.load(path, stream=False)
+def _download_file_content(path: str, /) -> bytes:
+    """Download and return a file from storage as bytes."""
+    data = get_workflow_file_runtime().storage_load(path, stream=False)
     if not isinstance(data, bytes):
         raise ValueError(f"file {path} is not a bytes object")
     return data
 
 
-def _get_encoded_string(f: File, /):
+def _get_encoded_string(f: File, /) -> str:
     match f.transfer_method:
         case FileTransferMethod.REMOTE_URL:
             if f.remote_url is None:
                 raise ValueError("Missing file remote_url")
-            response = ssrf_proxy.get(f.remote_url, follow_redirects=True)
+            response = get_workflow_file_runtime().http_get(f.remote_url, follow_redirects=True)
             response.raise_for_status()
             data = response.content
         case FileTransferMethod.LOCAL_FILE:
@@ -148,8 +113,7 @@ def _get_encoded_string(f: File, /):
         case FileTransferMethod.DATASOURCE_FILE:
             data = _download_file_content(f.storage_key)
 
-    encoded_string = base64.b64encode(data).decode("utf-8")
-    return encoded_string
+    return base64.b64encode(data).decode("utf-8")
 
 
 def _to_url(f: File, /):
@@ -162,21 +126,15 @@ def _to_url(f: File, /):
             raise ValueError("Missing file related_id")
         return f.remote_url or helpers.get_signed_file_url(upload_file_id=f.related_id)
     elif f.transfer_method == FileTransferMethod.TOOL_FILE:
-        # add sign url
         if f.related_id is None or f.extension is None:
             raise ValueError("Missing file related_id or extension")
-        return sign_tool_file(tool_file_id=f.related_id, extension=f.extension)
+        return helpers.get_signed_tool_file_url(tool_file_id=f.related_id, extension=f.extension)
     else:
         raise ValueError(f"Unsupported transfer method: {f.transfer_method}")
 
 
 class FileManager:
-    """
-    Adapter exposing file manager helpers behind FileManagerProtocol.
-
-    This is intentionally a thin wrapper over the existing module-level functions so callers can inject it
-    where a protocol-typed file manager is expected.
-    """
+    """Adapter exposing file manager helpers behind FileManagerProtocol."""
 
     def download(self, f: File, /) -> bytes:
         return download(f)

api/core/workflow/file/helpers.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import base64
 import hashlib
 import hmac
@@ -5,20 +7,21 @@ import os
 import time
 import urllib.parse
 
-from configs import dify_config
+from .runtime import get_workflow_file_runtime
 
 
-def get_signed_file_url(upload_file_id: str, as_attachment=False, for_external: bool = True) -> str:
-    base_url = dify_config.FILES_URL if for_external else (dify_config.INTERNAL_FILES_URL or dify_config.FILES_URL)
+def get_signed_file_url(upload_file_id: str, as_attachment: bool = False, for_external: bool = True) -> str:
+    runtime = get_workflow_file_runtime()
+    base_url = runtime.files_url if for_external else (runtime.internal_files_url or runtime.files_url)
     url = f"{base_url}/files/{upload_file_id}/file-preview"
 
     timestamp = str(int(time.time()))
     nonce = os.urandom(16).hex()
-    key = dify_config.SECRET_KEY.encode()
+    key = runtime.secret_key.encode()
     msg = f"file-preview|{upload_file_id}|{timestamp}|{nonce}"
     sign = hmac.new(key, msg.encode(), hashlib.sha256).digest()
     encoded_sign = base64.urlsafe_b64encode(sign).decode()
-    query = {"timestamp": timestamp, "nonce": nonce, "sign": encoded_sign}
+    query: dict[str, str] = {"timestamp": timestamp, "nonce": nonce, "sign": encoded_sign}
     if as_attachment:
         query["as_attachment"] = "true"
     query_string = urllib.parse.urlencode(query)
@@ -27,57 +30,63 @@ def get_signed_file_url(upload_file_id: str, as_attachment=False, for_external:
 
 
 def get_signed_file_url_for_plugin(filename: str, mimetype: str, tenant_id: str, user_id: str) -> str:
-    # Plugin access should use internal URL for Docker network communication
-    base_url = dify_config.INTERNAL_FILES_URL or dify_config.FILES_URL
+    runtime = get_workflow_file_runtime()
+    # Plugin access should use internal URL for Docker network communication.
+    base_url = runtime.internal_files_url or runtime.files_url
     url = f"{base_url}/files/upload/for-plugin"
     timestamp = str(int(time.time()))
     nonce = os.urandom(16).hex()
-    key = dify_config.SECRET_KEY.encode()
+    key = runtime.secret_key.encode()
     msg = f"upload|{filename}|{mimetype}|{tenant_id}|{user_id}|{timestamp}|{nonce}"
     sign = hmac.new(key, msg.encode(), hashlib.sha256).digest()
     encoded_sign = base64.urlsafe_b64encode(sign).decode()
     return f"{url}?timestamp={timestamp}&nonce={nonce}&sign={encoded_sign}&user_id={user_id}&tenant_id={tenant_id}"
 
 
+def get_signed_tool_file_url(tool_file_id: str, extension: str, for_external: bool = True) -> str:
+    runtime = get_workflow_file_runtime()
+    return runtime.sign_tool_file(tool_file_id=tool_file_id, extension=extension, for_external=for_external)
+
+
 def verify_plugin_file_signature(
     *, filename: str, mimetype: str, tenant_id: str, user_id: str, timestamp: str, nonce: str, sign: str
 ) -> bool:
+    runtime = get_workflow_file_runtime()
     data_to_sign = f"upload|{filename}|{mimetype}|{tenant_id}|{user_id}|{timestamp}|{nonce}"
-    secret_key = dify_config.SECRET_KEY.encode()
+    secret_key = runtime.secret_key.encode()
     recalculated_sign = hmac.new(secret_key, data_to_sign.encode(), hashlib.sha256).digest()
     recalculated_encoded_sign = base64.urlsafe_b64encode(recalculated_sign).decode()
 
-    # verify signature
     if sign != recalculated_encoded_sign:
         return False
 
     current_time = int(time.time())
-    return current_time - int(timestamp) <= dify_config.FILES_ACCESS_TIMEOUT
+    return current_time - int(timestamp) <= runtime.files_access_timeout
 
 
 def verify_image_signature(*, upload_file_id: str, timestamp: str, nonce: str, sign: str) -> bool:
+    runtime = get_workflow_file_runtime()
     data_to_sign = f"image-preview|{upload_file_id}|{timestamp}|{nonce}"
-    secret_key = dify_config.SECRET_KEY.encode()
+    secret_key = runtime.secret_key.encode()
     recalculated_sign = hmac.new(secret_key, data_to_sign.encode(), hashlib.sha256).digest()
     recalculated_encoded_sign = base64.urlsafe_b64encode(recalculated_sign).decode()
 
-    # verify signature
     if sign != recalculated_encoded_sign:
         return False
 
     current_time = int(time.time())
-    return current_time - int(timestamp) <= dify_config.FILES_ACCESS_TIMEOUT
+    return current_time - int(timestamp) <= runtime.files_access_timeout
 
 
 def verify_file_signature(*, upload_file_id: str, timestamp: str, nonce: str, sign: str) -> bool:
+    runtime = get_workflow_file_runtime()
     data_to_sign = f"file-preview|{upload_file_id}|{timestamp}|{nonce}"
-    secret_key = dify_config.SECRET_KEY.encode()
+    secret_key = runtime.secret_key.encode()
     recalculated_sign = hmac.new(secret_key, data_to_sign.encode(), hashlib.sha256).digest()
     recalculated_encoded_sign = base64.urlsafe_b64encode(recalculated_sign).decode()
 
-    # verify signature
     if sign != recalculated_encoded_sign:
         return False
 
     current_time = int(time.time())
-    return current_time - int(timestamp) <= dify_config.FILES_ACCESS_TIMEOUT
+    return current_time - int(timestamp) <= runtime.files_access_timeout

api/core/workflow/file/models.py

@@ -1,16 +1,26 @@
+from __future__ import annotations
+
 from collections.abc import Mapping, Sequence
 from typing import Any
 
 from pydantic import BaseModel, Field, model_validator
 
 from core.model_runtime.entities.message_entities import ImagePromptMessageContent
-from core.tools.signature import sign_tool_file
 
 from . import helpers
 from .constants import FILE_MODEL_IDENTITY
 from .enums import FileTransferMethod, FileType
 
 
+def sign_tool_file(*, tool_file_id: str, extension: str, for_external: bool = True) -> str:
+    """Compatibility shim for tests and legacy callers patching ``models.sign_tool_file``."""
+    return helpers.get_signed_tool_file_url(
+        tool_file_id=tool_file_id,
+        extension=extension,
+        for_external=for_external,
+    )
+
+
 class ImageConfig(BaseModel):
     """
     NOTE: This part of validation is deprecated, but still used in app features "Image Upload".
@@ -122,7 +132,11 @@ class File(BaseModel):
         elif self.transfer_method in [FileTransferMethod.TOOL_FILE, FileTransferMethod.DATASOURCE_FILE]:
             assert self.related_id is not None
             assert self.extension is not None
-            return sign_tool_file(tool_file_id=self.related_id, extension=self.extension, for_external=for_external)
+            return sign_tool_file(
+                tool_file_id=self.related_id,
+                extension=self.extension,
+                for_external=for_external,
+            )
         return None
 
     def to_plugin_parameter(self) -> dict[str, Any]:
@@ -137,7 +151,7 @@ class File(BaseModel):
         }
 
     @model_validator(mode="after")
-    def validate_after(self):
+    def validate_after(self) -> File:
         match self.transfer_method:
             case FileTransferMethod.REMOTE_URL:
                 if not self.remote_url:
@@ -160,5 +174,5 @@ class File(BaseModel):
         return self._storage_key
 
     @storage_key.setter
-    def storage_key(self, value: str):
+    def storage_key(self, value: str) -> None:
         self._storage_key = value

api/core/workflow/file/protocols.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+from collections.abc import Generator
+from typing import Protocol
+
+
+class HttpResponseProtocol(Protocol):
+    """Subset of response behavior needed by workflow file helpers."""
+
+    @property
+    def content(self) -> bytes: ...
+
+    def raise_for_status(self) -> object: ...
+
+
+class WorkflowFileRuntimeProtocol(Protocol):
+    """Runtime dependencies required by ``core.workflow.file``.
+
+    Implementations are expected to be provided by integration layers (for example,
+    ``core.app.workflow.file_runtime``) so the workflow package avoids importing
+    application infrastructure modules directly.
+    """
+
+    @property
+    def files_url(self) -> str: ...
+
+    @property
+    def internal_files_url(self) -> str | None: ...
+
+    @property
+    def secret_key(self) -> str: ...
+
+    @property
+    def files_access_timeout(self) -> int: ...
+
+    @property
+    def multimodal_send_format(self) -> str: ...
+
+    def http_get(self, url: str, *, follow_redirects: bool = True) -> HttpResponseProtocol: ...
+
+    def storage_load(self, path: str, *, stream: bool = False) -> bytes | Generator: ...
+
+    def sign_tool_file(self, *, tool_file_id: str, extension: str, for_external: bool = True) -> str: ...

api/core/workflow/file/runtime.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+from collections.abc import Generator
+from typing import NoReturn
+
+from .protocols import HttpResponseProtocol, WorkflowFileRuntimeProtocol
+
+
+class WorkflowFileRuntimeNotConfiguredError(RuntimeError):
+    """Raised when workflow file runtime dependencies were not configured."""
+
+
+class _UnconfiguredWorkflowFileRuntime(WorkflowFileRuntimeProtocol):
+    def _raise(self) -> NoReturn:
+        raise WorkflowFileRuntimeNotConfiguredError(
+            "workflow file runtime is not configured, call set_workflow_file_runtime(...) first"
+        )
+
+    @property
+    def files_url(self) -> str:
+        self._raise()
+
+    @property
+    def internal_files_url(self) -> str | None:
+        self._raise()
+
+    @property
+    def secret_key(self) -> str:
+        self._raise()
+
+    @property
+    def files_access_timeout(self) -> int:
+        self._raise()
+
+    @property
+    def multimodal_send_format(self) -> str:
+        self._raise()
+
+    def http_get(self, url: str, *, follow_redirects: bool = True) -> HttpResponseProtocol:
+        self._raise()
+
+    def storage_load(self, path: str, *, stream: bool = False) -> bytes | Generator:
+        self._raise()
+
+    def sign_tool_file(self, *, tool_file_id: str, extension: str, for_external: bool = True) -> str:
+        self._raise()
+
+
+_runtime: WorkflowFileRuntimeProtocol = _UnconfiguredWorkflowFileRuntime()
+
+
+def set_workflow_file_runtime(runtime: WorkflowFileRuntimeProtocol) -> None:
+    global _runtime
+    _runtime = runtime
+
+
+def get_workflow_file_runtime() -> WorkflowFileRuntimeProtocol:
+    return _runtime

api/core/workflow/file/tool_file_parser.py

@@ -0,0 +1,9 @@
+from collections.abc import Callable
+from typing import Any
+
+_tool_file_manager_factory: Callable[[], Any] | None = None
+
+
+def set_tool_file_manager_factory(factory: Callable[[], Any]):
+    global _tool_file_manager_factory
+    _tool_file_manager_factory = factory

api/core/workflow/node_events/node.py

@@ -3,10 +3,10 @@ from datetime import datetime
 
 from pydantic import Field
 
-from core.file import File
 from core.model_runtime.entities.llm_entities import LLMUsage
 from core.rag.entities.citation_metadata import RetrievalSourceMetadata
 from core.workflow.entities.pause_reason import PauseReason
+from core.workflow.file import File
 from core.workflow.node_events import NodeRunResult
 
 from .base import NodeEventBase

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

@@ -11,7 +11,6 @@ from sqlalchemy.orm import Session
 
 from core.agent.entities import AgentToolEntity
 from core.agent.plugin_entities import AgentStrategyParameter
-from core.file import File, FileTransferMethod
 from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance, ModelManager
 from core.model_runtime.entities.llm_entities import LLMUsage, LLMUsageMetadata
@@ -33,6 +32,7 @@ from core.workflow.enums import (
     WorkflowNodeExecutionMetadataKey,
     WorkflowNodeExecutionStatus,
 )
+from core.workflow.file import File, FileTransferMethod
 from core.workflow.node_events import (
     AgentLogEvent,
     NodeEventBase,

api/core/workflow/nodes/datasource/datasource_node.py

@@ -14,13 +14,13 @@ from core.datasource.entities.datasource_entities import (
 from core.datasource.online_document.online_document_plugin import OnlineDocumentDatasourcePlugin
 from core.datasource.online_drive.online_drive_plugin import OnlineDriveDatasourcePlugin
 from core.datasource.utils.message_transformer import DatasourceFileMessageTransformer
-from core.file import File
-from core.file.enums import FileTransferMethod, FileType
 from core.plugin.impl.exc import PluginDaemonClientSideError
 from core.variables.segments import ArrayAnySegment
 from core.variables.variables import ArrayAnyVariable
 from core.workflow.entities.workflow_node_execution import WorkflowNodeExecutionStatus
 from core.workflow.enums import NodeExecutionType, NodeType, SystemVariableKey
+from core.workflow.file import File
+from core.workflow.file.enums import FileTransferMethod, FileType
 from core.workflow.node_events import NodeRunResult, StreamChunkEvent, StreamCompletedEvent
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.base.variable_template_parser import VariableTemplateParser

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

@@ -1,4 +1,4 @@
-from .entities import DocumentExtractorNodeData
+from .entities import DocumentExtractorNodeData, UnstructuredApiConfig
 from .node import DocumentExtractorNode
 
-__all__ = ["DocumentExtractorNode", "DocumentExtractorNodeData"]
+__all__ = ["DocumentExtractorNode", "DocumentExtractorNodeData", "UnstructuredApiConfig"]

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

@@ -1,7 +1,14 @@
 from collections.abc import Sequence
+from dataclasses import dataclass
 
 from core.workflow.nodes.base import BaseNodeData
 
 
 class DocumentExtractorNodeData(BaseNodeData):
     variable_selector: Sequence[str]
+
+
+@dataclass(frozen=True)
+class UnstructuredApiConfig:
+    api_url: str | None = None
+    api_key: str = ""

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

@@ -5,7 +5,7 @@ import logging
 import os
 import tempfile
 from collections.abc import Mapping, Sequence
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import charset_normalizer
 import docx
@@ -20,20 +20,23 @@ from docx.oxml.text.paragraph import CT_P
 from docx.table import Table
 from docx.text.paragraph import Paragraph
 
-from configs import dify_config
-from core.file import File, FileTransferMethod, file_manager
 from core.helper import ssrf_proxy
 from core.variables import ArrayFileSegment
 from core.variables.segments import ArrayStringSegment, FileSegment
 from core.workflow.enums import NodeType, WorkflowNodeExecutionStatus
+from core.workflow.file import File, FileTransferMethod, file_manager
 from core.workflow.node_events import NodeRunResult
 from core.workflow.nodes.base.node import Node
 
-from .entities import DocumentExtractorNodeData
+from .entities import DocumentExtractorNodeData, UnstructuredApiConfig
 from .exc import DocumentExtractorError, FileDownloadError, TextExtractionError, UnsupportedFileTypeError
 
 logger = logging.getLogger(__name__)
 
+if TYPE_CHECKING:
+    from core.workflow.entities import GraphInitParams
+    from core.workflow.runtime import GraphRuntimeState
+
 
 class DocumentExtractorNode(Node[DocumentExtractorNodeData]):
     """
@@ -47,6 +50,23 @@ class DocumentExtractorNode(Node[DocumentExtractorNodeData]):
     def version(cls) -> str:
         return "1"
 
+    def __init__(
+        self,
+        id: str,
+        config: Mapping[str, Any],
+        graph_init_params: "GraphInitParams",
+        graph_runtime_state: "GraphRuntimeState",
+        *,
+        unstructured_api_config: UnstructuredApiConfig | None = None,
+    ) -> None:
+        super().__init__(
+            id=id,
+            config=config,
+            graph_init_params=graph_init_params,
+            graph_runtime_state=graph_runtime_state,
+        )
+        self._unstructured_api_config = unstructured_api_config or UnstructuredApiConfig()
+
     def _run(self):
         variable_selector = self.node_data.variable_selector
         variable = self.graph_runtime_state.variable_pool.get(variable_selector)
@@ -64,7 +84,10 @@ class DocumentExtractorNode(Node[DocumentExtractorNodeData]):
 
         try:
             if isinstance(value, list):
-                extracted_text_list = list(map(_extract_text_from_file, value))
+                extracted_text_list = [
+                    _extract_text_from_file(file, unstructured_api_config=self._unstructured_api_config)
+                    for file in value
+                ]
                 return NodeRunResult(
                     status=WorkflowNodeExecutionStatus.SUCCEEDED,
                     inputs=inputs,
@@ -72,7 +95,7 @@ class DocumentExtractorNode(Node[DocumentExtractorNodeData]):
                     outputs={"text": ArrayStringSegment(value=extracted_text_list)},
                 )
             elif isinstance(value, File):
-                extracted_text = _extract_text_from_file(value)
+                extracted_text = _extract_text_from_file(value, unstructured_api_config=self._unstructured_api_config)
                 return NodeRunResult(
                     status=WorkflowNodeExecutionStatus.SUCCEEDED,
                     inputs=inputs,
@@ -103,7 +126,12 @@ class DocumentExtractorNode(Node[DocumentExtractorNodeData]):
         return {node_id + ".files": typed_node_data.variable_selector}
 
 
-def _extract_text_by_mime_type(*, file_content: bytes, mime_type: str) -> str:
+def _extract_text_by_mime_type(
+    *,
+    file_content: bytes,
+    mime_type: str,
+    unstructured_api_config: UnstructuredApiConfig,
+) -> str:
     """Extract text from a file based on its MIME type."""
     match mime_type:
         case "text/plain" | "text/html" | "text/htm" | "text/markdown" | "text/xml":
@@ -111,7 +139,7 @@ def _extract_text_by_mime_type(*, file_content: bytes, mime_type: str) -> str:
         case "application/pdf":
             return _extract_text_from_pdf(file_content)
         case "application/msword":
-            return _extract_text_from_doc(file_content)
+            return _extract_text_from_doc(file_content, unstructured_api_config=unstructured_api_config)
         case "application/vnd.openxmlformats-officedocument.wordprocessingml.document":
             return _extract_text_from_docx(file_content)
         case "text/csv":
@@ -119,11 +147,11 @@ def _extract_text_by_mime_type(*, file_content: bytes, mime_type: str) -> str:
         case "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" | "application/vnd.ms-excel":
             return _extract_text_from_excel(file_content)
         case "application/vnd.ms-powerpoint":
-            return _extract_text_from_ppt(file_content)
+            return _extract_text_from_ppt(file_content, unstructured_api_config=unstructured_api_config)
         case "application/vnd.openxmlformats-officedocument.presentationml.presentation":
-            return _extract_text_from_pptx(file_content)
+            return _extract_text_from_pptx(file_content, unstructured_api_config=unstructured_api_config)
         case "application/epub+zip":
-            return _extract_text_from_epub(file_content)
+            return _extract_text_from_epub(file_content, unstructured_api_config=unstructured_api_config)
         case "message/rfc822":
             return _extract_text_from_eml(file_content)
         case "application/vnd.ms-outlook":
@@ -140,7 +168,12 @@ def _extract_text_by_mime_type(*, file_content: bytes, mime_type: str) -> str:
             raise UnsupportedFileTypeError(f"Unsupported MIME type: {mime_type}")
 
 
-def _extract_text_by_file_extension(*, file_content: bytes, file_extension: str) -> str:
+def _extract_text_by_file_extension(
+    *,
+    file_content: bytes,
+    file_extension: str,
+    unstructured_api_config: UnstructuredApiConfig,
+) -> str:
     """Extract text from a file based on its file extension."""
     match file_extension:
         case (
@@ -203,7 +236,7 @@ def _extract_text_by_file_extension(*, file_content: bytes, file_extension: str)
         case ".pdf":
             return _extract_text_from_pdf(file_content)
         case ".doc":
-            return _extract_text_from_doc(file_content)
+            return _extract_text_from_doc(file_content, unstructured_api_config=unstructured_api_config)
         case ".docx":
             return _extract_text_from_docx(file_content)
         case ".csv":
@@ -211,11 +244,11 @@ def _extract_text_by_file_extension(*, file_content: bytes, file_extension: str)
         case ".xls" | ".xlsx":
             return _extract_text_from_excel(file_content)
         case ".ppt":
-            return _extract_text_from_ppt(file_content)
+            return _extract_text_from_ppt(file_content, unstructured_api_config=unstructured_api_config)
         case ".pptx":
-            return _extract_text_from_pptx(file_content)
+            return _extract_text_from_pptx(file_content, unstructured_api_config=unstructured_api_config)
         case ".epub":
-            return _extract_text_from_epub(file_content)
+            return _extract_text_from_epub(file_content, unstructured_api_config=unstructured_api_config)
         case ".eml":
             return _extract_text_from_eml(file_content)
         case ".msg":
@@ -312,14 +345,15 @@ def _extract_text_from_pdf(file_content: bytes) -> str:
         raise TextExtractionError(f"Failed to extract text from PDF: {str(e)}") from e
 
 
-def _extract_text_from_doc(file_content: bytes) -> str:
+def _extract_text_from_doc(file_content: bytes, *, unstructured_api_config: UnstructuredApiConfig) -> str:
     """
     Extract text from a DOC file.
     """
     from unstructured.partition.api import partition_via_api
 
-    if not dify_config.UNSTRUCTURED_API_URL:
-        raise TextExtractionError("UNSTRUCTURED_API_URL must be set")
+    if not unstructured_api_config.api_url:
+        raise TextExtractionError("Unstructured API URL is not configured for DOC file processing.")
+    api_key = unstructured_api_config.api_key or ""
 
     try:
         with tempfile.NamedTemporaryFile(suffix=".doc", delete=False) as temp_file:
@@ -329,8 +363,8 @@ def _extract_text_from_doc(file_content: bytes) -> str:
                 elements = partition_via_api(
                     file=file,
                     metadata_filename=temp_file.name,
-                    api_url=dify_config.UNSTRUCTURED_API_URL,
-                    api_key=dify_config.UNSTRUCTURED_API_KEY,  # type: ignore
+                    api_url=unstructured_api_config.api_url,
+                    api_key=api_key,
                 )
             os.unlink(temp_file.name)
         return "\n".join([getattr(element, "text", "") for element in elements])
@@ -420,12 +454,20 @@ def _download_file_content(file: File) -> bytes:
         raise FileDownloadError(f"Error downloading file: {str(e)}") from e
 
 
-def _extract_text_from_file(file: File):
+def _extract_text_from_file(file: File, *, unstructured_api_config: UnstructuredApiConfig) -> str:
     file_content = _download_file_content(file)
     if file.extension:
-        extracted_text = _extract_text_by_file_extension(file_content=file_content, file_extension=file.extension)
+        extracted_text = _extract_text_by_file_extension(
+            file_content=file_content,
+            file_extension=file.extension,
+            unstructured_api_config=unstructured_api_config,
+        )
     elif file.mime_type:
-        extracted_text = _extract_text_by_mime_type(file_content=file_content, mime_type=file.mime_type)
+        extracted_text = _extract_text_by_mime_type(
+            file_content=file_content,
+            mime_type=file.mime_type,
+            unstructured_api_config=unstructured_api_config,
+        )
     else:
         raise UnsupportedFileTypeError("Unable to determine file type: MIME type or file extension is missing")
     return extracted_text
@@ -517,12 +559,14 @@ def _extract_text_from_excel(file_content: bytes) -> str:
         raise TextExtractionError(f"Failed to extract text from Excel file: {str(e)}") from e
 
 
-def _extract_text_from_ppt(file_content: bytes) -> str:
+def _extract_text_from_ppt(file_content: bytes, *, unstructured_api_config: UnstructuredApiConfig) -> str:
     from unstructured.partition.api import partition_via_api
     from unstructured.partition.ppt import partition_ppt
 
+    api_key = unstructured_api_config.api_key or ""
+
     try:
-        if dify_config.UNSTRUCTURED_API_URL:
+        if unstructured_api_config.api_url:
             with tempfile.NamedTemporaryFile(suffix=".ppt", delete=False) as temp_file:
                 temp_file.write(file_content)
                 temp_file.flush()
@@ -530,8 +574,8 @@ def _extract_text_from_ppt(file_content: bytes) -> str:
                     elements = partition_via_api(
                         file=file,
                         metadata_filename=temp_file.name,
-                        api_url=dify_config.UNSTRUCTURED_API_URL,
-                        api_key=dify_config.UNSTRUCTURED_API_KEY,  # type: ignore
+                        api_url=unstructured_api_config.api_url,
+                        api_key=api_key,
                     )
                 os.unlink(temp_file.name)
         else:
@@ -543,12 +587,14 @@ def _extract_text_from_ppt(file_content: bytes) -> str:
         raise TextExtractionError(f"Failed to extract text from PPTX: {str(e)}") from e
 
 
-def _extract_text_from_pptx(file_content: bytes) -> str:
+def _extract_text_from_pptx(file_content: bytes, *, unstructured_api_config: UnstructuredApiConfig) -> str:
     from unstructured.partition.api import partition_via_api
     from unstructured.partition.pptx import partition_pptx
 
+    api_key = unstructured_api_config.api_key or ""
+
     try:
-        if dify_config.UNSTRUCTURED_API_URL:
+        if unstructured_api_config.api_url:
             with tempfile.NamedTemporaryFile(suffix=".pptx", delete=False) as temp_file:
                 temp_file.write(file_content)
                 temp_file.flush()
@@ -556,8 +602,8 @@ def _extract_text_from_pptx(file_content: bytes) -> str:
                     elements = partition_via_api(
                         file=file,
                         metadata_filename=temp_file.name,
-                        api_url=dify_config.UNSTRUCTURED_API_URL,
-                        api_key=dify_config.UNSTRUCTURED_API_KEY,  # type: ignore
+                        api_url=unstructured_api_config.api_url,
+                        api_key=api_key,
                     )
                 os.unlink(temp_file.name)
         else:
@@ -568,12 +614,14 @@ def _extract_text_from_pptx(file_content: bytes) -> str:
         raise TextExtractionError(f"Failed to extract text from PPTX: {str(e)}") from e
 
 
-def _extract_text_from_epub(file_content: bytes) -> str:
+def _extract_text_from_epub(file_content: bytes, *, unstructured_api_config: UnstructuredApiConfig) -> str:
     from unstructured.partition.api import partition_via_api
     from unstructured.partition.epub import partition_epub
 
+    api_key = unstructured_api_config.api_key or ""
+
     try:
-        if dify_config.UNSTRUCTURED_API_URL:
+        if unstructured_api_config.api_url:
             with tempfile.NamedTemporaryFile(suffix=".epub", delete=False) as temp_file:
                 temp_file.write(file_content)
                 temp_file.flush()
@@ -581,8 +629,8 @@ def _extract_text_from_epub(file_content: bytes) -> str:
                     elements = partition_via_api(
                         file=file,
                         metadata_filename=temp_file.name,
-                        api_url=dify_config.UNSTRUCTURED_API_URL,
-                        api_key=dify_config.UNSTRUCTURED_API_KEY,  # type: ignore
+                        api_url=unstructured_api_config.api_url,
+                        api_key=api_key,
                     )
                 os.unlink(temp_file.name)
         else:

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

@@ -1,4 +1,22 @@
-from .entities import BodyData, HttpRequestNodeAuthorization, HttpRequestNodeBody, HttpRequestNodeData
+from .config import build_http_request_config, resolve_http_request_config
+from .entities import (
+    HTTP_REQUEST_CONFIG_FILTER_KEY,
+    BodyData,
+    HttpRequestNodeAuthorization,
+    HttpRequestNodeBody,
+    HttpRequestNodeConfig,
+    HttpRequestNodeData,
+)
 from .node import HttpRequestNode
 
-__all__ = ["BodyData", "HttpRequestNode", "HttpRequestNodeAuthorization", "HttpRequestNodeBody", "HttpRequestNodeData"]
+__all__ = [
+    "HTTP_REQUEST_CONFIG_FILTER_KEY",
+    "BodyData",
+    "HttpRequestNode",
+    "HttpRequestNodeAuthorization",
+    "HttpRequestNodeBody",
+    "HttpRequestNodeConfig",
+    "HttpRequestNodeData",
+    "build_http_request_config",
+    "resolve_http_request_config",
+]

api/core/workflow/nodes/http_request/config.py

@@ -0,0 +1,33 @@
+from collections.abc import Mapping
+
+from .entities import HTTP_REQUEST_CONFIG_FILTER_KEY, HttpRequestNodeConfig
+
+
+def build_http_request_config(
+    *,
+    max_connect_timeout: int = 10,
+    max_read_timeout: int = 600,
+    max_write_timeout: int = 600,
+    max_binary_size: int = 10 * 1024 * 1024,
+    max_text_size: int = 1 * 1024 * 1024,
+    ssl_verify: bool = True,
+    ssrf_default_max_retries: int = 3,
+) -> HttpRequestNodeConfig:
+    return HttpRequestNodeConfig(
+        max_connect_timeout=max_connect_timeout,
+        max_read_timeout=max_read_timeout,
+        max_write_timeout=max_write_timeout,
+        max_binary_size=max_binary_size,
+        max_text_size=max_text_size,
+        ssl_verify=ssl_verify,
+        ssrf_default_max_retries=ssrf_default_max_retries,
+    )
+
+
+def resolve_http_request_config(filters: Mapping[str, object] | None) -> HttpRequestNodeConfig:
+    if not filters:
+        raise ValueError("http_request_config is required to build HTTP request default config")
+    config = filters.get(HTTP_REQUEST_CONFIG_FILTER_KEY)
+    if not isinstance(config, HttpRequestNodeConfig):
+        raise ValueError("http_request_config must be an HttpRequestNodeConfig instance")
+    return config

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

@@ -1,5 +1,6 @@
 import mimetypes
 from collections.abc import Sequence
+from dataclasses import dataclass
 from email.message import Message
 from typing import Any, Literal
 
@@ -7,9 +8,10 @@ import charset_normalizer
 import httpx
 from pydantic import BaseModel, Field, ValidationInfo, field_validator
 
-from configs import dify_config
 from core.workflow.nodes.base import BaseNodeData
 
+HTTP_REQUEST_CONFIG_FILTER_KEY = "http_request_config"
+
 
 class HttpRequestNodeAuthorizationConfig(BaseModel):
     type: Literal["basic", "bearer", "custom"]
@@ -59,9 +61,27 @@ class HttpRequestNodeBody(BaseModel):
 
 
 class HttpRequestNodeTimeout(BaseModel):
-    connect: int = dify_config.HTTP_REQUEST_MAX_CONNECT_TIMEOUT
-    read: int = dify_config.HTTP_REQUEST_MAX_READ_TIMEOUT
-    write: int = dify_config.HTTP_REQUEST_MAX_WRITE_TIMEOUT
+    connect: int | None = None
+    read: int | None = None
+    write: int | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class HttpRequestNodeConfig:
+    max_connect_timeout: int
+    max_read_timeout: int
+    max_write_timeout: int
+    max_binary_size: int
+    max_text_size: int
+    ssl_verify: bool
+    ssrf_default_max_retries: int
+
+    def default_timeout(self) -> "HttpRequestNodeTimeout":
+        return HttpRequestNodeTimeout(
+            connect=self.max_connect_timeout,
+            read=self.max_read_timeout,
+            write=self.max_write_timeout,
+        )
 
 
 class HttpRequestNodeData(BaseNodeData):
@@ -91,7 +111,7 @@ class HttpRequestNodeData(BaseNodeData):
     params: str
     body: HttpRequestNodeBody | None = None
     timeout: HttpRequestNodeTimeout | None = None
-    ssl_verify: bool | None = dify_config.HTTP_REQUEST_NODE_SSL_VERIFY
+    ssl_verify: bool | None = None
 
 
 class Response:

api/core/workflow/nodes/http_request/executor.py

@@ -10,16 +10,16 @@ from urllib.parse import urlencode, urlparse
 import httpx
 from json_repair import repair_json
 
-from configs import dify_config
-from core.file.enums import FileTransferMethod
-from core.file.file_manager import file_manager as default_file_manager
 from core.helper.ssrf_proxy import ssrf_proxy
 from core.variables.segments import ArrayFileSegment, FileSegment
+from core.workflow.file.enums import FileTransferMethod
+from core.workflow.file.file_manager import file_manager as default_file_manager
 from core.workflow.runtime import VariablePool
 
 from ..protocols import FileManagerProtocol, HttpClientProtocol
 from .entities import (
     HttpRequestNodeAuthorization,
+    HttpRequestNodeConfig,
     HttpRequestNodeData,
     HttpRequestNodeTimeout,
     Response,
@@ -78,10 +78,13 @@ class Executor:
         node_data: HttpRequestNodeData,
         timeout: HttpRequestNodeTimeout,
         variable_pool: VariablePool,
-        max_retries: int = dify_config.SSRF_DEFAULT_MAX_RETRIES,
+        http_request_config: HttpRequestNodeConfig,
+        max_retries: int | None = None,
+        ssl_verify: bool | None = None,
         http_client: HttpClientProtocol | None = None,
         file_manager: FileManagerProtocol | None = None,
     ):
+        self._http_request_config = http_request_config
         # If authorization API key is present, convert the API key using the variable pool
         if node_data.authorization.type == "api-key":
             if node_data.authorization.config is None:
@@ -99,14 +102,20 @@ class Executor:
         self.method = node_data.method
         self.auth = node_data.authorization
         self.timeout = timeout
-        self.ssl_verify = node_data.ssl_verify
+        self.ssl_verify = ssl_verify if ssl_verify is not None else node_data.ssl_verify
+        if self.ssl_verify is None:
+            self.ssl_verify = self._http_request_config.ssl_verify
+        if not isinstance(self.ssl_verify, bool):
+            raise ValueError("ssl_verify must be a boolean")
         self.params = None
         self.headers = {}
         self.content = None
         self.files = None
         self.data = None
         self.json = None
-        self.max_retries = max_retries
+        self.max_retries = (
+            max_retries if max_retries is not None else self._http_request_config.ssrf_default_max_retries
+        )
         self._http_client = http_client or ssrf_proxy
         self._file_manager = file_manager or default_file_manager
 
@@ -319,9 +328,9 @@ class Executor:
         executor_response = Response(response)
 
         threshold_size = (
-            dify_config.HTTP_REQUEST_NODE_MAX_BINARY_SIZE
+            self._http_request_config.max_binary_size
             if executor_response.is_file
-            else dify_config.HTTP_REQUEST_NODE_MAX_TEXT_SIZE
+            else self._http_request_config.max_text_size
         )
         if executor_response.size > threshold_size:
             raise ResponseSizeError(
@@ -366,7 +375,9 @@ class Executor:
                 **request_args,
                 max_retries=self.max_retries,
             )
-        except (self._http_client.max_retries_exceeded_error, self._http_client.request_error) as e:
+        except self._http_client.max_retries_exceeded_error as e:
+            raise HttpRequestNodeError(f"Reached maximum retries for URL {self.url}") from e
+        except self._http_client.request_error as e:
             raise HttpRequestNodeError(str(e)) from e
         return response
 

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

@@ -3,13 +3,12 @@ import mimetypes
 from collections.abc import Callable, Mapping, Sequence
 from typing import TYPE_CHECKING, Any
 
-from configs import dify_config
-from core.file import File, FileTransferMethod
-from core.file.file_manager import file_manager as default_file_manager
 from core.helper.ssrf_proxy import ssrf_proxy
 from core.tools.tool_file_manager import ToolFileManager
 from core.variables.segments import ArrayFileSegment
 from core.workflow.enums import NodeType, WorkflowNodeExecutionStatus
+from core.workflow.file import File, FileTransferMethod
+from core.workflow.file.file_manager import file_manager as default_file_manager
 from core.workflow.node_events import NodeRunResult
 from core.workflow.nodes.base import variable_template_parser
 from core.workflow.nodes.base.entities import VariableSelector
@@ -18,19 +17,16 @@ from core.workflow.nodes.http_request.executor import Executor
 from core.workflow.nodes.protocols import FileManagerProtocol, HttpClientProtocol
 from factories import file_factory
 
+from .config import build_http_request_config, resolve_http_request_config
 from .entities import (
+    HTTP_REQUEST_CONFIG_FILTER_KEY,
+    HttpRequestNodeConfig,
     HttpRequestNodeData,
     HttpRequestNodeTimeout,
     Response,
 )
 from .exc import HttpRequestNodeError, RequestBodyError
 
-HTTP_REQUEST_DEFAULT_TIMEOUT = HttpRequestNodeTimeout(
-    connect=dify_config.HTTP_REQUEST_MAX_CONNECT_TIMEOUT,
-    read=dify_config.HTTP_REQUEST_MAX_READ_TIMEOUT,
-    write=dify_config.HTTP_REQUEST_MAX_WRITE_TIMEOUT,
-)
-
 logger = logging.getLogger(__name__)
 
 if TYPE_CHECKING:
@@ -48,6 +44,7 @@ class HttpRequestNode(Node[HttpRequestNodeData]):
         graph_init_params: "GraphInitParams",
         graph_runtime_state: "GraphRuntimeState",
         *,
+        http_request_config: HttpRequestNodeConfig,
         http_client: HttpClientProtocol | None = None,
         tool_file_manager_factory: Callable[[], ToolFileManager] = ToolFileManager,
         file_manager: FileManagerProtocol | None = None,
@@ -58,12 +55,18 @@ class HttpRequestNode(Node[HttpRequestNodeData]):
             graph_init_params=graph_init_params,
             graph_runtime_state=graph_runtime_state,
         )
+        self._http_request_config = http_request_config
         self._http_client = http_client or ssrf_proxy
         self._tool_file_manager_factory = tool_file_manager_factory
         self._file_manager = file_manager or default_file_manager
 
     @classmethod
     def get_default_config(cls, filters: Mapping[str, object] | None = None) -> Mapping[str, object]:
+        if not filters or HTTP_REQUEST_CONFIG_FILTER_KEY not in filters:
+            http_request_config = build_http_request_config()
+        else:
+            http_request_config = resolve_http_request_config(filters)
+        default_timeout = http_request_config.default_timeout()
         return {
             "type": "http-request",
             "config": {
@@ -73,15 +76,15 @@ class HttpRequestNode(Node[HttpRequestNodeData]):
                 },
                 "body": {"type": "none"},
                 "timeout": {
-                    **HTTP_REQUEST_DEFAULT_TIMEOUT.model_dump(),
-                    "max_connect_timeout": dify_config.HTTP_REQUEST_MAX_CONNECT_TIMEOUT,
-                    "max_read_timeout": dify_config.HTTP_REQUEST_MAX_READ_TIMEOUT,
-                    "max_write_timeout": dify_config.HTTP_REQUEST_MAX_WRITE_TIMEOUT,
+                    **default_timeout.model_dump(),
+                    "max_connect_timeout": http_request_config.max_connect_timeout,
+                    "max_read_timeout": http_request_config.max_read_timeout,
+                    "max_write_timeout": http_request_config.max_write_timeout,
                 },
-                "ssl_verify": dify_config.HTTP_REQUEST_NODE_SSL_VERIFY,
+                "ssl_verify": http_request_config.ssl_verify,
             },
             "retry_config": {
-                "max_retries": dify_config.SSRF_DEFAULT_MAX_RETRIES,
+                "max_retries": http_request_config.ssrf_default_max_retries,
                 "retry_interval": 0.5 * (2**2),
                 "retry_enabled": True,
             },
@@ -98,7 +101,9 @@ class HttpRequestNode(Node[HttpRequestNodeData]):
                 node_data=self.node_data,
                 timeout=self._get_request_timeout(self.node_data),
                 variable_pool=self.graph_runtime_state.variable_pool,
+                http_request_config=self._http_request_config,
                 max_retries=0,
+                ssl_verify=self.node_data.ssl_verify,
                 http_client=self._http_client,
                 file_manager=self._file_manager,
             )
@@ -142,16 +147,17 @@ class HttpRequestNode(Node[HttpRequestNodeData]):
                 error_type=type(e).__name__,
             )
 
-    @staticmethod
-    def _get_request_timeout(node_data: HttpRequestNodeData) -> HttpRequestNodeTimeout:
+    def _get_request_timeout(self, node_data: HttpRequestNodeData) -> HttpRequestNodeTimeout:
+        default_timeout = self._http_request_config.default_timeout()
         timeout = node_data.timeout
         if timeout is None:
-            return HTTP_REQUEST_DEFAULT_TIMEOUT
+            return default_timeout
 
-        timeout.connect = timeout.connect or HTTP_REQUEST_DEFAULT_TIMEOUT.connect
-        timeout.read = timeout.read or HTTP_REQUEST_DEFAULT_TIMEOUT.read
-        timeout.write = timeout.write or HTTP_REQUEST_DEFAULT_TIMEOUT.write
-        return timeout
+        return HttpRequestNodeTimeout(
+            connect=timeout.connect or default_timeout.connect,
+            read=timeout.read or default_timeout.read,
+            write=timeout.write or default_timeout.write,
+        )
 
     @classmethod
     def _extract_variable_selector_to_variable_mapping(

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

@@ -30,7 +30,7 @@ from .exc import (
 )
 
 if TYPE_CHECKING:
-    from core.file.models import File
+    from core.workflow.file.models import File
     from core.workflow.runtime import GraphRuntimeState
 
 logger = logging.getLogger(__name__)

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

@@ -1,10 +1,10 @@
 from collections.abc import Callable, Sequence
 from typing import Any, TypeAlias, TypeVar
 
-from core.file import File
 from core.variables import ArrayFileSegment, ArrayNumberSegment, ArrayStringSegment
 from core.variables.segments import ArrayAnySegment, ArrayBooleanSegment, ArraySegment
 from core.workflow.enums import NodeType, WorkflowNodeExecutionStatus
+from core.workflow.file import File
 from core.workflow.node_events import NodeRunResult
 from core.workflow.nodes.base.node import Node
 

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

@@ -4,10 +4,10 @@ import typing as tp
 from sqlalchemy import Engine
 
 from constants.mimetypes import DEFAULT_EXTENSION, DEFAULT_MIME_TYPE
-from core.file import File, FileTransferMethod, FileType
 from core.helper import ssrf_proxy
 from core.tools.signature import sign_tool_file
 from core.tools.tool_file_manager import ToolFileManager
+from core.workflow.file import File, FileTransferMethod, FileType
 from extensions.ext_database import db as global_db
 
 

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

@@ -7,7 +7,6 @@ from sqlalchemy.orm import Session
 from configs import dify_config
 from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
 from core.entities.provider_entities import ProviderQuotaType, QuotaUnit
-from core.file.models import File
 from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance, ModelManager
 from core.model_runtime.entities.llm_entities import LLMUsage
@@ -16,6 +15,7 @@ from core.model_runtime.model_providers.__base.large_language_model import Large
 from core.prompt.entities.advanced_prompt_entities import MemoryConfig
 from core.variables.segments import ArrayAnySegment, ArrayFileSegment, FileSegment, NoneSegment, StringSegment
 from core.workflow.enums import SystemVariableKey
+from core.workflow.file.models import File
 from core.workflow.nodes.llm.entities import ModelConfig
 from core.workflow.runtime import VariablePool
 from extensions.ext_database import db

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

@@ -12,7 +12,6 @@ from typing import TYPE_CHECKING, Any, Literal
 from sqlalchemy import select
 
 from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
-from core.file import File, FileTransferMethod, FileType, file_manager
 from core.helper.code_executor import CodeExecutor, CodeLanguage
 from core.llm_generator.output_parser.errors import OutputParserError
 from core.llm_generator.output_parser.structured_output import invoke_llm_with_structured_output
@@ -65,6 +64,7 @@ from core.workflow.enums import (
     WorkflowNodeExecutionMetadataKey,
     WorkflowNodeExecutionStatus,
 )
+from core.workflow.file import File, FileTransferMethod, FileType, file_manager
 from core.workflow.node_events import (
     ModelInvokeCompletedEvent,
     NodeEventBase,
@@ -101,7 +101,7 @@ from .exc import (
 from .file_saver import FileSaverImpl, LLMFileSaver
 
 if TYPE_CHECKING:
-    from core.file.models import File
+    from core.workflow.file.models import File
     from core.workflow.runtime import GraphRuntimeState
 
 logger = logging.getLogger(__name__)

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

@@ -71,9 +71,9 @@ class LoopNode(LLMUsageTrackingMixin, Node[LoopNodeData]):
         if self.node_data.loop_variables:
             value_processor: dict[Literal["constant", "variable"], Callable[[LoopVariableData], Segment | None]] = {
                 "constant": lambda var: self._get_segment_for_constant(var.var_type, var.value),
-                "variable": lambda var: self.graph_runtime_state.variable_pool.get(var.value)
-                if isinstance(var.value, list)
-                else None,
+                "variable": lambda var: (
+                    self.graph_runtime_state.variable_pool.get(var.value) if isinstance(var.value, list) else None
+                ),
             }
             for loop_variable in self.node_data.loop_variables:
                 if loop_variable.value_type not in value_processor:

api/core/workflow/nodes/parameter_extractor/parameter_extractor_node.py

@@ -6,7 +6,6 @@ from collections.abc import Mapping, Sequence
 from typing import Any, cast
 
 from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
-from core.file import File
 from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance
 from core.model_runtime.entities import ImagePromptMessageContent
@@ -28,6 +27,7 @@ from core.prompt.simple_prompt_transform import ModelMode
 from core.prompt.utils.prompt_message_util import PromptMessageUtil
 from core.variables.types import ArrayValidation, SegmentType
 from core.workflow.enums import NodeType, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
+from core.workflow.file import File
 from core.workflow.node_events import NodeRunResult
 from core.workflow.nodes.base import variable_template_parser
 from core.workflow.nodes.base.node import Node

api/core/workflow/nodes/protocols.py

@@ -2,7 +2,7 @@ from typing import Any, Protocol
 
 import httpx
 
-from core.file import File
+from core.workflow.file import File
 
 
 class HttpClientProtocol(Protocol):

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

@@ -39,7 +39,7 @@ from .template_prompts import (
 )
 
 if TYPE_CHECKING:
-    from core.file.models import File
+    from core.workflow.file.models import File
     from core.workflow.runtime import GraphRuntimeState