feat: oauth config init

- Introduced `TriggerProviderInfoApi` to retrieve information for a specific trigger provider, improving API capabilities.
- Added `get_trigger_provider` method in `TriggerProviderService` to fetch trigger provider details, enhancing data retrieval.
- Updated route configurations to include the new API endpoint for trigger provider information.

These changes enhance the functionality and usability of trigger provider interactions within the application.

.devcontainer/post_create_command.sh

@@ -1,5 +1,6 @@
 #!/bin/bash
 
+npm add -g pnpm@10.15.0
 corepack enable
 cd web && pnpm install
 pipx install uv

.github/workflows/autofix.yml

@@ -2,6 +2,8 @@ name: autofix.ci
 on:
   pull_request:
     branches: ["main"]
+  push:
+    branches: ["main"]
 permissions:
   contents: read
 

.github/workflows/style.yml

@@ -89,7 +89,9 @@ jobs:
       - name: Web style check
         if: steps.changed-files.outputs.any_changed == 'true'
         working-directory: ./web
-        run: pnpm run lint
+        run: |
+          pnpm run lint
+          pnpm run eslint
 
   docker-compose-template:
     name: Docker Compose Template

.gitignore

@@ -218,3 +218,6 @@ mise.toml
 .roo/
 api/.env.backup
 /clickzetta
+
+# mcp
+.serena

CLAUDE.md

@@ -59,6 +59,7 @@ pnpm test                         # Run Jest tests
 - Use type hints for all functions and class attributes
 - No `Any` types unless absolutely necessary
 - Implement special methods (`__repr__`, `__str__`) appropriately
+- **Logging**: Never use `str(e)` in `logger.exception()` calls. Use `logger.exception("message", exc_info=e)` instead
 
 ### TypeScript/JavaScript
 

api/.env.example

@@ -434,6 +434,9 @@ HTTP_REQUEST_NODE_MAX_BINARY_SIZE=10485760
 HTTP_REQUEST_NODE_MAX_TEXT_SIZE=1048576
 HTTP_REQUEST_NODE_SSL_VERIFY=True
 
+# Webhook request configuration
+WEBHOOK_REQUEST_BODY_MAX_SIZE=10485760
+
 # Respect X-* headers to redirect clients
 RESPECT_XFORWARD_HEADERS_ENABLED=false
 
@@ -460,16 +463,6 @@ WORKFLOW_CALL_MAX_DEPTH=5
 WORKFLOW_PARALLEL_DEPTH_LIMIT=3
 MAX_VARIABLE_SIZE=204800
 
-# GraphEngine Worker Pool Configuration
-# Minimum number of workers per GraphEngine instance (default: 1)
-GRAPH_ENGINE_MIN_WORKERS=1
-# Maximum number of workers per GraphEngine instance (default: 10)
-GRAPH_ENGINE_MAX_WORKERS=10
-# Queue depth threshold that triggers worker scale up (default: 3)
-GRAPH_ENGINE_SCALE_UP_THRESHOLD=3
-# Seconds of idle time before scaling down workers (default: 5.0)
-GRAPH_ENGINE_SCALE_DOWN_IDLE_TIME=5.0
-
 # Workflow storage configuration
 # Options: rdbms, hybrid
 # rdbms: Use only the relational database (default)
@@ -512,6 +505,12 @@ ENABLE_CLEAN_MESSAGES=false
 ENABLE_MAIL_CLEAN_DOCUMENT_NOTIFY_TASK=false
 ENABLE_DATASETS_QUEUE_MONITOR=false
 ENABLE_CHECK_UPGRADABLE_PLUGIN_TASK=true
+ENABLE_WORKFLOW_SCHEDULE_POLLER_TASK=true
+# Interval time in minutes for polling scheduled workflows(default: 1 min)
+WORKFLOW_SCHEDULE_POLLER_INTERVAL=1
+WORKFLOW_SCHEDULE_POLLER_BATCH_SIZE=100
+# Maximum number of scheduled workflows to dispatch per tick (0 for unlimited)
+WORKFLOW_SCHEDULE_MAX_DISPATCH_PER_TICK=0
 
 # Position configuration
 POSITION_TOOL_PINS=

api/.importlinter

@@ -1,112 +0,0 @@
-[importlinter]
-root_packages =
-    core
-    configs
-    controllers
-    models
-    tasks
-    services
-
-[importlinter:contract:workflow]
-name = Workflow
-type=layers
-layers =
-    graph_engine
-    graph_events
-    graph
-    nodes
-    node_events
-    entities
-containers =
-    core.workflow
-ignore_imports =
-    core.workflow.nodes.base.node -> core.workflow.graph_events
-    core.workflow.nodes.iteration.iteration_node -> core.workflow.graph_events
-    core.workflow.nodes.iteration.iteration_node -> core.workflow.graph_engine
-    core.workflow.nodes.iteration.iteration_node -> core.workflow.graph
-    core.workflow.nodes.iteration.iteration_node -> core.workflow.graph_engine.command_channels
-    core.workflow.nodes.loop.loop_node -> core.workflow.graph_events
-    core.workflow.nodes.loop.loop_node -> core.workflow.graph_engine
-    core.workflow.nodes.loop.loop_node -> core.workflow.graph
-    core.workflow.nodes.loop.loop_node -> core.workflow.graph_engine.command_channels
-    core.workflow.nodes.node_factory -> core.workflow.graph
-
-[importlinter:contract:rsc]
-name = RSC
-type = layers
-layers =
-    graph_engine
-    response_coordinator
-containers =
-    core.workflow.graph_engine
-
-[importlinter:contract:worker]
-name = Worker
-type = layers
-layers =
-    graph_engine
-    worker
-containers =
-    core.workflow.graph_engine
-
-[importlinter:contract:graph-engine-architecture]
-name = Graph Engine Architecture
-type = layers
-layers =
-    graph_engine
-    orchestration
-    command_processing
-    event_management
-    error_handling
-    graph_traversal
-    state_management
-    worker_management
-    domain
-containers =
-    core.workflow.graph_engine
-
-[importlinter:contract:domain-isolation]
-name = Domain Model Isolation
-type = forbidden
-source_modules =
-    core.workflow.graph_engine.domain
-forbidden_modules =
-    core.workflow.graph_engine.worker_management
-    core.workflow.graph_engine.command_channels
-    core.workflow.graph_engine.layers
-    core.workflow.graph_engine.protocols
-
-[importlinter:contract:worker-management]
-name = Worker Management
-type = forbidden
-source_modules =
-    core.workflow.graph_engine.worker_management
-forbidden_modules =
-    core.workflow.graph_engine.orchestration
-    core.workflow.graph_engine.command_processing
-    core.workflow.graph_engine.event_management
-
-[importlinter:contract:error-handling-strategies]
-name = Error Handling Strategies
-type = independence
-modules =
-    core.workflow.graph_engine.error_handling.abort_strategy
-    core.workflow.graph_engine.error_handling.retry_strategy
-    core.workflow.graph_engine.error_handling.fail_branch_strategy
-    core.workflow.graph_engine.error_handling.default_value_strategy
-
-[importlinter:contract:graph-traversal-components]
-name = Graph Traversal Components
-type = layers
-layers =
-    edge_processor
-    skip_propagator
-containers =
-    core.workflow.graph_engine.graph_traversal
-
-[importlinter:contract:command-channels]
-name = Command Channels Independence
-type = independence
-modules =
-    core.workflow.graph_engine.command_channels.in_memory_channel
-    core.workflow.graph_engine.command_channels.redis_channel

api/.vscode/launch.json.example

@@ -54,7 +54,7 @@
                 "--loglevel",
                 "DEBUG",
                 "-Q",
-                "dataset,generation,mail,ops_trace,app_deletion"
+                "dataset,generation,mail,ops_trace,app_deletion,workflow"
             ]
         }
     ]

api/commands.py

@@ -13,6 +13,7 @@ from sqlalchemy.exc import SQLAlchemyError
 
 from configs import dify_config
 from constants.languages import languages
+from core.plugin.entities.plugin import ToolProviderID
 from core.rag.datasource.vdb.vector_factory import Vector
 from core.rag.datasource.vdb.vector_type import VectorType
 from core.rag.index_processor.constant.built_in_field import BuiltInField
@@ -30,7 +31,6 @@ from models.dataset import Dataset, DatasetCollectionBinding, DatasetMetadata, D
 from models.dataset import Document as DatasetDocument
 from models.model import Account, App, AppAnnotationSetting, AppMode, Conversation, MessageAnnotation
 from models.provider import Provider, ProviderModel
-from models.provider_ids import ToolProviderID
 from models.tools import ToolOAuthSystemClient
 from services.account_service import AccountService, RegisterService, TenantService
 from services.clear_free_plan_tenant_expired_logs import ClearFreePlanTenantExpiredLogs
@@ -1207,6 +1207,55 @@ def setup_system_tool_oauth_client(provider, client_params):
     click.echo(click.style(f"OAuth client params setup successfully. id: {oauth_client.id}", fg="green"))
 
 
+@click.command("setup-system-trigger-oauth-client", help="Setup system trigger oauth client.")
+@click.option("--provider", prompt=True, help="Provider name")
+@click.option("--client-params", prompt=True, help="Client Params")
+def setup_system_trigger_oauth_client(provider, client_params):
+    """
+    Setup system trigger oauth client
+    """
+    from core.plugin.entities.plugin import TriggerProviderID
+    from models.trigger import TriggerOAuthSystemClient
+
+    provider_id = TriggerProviderID(provider)
+    provider_name = provider_id.provider_name
+    plugin_id = provider_id.plugin_id
+
+    try:
+        # json validate
+        click.echo(click.style(f"Validating client params: {client_params}", fg="yellow"))
+        client_params_dict = TypeAdapter(dict[str, Any]).validate_json(client_params)
+        click.echo(click.style("Client params validated successfully.", fg="green"))
+
+        click.echo(click.style(f"Encrypting client params: {client_params}", fg="yellow"))
+        click.echo(click.style(f"Using SECRET_KEY: `{dify_config.SECRET_KEY}`", fg="yellow"))
+        oauth_client_params = encrypt_system_oauth_params(client_params_dict)
+        click.echo(click.style("Client params encrypted successfully.", fg="green"))
+    except Exception as e:
+        click.echo(click.style(f"Error parsing client params: {str(e)}", fg="red"))
+        return
+
+    deleted_count = (
+        db.session.query(TriggerOAuthSystemClient)
+        .filter_by(
+            provider=provider_name,
+            plugin_id=plugin_id,
+        )
+        .delete()
+    )
+    if deleted_count > 0:
+        click.echo(click.style(f"Deleted {deleted_count} existing oauth client params.", fg="yellow"))
+
+    oauth_client = TriggerOAuthSystemClient(
+        provider=provider_name,
+        plugin_id=plugin_id,
+        encrypted_oauth_params=oauth_client_params,
+    )
+    db.session.add(oauth_client)
+    db.session.commit()
+    click.echo(click.style(f"OAuth client params setup successfully. id: {oauth_client.id}", fg="green"))
+
+
 def _find_orphaned_draft_variables(batch_size: int = 1000) -> list[str]:
     """
     Find draft variables that reference non-existent apps.

api/configs/feature/__init__.py

@@ -147,6 +147,17 @@ class CodeExecutionSandboxConfig(BaseSettings):
     )
 
 
+class TriggerConfig(BaseSettings):
+    """
+    Configuration for trigger
+    """
+
+    WEBHOOK_REQUEST_BODY_MAX_SIZE: PositiveInt = Field(
+        description="Maximum allowed size for webhook request bodies in bytes",
+        default=10485760,
+    )
+
+
 class PluginConfig(BaseSettings):
     """
     Plugin configs
@@ -529,28 +540,6 @@ class WorkflowConfig(BaseSettings):
         default=200 * 1024,
     )
 
-    # GraphEngine Worker Pool Configuration
-    GRAPH_ENGINE_MIN_WORKERS: PositiveInt = Field(
-        description="Minimum number of workers per GraphEngine instance",
-        default=1,
-    )
-
-    GRAPH_ENGINE_MAX_WORKERS: PositiveInt = Field(
-        description="Maximum number of workers per GraphEngine instance",
-        default=10,
-    )
-
-    GRAPH_ENGINE_SCALE_UP_THRESHOLD: PositiveInt = Field(
-        description="Queue depth threshold that triggers worker scale up",
-        default=3,
-    )
-
-    GRAPH_ENGINE_SCALE_DOWN_IDLE_TIME: float = Field(
-        description="Seconds of idle time before scaling down workers",
-        default=5.0,
-        ge=0.1,
-    )
-
 
 class WorkflowNodeExecutionConfig(BaseSettings):
     """
@@ -893,6 +882,22 @@ class CeleryScheduleTasksConfig(BaseSettings):
         description="Enable check upgradable plugin task",
         default=True,
     )
+    ENABLE_WORKFLOW_SCHEDULE_POLLER_TASK: bool = Field(
+        description="Enable workflow schedule poller task",
+        default=True,
+    )
+    WORKFLOW_SCHEDULE_POLLER_INTERVAL: int = Field(
+        description="Workflow schedule poller interval in minutes",
+        default=1,
+    )
+    WORKFLOW_SCHEDULE_POLLER_BATCH_SIZE: int = Field(
+        description="Maximum number of schedules to process in each poll batch",
+        default=100,
+    )
+    WORKFLOW_SCHEDULE_MAX_DISPATCH_PER_TICK: int = Field(
+        description="Maximum schedules to dispatch per tick (0=unlimited, circuit breaker)",
+        default=0,
+    )
 
 
 class PositionConfig(BaseSettings):
@@ -1016,6 +1021,7 @@ class FeatureConfig(
     AuthConfig,  # Changed from OAuthConfig to AuthConfig
     BillingConfig,
     CodeExecutionSandboxConfig,
+    TriggerConfig,
     PluginConfig,
     MarketplaceConfig,
     DataSetConfig,

api/contexts/__init__.py

@@ -8,6 +8,7 @@ if TYPE_CHECKING:
     from core.model_runtime.entities.model_entities import AIModelEntity
     from core.plugin.entities.plugin_daemon import PluginModelProviderEntity
     from core.tools.plugin_tool.provider import PluginToolProviderController
+    from core.trigger.provider import PluginTriggerProviderController
     from core.workflow.entities.variable_pool import VariablePool
 
 
@@ -33,3 +34,11 @@ plugin_model_schema_lock: RecyclableContextVar[Lock] = RecyclableContextVar(Cont
 plugin_model_schemas: RecyclableContextVar[dict[str, "AIModelEntity"]] = RecyclableContextVar(
     ContextVar("plugin_model_schemas")
 )
+
+plugin_trigger_providers: RecyclableContextVar[dict[str, "PluginTriggerProviderController"]] = RecyclableContextVar(
+    ContextVar("plugin_trigger_providers")
+)
+
+plugin_trigger_providers_lock: RecyclableContextVar[Lock] = RecyclableContextVar(
+    ContextVar("plugin_trigger_providers_lock")
+)

api/controllers/console/__init__.py

@@ -67,6 +67,7 @@ from .app import (
     workflow_draft_variable,
     workflow_run,
     workflow_statistic,
+    workflow_trigger,
 )
 
 # Import auth controllers
@@ -180,5 +181,6 @@ from .workspace import (
     models,
     plugin,
     tool_providers,
+    trigger_providers,
     workspace,
 )

api/controllers/console/app/generator.py

@@ -12,14 +12,12 @@ from controllers.console.app.error import (
 )
 from controllers.console.wraps import account_initialization_required, setup_required
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
+from core.helper.code_executor.code_node_provider import CodeNodeProvider
 from core.helper.code_executor.javascript.javascript_code_provider import JavascriptCodeProvider
 from core.helper.code_executor.python3.python3_code_provider import Python3CodeProvider
 from core.llm_generator.llm_generator import LLMGenerator
 from core.model_runtime.errors.invoke import InvokeError
-from extensions.ext_database import db
 from libs.login import login_required
-from models import App
-from services.workflow_service import WorkflowService
 
 
 class RuleGenerateApi(Resource):
@@ -128,16 +126,17 @@ class InstructionGenerateApi(Resource):
         parser.add_argument("model_config", type=dict, required=True, nullable=False, location="json")
         parser.add_argument("ideal_output", type=str, required=False, default="", location="json")
         args = parser.parse_args()
-        code_template = (
-            Python3CodeProvider.get_default_code()
-            if args["language"] == "python"
-            else (JavascriptCodeProvider.get_default_code())
-            if args["language"] == "javascript"
-            else ""
+        providers: list[type[CodeNodeProvider]] = [Python3CodeProvider, JavascriptCodeProvider]
+        code_provider: type[CodeNodeProvider] | None = next(
+            (p for p in providers if p.is_accept_language(args["language"])), None
         )
+        code_template = code_provider.get_default_code() if code_provider else ""
         try:
             # Generate from nothing for a workflow node
             if (args["current"] == code_template or args["current"] == "") and args["node_id"] != "":
+                from models import App, db
+                from services.workflow_service import WorkflowService
+
                 app = db.session.query(App).where(App.id == args["flow_id"]).first()
                 if not app:
                     return {"error": f"app {args['flow_id']} not found"}, 400

api/controllers/console/app/workflow.py

@@ -24,7 +24,7 @@ from core.app.apps.base_app_queue_manager import AppQueueManager
 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.workflow.graph_engine.manager import GraphEngineManager
+from core.model_runtime.utils.encoders import jsonable_encoder
 from extensions.ext_database import db
 from factories import file_factory, variable_factory
 from fields.workflow_fields import workflow_fields, workflow_pagination_fields
@@ -39,6 +39,7 @@ from models.workflow import Workflow
 from services.app_generate_service import AppGenerateService
 from services.errors.app import WorkflowHashNotEqualError
 from services.errors.llm import InvokeRateLimitError
+from services.trigger_debug_service import TriggerDebugService
 from services.workflow_service import DraftWorkflowDeletionError, WorkflowInUseError, WorkflowService
 
 logger = logging.getLogger(__name__)
@@ -414,12 +415,7 @@ class WorkflowTaskStopApi(Resource):
         if not current_user.is_editor:
             raise Forbidden()
 
-        # Stop using both mechanisms for backward compatibility
-        # Legacy stop flag mechanism (without user check)
-        AppQueueManager.set_stop_flag_no_user_check(task_id)
-
-        # New graph engine command channel mechanism
-        GraphEngineManager.send_stop_command(task_id)
+        AppQueueManager.set_stop_flag(task_id, InvokeFrom.DEBUGGER, current_user.id)
 
         return {"result": "success"}
 
@@ -812,6 +808,132 @@ class DraftWorkflowNodeLastRunApi(Resource):
         return node_exec
 
 
+class DraftWorkflowTriggerNodeApi(Resource):
+    """
+    Single node debug - Polling API for trigger events
+    Path: /apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/trigger
+    """
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.WORKFLOW])
+    def post(self, app_model: App, node_id: str):
+        """
+        Poll for trigger events and execute single node when event arrives
+        """
+        if not isinstance(current_user, Account) or not current_user.is_editor:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("trigger_name", type=str, required=True, location="json")
+        parser.add_argument("subscription_id", type=str, required=True, location="json")
+        args = parser.parse_args()
+        trigger_name = args["trigger_name"]
+        subscription_id = args["subscription_id"]
+        event = TriggerDebugService.poll_event(
+            tenant_id=app_model.tenant_id,
+            user_id=current_user.id,
+            app_id=app_model.id,
+            subscription_id=subscription_id,
+            node_id=node_id,
+            trigger_name=trigger_name,
+        )
+        if not event:
+            return jsonable_encoder({"status": "waiting"})
+
+        try:
+            workflow_service = WorkflowService()
+            draft_workflow = workflow_service.get_draft_workflow(app_model)
+            if not draft_workflow:
+                raise ValueError("Workflow not found")
+
+            user_inputs = event.model_dump()
+            node_execution = workflow_service.run_draft_workflow_node(
+                app_model=app_model,
+                draft_workflow=draft_workflow,
+                node_id=node_id,
+                user_inputs=user_inputs,
+                account=current_user,
+                query="",
+                files=[],
+            )
+            return jsonable_encoder(node_execution)
+        except Exception:
+            logger.exception("Error running draft workflow trigger node")
+            return jsonable_encoder(
+                {
+                    "status": "error",
+                }
+            ), 500
+
+
+class DraftWorkflowTriggerRunApi(Resource):
+    """
+    Full workflow debug - Polling API for trigger events
+    Path: /apps/<uuid:app_id>/workflows/draft/trigger/run
+    """
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.WORKFLOW])
+    def post(self, app_model: App):
+        """
+        Poll for trigger events and execute full workflow when event arrives
+        """
+        if not isinstance(current_user, Account) or not current_user.is_editor:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("node_id", type=str, required=True, location="json", nullable=False)
+        parser.add_argument("trigger_name", type=str, required=True, location="json", nullable=False)
+        parser.add_argument("subscription_id", type=str, required=True, location="json", nullable=False)
+        args = parser.parse_args()
+        node_id = args["node_id"]
+        trigger_name = args["trigger_name"]
+        subscription_id = args["subscription_id"]
+
+        event = TriggerDebugService.poll_event(
+            tenant_id=app_model.tenant_id,
+            user_id=current_user.id,
+            app_id=app_model.id,
+            subscription_id=subscription_id,
+            node_id=node_id,
+            trigger_name=trigger_name,
+        )
+        if not event:
+            return jsonable_encoder({"status": "waiting"})
+
+        workflow_args = {
+            "inputs": event.model_dump(),
+            "query": "",
+            "files": [],
+        }
+        external_trace_id = get_external_trace_id(request)
+        if external_trace_id:
+            workflow_args["external_trace_id"] = external_trace_id
+
+        try:
+            response = AppGenerateService.generate(
+                app_model=app_model,
+                user=current_user,
+                args=workflow_args,
+                invoke_from=InvokeFrom.DEBUGGER,
+                streaming=True,
+            )
+            return helper.compact_generate_response(response)
+        except InvokeRateLimitError as ex:
+            raise InvokeRateLimitHttpError(ex.description)
+        except Exception:
+            logger.exception("Error running draft workflow trigger run")
+            return jsonable_encoder(
+                {
+                    "status": "error",
+                }
+            ), 500
+
+
 api.add_resource(
     DraftWorkflowApi,
     "/apps/<uuid:app_id>/workflows/draft",
@@ -836,6 +958,14 @@ api.add_resource(
     DraftWorkflowNodeRunApi,
     "/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/run",
 )
+api.add_resource(
+    DraftWorkflowTriggerNodeApi,
+    "/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/trigger",
+)
+api.add_resource(
+    DraftWorkflowTriggerRunApi,
+    "/apps/<uuid:app_id>/workflows/draft/trigger/run",
+)
 api.add_resource(
     AdvancedChatDraftRunIterationNodeApi,
     "/apps/<uuid:app_id>/advanced-chat/workflows/draft/iteration/nodes/<string:node_id>/run",

api/controllers/console/app/workflow_app_log.py

@@ -6,7 +6,7 @@ from sqlalchemy.orm import Session
 from controllers.console import api
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
-from core.workflow.enums import WorkflowExecutionStatus
+from core.workflow.entities.workflow_execution import WorkflowExecutionStatus
 from extensions.ext_database import db
 from fields.workflow_app_log_fields import workflow_app_log_pagination_fields
 from libs.login import login_required
@@ -27,7 +27,9 @@ class WorkflowAppLogApi(Resource):
         """
         parser = reqparse.RequestParser()
         parser.add_argument("keyword", type=str, location="args")
-        parser.add_argument("status", type=str, choices=["succeeded", "failed", "stopped"], location="args")
+        parser.add_argument(
+            "status", type=str, choices=["succeeded", "failed", "stopped", "partial-succeeded"], location="args"
+        )
         parser.add_argument(
             "created_at__before", type=str, location="args", help="Filter logs created before this timestamp"
         )

api/controllers/console/app/workflow_draft_variable.py

@@ -17,11 +17,10 @@ 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 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
 from libs.login import current_user, login_required
-from models import App, AppMode
+from models import App, AppMode, db
 from models.account import Account
 from models.workflow import WorkflowDraftVariable
 from services.workflow_draft_variable_service import WorkflowDraftVariableList, WorkflowDraftVariableService

api/controllers/console/app/workflow_trigger.py

@@ -0,0 +1,249 @@
+import logging
+
+from flask_restx import Resource, marshal_with, reqparse
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import Forbidden, NotFound
+
+from configs import dify_config
+from controllers.console import api
+from controllers.console.app.wraps import get_app_model
+from controllers.console.wraps import account_initialization_required, setup_required
+from core.model_runtime.utils.encoders import jsonable_encoder
+from extensions.ext_database import db
+from fields.workflow_trigger_fields import trigger_fields, triggers_list_fields, webhook_trigger_fields
+from libs.login import current_user, login_required
+from models.model import Account, AppMode
+from models.workflow import AppTrigger, AppTriggerStatus, WorkflowWebhookTrigger
+
+logger = logging.getLogger(__name__)
+
+from services.workflow_plugin_trigger_service import WorkflowPluginTriggerService
+
+
+class PluginTriggerApi(Resource):
+    """Workflow Plugin Trigger API"""
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=AppMode.WORKFLOW)
+    def post(self, app_model):
+        """Create plugin trigger"""
+        parser = reqparse.RequestParser()
+        parser.add_argument("node_id", type=str, required=False, location="json")
+        parser.add_argument("provider_id", type=str, required=False, location="json")
+        parser.add_argument("trigger_name", type=str, required=False, location="json")
+        parser.add_argument("subscription_id", type=str, required=False, location="json")
+        args = parser.parse_args()
+
+        assert isinstance(current_user, Account)
+        assert current_user.current_tenant_id is not None
+        if not current_user.is_editor:
+            raise Forbidden()
+
+        plugin_trigger = WorkflowPluginTriggerService.create_plugin_trigger(
+            app_id=app_model.id,
+            tenant_id=current_user.current_tenant_id,
+            node_id=args["node_id"],
+            provider_id=args["provider_id"],
+            trigger_name=args["trigger_name"],
+            subscription_id=args["subscription_id"],
+        )
+
+        return jsonable_encoder(plugin_trigger)
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=AppMode.WORKFLOW)
+    def get(self, app_model):
+        """Get plugin trigger"""
+        parser = reqparse.RequestParser()
+        parser.add_argument("node_id", type=str, required=True, help="Node ID is required")
+        args = parser.parse_args()
+
+        plugin_trigger = WorkflowPluginTriggerService.get_plugin_trigger(
+            app_id=app_model.id,
+            node_id=args["node_id"],
+        )
+
+        return jsonable_encoder(plugin_trigger)
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=AppMode.WORKFLOW)
+    def put(self, app_model):
+        """Update plugin trigger"""
+        parser = reqparse.RequestParser()
+        parser.add_argument("node_id", type=str, required=True, help="Node ID is required")
+        parser.add_argument("subscription_id", type=str, required=True, location="json", help="Subscription ID")
+        args = parser.parse_args()
+
+        assert isinstance(current_user, Account)
+        assert current_user.current_tenant_id is not None
+        if not current_user.is_editor:
+            raise Forbidden()
+
+        plugin_trigger = WorkflowPluginTriggerService.update_plugin_trigger(
+            app_id=app_model.id,
+            node_id=args["node_id"],
+            subscription_id=args["subscription_id"],
+        )
+
+        return jsonable_encoder(plugin_trigger)
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=AppMode.WORKFLOW)
+    def delete(self, app_model):
+        """Delete plugin trigger"""
+        parser = reqparse.RequestParser()
+        parser.add_argument("node_id", type=str, required=True, help="Node ID is required")
+        args = parser.parse_args()
+
+        assert isinstance(current_user, Account)
+        assert current_user.current_tenant_id is not None
+        if not current_user.is_editor:
+            raise Forbidden()
+
+        WorkflowPluginTriggerService.delete_plugin_trigger(
+            app_id=app_model.id,
+            node_id=args["node_id"],
+        )
+
+        return {"result": "success"}, 204
+
+
+class WebhookTriggerApi(Resource):
+    """Webhook Trigger API"""
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=AppMode.WORKFLOW)
+    @marshal_with(webhook_trigger_fields)
+    def get(self, app_model):
+        """Get webhook trigger for a node"""
+        parser = reqparse.RequestParser()
+        parser.add_argument("node_id", type=str, required=True, help="Node ID is required")
+        args = parser.parse_args()
+
+        node_id = args["node_id"]
+
+        with Session(db.engine) as session:
+            # Get webhook trigger for this app and node
+            webhook_trigger = (
+                session.query(WorkflowWebhookTrigger)
+                .filter(
+                    WorkflowWebhookTrigger.app_id == app_model.id,
+                    WorkflowWebhookTrigger.node_id == node_id,
+                )
+                .first()
+            )
+
+            if not webhook_trigger:
+                raise NotFound("Webhook trigger not found for this node")
+
+            # Add computed fields for marshal_with
+            base_url = dify_config.SERVICE_API_URL
+            webhook_trigger.webhook_url = f"{base_url}/triggers/webhook/{webhook_trigger.webhook_id}"  # type: ignore
+            webhook_trigger.webhook_debug_url = f"{base_url}/triggers/webhook-debug/{webhook_trigger.webhook_id}"  # type: ignore
+
+            return webhook_trigger
+
+
+class AppTriggersApi(Resource):
+    """App Triggers list API"""
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=AppMode.WORKFLOW)
+    @marshal_with(triggers_list_fields)
+    def get(self, app_model):
+        """Get app triggers list"""
+        assert isinstance(current_user, Account)
+        assert current_user.current_tenant_id is not None
+
+        with Session(db.engine) as session:
+            # Get all triggers for this app using select API
+            triggers = (
+                session.execute(
+                    select(AppTrigger)
+                    .where(
+                        AppTrigger.tenant_id == current_user.current_tenant_id,
+                        AppTrigger.app_id == app_model.id,
+                    )
+                    .order_by(AppTrigger.created_at.desc(), AppTrigger.id.desc())
+                )
+                .scalars()
+                .all()
+            )
+
+        # Add computed icon field for each trigger
+        url_prefix = dify_config.CONSOLE_API_URL + "/console/api/workspaces/current/tool-provider/builtin/"
+        for trigger in triggers:
+            if trigger.trigger_type == "trigger-plugin":
+                trigger.icon = url_prefix + trigger.provider_name + "/icon"  # type: ignore
+            else:
+                trigger.icon = ""  # type: ignore
+
+        return {"data": triggers}
+
+
+class AppTriggerEnableApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=AppMode.WORKFLOW)
+    @marshal_with(trigger_fields)
+    def post(self, app_model):
+        """Update app trigger (enable/disable)"""
+        parser = reqparse.RequestParser()
+        parser.add_argument("trigger_id", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("enable_trigger", type=bool, required=True, nullable=False, location="json")
+        args = parser.parse_args()
+
+        assert isinstance(current_user, Account)
+        assert current_user.current_tenant_id is not None
+        if not current_user.is_editor:
+            raise Forbidden()
+
+        trigger_id = args["trigger_id"]
+
+        with Session(db.engine) as session:
+            # Find the trigger using select
+            trigger = session.execute(
+                select(AppTrigger).where(
+                    AppTrigger.id == trigger_id,
+                    AppTrigger.tenant_id == current_user.current_tenant_id,
+                    AppTrigger.app_id == app_model.id,
+                )
+            ).scalar_one_or_none()
+
+            if not trigger:
+                raise NotFound("Trigger not found")
+
+            # Update status based on enable_trigger boolean
+            trigger.status = AppTriggerStatus.ENABLED if args["enable_trigger"] else AppTriggerStatus.DISABLED
+
+            session.commit()
+            session.refresh(trigger)
+
+        # Add computed icon field
+        url_prefix = dify_config.CONSOLE_API_URL + "/console/api/workspaces/current/tool-provider/builtin/"
+        if trigger.trigger_type == "trigger-plugin":
+            trigger.icon = url_prefix + trigger.provider_name + "/icon"  # type: ignore
+        else:
+            trigger.icon = ""  # type: ignore
+
+        return trigger
+
+
+api.add_resource(WebhookTriggerApi, "/apps/<uuid:app_id>/workflows/triggers/webhook")
+api.add_resource(PluginTriggerApi, "/apps/<uuid:app_id>/workflows/triggers/plugin")
+api.add_resource(AppTriggersApi, "/apps/<uuid:app_id>/triggers")
+api.add_resource(AppTriggerEnableApi, "/apps/<uuid:app_id>/trigger-enable")

api/controllers/console/auth/oauth_server.py

@@ -44,22 +44,19 @@ def oauth_server_access_token_required(view):
         if not oauth_provider_app or not isinstance(oauth_provider_app, OAuthProviderApp):
             raise BadRequest("Invalid oauth_provider_app")
 
-        if not request.headers.get("Authorization"):
-            raise BadRequest("Authorization is required")
-
         authorization_header = request.headers.get("Authorization")
         if not authorization_header:
             raise BadRequest("Authorization header is required")
 
-        parts = authorization_header.split(" ")
+        parts = authorization_header.strip().split(" ")
         if len(parts) != 2:
             raise BadRequest("Invalid Authorization header format")
 
-        token_type = parts[0]
-        if token_type != "Bearer":
+        token_type = parts[0].strip()
+        if token_type.lower() != "bearer":
             raise BadRequest("token_type is invalid")
 
-        access_token = parts[1]
+        access_token = parts[1].strip()
         if not access_token:
             raise BadRequest("access_token is required")
 
@@ -125,7 +122,10 @@ class OAuthServerUserTokenApi(Resource):
         parser.add_argument("refresh_token", type=str, required=False, location="json")
         parsed_args = parser.parse_args()
 
-        grant_type = OAuthGrantType(parsed_args["grant_type"])
+        try:
+            grant_type = OAuthGrantType(parsed_args["grant_type"])
+        except ValueError:
+            raise BadRequest("invalid grant_type")
 
         if grant_type == OAuthGrantType.AUTHORIZATION_CODE:
             if not parsed_args["code"]:
@@ -163,8 +163,6 @@ class OAuthServerUserTokenApi(Resource):
                     "refresh_token": refresh_token,
                 }
             )
-        else:
-            raise BadRequest("invalid grant_type")
 
 
 class OAuthServerUserAccountApi(Resource):

api/controllers/console/datasets/datasets.py

@@ -19,6 +19,7 @@ from controllers.console.wraps import (
 from core.errors.error import LLMBadRequestError, ProviderTokenNotInitError
 from core.indexing_runner import IndexingRunner
 from core.model_runtime.entities.model_entities import ModelType
+from core.plugin.entities.plugin import ModelProviderID
 from core.provider_manager import ProviderManager
 from core.rag.datasource.vdb.vector_type import VectorType
 from core.rag.extractor.entity.extract_setting import ExtractSetting
@@ -30,7 +31,6 @@ from fields.document_fields import document_status_fields
 from libs.login import login_required
 from models import ApiToken, Dataset, Document, DocumentSegment, UploadFile
 from models.dataset import DatasetPermissionEnum
-from models.provider_ids import ModelProviderID
 from services.dataset_service import DatasetPermissionService, DatasetService, DocumentService
 
 

api/controllers/console/explore/workflow.py

@@ -20,7 +20,6 @@ from core.errors.error import (
     QuotaExceededError,
 )
 from core.model_runtime.errors.invoke import InvokeError
-from core.workflow.graph_engine.manager import GraphEngineManager
 from libs import helper
 from libs.login import current_user
 from models.model import AppMode, InstalledApp
@@ -79,11 +78,6 @@ class InstalledAppWorkflowTaskStopApi(InstalledAppResource):
             raise NotWorkflowAppError()
         assert current_user is not None
 
-        # Stop using both mechanisms for backward compatibility
-        # Legacy stop flag mechanism (without user check)
-        AppQueueManager.set_stop_flag_no_user_check(task_id)
-
-        # New graph engine command channel mechanism
-        GraphEngineManager.send_stop_command(task_id)
+        AppQueueManager.set_stop_flag(task_id, InvokeFrom.EXPLORE, current_user.id)
 
         return {"result": "success"}

api/controllers/console/workspace/plugin.py

@@ -516,18 +516,20 @@ class PluginFetchDynamicSelectOptionsApi(Resource):
         parser.add_argument("provider", type=str, required=True, location="args")
         parser.add_argument("action", type=str, required=True, location="args")
         parser.add_argument("parameter", type=str, required=True, location="args")
+        parser.add_argument("credential_id", type=str, required=False, location="args")
         parser.add_argument("provider_type", type=str, required=True, location="args")
         args = parser.parse_args()
 
         try:
             options = PluginParameterService.get_dynamic_select_options(
-                tenant_id,
-                user_id,
-                args["plugin_id"],
-                args["provider"],
-                args["action"],
-                args["parameter"],
-                args["provider_type"],
+                tenant_id=tenant_id,
+                user_id=user_id,
+                plugin_id=args["plugin_id"],
+                provider=args["provider"],
+                action=args["action"],
+                parameter=args["parameter"],
+                credential_id=args["credential_id"],
+                provider_type=args["provider_type"],
             )
         except PluginDaemonClientSideError as e:
             raise ValueError(e)

api/controllers/console/workspace/tool_providers.py

@@ -21,11 +21,11 @@ from core.mcp.auth.auth_provider import OAuthClientProvider
 from core.mcp.error import MCPAuthError, MCPError
 from core.mcp.mcp_client import MCPClient
 from core.model_runtime.utils.encoders import jsonable_encoder
+from core.plugin.entities.plugin import ToolProviderID
+from core.plugin.entities.plugin_daemon import CredentialType
 from core.plugin.impl.oauth import OAuthHandler
-from core.tools.entities.tool_entities import CredentialType
 from libs.helper import StrLen, alphanumeric, uuid_value
 from libs.login import login_required
-from models.provider_ids import ToolProviderID
 from services.plugin.oauth_service import OAuthProxyService
 from services.tools.api_tools_manage_service import ApiToolManageService
 from services.tools.builtin_tools_manage_service import BuiltinToolManageService

api/controllers/console/workspace/trigger_providers.py

@@ -0,0 +1,589 @@
+import logging
+
+from flask import make_response, redirect, request
+from flask_restx import Resource, reqparse
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import BadRequest, Forbidden
+
+from configs import dify_config
+from controllers.console import api
+from controllers.console.wraps import account_initialization_required, setup_required
+from core.model_runtime.utils.encoders import jsonable_encoder
+from core.plugin.entities.plugin import TriggerProviderID
+from core.plugin.entities.plugin_daemon import CredentialType
+from core.plugin.impl.oauth import OAuthHandler
+from core.trigger.entities.entities import SubscriptionBuilderUpdater
+from core.trigger.trigger_manager import TriggerManager
+from extensions.ext_database import db
+from libs.login import current_user, login_required
+from models.account import Account
+from services.plugin.oauth_service import OAuthProxyService
+from services.trigger.trigger_provider_service import TriggerProviderService
+from services.trigger.trigger_subscription_builder_service import TriggerSubscriptionBuilderService
+from services.workflow_plugin_trigger_service import WorkflowPluginTriggerService
+
+logger = logging.getLogger(__name__)
+
+
+class TriggerProviderListApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self):
+        """List all trigger providers for the current tenant"""
+        user = current_user
+        assert isinstance(user, Account)
+        assert user.current_tenant_id is not None
+        return jsonable_encoder(TriggerProviderService.list_trigger_providers(user.current_tenant_id))
+
+
+class TriggerProviderInfoApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, provider):
+        """Get info for a trigger provider"""
+        user = current_user
+        assert isinstance(user, Account)
+        assert user.current_tenant_id is not None
+        return jsonable_encoder(
+            TriggerProviderService.get_trigger_provider(user.current_tenant_id, TriggerProviderID(provider))
+        )
+
+
+class TriggerSubscriptionListApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, provider):
+        """List all trigger subscriptions for the current tenant's provider"""
+        user = current_user
+        assert isinstance(user, Account)
+        assert user.current_tenant_id is not None
+        if not user.is_admin_or_owner:
+            raise Forbidden()
+
+        try:
+            return jsonable_encoder(
+                TriggerProviderService.list_trigger_provider_subscriptions(
+                    tenant_id=user.current_tenant_id, provider_id=TriggerProviderID(provider)
+                )
+            )
+        except Exception as e:
+            logger.exception("Error listing trigger providers", exc_info=e)
+            raise
+
+
+class TriggerSubscriptionBuilderCreateApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, provider):
+        """Add a new subscription instance for a trigger provider"""
+        user = current_user
+        assert isinstance(user, Account)
+        assert user.current_tenant_id is not None
+        if not user.is_admin_or_owner:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("credential_type", type=str, required=False, nullable=True, location="json")
+        args = parser.parse_args()
+
+        try:
+            credential_type = CredentialType.of(args.get("credential_type") or CredentialType.UNAUTHORIZED.value)
+            subscription_builder = TriggerSubscriptionBuilderService.create_trigger_subscription_builder(
+                tenant_id=user.current_tenant_id,
+                user_id=user.id,
+                provider_id=TriggerProviderID(provider),
+                credential_type=credential_type,
+            )
+            return jsonable_encoder({"subscription_builder": subscription_builder})
+        except ValueError as e:
+            raise BadRequest(str(e))
+        except Exception as e:
+            logger.exception("Error adding provider credential", exc_info=e)
+            raise
+
+
+class TriggerSubscriptionBuilderGetApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, provider, subscription_builder_id):
+        """Get a subscription instance for a trigger provider"""
+        return jsonable_encoder(
+            TriggerSubscriptionBuilderService.get_subscription_builder_by_id(subscription_builder_id)
+        )
+
+
+class TriggerSubscriptionBuilderVerifyApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, provider, subscription_builder_id):
+        """Verify a subscription instance for a trigger provider"""
+        user = current_user
+        assert isinstance(user, Account)
+        assert user.current_tenant_id is not None
+        if not user.is_admin_or_owner:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        # The credentials of the subscription builder
+        parser.add_argument("credentials", type=dict, required=False, nullable=True, location="json")
+        args = parser.parse_args()
+
+        try:
+            TriggerSubscriptionBuilderService.update_trigger_subscription_builder(
+                tenant_id=user.current_tenant_id,
+                provider_id=TriggerProviderID(provider),
+                subscription_builder_id=subscription_builder_id,
+                subscription_builder_updater=SubscriptionBuilderUpdater(
+                    credentials=args.get("credentials", None),
+                ),
+            )
+            return TriggerSubscriptionBuilderService.verify_trigger_subscription_builder(
+                tenant_id=user.current_tenant_id,
+                user_id=user.id,
+                provider_id=TriggerProviderID(provider),
+                subscription_builder_id=subscription_builder_id,
+            )
+        except Exception as e:
+            logger.exception("Error verifying provider credential", exc_info=e)
+            raise
+
+
+class TriggerSubscriptionBuilderUpdateApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, provider, subscription_builder_id):
+        """Update a subscription instance for a trigger provider"""
+        user = current_user
+        assert isinstance(user, Account)
+        assert user.current_tenant_id is not None
+
+        parser = reqparse.RequestParser()
+        # The name of the subscription builder
+        parser.add_argument("name", type=str, required=False, nullable=True, location="json")
+        # The parameters of the subscription builder
+        parser.add_argument("parameters", type=dict, required=False, nullable=True, location="json")
+        # The properties of the subscription builder
+        parser.add_argument("properties", type=dict, required=False, nullable=True, location="json")
+        # The credentials of the subscription builder
+        parser.add_argument("credentials", type=dict, required=False, nullable=True, location="json")
+        args = parser.parse_args()
+        try:
+            return jsonable_encoder(
+                TriggerSubscriptionBuilderService.update_trigger_subscription_builder(
+                    tenant_id=user.current_tenant_id,
+                    provider_id=TriggerProviderID(provider),
+                    subscription_builder_id=subscription_builder_id,
+                    subscription_builder_updater=SubscriptionBuilderUpdater(
+                        name=args.get("name", None),
+                        parameters=args.get("parameters", None),
+                        properties=args.get("properties", None),
+                        credentials=args.get("credentials", None),
+                    ),
+                )
+            )
+        except Exception as e:
+            logger.exception("Error updating provider credential", exc_info=e)
+            raise
+
+
+class TriggerSubscriptionBuilderLogsApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, provider, subscription_builder_id):
+        """Get the request logs for a subscription instance for a trigger provider"""
+        user = current_user
+        assert isinstance(user, Account)
+        assert user.current_tenant_id is not None
+
+        try:
+            logs = TriggerSubscriptionBuilderService.list_logs(subscription_builder_id)
+            return jsonable_encoder({"logs": [log.model_dump(mode="json") for log in logs]})
+        except Exception as e:
+            logger.exception("Error getting request logs for subscription builder", exc_info=e)
+            raise
+
+
+class TriggerSubscriptionBuilderBuildApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, provider, subscription_builder_id):
+        """Build a subscription instance for a trigger provider"""
+        user = current_user
+        assert isinstance(user, Account)
+        assert user.current_tenant_id is not None
+        if not user.is_admin_or_owner:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        # The name of the subscription builder
+        parser.add_argument("name", type=str, required=False, nullable=True, location="json")
+        # The parameters of the subscription builder
+        parser.add_argument("parameters", type=dict, required=False, nullable=True, location="json")
+        # The properties of the subscription builder
+        parser.add_argument("properties", type=dict, required=False, nullable=True, location="json")
+        # The credentials of the subscription builder
+        parser.add_argument("credentials", type=dict, required=False, nullable=True, location="json")
+        args = parser.parse_args()
+        try:
+            TriggerSubscriptionBuilderService.update_trigger_subscription_builder(
+                tenant_id=user.current_tenant_id,
+                provider_id=TriggerProviderID(provider),
+                subscription_builder_id=subscription_builder_id,
+                subscription_builder_updater=SubscriptionBuilderUpdater(
+                    name=args.get("name", None),
+                    parameters=args.get("parameters", None),
+                    properties=args.get("properties", None),
+                ),
+            )
+            TriggerSubscriptionBuilderService.build_trigger_subscription_builder(
+                tenant_id=user.current_tenant_id,
+                user_id=user.id,
+                provider_id=TriggerProviderID(provider),
+                subscription_builder_id=subscription_builder_id,
+            )
+            return 200
+        except ValueError as e:
+            raise BadRequest(str(e))
+        except Exception as e:
+            logger.exception("Error building provider credential", exc_info=e)
+            raise
+
+
+class TriggerSubscriptionDeleteApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, subscription_id):
+        """Delete a subscription instance"""
+        user = current_user
+        assert isinstance(user, Account)
+        assert user.current_tenant_id is not None
+        if not user.is_admin_or_owner:
+            raise Forbidden()
+
+        try:
+            with Session(db.engine) as session:
+                # Delete trigger provider subscription
+                TriggerProviderService.delete_trigger_provider(
+                    session=session,
+                    tenant_id=user.current_tenant_id,
+                    subscription_id=subscription_id,
+                )
+                # Delete plugin triggers
+                WorkflowPluginTriggerService.delete_plugin_trigger_by_subscription(
+                    session=session,
+                    tenant_id=user.current_tenant_id,
+                    subscription_id=subscription_id,
+                )
+                session.commit()
+            return {"result": "success"}
+        except ValueError as e:
+            raise BadRequest(str(e))
+        except Exception as e:
+            logger.exception("Error deleting provider credential", exc_info=e)
+            raise
+
+
+class TriggerOAuthAuthorizeApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, provider):
+        """Initiate OAuth authorization flow for a trigger provider"""
+        user = current_user
+        assert isinstance(user, Account)
+        assert user.current_tenant_id is not None
+
+        try:
+            provider_id = TriggerProviderID(provider)
+            plugin_id = provider_id.plugin_id
+            provider_name = provider_id.provider_name
+            tenant_id = user.current_tenant_id
+
+            # Get OAuth client configuration
+            oauth_client_params = TriggerProviderService.get_oauth_client(
+                tenant_id=tenant_id,
+                provider_id=provider_id,
+            )
+
+            if oauth_client_params is None:
+                raise Forbidden("No OAuth client configuration found for this trigger provider")
+
+            # Create subscription builder
+            subscription_builder = TriggerSubscriptionBuilderService.create_trigger_subscription_builder(
+                tenant_id=tenant_id,
+                user_id=user.id,
+                provider_id=provider_id,
+                credential_type=CredentialType.OAUTH2,
+            )
+
+            # Create OAuth handler and proxy context
+            oauth_handler = OAuthHandler()
+            context_id = OAuthProxyService.create_proxy_context(
+                user_id=user.id,
+                tenant_id=tenant_id,
+                plugin_id=plugin_id,
+                provider=provider_name,
+                extra_data={
+                    "subscription_builder_id": subscription_builder.id,
+                },
+            )
+
+            # Build redirect URI for callback
+            redirect_uri = f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{provider}/trigger/callback"
+
+            # Get authorization URL
+            authorization_url_response = oauth_handler.get_authorization_url(
+                tenant_id=tenant_id,
+                user_id=user.id,
+                plugin_id=plugin_id,
+                provider=provider_name,
+                redirect_uri=redirect_uri,
+                system_credentials=oauth_client_params,
+            )
+
+            # Create response with cookie
+            response = make_response(
+                jsonable_encoder(
+                    {
+                        "authorization_url": authorization_url_response.authorization_url,
+                        "subscription_builder_id": subscription_builder.id,
+                        "subscription_builder": subscription_builder,
+                    }
+                )
+            )
+            response.set_cookie(
+                "context_id",
+                context_id,
+                httponly=True,
+                samesite="Lax",
+                max_age=OAuthProxyService.__MAX_AGE__,
+            )
+
+            return response
+
+        except Exception as e:
+            logger.exception("Error initiating OAuth flow", exc_info=e)
+            raise
+
+
+class TriggerOAuthCallbackApi(Resource):
+    @setup_required
+    def get(self, provider):
+        """Handle OAuth callback for trigger provider"""
+        context_id = request.cookies.get("context_id")
+        if not context_id:
+            raise Forbidden("context_id not found")
+
+        # Use and validate proxy context
+        context = OAuthProxyService.use_proxy_context(context_id)
+        if context is None:
+            raise Forbidden("Invalid context_id")
+
+        # Parse provider ID
+        provider_id = TriggerProviderID(provider)
+        plugin_id = provider_id.plugin_id
+        provider_name = provider_id.provider_name
+        user_id = context.get("user_id")
+        tenant_id = context.get("tenant_id")
+        subscription_builder_id = context.get("subscription_builder_id")
+
+        # Get OAuth client configuration
+        oauth_client_params = TriggerProviderService.get_oauth_client(
+            tenant_id=tenant_id,
+            provider_id=provider_id,
+        )
+
+        if oauth_client_params is None:
+            raise Forbidden("No OAuth client configuration found for this trigger provider")
+
+        # Get OAuth credentials from callback
+        oauth_handler = OAuthHandler()
+        redirect_uri = f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{provider}/trigger/callback"
+
+        credentials_response = oauth_handler.get_credentials(
+            tenant_id=tenant_id,
+            user_id=user_id,
+            plugin_id=plugin_id,
+            provider=provider_name,
+            redirect_uri=redirect_uri,
+            system_credentials=oauth_client_params,
+            request=request,
+        )
+
+        credentials = credentials_response.credentials
+        expires_at = credentials_response.expires_at
+
+        if not credentials:
+            raise Exception("Failed to get OAuth credentials")
+
+        # Update subscription builder
+        TriggerSubscriptionBuilderService.update_trigger_subscription_builder(
+            tenant_id=tenant_id,
+            provider_id=provider_id,
+            subscription_builder_id=subscription_builder_id,
+            subscription_builder_updater=SubscriptionBuilderUpdater(
+                credentials=credentials,
+                credential_expires_at=expires_at,
+            ),
+        )
+        # Redirect to OAuth callback page
+        return redirect(f"{dify_config.CONSOLE_WEB_URL}/oauth-callback")
+
+
+class TriggerOAuthClientManageApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, provider):
+        """Get OAuth client configuration for a provider"""
+        user = current_user
+        assert isinstance(user, Account)
+        assert user.current_tenant_id is not None
+        if not user.is_admin_or_owner:
+            raise Forbidden()
+
+        try:
+            provider_id = TriggerProviderID(provider)
+
+            # Get custom OAuth client params if exists
+            custom_params = TriggerProviderService.get_custom_oauth_client_params(
+                tenant_id=user.current_tenant_id,
+                provider_id=provider_id,
+            )
+
+            # Check if custom client is enabled
+            is_custom_enabled = TriggerProviderService.is_oauth_custom_client_enabled(
+                tenant_id=user.current_tenant_id,
+                provider_id=provider_id,
+            )
+
+            # Check if there's a system OAuth client
+            system_client = TriggerProviderService.get_oauth_client(
+                tenant_id=user.current_tenant_id,
+                provider_id=provider_id,
+            )
+            provider_controller = TriggerManager.get_trigger_provider(user.current_tenant_id, provider_id)
+            redirect_uri = f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{provider}/trigger/callback"
+            return jsonable_encoder(
+                {
+                    "configured": bool(custom_params or system_client),
+                    "oauth_client_schema": provider_controller.get_oauth_client_schema(),
+                    "custom_configured": bool(custom_params),
+                    "custom_enabled": is_custom_enabled,
+                    "redirect_uri": redirect_uri,
+                    "params": custom_params if custom_params else {},
+                }
+            )
+
+        except Exception as e:
+            logger.exception("Error getting OAuth client", exc_info=e)
+            raise
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, provider):
+        """Configure custom OAuth client for a provider"""
+        user = current_user
+        assert isinstance(user, Account)
+        assert user.current_tenant_id is not None
+        if not user.is_admin_or_owner:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("client_params", type=dict, required=False, nullable=True, location="json")
+        parser.add_argument("enabled", type=bool, required=False, nullable=True, location="json")
+        args = parser.parse_args()
+
+        try:
+            provider_id = TriggerProviderID(provider)
+            return TriggerProviderService.save_custom_oauth_client_params(
+                tenant_id=user.current_tenant_id,
+                provider_id=provider_id,
+                client_params=args.get("client_params"),
+                enabled=args.get("enabled"),
+            )
+
+        except ValueError as e:
+            raise BadRequest(str(e))
+        except Exception as e:
+            logger.exception("Error configuring OAuth client", exc_info=e)
+            raise
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def delete(self, provider):
+        """Remove custom OAuth client configuration"""
+        user = current_user
+        assert isinstance(user, Account)
+        assert user.current_tenant_id is not None
+        if not user.is_admin_or_owner:
+            raise Forbidden()
+
+        try:
+            provider_id = TriggerProviderID(provider)
+
+            return TriggerProviderService.delete_custom_oauth_client_params(
+                tenant_id=user.current_tenant_id,
+                provider_id=provider_id,
+            )
+        except ValueError as e:
+            raise BadRequest(str(e))
+        except Exception as e:
+            logger.exception("Error removing OAuth client", exc_info=e)
+            raise
+
+
+# Trigger Subscription
+api.add_resource(TriggerProviderListApi, "/workspaces/current/triggers")
+api.add_resource(TriggerProviderInfoApi, "/workspaces/current/trigger-provider/<path:provider>/info")
+api.add_resource(TriggerSubscriptionListApi, "/workspaces/current/trigger-provider/<path:provider>/subscriptions/list")
+api.add_resource(
+    TriggerSubscriptionDeleteApi,
+    "/workspaces/current/trigger-provider/<path:subscription_id>/subscriptions/delete",
+)
+
+# Trigger Subscription Builder
+api.add_resource(
+    TriggerSubscriptionBuilderCreateApi,
+    "/workspaces/current/trigger-provider/<path:provider>/subscriptions/builder/create",
+)
+api.add_resource(
+    TriggerSubscriptionBuilderGetApi,
+    "/workspaces/current/trigger-provider/<path:provider>/subscriptions/builder/<path:subscription_builder_id>",
+)
+api.add_resource(
+    TriggerSubscriptionBuilderUpdateApi,
+    "/workspaces/current/trigger-provider/<path:provider>/subscriptions/builder/update/<path:subscription_builder_id>",
+)
+api.add_resource(
+    TriggerSubscriptionBuilderVerifyApi,
+    "/workspaces/current/trigger-provider/<path:provider>/subscriptions/builder/verify/<path:subscription_builder_id>",
+)
+api.add_resource(
+    TriggerSubscriptionBuilderBuildApi,
+    "/workspaces/current/trigger-provider/<path:provider>/subscriptions/builder/build/<path:subscription_builder_id>",
+)
+api.add_resource(
+    TriggerSubscriptionBuilderLogsApi,
+    "/workspaces/current/trigger-provider/<path:provider>/subscriptions/builder/logs/<path:subscription_builder_id>",
+)
+
+
+# OAuth
+api.add_resource(
+    TriggerOAuthAuthorizeApi, "/workspaces/current/trigger-provider/<path:provider>/subscriptions/oauth/authorize"
+)
+api.add_resource(TriggerOAuthCallbackApi, "/oauth/plugin/<path:provider>/trigger/callback")
+api.add_resource(TriggerOAuthClientManageApi, "/workspaces/current/trigger-provider/<path:provider>/oauth/client")

api/controllers/files/tool_files.py

@@ -8,7 +8,7 @@ from controllers.common.errors import UnsupportedFileTypeError
 from controllers.files import files_ns
 from core.tools.signature import verify_tool_file_signature
 from core.tools.tool_file_manager import ToolFileManager
-from extensions.ext_database import db as global_db
+from models import db as global_db
 
 
 @files_ns.route("/tools/<uuid:file_id>.<string:extension>")

api/controllers/inner_api/wraps.py

@@ -1,8 +1,12 @@
 from base64 import b64encode
+from collections.abc import Callable
 from functools import wraps
 from hashlib import sha1
 from hmac import new as hmac_new
+from typing import ParamSpec, TypeVar
 
+P = ParamSpec("P")
+R = TypeVar("R")
 from flask import abort, request
 
 from configs import dify_config
@@ -10,9 +14,9 @@ from extensions.ext_database import db
 from models.model import EndUser
 
 
-def billing_inner_api_only(view):
+def billing_inner_api_only(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         if not dify_config.INNER_API:
             abort(404)
 
@@ -26,9 +30,9 @@ def billing_inner_api_only(view):
     return decorated
 
 
-def enterprise_inner_api_only(view):
+def enterprise_inner_api_only(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         if not dify_config.INNER_API:
             abort(404)
 
@@ -78,9 +82,9 @@ def enterprise_inner_api_user_auth(view):
     return decorated
 
 
-def plugin_inner_api_only(view):
+def plugin_inner_api_only(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         if not dify_config.PLUGIN_DAEMON_KEY:
             abort(404)
 

api/controllers/mcp/mcp.py

@@ -9,10 +9,9 @@ from controllers.console.app.mcp_server import AppMCPServerStatus
 from controllers.mcp import mcp_ns
 from core.app.app_config.entities import VariableEntity
 from core.mcp import types as mcp_types
-from core.mcp.server.streamable_http import handle_mcp_request
 from extensions.ext_database import db
 from libs import helper
-from models.model import App, AppMCPServer, AppMode, EndUser
+from models.model import App, AppMCPServer, AppMode
 
 
 class MCPRequestError(Exception):
@@ -195,50 +194,6 @@ class MCPAppApi(Resource):
             except ValidationError as e:
                 raise MCPRequestError(mcp_types.INVALID_PARAMS, f"Invalid MCP request: {str(e)}")
 
-    def _retrieve_end_user(self, tenant_id: str, mcp_server_id: str, session: Session) -> EndUser | None:
-        """Get end user from existing session - optimized query"""
-        return (
-            session.query(EndUser)
-            .where(EndUser.tenant_id == tenant_id)
-            .where(EndUser.session_id == mcp_server_id)
-            .where(EndUser.type == "mcp")
-            .first()
-        )
-
-    def _create_end_user(
-        self, client_name: str, tenant_id: str, app_id: str, mcp_server_id: str, session: Session
-    ) -> EndUser:
-        """Create end user in existing session"""
-        end_user = EndUser(
-            tenant_id=tenant_id,
-            app_id=app_id,
-            type="mcp",
-            name=client_name,
-            session_id=mcp_server_id,
-        )
-        session.add(end_user)
-        session.flush()  # Use flush instead of commit to keep transaction open
-        session.refresh(end_user)
-        return end_user
-
-    def _handle_mcp_request(
-        self,
-        app: App,
-        mcp_server: AppMCPServer,
-        mcp_request: mcp_types.ClientRequest,
-        user_input_form: list[VariableEntity],
-        session: Session,
-        request_id: Union[int, str],
-    ) -> mcp_types.JSONRPCResponse | mcp_types.JSONRPCError | None:
-        """Handle MCP request and return response"""
-        end_user = self._retrieve_end_user(mcp_server.tenant_id, mcp_server.id, session)
-
-        if not end_user and isinstance(mcp_request.root, mcp_types.InitializeRequest):
-            client_info = mcp_request.root.params.clientInfo
-            client_name = f"{client_info.name}@{client_info.version}"
-            # Commit the session before creating end user to avoid transaction conflicts
-            session.commit()
-            with Session(db.engine, expire_on_commit=False) as create_session, create_session.begin():
-                end_user = self._create_end_user(client_name, app.tenant_id, app.id, mcp_server.id, create_session)
-
-        return handle_mcp_request(app, mcp_request, user_input_form, mcp_server, end_user, request_id)
+        mcp_server_handler = MCPServerStreamableHTTPRequestHandler(app, request, converted_user_input_form)
+        response = mcp_server_handler.handle()
+        return helper.compact_generate_response(response)

api/controllers/service_api/app/workflow.py

@@ -26,8 +26,7 @@ from core.errors.error import (
 )
 from core.helper.trace_id_helper import get_external_trace_id
 from core.model_runtime.errors.invoke import InvokeError
-from core.workflow.enums import WorkflowExecutionStatus
-from core.workflow.graph_engine.manager import GraphEngineManager
+from core.workflow.entities.workflow_execution import WorkflowExecutionStatus
 from extensions.ext_database import db
 from fields.workflow_app_log_fields import build_workflow_app_log_pagination_model
 from libs import helper
@@ -263,12 +262,7 @@ class WorkflowTaskStopApi(Resource):
         if app_mode != AppMode.WORKFLOW:
             raise NotWorkflowAppError()
 
-        # Stop using both mechanisms for backward compatibility
-        # Legacy stop flag mechanism (without user check)
-        AppQueueManager.set_stop_flag_no_user_check(task_id)
-
-        # New graph engine command channel mechanism
-        GraphEngineManager.send_stop_command(task_id)
+        AppQueueManager.set_stop_flag(task_id, InvokeFrom.SERVICE_API, end_user.id)
 
         return {"result": "success"}
 

api/controllers/service_api/dataset/dataset.py

@@ -13,13 +13,13 @@ from controllers.service_api.wraps import (
     validate_dataset_token,
 )
 from core.model_runtime.entities.model_entities import ModelType
+from core.plugin.entities.plugin import ModelProviderID
 from core.provider_manager import ProviderManager
 from fields.dataset_fields import dataset_detail_fields
 from fields.tag_fields import build_dataset_tag_fields
 from libs.login import current_user
 from models.account import Account
 from models.dataset import Dataset, DatasetPermissionEnum
-from models.provider_ids import ModelProviderID
 from services.dataset_service import DatasetPermissionService, DatasetService, DocumentService
 from services.entities.knowledge_entities.knowledge_entities import RetrievalModel
 from services.tag_service import TagService

api/controllers/service_api/wraps.py

@@ -1,7 +1,7 @@
 import time
 from collections.abc import Callable
 from datetime import timedelta
-from enum import Enum
+from enum import StrEnum, auto
 from functools import wraps
 from typing import Optional
 
@@ -23,14 +23,14 @@ from models.model import ApiToken, App, EndUser
 from services.feature_service import FeatureService
 
 
-class WhereisUserArg(Enum):
+class WhereisUserArg(StrEnum):
     """
     Enum for whereis_user_arg.
     """
 
-    QUERY = "query"
-    JSON = "json"
-    FORM = "form"
+    QUERY = auto()
+    JSON = auto()
+    FORM = auto()
 
 
 class FetchUserArg(BaseModel):

api/controllers/trigger/__init__.py

@@ -0,0 +1,7 @@
+from flask import Blueprint
+
+# Create trigger blueprint
+bp = Blueprint("trigger", __name__, url_prefix="/triggers")
+
+# Import routes after blueprint creation to avoid circular imports
+from . import trigger, webhook

api/controllers/trigger/trigger.py

@@ -0,0 +1,41 @@
+import logging
+import re
+
+from flask import jsonify, request
+from werkzeug.exceptions import NotFound
+
+from controllers.trigger import bp
+from services.trigger.trigger_subscription_builder_service import TriggerSubscriptionBuilderService
+from services.trigger_service import TriggerService
+
+logger = logging.getLogger(__name__)
+
+UUID_PATTERN = r"^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$"
+UUID_MATCHER = re.compile(UUID_PATTERN)
+
+
+@bp.route("/plugin/<string:endpoint_id>", methods=["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"])
+def trigger_endpoint(endpoint_id: str):
+    """
+    Handle endpoint trigger calls.
+    """
+    # endpoint_id must be UUID
+    if not UUID_MATCHER.match(endpoint_id):
+        raise NotFound("Invalid endpoint ID")
+    handling_chain = [
+        TriggerService.process_endpoint,
+        TriggerSubscriptionBuilderService.process_builder_validation_endpoint,
+    ]
+    try:
+        for handler in handling_chain:
+            response = handler(endpoint_id, request)
+            if response:
+                break
+        if not response:
+            raise NotFound("Endpoint not found")
+        return response
+    except ValueError as e:
+        raise NotFound(str(e))
+    except Exception as e:
+        logger.exception("Webhook processing failed for {endpoint_id}")
+        return jsonify({"error": "Internal server error", "message": str(e)}), 500

api/controllers/trigger/webhook.py

@@ -0,0 +1,46 @@
+import logging
+
+from flask import jsonify
+from werkzeug.exceptions import NotFound, RequestEntityTooLarge
+
+from controllers.trigger import bp
+from services.webhook_service import WebhookService
+
+logger = logging.getLogger(__name__)
+
+
+@bp.route("/webhook/<string:webhook_id>", methods=["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"])
+@bp.route("/webhook-debug/<string:webhook_id>", methods=["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"])
+def handle_webhook(webhook_id: str):
+    """
+    Handle webhook trigger calls.
+
+    This endpoint receives webhook calls and processes them according to the
+    configured webhook trigger settings.
+    """
+    try:
+        # Get webhook trigger, workflow, and node configuration
+        webhook_trigger, workflow, node_config = WebhookService.get_webhook_trigger_and_workflow(webhook_id)
+
+        # Extract request data
+        webhook_data = WebhookService.extract_webhook_data(webhook_trigger)
+
+        # Validate request against node configuration
+        validation_result = WebhookService.validate_webhook_request(webhook_data, node_config)
+        if not validation_result["valid"]:
+            return jsonify({"error": "Bad Request", "message": validation_result["error"]}), 400
+
+        # Process webhook call (send to Celery)
+        WebhookService.trigger_workflow_execution(webhook_trigger, webhook_data, workflow)
+
+        # Return configured response
+        response_data, status_code = WebhookService.generate_webhook_response(node_config)
+        return jsonify(response_data), status_code
+
+    except ValueError as e:
+        raise NotFound(str(e))
+    except RequestEntityTooLarge:
+        raise
+    except Exception as e:
+        logger.exception("Webhook processing failed for %s", webhook_id)
+        return jsonify({"error": "Internal server error", "message": str(e)}), 500

api/controllers/web/workflow.py

@@ -21,7 +21,6 @@ from core.errors.error import (
     QuotaExceededError,
 )
 from core.model_runtime.errors.invoke import InvokeError
-from core.workflow.graph_engine.manager import GraphEngineManager
 from libs import helper
 from models.model import App, AppMode, EndUser
 from services.app_generate_service import AppGenerateService
@@ -111,12 +110,7 @@ class WorkflowTaskStopApi(WebApiResource):
         if app_mode != AppMode.WORKFLOW:
             raise NotWorkflowAppError()
 
-        # Stop using both mechanisms for backward compatibility
-        # Legacy stop flag mechanism (without user check)
-        AppQueueManager.set_stop_flag_no_user_check(task_id)
-
-        # New graph engine command channel mechanism
-        GraphEngineManager.send_stop_command(task_id)
+        AppQueueManager.set_stop_flag(task_id, InvokeFrom.WEB_APP, end_user.id)
 
         return {"result": "success"}
 

api/core/app/app_config/easy_ui_based_app/model_config/manager.py

@@ -4,8 +4,8 @@ from typing import Any
 from core.app.app_config.entities import ModelConfigEntity
 from core.model_runtime.entities.model_entities import ModelPropertyKey, ModelType
 from core.model_runtime.model_providers.model_provider_factory import ModelProviderFactory
+from core.plugin.entities.plugin import ModelProviderID
 from core.provider_manager import ProviderManager
-from models.provider_ids import ModelProviderID
 
 
 class ModelConfigManager:

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

@@ -1,11 +1,11 @@
 import logging
-import time
 from collections.abc import Mapping
 from typing import Any, Optional, cast
 
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
+from configs import dify_config
 from core.app.apps.advanced_chat.app_config_manager import AdvancedChatAppConfig
 from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.apps.workflow_app_runner import WorkflowBasedAppRunner
@@ -23,17 +23,16 @@ from core.app.features.annotation_reply.annotation_reply import AnnotationReplyF
 from core.moderation.base import ModerationError
 from core.moderation.input_moderation import InputModeration
 from core.variables.variables import VariableUnion
-from core.workflow.entities import GraphRuntimeState, VariablePool
-from core.workflow.graph_engine.command_channels.redis_channel import RedisChannel
+from core.workflow.callbacks import WorkflowCallback, WorkflowLoggingCallback
+from core.workflow.entities.variable_pool import VariablePool
 from core.workflow.system_variable import SystemVariable
 from core.workflow.variable_loader import VariableLoader
 from core.workflow.workflow_entry import WorkflowEntry
 from extensions.ext_database import db
-from extensions.ext_redis import redis_client
 from models import Workflow
 from models.enums import UserFrom
 from models.model import App, Conversation, Message, MessageAnnotation
-from models.workflow import ConversationVariable
+from models.workflow import ConversationVariable, WorkflowType
 
 logger = logging.getLogger(__name__)
 
@@ -77,32 +76,24 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
         if not app_record:
             raise ValueError("App not found")
 
+        workflow_callbacks: list[WorkflowCallback] = []
+        if dify_config.DEBUG:
+            workflow_callbacks.append(WorkflowLoggingCallback())
+
         if self.application_generate_entity.single_iteration_run:
             # if only single iteration run is requested
-            graph_runtime_state = GraphRuntimeState(
-                variable_pool=VariablePool.empty(),
-                start_at=time.time(),
-            )
             graph, variable_pool = self._get_graph_and_variable_pool_of_single_iteration(
                 workflow=self._workflow,
                 node_id=self.application_generate_entity.single_iteration_run.node_id,
                 user_inputs=dict(self.application_generate_entity.single_iteration_run.inputs),
-                graph_runtime_state=graph_runtime_state,
             )
-            graph_runtime_state.variable_pool = variable_pool
         elif self.application_generate_entity.single_loop_run:
             # if only single loop run is requested
-            graph_runtime_state = GraphRuntimeState(
-                variable_pool=VariablePool.empty(),
-                start_at=time.time(),
-            )
             graph, variable_pool = self._get_graph_and_variable_pool_of_single_loop(
                 workflow=self._workflow,
                 node_id=self.application_generate_entity.single_loop_run.node_id,
                 user_inputs=dict(self.application_generate_entity.single_loop_run.inputs),
-                graph_runtime_state=graph_runtime_state,
             )
-            graph_runtime_state.variable_pool = variable_pool
         else:
             inputs = self.application_generate_entity.inputs
             query = self.application_generate_entity.query
@@ -149,31 +140,20 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
                 environment_variables=self._workflow.environment_variables,
                 # Based on the definition of `VariableUnion`,
                 # `list[Variable]` can be safely used as `list[VariableUnion]` since they are compatible.
-                conversation_variables=cast(list[VariableUnion], conversation_variables),
+                conversation_variables=conversation_variables,
             )
 
             # init graph
-            graph_runtime_state = GraphRuntimeState(variable_pool=variable_pool, start_at=time.time())
-            graph = self._init_graph(
-                graph_config=self._workflow.graph_dict,
-                graph_runtime_state=graph_runtime_state,
-                workflow_id=self._workflow.id,
-                tenant_id=self._workflow.tenant_id,
-                user_id=self.application_generate_entity.user_id,
-            )
+            graph = self._init_graph(graph_config=self._workflow.graph_dict)
 
         db.session.close()
 
         # RUN WORKFLOW
-        # Create Redis command channel for this workflow execution
-        task_id = self.application_generate_entity.task_id
-        channel_key = f"workflow:{task_id}:commands"
-        command_channel = RedisChannel(redis_client, channel_key)
-
         workflow_entry = WorkflowEntry(
             tenant_id=self._workflow.tenant_id,
             app_id=self._workflow.app_id,
             workflow_id=self._workflow.id,
+            workflow_type=WorkflowType.value_of(self._workflow.type),
             graph=graph,
             graph_config=self._workflow.graph_dict,
             user_id=self.application_generate_entity.user_id,
@@ -184,11 +164,12 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
             ),
             invoke_from=self.application_generate_entity.invoke_from,
             call_depth=self.application_generate_entity.call_depth,
-            graph_runtime_state=graph_runtime_state,
-            command_channel=command_channel,
+            variable_pool=variable_pool,
         )
 
-        generator = workflow_entry.run()
+        generator = workflow_entry.run(
+            callbacks=workflow_callbacks,
+        )
 
         for event in generator:
             self._handle_event(workflow_entry, event)

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

@@ -31,9 +31,14 @@ from core.app.entities.queue_entities import (
     QueueMessageReplaceEvent,
     QueueNodeExceptionEvent,
     QueueNodeFailedEvent,
+    QueueNodeInIterationFailedEvent,
+    QueueNodeInLoopFailedEvent,
     QueueNodeRetryEvent,
     QueueNodeStartedEvent,
     QueueNodeSucceededEvent,
+    QueueParallelBranchRunFailedEvent,
+    QueueParallelBranchRunStartedEvent,
+    QueueParallelBranchRunSucceededEvent,
     QueuePingEvent,
     QueueRetrieverResourcesEvent,
     QueueStopEvent,
@@ -60,8 +65,8 @@ from core.app.task_pipeline.message_cycle_manager import MessageCycleManager
 from core.base.tts import AppGeneratorTTSPublisher, AudioTrunk
 from core.model_runtime.entities.llm_entities import LLMUsage
 from core.ops.ops_trace_manager import TraceQueueManager
-from core.workflow.entities import GraphRuntimeState
-from core.workflow.enums import WorkflowExecutionStatus, WorkflowType
+from core.workflow.entities.workflow_execution import WorkflowExecutionStatus, WorkflowType
+from core.workflow.graph_engine.entities.graph_runtime_state import GraphRuntimeState
 from core.workflow.nodes import NodeType
 from core.workflow.repositories.draft_variable_repository import DraftVariableSaverFactory
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
@@ -390,7 +395,9 @@ class AdvancedChatAppGenerateTaskPipeline:
 
     def _handle_node_failed_events(
         self,
-        event: Union[QueueNodeFailedEvent, QueueNodeExceptionEvent],
+        event: Union[
+            QueueNodeFailedEvent, QueueNodeInIterationFailedEvent, QueueNodeInLoopFailedEvent, QueueNodeExceptionEvent
+        ],
         **kwargs,
     ) -> Generator[StreamResponse, None, None]:
         """Handle various node failure events."""
@@ -435,6 +442,32 @@ class AdvancedChatAppGenerateTaskPipeline:
             answer=delta_text, message_id=self._message_id, from_variable_selector=event.from_variable_selector
         )
 
+    def _handle_parallel_branch_started_event(
+        self, event: QueueParallelBranchRunStartedEvent, **kwargs
+    ) -> Generator[StreamResponse, None, None]:
+        """Handle parallel branch started events."""
+        self._ensure_workflow_initialized()
+
+        parallel_start_resp = self._workflow_response_converter.workflow_parallel_branch_start_to_stream_response(
+            task_id=self._application_generate_entity.task_id,
+            workflow_execution_id=self._workflow_run_id,
+            event=event,
+        )
+        yield parallel_start_resp
+
+    def _handle_parallel_branch_finished_events(
+        self, event: Union[QueueParallelBranchRunSucceededEvent, QueueParallelBranchRunFailedEvent], **kwargs
+    ) -> Generator[StreamResponse, None, None]:
+        """Handle parallel branch finished events."""
+        self._ensure_workflow_initialized()
+
+        parallel_finish_resp = self._workflow_response_converter.workflow_parallel_branch_finished_to_stream_response(
+            task_id=self._application_generate_entity.task_id,
+            workflow_execution_id=self._workflow_run_id,
+            event=event,
+        )
+        yield parallel_finish_resp
+
     def _handle_iteration_start_event(
         self, event: QueueIterationStartEvent, **kwargs
     ) -> Generator[StreamResponse, None, None]:
@@ -726,6 +759,8 @@ class AdvancedChatAppGenerateTaskPipeline:
             QueueNodeRetryEvent: self._handle_node_retry_event,
             QueueNodeStartedEvent: self._handle_node_started_event,
             QueueNodeSucceededEvent: self._handle_node_succeeded_event,
+            # Parallel branch events
+            QueueParallelBranchRunStartedEvent: self._handle_parallel_branch_started_event,
             # Iteration events
             QueueIterationStartEvent: self._handle_iteration_start_event,
             QueueIterationNextEvent: self._handle_iteration_next_event,
@@ -773,6 +808,8 @@ class AdvancedChatAppGenerateTaskPipeline:
             event,
             (
                 QueueNodeFailedEvent,
+                QueueNodeInIterationFailedEvent,
+                QueueNodeInLoopFailedEvent,
                 QueueNodeExceptionEvent,
             ),
         ):
@@ -785,6 +822,17 @@ class AdvancedChatAppGenerateTaskPipeline:
             )
             return
 
+        # Handle parallel branch finished events with isinstance check
+        if isinstance(event, (QueueParallelBranchRunSucceededEvent, QueueParallelBranchRunFailedEvent)):
+            yield from self._handle_parallel_branch_finished_events(
+                event,
+                graph_runtime_state=graph_runtime_state,
+                tts_publisher=tts_publisher,
+                trace_manager=trace_manager,
+                queue_message=queue_message,
+            )
+            return
+
         # For unhandled events, we continue (original behavior)
         return
 
@@ -808,6 +856,11 @@ class AdvancedChatAppGenerateTaskPipeline:
                     graph_runtime_state = event.graph_runtime_state
                     yield from self._handle_workflow_started_event(event)
 
+                case QueueTextChunkEvent():
+                    yield from self._handle_text_chunk_event(
+                        event, tts_publisher=tts_publisher, queue_message=queue_message
+                    )
+
                 case QueueErrorEvent():
                     yield from self._handle_error_event(event)
                     break

api/core/app/apps/base_app_generator.py

@@ -6,7 +6,7 @@ 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.nodes.enums import NodeType
 from core.workflow.repositories.draft_variable_repository import (
     DraftVariableSaver,
     DraftVariableSaverFactory,

api/core/app/apps/base_app_queue_manager.py

@@ -1,7 +1,7 @@
 import queue
 import time
 from abc import abstractmethod
-from enum import Enum
+from enum import IntEnum, auto
 from typing import Any, Optional
 
 from sqlalchemy.orm import DeclarativeMeta
@@ -19,9 +19,9 @@ from core.app.entities.queue_entities import (
 from extensions.ext_redis import redis_client
 
 
-class PublishFrom(Enum):
-    APPLICATION_MANAGER = 1
-    TASK_PIPELINE = 2
+class PublishFrom(IntEnum):
+    APPLICATION_MANAGER = auto()
+    TASK_PIPELINE = auto()
 
 
 class AppQueueManager:
@@ -126,21 +126,6 @@ class AppQueueManager:
         stopped_cache_key = cls._generate_stopped_cache_key(task_id)
         redis_client.setex(stopped_cache_key, 600, 1)
 
-    @classmethod
-    def set_stop_flag_no_user_check(cls, task_id: str) -> None:
-        """
-        Set task stop flag without user permission check.
-        This method allows stopping workflows without user context.
-
-        :param task_id: The task ID to stop
-        :return:
-        """
-        if not task_id:
-            return
-
-        stopped_cache_key = cls._generate_stopped_cache_key(task_id)
-        redis_client.setex(stopped_cache_key, 600, 1)
-
     def _is_stopped(self) -> bool:
         """
         Check if task is stopped

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

@@ -1,7 +1,7 @@
 import time
 from collections.abc import Mapping, Sequence
 from datetime import UTC, datetime
-from typing import Any, Optional, Union
+from typing import Any, Optional, Union, cast
 
 from sqlalchemy.orm import Session
 
@@ -16,9 +16,14 @@ from core.app.entities.queue_entities import (
     QueueLoopStartEvent,
     QueueNodeExceptionEvent,
     QueueNodeFailedEvent,
+    QueueNodeInIterationFailedEvent,
+    QueueNodeInLoopFailedEvent,
     QueueNodeRetryEvent,
     QueueNodeStartedEvent,
     QueueNodeSucceededEvent,
+    QueueParallelBranchRunFailedEvent,
+    QueueParallelBranchRunStartedEvent,
+    QueueParallelBranchRunSucceededEvent,
 )
 from core.app.entities.task_entities import (
     AgentLogStreamResponse,
@@ -31,16 +36,18 @@ from core.app.entities.task_entities import (
     NodeFinishStreamResponse,
     NodeRetryStreamResponse,
     NodeStartStreamResponse,
+    ParallelBranchFinishedStreamResponse,
+    ParallelBranchStartStreamResponse,
     WorkflowFinishStreamResponse,
     WorkflowStartStreamResponse,
 )
 from core.file import FILE_MODEL_IDENTITY, File
-from core.tools.entities.tool_entities import ToolProviderType
 from core.tools.tool_manager import ToolManager
 from core.variables.segments import ArrayFileSegment, FileSegment, Segment
-from core.workflow.entities import WorkflowExecution, WorkflowNodeExecution
-from core.workflow.enums import WorkflowNodeExecutionStatus
+from core.workflow.entities.workflow_execution import WorkflowExecution
+from core.workflow.entities.workflow_node_execution import WorkflowNodeExecution, WorkflowNodeExecutionStatus
 from core.workflow.nodes import NodeType
+from core.workflow.nodes.tool.entities import ToolNodeData
 from core.workflow.workflow_type_encoder import WorkflowRuntimeTypeConverter
 from libs.datetime_utils import naive_utc_now
 from models import (
@@ -164,10 +171,11 @@ class WorkflowResponseConverter:
 
         # extras logic
         if event.node_type == NodeType.TOOL:
+            node_data = cast(ToolNodeData, event.node_data)
             response.data.extras["icon"] = ToolManager.get_tool_icon(
                 tenant_id=self._application_generate_entity.app_config.tenant_id,
-                provider_type=ToolProviderType(event.provider_type),
-                provider_id=event.provider_id,
+                provider_type=node_data.provider_type,
+                provider_id=node_data.provider_id,
             )
 
         return response
@@ -175,7 +183,11 @@ class WorkflowResponseConverter:
     def workflow_node_finish_to_stream_response(
         self,
         *,
-        event: QueueNodeSucceededEvent | QueueNodeFailedEvent | QueueNodeExceptionEvent,
+        event: QueueNodeSucceededEvent
+        | QueueNodeFailedEvent
+        | QueueNodeInIterationFailedEvent
+        | QueueNodeInLoopFailedEvent
+        | QueueNodeExceptionEvent,
         task_id: str,
         workflow_node_execution: WorkflowNodeExecution,
     ) -> Optional[NodeFinishStreamResponse]:
@@ -209,6 +221,9 @@ class WorkflowResponseConverter:
                 finished_at=int(workflow_node_execution.finished_at.timestamp()),
                 files=self.fetch_files_from_node_outputs(workflow_node_execution.outputs or {}),
                 parallel_id=event.parallel_id,
+                parallel_start_node_id=event.parallel_start_node_id,
+                parent_parallel_id=event.parent_parallel_id,
+                parent_parallel_start_node_id=event.parent_parallel_start_node_id,
                 iteration_id=event.in_iteration_id,
                 loop_id=event.in_loop_id,
             ),
@@ -260,6 +275,50 @@ class WorkflowResponseConverter:
             ),
         )
 
+    def workflow_parallel_branch_start_to_stream_response(
+        self,
+        *,
+        task_id: str,
+        workflow_execution_id: str,
+        event: QueueParallelBranchRunStartedEvent,
+    ) -> ParallelBranchStartStreamResponse:
+        return ParallelBranchStartStreamResponse(
+            task_id=task_id,
+            workflow_run_id=workflow_execution_id,
+            data=ParallelBranchStartStreamResponse.Data(
+                parallel_id=event.parallel_id,
+                parallel_branch_id=event.parallel_start_node_id,
+                parent_parallel_id=event.parent_parallel_id,
+                parent_parallel_start_node_id=event.parent_parallel_start_node_id,
+                iteration_id=event.in_iteration_id,
+                loop_id=event.in_loop_id,
+                created_at=int(time.time()),
+            ),
+        )
+
+    def workflow_parallel_branch_finished_to_stream_response(
+        self,
+        *,
+        task_id: str,
+        workflow_execution_id: str,
+        event: QueueParallelBranchRunSucceededEvent | QueueParallelBranchRunFailedEvent,
+    ) -> ParallelBranchFinishedStreamResponse:
+        return ParallelBranchFinishedStreamResponse(
+            task_id=task_id,
+            workflow_run_id=workflow_execution_id,
+            data=ParallelBranchFinishedStreamResponse.Data(
+                parallel_id=event.parallel_id,
+                parallel_branch_id=event.parallel_start_node_id,
+                parent_parallel_id=event.parent_parallel_id,
+                parent_parallel_start_node_id=event.parent_parallel_start_node_id,
+                iteration_id=event.in_iteration_id,
+                loop_id=event.in_loop_id,
+                status="succeeded" if isinstance(event, QueueParallelBranchRunSucceededEvent) else "failed",
+                error=event.error if isinstance(event, QueueParallelBranchRunFailedEvent) else None,
+                created_at=int(time.time()),
+            ),
+        )
+
     def workflow_iteration_start_to_stream_response(
         self,
         *,
@@ -274,11 +333,13 @@ class WorkflowResponseConverter:
                 id=event.node_id,
                 node_id=event.node_id,
                 node_type=event.node_type.value,
-                title=event.node_title,
+                title=event.node_data.title,
                 created_at=int(time.time()),
                 extras={},
                 inputs=event.inputs or {},
                 metadata=event.metadata or {},
+                parallel_id=event.parallel_id,
+                parallel_start_node_id=event.parallel_start_node_id,
             ),
         )
 
@@ -296,10 +357,15 @@ class WorkflowResponseConverter:
                 id=event.node_id,
                 node_id=event.node_id,
                 node_type=event.node_type.value,
-                title=event.node_title,
+                title=event.node_data.title,
                 index=event.index,
+                pre_iteration_output=event.output,
                 created_at=int(time.time()),
                 extras={},
+                parallel_id=event.parallel_id,
+                parallel_start_node_id=event.parallel_start_node_id,
+                parallel_mode_run_id=event.parallel_mode_run_id,
+                duration=event.duration,
             ),
         )
 
@@ -318,7 +384,7 @@ class WorkflowResponseConverter:
                 id=event.node_id,
                 node_id=event.node_id,
                 node_type=event.node_type.value,
-                title=event.node_title,
+                title=event.node_data.title,
                 outputs=json_converter.to_json_encodable(event.outputs),
                 created_at=int(time.time()),
                 extras={},
@@ -332,6 +398,8 @@ class WorkflowResponseConverter:
                 execution_metadata=event.metadata,
                 finished_at=int(time.time()),
                 steps=event.steps,
+                parallel_id=event.parallel_id,
+                parallel_start_node_id=event.parallel_start_node_id,
             ),
         )
 
@@ -345,7 +413,7 @@ class WorkflowResponseConverter:
                 id=event.node_id,
                 node_id=event.node_id,
                 node_type=event.node_type.value,
-                title=event.node_title,
+                title=event.node_data.title,
                 created_at=int(time.time()),
                 extras={},
                 inputs=event.inputs or {},
@@ -369,7 +437,7 @@ class WorkflowResponseConverter:
                 id=event.node_id,
                 node_id=event.node_id,
                 node_type=event.node_type.value,
-                title=event.node_title,
+                title=event.node_data.title,
                 index=event.index,
                 pre_loop_output=event.output,
                 created_at=int(time.time()),
@@ -377,6 +445,7 @@ class WorkflowResponseConverter:
                 parallel_id=event.parallel_id,
                 parallel_start_node_id=event.parallel_start_node_id,
                 parallel_mode_run_id=event.parallel_mode_run_id,
+                duration=event.duration,
             ),
         )
 
@@ -394,7 +463,7 @@ class WorkflowResponseConverter:
                 id=event.node_id,
                 node_id=event.node_id,
                 node_type=event.node_type.value,
-                title=event.node_title,
+                title=event.node_data.title,
                 outputs=WorkflowRuntimeTypeConverter().to_json_encodable(event.outputs),
                 created_at=int(time.time()),
                 extras={},

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

@@ -3,7 +3,7 @@ import logging
 import threading
 import uuid
 from collections.abc import Generator, Mapping, Sequence
-from typing import Any, Literal, Union, overload
+from typing import Any, Literal, Optional, Union, overload
 
 from flask import Flask, current_app
 from pydantic import ValidationError
@@ -53,6 +53,9 @@ class WorkflowAppGenerator(BaseAppGenerator):
         invoke_from: InvokeFrom,
         streaming: Literal[True],
         call_depth: int,
+        workflow_thread_pool_id: Optional[str],
+        triggered_from: Optional[WorkflowRunTriggeredFrom] = None,
+        root_node_id: Optional[str] = None,
     ) -> Generator[Mapping | str, None, None]: ...
 
     @overload
@@ -66,6 +69,9 @@ class WorkflowAppGenerator(BaseAppGenerator):
         invoke_from: InvokeFrom,
         streaming: Literal[False],
         call_depth: int,
+        workflow_thread_pool_id: Optional[str],
+        triggered_from: Optional[WorkflowRunTriggeredFrom] = None,
+        root_node_id: Optional[str] = None,
     ) -> Mapping[str, Any]: ...
 
     @overload
@@ -79,6 +85,9 @@ class WorkflowAppGenerator(BaseAppGenerator):
         invoke_from: InvokeFrom,
         streaming: bool,
         call_depth: int,
+        workflow_thread_pool_id: Optional[str],
+        triggered_from: Optional[WorkflowRunTriggeredFrom] = None,
+        root_node_id: Optional[str] = None,
     ) -> Union[Mapping[str, Any], Generator[Mapping | str, None, None]]: ...
 
     def generate(
@@ -91,6 +100,9 @@ class WorkflowAppGenerator(BaseAppGenerator):
         invoke_from: InvokeFrom,
         streaming: bool = True,
         call_depth: int = 0,
+        workflow_thread_pool_id: Optional[str] = None,
+        triggered_from: Optional[WorkflowRunTriggeredFrom] = None,
+        root_node_id: Optional[str] = None,
     ) -> Union[Mapping[str, Any], Generator[Mapping | str, None, None]]:
         files: Sequence[Mapping[str, Any]] = args.get("files") or []
 
@@ -126,17 +138,20 @@ class WorkflowAppGenerator(BaseAppGenerator):
             **extract_external_trace_id_from_args(args),
         }
         workflow_run_id = str(uuid.uuid4())
+        if triggered_from in (WorkflowRunTriggeredFrom.DEBUGGING, WorkflowRunTriggeredFrom.APP_RUN):
+            # start node get inputs
+            inputs = self._prepare_user_inputs(
+                user_inputs=inputs,
+                variables=app_config.variables,
+                tenant_id=app_model.tenant_id,
+                strict_type_validation=True if invoke_from == InvokeFrom.SERVICE_API else False,
+            )
         # init application generate entity
         application_generate_entity = WorkflowAppGenerateEntity(
             task_id=str(uuid.uuid4()),
             app_config=app_config,
             file_upload_config=file_extra_config,
-            inputs=self._prepare_user_inputs(
-                user_inputs=inputs,
-                variables=app_config.variables,
-                tenant_id=app_model.tenant_id,
-                strict_type_validation=True if invoke_from == InvokeFrom.SERVICE_API else False,
-            ),
+            inputs=inputs,
             files=list(system_files),
             user_id=user.id,
             stream=streaming,
@@ -155,7 +170,10 @@ class WorkflowAppGenerator(BaseAppGenerator):
         # Create session factory
         session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
         # Create workflow execution(aka workflow run) repository
-        if invoke_from == InvokeFrom.DEBUGGER:
+        if triggered_from is not None:
+            # Use explicitly provided triggered_from (for async triggers)
+            workflow_triggered_from = triggered_from
+        elif invoke_from == InvokeFrom.DEBUGGER:
             workflow_triggered_from = WorkflowRunTriggeredFrom.DEBUGGING
         else:
             workflow_triggered_from = WorkflowRunTriggeredFrom.APP_RUN
@@ -182,6 +200,8 @@ class WorkflowAppGenerator(BaseAppGenerator):
             workflow_execution_repository=workflow_execution_repository,
             workflow_node_execution_repository=workflow_node_execution_repository,
             streaming=streaming,
+            workflow_thread_pool_id=workflow_thread_pool_id,
+            root_node_id=root_node_id,
         )
 
     def _generate(
@@ -195,7 +215,9 @@ class WorkflowAppGenerator(BaseAppGenerator):
         workflow_execution_repository: WorkflowExecutionRepository,
         workflow_node_execution_repository: WorkflowNodeExecutionRepository,
         streaming: bool = True,
+        workflow_thread_pool_id: Optional[str] = None,
         variable_loader: VariableLoader = DUMMY_VARIABLE_LOADER,
+        root_node_id: Optional[str] = None,
     ) -> Union[Mapping[str, Any], Generator[str | Mapping[str, Any], None, None]]:
         """
         Generate App response.
@@ -208,6 +230,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
         :param workflow_execution_repository: repository for workflow execution
         :param workflow_node_execution_repository: repository for workflow node execution
         :param streaming: is stream
+        :param workflow_thread_pool_id: workflow thread pool id
         """
         # init queue manager
         queue_manager = WorkflowAppQueueManager(
@@ -230,7 +253,9 @@ class WorkflowAppGenerator(BaseAppGenerator):
                 "application_generate_entity": application_generate_entity,
                 "queue_manager": queue_manager,
                 "context": context,
+                "workflow_thread_pool_id": workflow_thread_pool_id,
                 "variable_loader": variable_loader,
+                "root_node_id": root_node_id,
             },
         )
 
@@ -426,7 +451,18 @@ class WorkflowAppGenerator(BaseAppGenerator):
         queue_manager: AppQueueManager,
         context: contextvars.Context,
         variable_loader: VariableLoader,
+        workflow_thread_pool_id: Optional[str] = None,
+        root_node_id: Optional[str] = None,
     ) -> None:
+        """
+        Generate worker in a new thread.
+        :param flask_app: Flask app
+        :param application_generate_entity: application generate entity
+        :param queue_manager: queue manager
+        :param workflow_thread_pool_id: workflow thread pool id
+        :return:
+        """
+
         with preserve_flask_contexts(flask_app, context_vars=context):
             with Session(db.engine, expire_on_commit=False) as session:
                 workflow = session.scalar(
@@ -456,9 +492,11 @@ class WorkflowAppGenerator(BaseAppGenerator):
             runner = WorkflowAppRunner(
                 application_generate_entity=application_generate_entity,
                 queue_manager=queue_manager,
+                workflow_thread_pool_id=workflow_thread_pool_id,
                 variable_loader=variable_loader,
                 workflow=workflow,
                 system_user_id=system_user_id,
+                root_node_id=root_node_id,
             )
 
             try:

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

@@ -1,7 +1,7 @@
 import logging
-import time
-from typing import cast
+from typing import Optional, cast
 
+from configs import dify_config
 from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.apps.workflow.app_config_manager import WorkflowAppConfig
 from core.app.apps.workflow_app_runner import WorkflowBasedAppRunner
@@ -9,14 +9,13 @@ from core.app.entities.app_invoke_entities import (
     InvokeFrom,
     WorkflowAppGenerateEntity,
 )
-from core.workflow.entities import GraphRuntimeState, VariablePool
-from core.workflow.graph_engine.command_channels.redis_channel import RedisChannel
+from core.workflow.callbacks import WorkflowCallback, WorkflowLoggingCallback
+from core.workflow.entities.variable_pool import VariablePool
 from core.workflow.system_variable import SystemVariable
 from core.workflow.variable_loader import VariableLoader
 from core.workflow.workflow_entry import WorkflowEntry
-from extensions.ext_redis import redis_client
 from models.enums import UserFrom
-from models.workflow import Workflow
+from models.workflow import Workflow, WorkflowType
 
 logger = logging.getLogger(__name__)
 
@@ -32,8 +31,10 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
         application_generate_entity: WorkflowAppGenerateEntity,
         queue_manager: AppQueueManager,
         variable_loader: VariableLoader,
+        workflow_thread_pool_id: Optional[str] = None,
         workflow: Workflow,
         system_user_id: str,
+        root_node_id: Optional[str] = None,
     ) -> None:
         super().__init__(
             queue_manager=queue_manager,
@@ -41,8 +42,10 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
             app_id=application_generate_entity.app_config.app_id,
         )
         self.application_generate_entity = application_generate_entity
+        self.workflow_thread_pool_id = workflow_thread_pool_id
         self._workflow = workflow
         self._sys_user_id = system_user_id
+        self._root_node_id = root_node_id
 
     def run(self) -> None:
         """
@@ -51,30 +54,24 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
         app_config = self.application_generate_entity.app_config
         app_config = cast(WorkflowAppConfig, app_config)
 
+        workflow_callbacks: list[WorkflowCallback] = []
+        if dify_config.DEBUG:
+            workflow_callbacks.append(WorkflowLoggingCallback())
+
         # if only single iteration run is requested
         if self.application_generate_entity.single_iteration_run:
             # if only single iteration run is requested
-            graph_runtime_state = GraphRuntimeState(
-                variable_pool=VariablePool.empty(),
-                start_at=time.time(),
-            )
             graph, variable_pool = self._get_graph_and_variable_pool_of_single_iteration(
                 workflow=self._workflow,
                 node_id=self.application_generate_entity.single_iteration_run.node_id,
                 user_inputs=self.application_generate_entity.single_iteration_run.inputs,
-                graph_runtime_state=graph_runtime_state,
             )
         elif self.application_generate_entity.single_loop_run:
             # if only single loop run is requested
-            graph_runtime_state = GraphRuntimeState(
-                variable_pool=VariablePool.empty(),
-                start_at=time.time(),
-            )
             graph, variable_pool = self._get_graph_and_variable_pool_of_single_loop(
                 workflow=self._workflow,
                 node_id=self.application_generate_entity.single_loop_run.node_id,
                 user_inputs=self.application_generate_entity.single_loop_run.inputs,
-                graph_runtime_state=graph_runtime_state,
             )
         else:
             inputs = self.application_generate_entity.inputs
@@ -97,26 +94,15 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
                 conversation_variables=[],
             )
 
-            graph_runtime_state = GraphRuntimeState(variable_pool=variable_pool, start_at=time.perf_counter())
-
             # init graph
-            graph = self._init_graph(
-                graph_config=self._workflow.graph_dict,
-                graph_runtime_state=graph_runtime_state,
-                workflow_id=self._workflow.id,
-                tenant_id=self._workflow.tenant_id,
-            )
+            graph = self._init_graph(graph_config=self._workflow.graph_dict, root_node_id=self._root_node_id)
 
         # RUN WORKFLOW
-        # Create Redis command channel for this workflow execution
-        task_id = self.application_generate_entity.task_id
-        channel_key = f"workflow:{task_id}:commands"
-        command_channel = RedisChannel(redis_client, channel_key)
-
         workflow_entry = WorkflowEntry(
             tenant_id=self._workflow.tenant_id,
             app_id=self._workflow.app_id,
             workflow_id=self._workflow.id,
+            workflow_type=WorkflowType.value_of(self._workflow.type),
             graph=graph,
             graph_config=self._workflow.graph_dict,
             user_id=self.application_generate_entity.user_id,
@@ -127,11 +113,11 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
             ),
             invoke_from=self.application_generate_entity.invoke_from,
             call_depth=self.application_generate_entity.call_depth,
-            graph_runtime_state=graph_runtime_state,
-            command_channel=command_channel,
+            variable_pool=variable_pool,
+            thread_pool_id=self.workflow_thread_pool_id,
         )
 
-        generator = workflow_entry.run()
+        generator = workflow_entry.run(callbacks=workflow_callbacks)
 
         for event in generator:
             self._handle_event(workflow_entry, event)

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

@@ -2,7 +2,7 @@ import logging
 import time
 from collections.abc import Callable, Generator
 from contextlib import contextmanager
-from typing import Optional, Union
+from typing import Any, Optional, Union
 
 from sqlalchemy.orm import Session
 
@@ -14,7 +14,6 @@ from core.app.entities.app_invoke_entities import (
     WorkflowAppGenerateEntity,
 )
 from core.app.entities.queue_entities import (
-    AppQueueEvent,
     MessageQueueMessage,
     QueueAgentLogEvent,
     QueueErrorEvent,
@@ -26,9 +25,14 @@ from core.app.entities.queue_entities import (
     QueueLoopStartEvent,
     QueueNodeExceptionEvent,
     QueueNodeFailedEvent,
+    QueueNodeInIterationFailedEvent,
+    QueueNodeInLoopFailedEvent,
     QueueNodeRetryEvent,
     QueueNodeStartedEvent,
     QueueNodeSucceededEvent,
+    QueueParallelBranchRunFailedEvent,
+    QueueParallelBranchRunStartedEvent,
+    QueueParallelBranchRunSucceededEvent,
     QueuePingEvent,
     QueueStopEvent,
     QueueTextChunkEvent,
@@ -53,8 +57,8 @@ from core.app.entities.task_entities import (
 from core.app.task_pipeline.based_generate_task_pipeline import BasedGenerateTaskPipeline
 from core.base.tts import AppGeneratorTTSPublisher, AudioTrunk
 from core.ops.ops_trace_manager import TraceQueueManager
-from core.workflow.entities import GraphRuntimeState, WorkflowExecution
-from core.workflow.enums import WorkflowExecutionStatus, WorkflowType
+from core.workflow.entities.workflow_execution import WorkflowExecution, WorkflowExecutionStatus, WorkflowType
+from core.workflow.graph_engine.entities.graph_runtime_state import GraphRuntimeState
 from core.workflow.repositories.draft_variable_repository import DraftVariableSaverFactory
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
@@ -346,7 +350,9 @@ class WorkflowAppGenerateTaskPipeline:
 
     def _handle_node_failed_events(
         self,
-        event: Union[QueueNodeFailedEvent, QueueNodeExceptionEvent],
+        event: Union[
+            QueueNodeFailedEvent, QueueNodeInIterationFailedEvent, QueueNodeInLoopFailedEvent, QueueNodeExceptionEvent
+        ],
         **kwargs,
     ) -> Generator[StreamResponse, None, None]:
         """Handle various node failure events."""
@@ -365,6 +371,32 @@ class WorkflowAppGenerateTaskPipeline:
         if node_failed_response:
             yield node_failed_response
 
+    def _handle_parallel_branch_started_event(
+        self, event: QueueParallelBranchRunStartedEvent, **kwargs
+    ) -> Generator[StreamResponse, None, None]:
+        """Handle parallel branch started events."""
+        self._ensure_workflow_initialized()
+
+        parallel_start_resp = self._workflow_response_converter.workflow_parallel_branch_start_to_stream_response(
+            task_id=self._application_generate_entity.task_id,
+            workflow_execution_id=self._workflow_run_id,
+            event=event,
+        )
+        yield parallel_start_resp
+
+    def _handle_parallel_branch_finished_events(
+        self, event: Union[QueueParallelBranchRunSucceededEvent, QueueParallelBranchRunFailedEvent], **kwargs
+    ) -> Generator[StreamResponse, None, None]:
+        """Handle parallel branch finished events."""
+        self._ensure_workflow_initialized()
+
+        parallel_finish_resp = self._workflow_response_converter.workflow_parallel_branch_finished_to_stream_response(
+            task_id=self._application_generate_entity.task_id,
+            workflow_execution_id=self._workflow_run_id,
+            event=event,
+        )
+        yield parallel_finish_resp
+
     def _handle_iteration_start_event(
         self, event: QueueIterationStartEvent, **kwargs
     ) -> Generator[StreamResponse, None, None]:
@@ -586,6 +618,8 @@ class WorkflowAppGenerateTaskPipeline:
             QueueNodeRetryEvent: self._handle_node_retry_event,
             QueueNodeStartedEvent: self._handle_node_started_event,
             QueueNodeSucceededEvent: self._handle_node_succeeded_event,
+            # Parallel branch events
+            QueueParallelBranchRunStartedEvent: self._handle_parallel_branch_started_event,
             # Iteration events
             QueueIterationStartEvent: self._handle_iteration_start_event,
             QueueIterationNextEvent: self._handle_iteration_next_event,
@@ -600,7 +634,7 @@ class WorkflowAppGenerateTaskPipeline:
 
     def _dispatch_event(
         self,
-        event: AppQueueEvent,
+        event: Any,
         *,
         graph_runtime_state: Optional[GraphRuntimeState] = None,
         tts_publisher: Optional[AppGeneratorTTSPublisher] = None,
@@ -627,6 +661,8 @@ class WorkflowAppGenerateTaskPipeline:
             event,
             (
                 QueueNodeFailedEvent,
+                QueueNodeInIterationFailedEvent,
+                QueueNodeInLoopFailedEvent,
                 QueueNodeExceptionEvent,
             ),
         ):
@@ -639,6 +675,17 @@ class WorkflowAppGenerateTaskPipeline:
             )
             return
 
+        # Handle parallel branch finished events with isinstance check
+        if isinstance(event, (QueueParallelBranchRunSucceededEvent, QueueParallelBranchRunFailedEvent)):
+            yield from self._handle_parallel_branch_finished_events(
+                event,
+                graph_runtime_state=graph_runtime_state,
+                tts_publisher=tts_publisher,
+                trace_manager=trace_manager,
+                queue_message=queue_message,
+            )
+            return
+
         # Handle workflow failed and stop events with isinstance check
         if isinstance(event, (QueueWorkflowFailedEvent, QueueStopEvent)):
             yield from self._handle_workflow_failed_and_stop_events(

api/core/app/apps/workflow_app_runner.py

@@ -1,8 +1,7 @@
 from collections.abc import Mapping
-from typing import Any, cast
+from typing import Any, Optional, cast
 
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
-from core.app.entities.app_invoke_entities import InvokeFrom
 from core.app.entities.queue_entities import (
     AppQueueEvent,
     QueueAgentLogEvent,
@@ -14,9 +13,14 @@ from core.app.entities.queue_entities import (
     QueueLoopStartEvent,
     QueueNodeExceptionEvent,
     QueueNodeFailedEvent,
+    QueueNodeInIterationFailedEvent,
+    QueueNodeInLoopFailedEvent,
     QueueNodeRetryEvent,
     QueueNodeStartedEvent,
     QueueNodeSucceededEvent,
+    QueueParallelBranchRunFailedEvent,
+    QueueParallelBranchRunStartedEvent,
+    QueueParallelBranchRunSucceededEvent,
     QueueRetrieverResourcesEvent,
     QueueTextChunkEvent,
     QueueWorkflowFailedEvent,
@@ -24,39 +28,42 @@ from core.app.entities.queue_entities import (
     QueueWorkflowStartedEvent,
     QueueWorkflowSucceededEvent,
 )
-from core.workflow.entities import GraphInitParams, GraphRuntimeState, VariablePool
-from core.workflow.graph import Graph
-from core.workflow.graph_events import (
+from core.workflow.entities.variable_pool import VariablePool
+from core.workflow.entities.workflow_node_execution import WorkflowNodeExecutionMetadataKey
+from core.workflow.graph_engine.entities.event import (
+    AgentLogEvent,
     GraphEngineEvent,
     GraphRunFailedEvent,
     GraphRunPartialSucceededEvent,
     GraphRunStartedEvent,
     GraphRunSucceededEvent,
-    NodeRunAgentLogEvent,
+    IterationRunFailedEvent,
+    IterationRunNextEvent,
+    IterationRunStartedEvent,
+    IterationRunSucceededEvent,
+    LoopRunFailedEvent,
+    LoopRunNextEvent,
+    LoopRunStartedEvent,
+    LoopRunSucceededEvent,
+    NodeInIterationFailedEvent,
+    NodeInLoopFailedEvent,
     NodeRunExceptionEvent,
     NodeRunFailedEvent,
-    NodeRunIterationFailedEvent,
-    NodeRunIterationNextEvent,
-    NodeRunIterationStartedEvent,
-    NodeRunIterationSucceededEvent,
-    NodeRunLoopFailedEvent,
-    NodeRunLoopNextEvent,
-    NodeRunLoopStartedEvent,
-    NodeRunLoopSucceededEvent,
     NodeRunRetrieverResourceEvent,
     NodeRunRetryEvent,
     NodeRunStartedEvent,
     NodeRunStreamChunkEvent,
     NodeRunSucceededEvent,
+    ParallelBranchRunFailedEvent,
+    ParallelBranchRunStartedEvent,
+    ParallelBranchRunSucceededEvent,
 )
-from core.workflow.graph_events.graph import GraphRunAbortedEvent
+from core.workflow.graph_engine.entities.graph import Graph
 from core.workflow.nodes import NodeType
-from core.workflow.nodes.node_factory import DifyNodeFactory
 from core.workflow.nodes.node_mapping import NODE_TYPE_CLASSES_MAPPING
 from core.workflow.system_variable import SystemVariable
 from core.workflow.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader, load_into_variable_pool
 from core.workflow.workflow_entry import WorkflowEntry
-from models.enums import UserFrom
 from models.workflow import Workflow
 
 
@@ -72,14 +79,7 @@ class WorkflowBasedAppRunner:
         self._variable_loader = variable_loader
         self._app_id = app_id
 
-    def _init_graph(
-        self,
-        graph_config: Mapping[str, Any],
-        graph_runtime_state: GraphRuntimeState,
-        workflow_id: str = "",
-        tenant_id: str = "",
-        user_id: str = "",
-    ) -> Graph:
+    def _init_graph(self, graph_config: Mapping[str, Any], root_node_id: Optional[str] = None) -> Graph:
         """
         Init graph
         """
@@ -91,28 +91,8 @@ class WorkflowBasedAppRunner:
 
         if not isinstance(graph_config.get("edges"), list):
             raise ValueError("edges in workflow graph must be a list")
-
-        # Create required parameters for Graph.init
-        graph_init_params = GraphInitParams(
-            tenant_id=tenant_id or "",
-            app_id=self._app_id,
-            workflow_id=workflow_id,
-            graph_config=graph_config,
-            user_id=user_id,
-            user_from=UserFrom.ACCOUNT.value,
-            invoke_from=InvokeFrom.SERVICE_API.value,
-            call_depth=0,
-        )
-
-        # Use the provided graph_runtime_state for consistent state management
-
-        node_factory = DifyNodeFactory(
-            graph_init_params=graph_init_params,
-            graph_runtime_state=graph_runtime_state,
-        )
-
         # init graph
-        graph = Graph.init(graph_config=graph_config, node_factory=node_factory)
+        graph = Graph.init(graph_config=graph_config, root_node_id=root_node_id)
 
         if not graph:
             raise ValueError("graph not found in workflow")
@@ -124,7 +104,6 @@ class WorkflowBasedAppRunner:
         workflow: Workflow,
         node_id: str,
         user_inputs: dict,
-        graph_runtime_state: GraphRuntimeState,
     ) -> tuple[Graph, VariablePool]:
         """
         Get variable pool of single iteration
@@ -149,9 +128,7 @@ class WorkflowBasedAppRunner:
         node_configs = [
             node
             for node in graph_config.get("nodes", [])
-            if node.get("id") == node_id
-            or node.get("data", {}).get("iteration_id", "") == node_id
-            or node.get("id") == f"{node_id}start"
+            if node.get("id") == node_id or node.get("data", {}).get("iteration_id", "") == node_id
         ]
 
         graph_config["nodes"] = node_configs
@@ -168,25 +145,8 @@ class WorkflowBasedAppRunner:
 
         graph_config["edges"] = edge_configs
 
-        # Create required parameters for Graph.init
-        graph_init_params = GraphInitParams(
-            tenant_id=workflow.tenant_id,
-            app_id=self._app_id,
-            workflow_id=workflow.id,
-            graph_config=graph_config,
-            user_id="",
-            user_from=UserFrom.ACCOUNT.value,
-            invoke_from=InvokeFrom.SERVICE_API.value,
-            call_depth=0,
-        )
-
-        node_factory = DifyNodeFactory(
-            graph_init_params=graph_init_params,
-            graph_runtime_state=graph_runtime_state,
-        )
-
         # init graph
-        graph = Graph.init(graph_config=graph_config, node_factory=node_factory, root_node_id=node_id)
+        graph = Graph.init(graph_config=graph_config, root_node_id=node_id)
 
         if not graph:
             raise ValueError("graph not found in workflow")
@@ -241,7 +201,6 @@ class WorkflowBasedAppRunner:
         workflow: Workflow,
         node_id: str,
         user_inputs: dict,
-        graph_runtime_state: GraphRuntimeState,
     ) -> tuple[Graph, VariablePool]:
         """
         Get variable pool of single loop
@@ -266,9 +225,7 @@ class WorkflowBasedAppRunner:
         node_configs = [
             node
             for node in graph_config.get("nodes", [])
-            if node.get("id") == node_id
-            or node.get("data", {}).get("loop_id", "") == node_id
-            or node.get("id") == f"{node_id}start"
+            if node.get("id") == node_id or node.get("data", {}).get("loop_id", "") == node_id
         ]
 
         graph_config["nodes"] = node_configs
@@ -285,25 +242,8 @@ class WorkflowBasedAppRunner:
 
         graph_config["edges"] = edge_configs
 
-        # Create required parameters for Graph.init
-        graph_init_params = GraphInitParams(
-            tenant_id=workflow.tenant_id,
-            app_id=self._app_id,
-            workflow_id=workflow.id,
-            graph_config=graph_config,
-            user_id="",
-            user_from=UserFrom.ACCOUNT.value,
-            invoke_from=InvokeFrom.SERVICE_API.value,
-            call_depth=0,
-        )
-
-        node_factory = DifyNodeFactory(
-            graph_init_params=graph_init_params,
-            graph_runtime_state=graph_runtime_state,
-        )
-
         # init graph
-        graph = Graph.init(graph_config=graph_config, node_factory=node_factory, root_node_id=node_id)
+        graph = Graph.init(graph_config=graph_config, root_node_id=node_id)
 
         if not graph:
             raise ValueError("graph not found in workflow")
@@ -370,21 +310,29 @@ class WorkflowBasedAppRunner:
             )
         elif isinstance(event, GraphRunFailedEvent):
             self._publish_event(QueueWorkflowFailedEvent(error=event.error, exceptions_count=event.exceptions_count))
-        elif isinstance(event, GraphRunAbortedEvent):
-            self._publish_event(QueueWorkflowFailedEvent(error=event.reason or "Unknown error", exceptions_count=0))
         elif isinstance(event, NodeRunRetryEvent):
-            node_run_result = event.node_run_result
-            inputs = node_run_result.inputs
-            process_data = node_run_result.process_data
-            outputs = node_run_result.outputs
-            execution_metadata = node_run_result.metadata
+            node_run_result = event.route_node_state.node_run_result
+            inputs: Mapping[str, Any] | None = {}
+            process_data: Mapping[str, Any] | None = {}
+            outputs: Mapping[str, Any] | None = {}
+            execution_metadata: Mapping[WorkflowNodeExecutionMetadataKey, Any] | None = {}
+            if node_run_result:
+                inputs = node_run_result.inputs
+                process_data = node_run_result.process_data
+                outputs = node_run_result.outputs
+                execution_metadata = node_run_result.metadata
             self._publish_event(
                 QueueNodeRetryEvent(
                     node_execution_id=event.id,
                     node_id=event.node_id,
-                    node_title=event.node_title,
                     node_type=event.node_type,
+                    node_data=event.node_data,
+                    parallel_id=event.parallel_id,
+                    parallel_start_node_id=event.parallel_start_node_id,
+                    parent_parallel_id=event.parent_parallel_id,
+                    parent_parallel_start_node_id=event.parent_parallel_start_node_id,
                     start_at=event.start_at,
+                    node_run_index=event.route_node_state.index,
                     predecessor_node_id=event.predecessor_node_id,
                     in_iteration_id=event.in_iteration_id,
                     in_loop_id=event.in_loop_id,
@@ -395,8 +343,6 @@ class WorkflowBasedAppRunner:
                     error=event.error,
                     execution_metadata=execution_metadata,
                     retry_index=event.retry_index,
-                    provider_type=event.provider_type,
-                    provider_id=event.provider_id,
                 )
             )
         elif isinstance(event, NodeRunStartedEvent):
@@ -404,30 +350,44 @@ class WorkflowBasedAppRunner:
                 QueueNodeStartedEvent(
                     node_execution_id=event.id,
                     node_id=event.node_id,
-                    node_title=event.node_title,
                     node_type=event.node_type,
-                    start_at=event.start_at,
+                    node_data=event.node_data,
+                    parallel_id=event.parallel_id,
+                    parallel_start_node_id=event.parallel_start_node_id,
+                    parent_parallel_id=event.parent_parallel_id,
+                    parent_parallel_start_node_id=event.parent_parallel_start_node_id,
+                    start_at=event.route_node_state.start_at,
+                    node_run_index=event.route_node_state.index,
                     predecessor_node_id=event.predecessor_node_id,
                     in_iteration_id=event.in_iteration_id,
                     in_loop_id=event.in_loop_id,
                     parallel_mode_run_id=event.parallel_mode_run_id,
                     agent_strategy=event.agent_strategy,
-                    provider_type=event.provider_type,
-                    provider_id=event.provider_id,
                 )
             )
         elif isinstance(event, NodeRunSucceededEvent):
-            node_run_result = event.node_run_result
-            inputs = node_run_result.inputs
-            process_data = node_run_result.process_data
-            outputs = node_run_result.outputs
-            execution_metadata = node_run_result.metadata
+            node_run_result = event.route_node_state.node_run_result
+            if node_run_result:
+                inputs = node_run_result.inputs
+                process_data = node_run_result.process_data
+                outputs = node_run_result.outputs
+                execution_metadata = node_run_result.metadata
+            else:
+                inputs = {}
+                process_data = {}
+                outputs = {}
+                execution_metadata = {}
             self._publish_event(
                 QueueNodeSucceededEvent(
                     node_execution_id=event.id,
                     node_id=event.node_id,
                     node_type=event.node_type,
-                    start_at=event.start_at,
+                    node_data=event.node_data,
+                    parallel_id=event.parallel_id,
+                    parallel_start_node_id=event.parallel_start_node_id,
+                    parent_parallel_id=event.parent_parallel_id,
+                    parent_parallel_start_node_id=event.parent_parallel_start_node_id,
+                    start_at=event.route_node_state.start_at,
                     inputs=inputs,
                     process_data=process_data,
                     outputs=outputs,
@@ -436,18 +396,34 @@ class WorkflowBasedAppRunner:
                     in_loop_id=event.in_loop_id,
                 )
             )
+
         elif isinstance(event, NodeRunFailedEvent):
             self._publish_event(
                 QueueNodeFailedEvent(
                     node_execution_id=event.id,
                     node_id=event.node_id,
                     node_type=event.node_type,
-                    start_at=event.start_at,
-                    inputs=event.node_run_result.inputs,
-                    process_data=event.node_run_result.process_data,
-                    outputs=event.node_run_result.outputs,
-                    error=event.node_run_result.error or "Unknown error",
-                    execution_metadata=event.node_run_result.metadata,
+                    node_data=event.node_data,
+                    parallel_id=event.parallel_id,
+                    parallel_start_node_id=event.parallel_start_node_id,
+                    parent_parallel_id=event.parent_parallel_id,
+                    parent_parallel_start_node_id=event.parent_parallel_start_node_id,
+                    start_at=event.route_node_state.start_at,
+                    inputs=event.route_node_state.node_run_result.inputs
+                    if event.route_node_state.node_run_result
+                    else {},
+                    process_data=event.route_node_state.node_run_result.process_data
+                    if event.route_node_state.node_run_result
+                    else {},
+                    outputs=event.route_node_state.node_run_result.outputs or {}
+                    if event.route_node_state.node_run_result
+                    else {},
+                    error=event.route_node_state.node_run_result.error
+                    if event.route_node_state.node_run_result and event.route_node_state.node_run_result.error
+                    else "Unknown error",
+                    execution_metadata=event.route_node_state.node_run_result.metadata
+                    if event.route_node_state.node_run_result
+                    else {},
                     in_iteration_id=event.in_iteration_id,
                     in_loop_id=event.in_loop_id,
                 )
@@ -458,21 +434,93 @@ class WorkflowBasedAppRunner:
                     node_execution_id=event.id,
                     node_id=event.node_id,
                     node_type=event.node_type,
-                    start_at=event.start_at,
-                    inputs=event.node_run_result.inputs,
-                    process_data=event.node_run_result.process_data,
-                    outputs=event.node_run_result.outputs,
-                    error=event.node_run_result.error or "Unknown error",
-                    execution_metadata=event.node_run_result.metadata,
+                    node_data=event.node_data,
+                    parallel_id=event.parallel_id,
+                    parallel_start_node_id=event.parallel_start_node_id,
+                    parent_parallel_id=event.parent_parallel_id,
+                    parent_parallel_start_node_id=event.parent_parallel_start_node_id,
+                    start_at=event.route_node_state.start_at,
+                    inputs=event.route_node_state.node_run_result.inputs
+                    if event.route_node_state.node_run_result
+                    else {},
+                    process_data=event.route_node_state.node_run_result.process_data
+                    if event.route_node_state.node_run_result
+                    else {},
+                    outputs=event.route_node_state.node_run_result.outputs
+                    if event.route_node_state.node_run_result
+                    else {},
+                    error=event.route_node_state.node_run_result.error
+                    if event.route_node_state.node_run_result and event.route_node_state.node_run_result.error
+                    else "Unknown error",
+                    execution_metadata=event.route_node_state.node_run_result.metadata
+                    if event.route_node_state.node_run_result
+                    else {},
                     in_iteration_id=event.in_iteration_id,
                     in_loop_id=event.in_loop_id,
                 )
             )
+
+        elif isinstance(event, NodeInIterationFailedEvent):
+            self._publish_event(
+                QueueNodeInIterationFailedEvent(
+                    node_execution_id=event.id,
+                    node_id=event.node_id,
+                    node_type=event.node_type,
+                    node_data=event.node_data,
+                    parallel_id=event.parallel_id,
+                    parallel_start_node_id=event.parallel_start_node_id,
+                    parent_parallel_id=event.parent_parallel_id,
+                    parent_parallel_start_node_id=event.parent_parallel_start_node_id,
+                    start_at=event.route_node_state.start_at,
+                    inputs=event.route_node_state.node_run_result.inputs
+                    if event.route_node_state.node_run_result
+                    else {},
+                    process_data=event.route_node_state.node_run_result.process_data
+                    if event.route_node_state.node_run_result
+                    else {},
+                    outputs=event.route_node_state.node_run_result.outputs or {}
+                    if event.route_node_state.node_run_result
+                    else {},
+                    execution_metadata=event.route_node_state.node_run_result.metadata
+                    if event.route_node_state.node_run_result
+                    else {},
+                    in_iteration_id=event.in_iteration_id,
+                    error=event.error,
+                )
+            )
+        elif isinstance(event, NodeInLoopFailedEvent):
+            self._publish_event(
+                QueueNodeInLoopFailedEvent(
+                    node_execution_id=event.id,
+                    node_id=event.node_id,
+                    node_type=event.node_type,
+                    node_data=event.node_data,
+                    parallel_id=event.parallel_id,
+                    parallel_start_node_id=event.parallel_start_node_id,
+                    parent_parallel_id=event.parent_parallel_id,
+                    parent_parallel_start_node_id=event.parent_parallel_start_node_id,
+                    start_at=event.route_node_state.start_at,
+                    inputs=event.route_node_state.node_run_result.inputs
+                    if event.route_node_state.node_run_result
+                    else {},
+                    process_data=event.route_node_state.node_run_result.process_data
+                    if event.route_node_state.node_run_result
+                    else {},
+                    outputs=event.route_node_state.node_run_result.outputs or {}
+                    if event.route_node_state.node_run_result
+                    else {},
+                    execution_metadata=event.route_node_state.node_run_result.metadata
+                    if event.route_node_state.node_run_result
+                    else {},
+                    in_loop_id=event.in_loop_id,
+                    error=event.error,
+                )
+            )
         elif isinstance(event, NodeRunStreamChunkEvent):
             self._publish_event(
                 QueueTextChunkEvent(
-                    text=event.chunk,
-                    from_variable_selector=list(event.selector),
+                    text=event.chunk_content,
+                    from_variable_selector=event.from_variable_selector,
                     in_iteration_id=event.in_iteration_id,
                     in_loop_id=event.in_loop_id,
                 )
@@ -485,10 +533,10 @@ class WorkflowBasedAppRunner:
                     in_loop_id=event.in_loop_id,
                 )
             )
-        elif isinstance(event, NodeRunAgentLogEvent):
+        elif isinstance(event, AgentLogEvent):
             self._publish_event(
                 QueueAgentLogEvent(
-                    id=event.message_id,
+                    id=event.id,
                     label=event.label,
                     node_execution_id=event.node_execution_id,
                     parent_id=event.parent_id,
@@ -499,13 +547,51 @@ class WorkflowBasedAppRunner:
                     node_id=event.node_id,
                 )
             )
-        elif isinstance(event, NodeRunIterationStartedEvent):
+        elif isinstance(event, ParallelBranchRunStartedEvent):
+            self._publish_event(
+                QueueParallelBranchRunStartedEvent(
+                    parallel_id=event.parallel_id,
+                    parallel_start_node_id=event.parallel_start_node_id,
+                    parent_parallel_id=event.parent_parallel_id,
+                    parent_parallel_start_node_id=event.parent_parallel_start_node_id,
+                    in_iteration_id=event.in_iteration_id,
+                    in_loop_id=event.in_loop_id,
+                )
+            )
+        elif isinstance(event, ParallelBranchRunSucceededEvent):
+            self._publish_event(
+                QueueParallelBranchRunSucceededEvent(
+                    parallel_id=event.parallel_id,
+                    parallel_start_node_id=event.parallel_start_node_id,
+                    parent_parallel_id=event.parent_parallel_id,
+                    parent_parallel_start_node_id=event.parent_parallel_start_node_id,
+                    in_iteration_id=event.in_iteration_id,
+                    in_loop_id=event.in_loop_id,
+                )
+            )
+        elif isinstance(event, ParallelBranchRunFailedEvent):
+            self._publish_event(
+                QueueParallelBranchRunFailedEvent(
+                    parallel_id=event.parallel_id,
+                    parallel_start_node_id=event.parallel_start_node_id,
+                    parent_parallel_id=event.parent_parallel_id,
+                    parent_parallel_start_node_id=event.parent_parallel_start_node_id,
+                    in_iteration_id=event.in_iteration_id,
+                    in_loop_id=event.in_loop_id,
+                    error=event.error,
+                )
+            )
+        elif isinstance(event, IterationRunStartedEvent):
             self._publish_event(
                 QueueIterationStartEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
-                    node_type=event.node_type,
-                    node_title=event.node_title,
+                    node_execution_id=event.iteration_id,
+                    node_id=event.iteration_node_id,
+                    node_type=event.iteration_node_type,
+                    node_data=event.iteration_node_data,
+                    parallel_id=event.parallel_id,
+                    parallel_start_node_id=event.parallel_start_node_id,
+                    parent_parallel_id=event.parent_parallel_id,
+                    parent_parallel_start_node_id=event.parent_parallel_start_node_id,
                     start_at=event.start_at,
                     node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
                     inputs=event.inputs,
@@ -513,41 +599,55 @@ class WorkflowBasedAppRunner:
                     metadata=event.metadata,
                 )
             )
-        elif isinstance(event, NodeRunIterationNextEvent):
+        elif isinstance(event, IterationRunNextEvent):
             self._publish_event(
                 QueueIterationNextEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
-                    node_type=event.node_type,
-                    node_title=event.node_title,
+                    node_execution_id=event.iteration_id,
+                    node_id=event.iteration_node_id,
+                    node_type=event.iteration_node_type,
+                    node_data=event.iteration_node_data,
+                    parallel_id=event.parallel_id,
+                    parallel_start_node_id=event.parallel_start_node_id,
+                    parent_parallel_id=event.parent_parallel_id,
+                    parent_parallel_start_node_id=event.parent_parallel_start_node_id,
                     index=event.index,
                     node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
                     output=event.pre_iteration_output,
+                    parallel_mode_run_id=event.parallel_mode_run_id,
+                    duration=event.duration,
                 )
             )
-        elif isinstance(event, (NodeRunIterationSucceededEvent | NodeRunIterationFailedEvent)):
+        elif isinstance(event, (IterationRunSucceededEvent | IterationRunFailedEvent)):
             self._publish_event(
                 QueueIterationCompletedEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
-                    node_type=event.node_type,
-                    node_title=event.node_title,
+                    node_execution_id=event.iteration_id,
+                    node_id=event.iteration_node_id,
+                    node_type=event.iteration_node_type,
+                    node_data=event.iteration_node_data,
+                    parallel_id=event.parallel_id,
+                    parallel_start_node_id=event.parallel_start_node_id,
+                    parent_parallel_id=event.parent_parallel_id,
+                    parent_parallel_start_node_id=event.parent_parallel_start_node_id,
                     start_at=event.start_at,
                     node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
                     inputs=event.inputs,
                     outputs=event.outputs,
                     metadata=event.metadata,
                     steps=event.steps,
-                    error=event.error if isinstance(event, NodeRunIterationFailedEvent) else None,
+                    error=event.error if isinstance(event, IterationRunFailedEvent) else None,
                 )
             )
-        elif isinstance(event, NodeRunLoopStartedEvent):
+        elif isinstance(event, LoopRunStartedEvent):
             self._publish_event(
                 QueueLoopStartEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
-                    node_type=event.node_type,
-                    node_title=event.node_title,
+                    node_execution_id=event.loop_id,
+                    node_id=event.loop_node_id,
+                    node_type=event.loop_node_type,
+                    node_data=event.loop_node_data,
+                    parallel_id=event.parallel_id,
+                    parallel_start_node_id=event.parallel_start_node_id,
+                    parent_parallel_id=event.parent_parallel_id,
+                    parent_parallel_start_node_id=event.parent_parallel_start_node_id,
                     start_at=event.start_at,
                     node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
                     inputs=event.inputs,
@@ -555,32 +655,42 @@ class WorkflowBasedAppRunner:
                     metadata=event.metadata,
                 )
             )
-        elif isinstance(event, NodeRunLoopNextEvent):
+        elif isinstance(event, LoopRunNextEvent):
             self._publish_event(
                 QueueLoopNextEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
-                    node_type=event.node_type,
-                    node_title=event.node_title,
+                    node_execution_id=event.loop_id,
+                    node_id=event.loop_node_id,
+                    node_type=event.loop_node_type,
+                    node_data=event.loop_node_data,
+                    parallel_id=event.parallel_id,
+                    parallel_start_node_id=event.parallel_start_node_id,
+                    parent_parallel_id=event.parent_parallel_id,
+                    parent_parallel_start_node_id=event.parent_parallel_start_node_id,
                     index=event.index,
                     node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
                     output=event.pre_loop_output,
+                    parallel_mode_run_id=event.parallel_mode_run_id,
+                    duration=event.duration,
                 )
             )
-        elif isinstance(event, (NodeRunLoopSucceededEvent | NodeRunLoopFailedEvent)):
+        elif isinstance(event, (LoopRunSucceededEvent | LoopRunFailedEvent)):
             self._publish_event(
                 QueueLoopCompletedEvent(
-                    node_execution_id=event.id,
-                    node_id=event.node_id,
-                    node_type=event.node_type,
-                    node_title=event.node_title,
+                    node_execution_id=event.loop_id,
+                    node_id=event.loop_node_id,
+                    node_type=event.loop_node_type,
+                    node_data=event.loop_node_data,
+                    parallel_id=event.parallel_id,
+                    parallel_start_node_id=event.parallel_start_node_id,
+                    parent_parallel_id=event.parent_parallel_id,
+                    parent_parallel_start_node_id=event.parent_parallel_start_node_id,
                     start_at=event.start_at,
                     node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
                     inputs=event.inputs,
                     outputs=event.outputs,
                     metadata=event.metadata,
                     steps=event.steps,
-                    error=event.error if isinstance(event, NodeRunLoopFailedEvent) else None,
+                    error=event.error if isinstance(event, LoopRunFailedEvent) else None,
                 )
             )
 

api/core/app/entities/app_invoke_entities.py

@@ -1,5 +1,5 @@
 from collections.abc import Mapping, Sequence
-from enum import StrEnum
+from enum import Enum
 from typing import Any, Optional
 
 from pydantic import BaseModel, ConfigDict, Field, ValidationInfo, field_validator
@@ -11,7 +11,7 @@ from core.file import File, FileUploadConfig
 from core.model_runtime.entities.model_entities import AIModelEntity
 
 
-class InvokeFrom(StrEnum):
+class InvokeFrom(Enum):
     """
     Invoke From.
     """

api/core/app/entities/queue_entities.py

@@ -7,9 +7,11 @@ from pydantic import BaseModel
 
 from core.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk
 from core.rag.entities.citation_metadata import RetrievalSourceMetadata
-from core.workflow.entities import AgentNodeStrategyInit, GraphRuntimeState
-from core.workflow.enums import WorkflowNodeExecutionMetadataKey
+from core.workflow.entities.node_entities import AgentNodeStrategyInit
+from core.workflow.entities.workflow_node_execution import WorkflowNodeExecutionMetadataKey
+from core.workflow.graph_engine.entities.graph_runtime_state import GraphRuntimeState
 from core.workflow.nodes import NodeType
+from core.workflow.nodes.base import BaseNodeData
 
 
 class QueueEvent(StrEnum):
@@ -41,6 +43,9 @@ class QueueEvent(StrEnum):
     ANNOTATION_REPLY = "annotation_reply"
     AGENT_THOUGHT = "agent_thought"
     MESSAGE_FILE = "message_file"
+    PARALLEL_BRANCH_RUN_STARTED = "parallel_branch_run_started"
+    PARALLEL_BRANCH_RUN_SUCCEEDED = "parallel_branch_run_succeeded"
+    PARALLEL_BRANCH_RUN_FAILED = "parallel_branch_run_failed"
     AGENT_LOG = "agent_log"
     ERROR = "error"
     PING = "ping"
@@ -75,7 +80,15 @@ class QueueIterationStartEvent(AppQueueEvent):
     node_execution_id: str
     node_id: str
     node_type: NodeType
-    node_title: str
+    node_data: BaseNodeData
+    parallel_id: Optional[str] = None
+    """parallel id if node is in parallel"""
+    parallel_start_node_id: Optional[str] = None
+    """parallel start node id if node is in parallel"""
+    parent_parallel_id: Optional[str] = None
+    """parent parallel id if node is in parallel"""
+    parent_parallel_start_node_id: Optional[str] = None
+    """parent parallel start node id if node is in parallel"""
     start_at: datetime
 
     node_run_index: int
@@ -95,9 +108,20 @@ class QueueIterationNextEvent(AppQueueEvent):
     node_execution_id: str
     node_id: str
     node_type: NodeType
-    node_title: str
+    node_data: BaseNodeData
+    parallel_id: Optional[str] = None
+    """parallel id if node is in parallel"""
+    parallel_start_node_id: Optional[str] = None
+    """parallel start node id if node is in parallel"""
+    parent_parallel_id: Optional[str] = None
+    """parent parallel id if node is in parallel"""
+    parent_parallel_start_node_id: Optional[str] = None
+    """parent parallel start node id if node is in parallel"""
+    parallel_mode_run_id: Optional[str] = None
+    """iteration run in parallel mode run id"""
     node_run_index: int
     output: Optional[Any] = None  # output for the current iteration
+    duration: Optional[float] = None
 
 
 class QueueIterationCompletedEvent(AppQueueEvent):
@@ -110,7 +134,15 @@ class QueueIterationCompletedEvent(AppQueueEvent):
     node_execution_id: str
     node_id: str
     node_type: NodeType
-    node_title: str
+    node_data: BaseNodeData
+    parallel_id: Optional[str] = None
+    """parallel id if node is in parallel"""
+    parallel_start_node_id: Optional[str] = None
+    """parallel start node id if node is in parallel"""
+    parent_parallel_id: Optional[str] = None
+    """parent parallel id if node is in parallel"""
+    parent_parallel_start_node_id: Optional[str] = None
+    """parent parallel start node id if node is in parallel"""
     start_at: datetime
 
     node_run_index: int
@@ -131,7 +163,7 @@ class QueueLoopStartEvent(AppQueueEvent):
     node_execution_id: str
     node_id: str
     node_type: NodeType
-    node_title: str
+    node_data: BaseNodeData
     parallel_id: Optional[str] = None
     """parallel id if node is in parallel"""
     parallel_start_node_id: Optional[str] = None
@@ -159,7 +191,7 @@ class QueueLoopNextEvent(AppQueueEvent):
     node_execution_id: str
     node_id: str
     node_type: NodeType
-    node_title: str
+    node_data: BaseNodeData
     parallel_id: Optional[str] = None
     """parallel id if node is in parallel"""
     parallel_start_node_id: Optional[str] = None
@@ -172,6 +204,7 @@ class QueueLoopNextEvent(AppQueueEvent):
     """iteration run in parallel mode run id"""
     node_run_index: int
     output: Optional[Any] = None  # output for the current loop
+    duration: Optional[float] = None
 
 
 class QueueLoopCompletedEvent(AppQueueEvent):
@@ -184,7 +217,7 @@ class QueueLoopCompletedEvent(AppQueueEvent):
     node_execution_id: str
     node_id: str
     node_type: NodeType
-    node_title: str
+    node_data: BaseNodeData
     parallel_id: Optional[str] = None
     """parallel id if node is in parallel"""
     parallel_start_node_id: Optional[str] = None
@@ -331,24 +364,27 @@ class QueueNodeStartedEvent(AppQueueEvent):
 
     node_execution_id: str
     node_id: str
-    node_title: str
     node_type: NodeType
-    node_run_index: int = 1  # FIXME(-LAN-): may not used
+    node_data: BaseNodeData
+    node_run_index: int = 1
     predecessor_node_id: Optional[str] = None
     parallel_id: Optional[str] = None
+    """parallel id if node is in parallel"""
     parallel_start_node_id: Optional[str] = None
+    """parallel start node id if node is in parallel"""
     parent_parallel_id: Optional[str] = None
+    """parent parallel id if node is in parallel"""
     parent_parallel_start_node_id: Optional[str] = None
+    """parent parallel start node id if node is in parallel"""
     in_iteration_id: Optional[str] = None
+    """iteration id if node is in iteration"""
     in_loop_id: Optional[str] = None
+    """loop id if node is in loop"""
     start_at: datetime
     parallel_mode_run_id: Optional[str] = None
+    """iteration run in parallel mode run id"""
     agent_strategy: Optional[AgentNodeStrategyInit] = None
 
-    # FIXME(-LAN-): only for ToolNode, need to refactor
-    provider_type: str  # should be a core.tools.entities.tool_entities.ToolProviderType
-    provider_id: str
-
 
 class QueueNodeSucceededEvent(AppQueueEvent):
     """
@@ -360,6 +396,7 @@ class QueueNodeSucceededEvent(AppQueueEvent):
     node_execution_id: str
     node_id: str
     node_type: NodeType
+    node_data: BaseNodeData
     parallel_id: Optional[str] = None
     """parallel id if node is in parallel"""
     parallel_start_node_id: Optional[str] = None
@@ -380,6 +417,10 @@ class QueueNodeSucceededEvent(AppQueueEvent):
     execution_metadata: Optional[Mapping[WorkflowNodeExecutionMetadataKey, Any]] = None
 
     error: Optional[str] = None
+    """single iteration duration map"""
+    iteration_duration_map: Optional[dict[str, float]] = None
+    """single loop duration map"""
+    loop_duration_map: Optional[dict[str, float]] = None
 
 
 class QueueAgentLogEvent(AppQueueEvent):
@@ -413,6 +454,72 @@ class QueueNodeRetryEvent(QueueNodeStartedEvent):
     retry_index: int  # retry index
 
 
+class QueueNodeInIterationFailedEvent(AppQueueEvent):
+    """
+    QueueNodeInIterationFailedEvent entity
+    """
+
+    event: QueueEvent = QueueEvent.NODE_FAILED
+
+    node_execution_id: str
+    node_id: str
+    node_type: NodeType
+    node_data: BaseNodeData
+    parallel_id: Optional[str] = None
+    """parallel id if node is in parallel"""
+    parallel_start_node_id: Optional[str] = None
+    """parallel start node id if node is in parallel"""
+    parent_parallel_id: Optional[str] = None
+    """parent parallel id if node is in parallel"""
+    parent_parallel_start_node_id: Optional[str] = None
+    """parent parallel start node id if node is in parallel"""
+    in_iteration_id: Optional[str] = None
+    """iteration id if node is in iteration"""
+    in_loop_id: Optional[str] = None
+    """loop id if node is in loop"""
+    start_at: datetime
+
+    inputs: Optional[Mapping[str, Any]] = None
+    process_data: Optional[Mapping[str, Any]] = None
+    outputs: Optional[Mapping[str, Any]] = None
+    execution_metadata: Optional[Mapping[WorkflowNodeExecutionMetadataKey, Any]] = None
+
+    error: str
+
+
+class QueueNodeInLoopFailedEvent(AppQueueEvent):
+    """
+    QueueNodeInLoopFailedEvent entity
+    """
+
+    event: QueueEvent = QueueEvent.NODE_FAILED
+
+    node_execution_id: str
+    node_id: str
+    node_type: NodeType
+    node_data: BaseNodeData
+    parallel_id: Optional[str] = None
+    """parallel id if node is in parallel"""
+    parallel_start_node_id: Optional[str] = None
+    """parallel start node id if node is in parallel"""
+    parent_parallel_id: Optional[str] = None
+    """parent parallel id if node is in parallel"""
+    parent_parallel_start_node_id: Optional[str] = None
+    """parent parallel start node id if node is in parallel"""
+    in_iteration_id: Optional[str] = None
+    """iteration id if node is in iteration"""
+    in_loop_id: Optional[str] = None
+    """loop id if node is in loop"""
+    start_at: datetime
+
+    inputs: Optional[Mapping[str, Any]] = None
+    process_data: Optional[Mapping[str, Any]] = None
+    outputs: Optional[Mapping[str, Any]] = None
+    execution_metadata: Optional[Mapping[WorkflowNodeExecutionMetadataKey, Any]] = None
+
+    error: str
+
+
 class QueueNodeExceptionEvent(AppQueueEvent):
     """
     QueueNodeExceptionEvent entity
@@ -423,6 +530,7 @@ class QueueNodeExceptionEvent(AppQueueEvent):
     node_execution_id: str
     node_id: str
     node_type: NodeType
+    node_data: BaseNodeData
     parallel_id: Optional[str] = None
     """parallel id if node is in parallel"""
     parallel_start_node_id: Optional[str] = None
@@ -455,7 +563,15 @@ class QueueNodeFailedEvent(AppQueueEvent):
     node_execution_id: str
     node_id: str
     node_type: NodeType
+    node_data: BaseNodeData
     parallel_id: Optional[str] = None
+    """parallel id if node is in parallel"""
+    parallel_start_node_id: Optional[str] = None
+    """parallel start node id if node is in parallel"""
+    parent_parallel_id: Optional[str] = None
+    """parent parallel id if node is in parallel"""
+    parent_parallel_start_node_id: Optional[str] = None
+    """parent parallel start node id if node is in parallel"""
     in_iteration_id: Optional[str] = None
     """iteration id if node is in iteration"""
     in_loop_id: Optional[str] = None
@@ -562,3 +678,61 @@ class WorkflowQueueMessage(QueueMessage):
     """
 
     pass
+
+
+class QueueParallelBranchRunStartedEvent(AppQueueEvent):
+    """
+    QueueParallelBranchRunStartedEvent entity
+    """
+
+    event: QueueEvent = QueueEvent.PARALLEL_BRANCH_RUN_STARTED
+
+    parallel_id: str
+    parallel_start_node_id: str
+    parent_parallel_id: Optional[str] = None
+    """parent parallel id if node is in parallel"""
+    parent_parallel_start_node_id: Optional[str] = None
+    """parent parallel start node id if node is in parallel"""
+    in_iteration_id: Optional[str] = None
+    """iteration id if node is in iteration"""
+    in_loop_id: Optional[str] = None
+    """loop id if node is in loop"""
+
+
+class QueueParallelBranchRunSucceededEvent(AppQueueEvent):
+    """
+    QueueParallelBranchRunSucceededEvent entity
+    """
+
+    event: QueueEvent = QueueEvent.PARALLEL_BRANCH_RUN_SUCCEEDED
+
+    parallel_id: str
+    parallel_start_node_id: str
+    parent_parallel_id: Optional[str] = None
+    """parent parallel id if node is in parallel"""
+    parent_parallel_start_node_id: Optional[str] = None
+    """parent parallel start node id if node is in parallel"""
+    in_iteration_id: Optional[str] = None
+    """iteration id if node is in iteration"""
+    in_loop_id: Optional[str] = None
+    """loop id if node is in loop"""
+
+
+class QueueParallelBranchRunFailedEvent(AppQueueEvent):
+    """
+    QueueParallelBranchRunFailedEvent entity
+    """
+
+    event: QueueEvent = QueueEvent.PARALLEL_BRANCH_RUN_FAILED
+
+    parallel_id: str
+    parallel_start_node_id: str
+    parent_parallel_id: Optional[str] = None
+    """parent parallel id if node is in parallel"""
+    parent_parallel_start_node_id: Optional[str] = None
+    """parent parallel start node id if node is in parallel"""
+    in_iteration_id: Optional[str] = None
+    """iteration id if node is in iteration"""
+    in_loop_id: Optional[str] = None
+    """loop id if node is in loop"""
+    error: str

api/core/app/entities/task_entities.py

@@ -7,8 +7,8 @@ from pydantic import BaseModel, ConfigDict, Field
 from core.model_runtime.entities.llm_entities import LLMResult, LLMUsage
 from core.model_runtime.utils.encoders import jsonable_encoder
 from core.rag.entities.citation_metadata import RetrievalSourceMetadata
-from core.workflow.entities import AgentNodeStrategyInit
-from core.workflow.enums import WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
+from core.workflow.entities.node_entities import AgentNodeStrategyInit
+from core.workflow.entities.workflow_node_execution import WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
 
 
 class AnnotationReplyAccount(BaseModel):
@@ -71,6 +71,8 @@ class StreamEvent(Enum):
     NODE_STARTED = "node_started"
     NODE_FINISHED = "node_finished"
     NODE_RETRY = "node_retry"
+    PARALLEL_BRANCH_STARTED = "parallel_branch_started"
+    PARALLEL_BRANCH_FINISHED = "parallel_branch_finished"
     ITERATION_STARTED = "iteration_started"
     ITERATION_NEXT = "iteration_next"
     ITERATION_COMPLETED = "iteration_completed"
@@ -438,6 +440,54 @@ class NodeRetryStreamResponse(StreamResponse):
         }
 
 
+class ParallelBranchStartStreamResponse(StreamResponse):
+    """
+    ParallelBranchStartStreamResponse entity
+    """
+
+    class Data(BaseModel):
+        """
+        Data entity
+        """
+
+        parallel_id: str
+        parallel_branch_id: str
+        parent_parallel_id: Optional[str] = None
+        parent_parallel_start_node_id: Optional[str] = None
+        iteration_id: Optional[str] = None
+        loop_id: Optional[str] = None
+        created_at: int
+
+    event: StreamEvent = StreamEvent.PARALLEL_BRANCH_STARTED
+    workflow_run_id: str
+    data: Data
+
+
+class ParallelBranchFinishedStreamResponse(StreamResponse):
+    """
+    ParallelBranchFinishedStreamResponse entity
+    """
+
+    class Data(BaseModel):
+        """
+        Data entity
+        """
+
+        parallel_id: str
+        parallel_branch_id: str
+        parent_parallel_id: Optional[str] = None
+        parent_parallel_start_node_id: Optional[str] = None
+        iteration_id: Optional[str] = None
+        loop_id: Optional[str] = None
+        status: str
+        error: Optional[str] = None
+        created_at: int
+
+    event: StreamEvent = StreamEvent.PARALLEL_BRANCH_FINISHED
+    workflow_run_id: str
+    data: Data
+
+
 class IterationNodeStartStreamResponse(StreamResponse):
     """
     NodeStartStreamResponse entity
@@ -456,6 +506,8 @@ class IterationNodeStartStreamResponse(StreamResponse):
         extras: dict = Field(default_factory=dict)
         metadata: Mapping = {}
         inputs: Mapping = {}
+        parallel_id: Optional[str] = None
+        parallel_start_node_id: Optional[str] = None
 
     event: StreamEvent = StreamEvent.ITERATION_STARTED
     workflow_run_id: str
@@ -478,7 +530,12 @@ class IterationNodeNextStreamResponse(StreamResponse):
         title: str
         index: int
         created_at: int
+        pre_iteration_output: Optional[Any] = None
         extras: dict = Field(default_factory=dict)
+        parallel_id: Optional[str] = None
+        parallel_start_node_id: Optional[str] = None
+        parallel_mode_run_id: Optional[str] = None
+        duration: Optional[float] = None
 
     event: StreamEvent = StreamEvent.ITERATION_NEXT
     workflow_run_id: str
@@ -510,6 +567,8 @@ class IterationNodeCompletedStreamResponse(StreamResponse):
         execution_metadata: Optional[Mapping] = None
         finished_at: int
         steps: int
+        parallel_id: Optional[str] = None
+        parallel_start_node_id: Optional[str] = None
 
     event: StreamEvent = StreamEvent.ITERATION_COMPLETED
     workflow_run_id: str
@@ -563,6 +622,7 @@ class LoopNodeNextStreamResponse(StreamResponse):
         parallel_id: Optional[str] = None
         parallel_start_node_id: Optional[str] = None
         parallel_mode_run_id: Optional[str] = None
+        duration: Optional[float] = None
 
     event: StreamEvent = StreamEvent.LOOP_NEXT
     workflow_run_id: str

api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py

@@ -472,9 +472,10 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
         :param event: agent thought event
         :return:
         """
-        agent_thought: Optional[MessageAgentThought] = (
-            db.session.query(MessageAgentThought).where(MessageAgentThought.id == event.agent_thought_id).first()
-        )
+        with Session(db.engine, expire_on_commit=False) as session:
+            agent_thought: Optional[MessageAgentThought] = (
+                session.query(MessageAgentThought).where(MessageAgentThought.id == event.agent_thought_id).first()
+            )
 
         if agent_thought:
             return AgentThoughtStreamResponse(

api/core/entities/provider_configuration.py

@@ -28,6 +28,7 @@ from core.model_runtime.entities.provider_entities import (
 )
 from core.model_runtime.model_providers.__base.ai_model import AIModel
 from core.model_runtime.model_providers.model_provider_factory import ModelProviderFactory
+from core.plugin.entities.plugin import ModelProviderID
 from extensions.ext_database import db
 from libs.datetime_utils import naive_utc_now
 from models.provider import (
@@ -40,7 +41,6 @@ from models.provider import (
     ProviderType,
     TenantPreferredModelProvider,
 )
-from models.provider_ids import ModelProviderID
 
 logger = logging.getLogger(__name__)
 
@@ -627,7 +627,6 @@ class ProviderConfiguration(BaseModel):
         Get custom model credentials.
         """
         # get provider model
-
         model_provider_id = ModelProviderID(self.provider.provider)
         provider_names = [self.provider.provider]
         if model_provider_id.is_langgenius():
@@ -1125,7 +1124,6 @@ class ProviderConfiguration(BaseModel):
         """
         Get provider model setting.
         """
-
         model_provider_id = ModelProviderID(self.provider.provider)
         provider_names = [self.provider.provider]
         if model_provider_id.is_langgenius():
@@ -1209,7 +1207,6 @@ class ProviderConfiguration(BaseModel):
         :param model: model name
         :return:
         """
-
         model_provider_id = ModelProviderID(self.provider.provider)
         provider_names = [self.provider.provider]
         if model_provider_id.is_langgenius():

api/core/entities/provider_entities.py

@@ -192,7 +192,7 @@ class ProviderConfig(BasicProviderConfig):
 
     scope: AppSelectorScope | ModelSelectorScope | ToolSelectorScope | None = None
     required: bool = False
-    default: Optional[Union[int, str, float, bool]] = None
+    default: Optional[Union[int, str, float, bool, list]] = None
     options: Optional[list[Option]] = None
     label: Optional[I18nObject] = None
     help: Optional[I18nObject] = None

api/core/helper/encrypter.py

@@ -3,7 +3,7 @@ import base64
 from libs import rsa
 
 
-def obfuscated_token(token: str):
+def obfuscated_token(token: str) -> str:
     if not token:
         return token
     if len(token) <= 8:
@@ -12,8 +12,8 @@ def obfuscated_token(token: str):
 
 
 def encrypt_token(tenant_id: str, token: str):
-    from extensions.ext_database import db
     from models.account import Tenant
+    from models.engine import db
 
     if not (tenant := db.session.query(Tenant).where(Tenant.id == tenant_id).first()):
         raise ValueError(f"Tenant with id {tenant_id} not found")

api/core/helper/marketplace.py

@@ -1,6 +1,6 @@
 from collections.abc import Sequence
 
-import requests
+import httpx
 from yarl import URL
 
 from configs import dify_config
@@ -23,7 +23,7 @@ def batch_fetch_plugin_manifests(plugin_ids: list[str]) -> Sequence[MarketplaceP
         return []
 
     url = str(marketplace_api_url / "api/v1/plugins/batch")
-    response = requests.post(url, json={"plugin_ids": plugin_ids})
+    response = httpx.post(url, json={"plugin_ids": plugin_ids})
     response.raise_for_status()
 
     return [MarketplacePluginDeclaration(**plugin) for plugin in response.json()["data"]["plugins"]]
@@ -36,7 +36,7 @@ def batch_fetch_plugin_manifests_ignore_deserialization_error(
         return []
 
     url = str(marketplace_api_url / "api/v1/plugins/batch")
-    response = requests.post(url, json={"plugin_ids": plugin_ids})
+    response = httpx.post(url, json={"plugin_ids": plugin_ids})
     response.raise_for_status()
     result: list[MarketplacePluginDeclaration] = []
     for plugin in response.json()["data"]["plugins"]:
@@ -50,5 +50,5 @@ def batch_fetch_plugin_manifests_ignore_deserialization_error(
 
 def record_install_plugin_event(plugin_unique_identifier: str):
     url = str(marketplace_api_url / "api/v1/stats/plugins/install_count")
-    response = requests.post(url, json={"unique_identifier": plugin_unique_identifier})
+    response = httpx.post(url, json={"unique_identifier": plugin_unique_identifier})
     response.raise_for_status()

api/core/helper/provider_encryption.py

@@ -0,0 +1,128 @@
+import contextlib
+from copy import deepcopy
+from typing import Any, Optional, Protocol
+
+from core.entities.provider_entities import BasicProviderConfig
+from core.helper import encrypter
+
+
+class ProviderConfigCache(Protocol):
+    """
+    Interface for provider configuration cache operations
+    """
+
+    def get(self) -> Optional[dict]:
+        """Get cached provider configuration"""
+        ...
+
+    def set(self, config: dict[str, Any]) -> None:
+        """Cache provider configuration"""
+        ...
+
+    def delete(self) -> None:
+        """Delete cached provider configuration"""
+        ...
+
+
+class ProviderConfigEncrypter:
+    tenant_id: str
+    config: list[BasicProviderConfig]
+    provider_config_cache: ProviderConfigCache
+
+    def __init__(
+        self,
+        tenant_id: str,
+        config: list[BasicProviderConfig],
+        provider_config_cache: ProviderConfigCache,
+    ):
+        self.tenant_id = tenant_id
+        self.config = config
+        self.provider_config_cache = provider_config_cache
+
+    def _deep_copy(self, data: dict[str, str]) -> dict[str, str]:
+        """
+        deep copy data
+        """
+        return deepcopy(data)
+
+    def encrypt(self, data: dict[str, str]) -> dict[str, str]:
+        """
+        encrypt tool credentials with tenant id
+
+        return a deep copy of credentials with encrypted values
+        """
+        data = self._deep_copy(data)
+
+        # get fields need to be decrypted
+        fields = dict[str, BasicProviderConfig]()
+        for credential in self.config:
+            fields[credential.name] = credential
+
+        for field_name, field in fields.items():
+            if field.type == BasicProviderConfig.Type.SECRET_INPUT:
+                if field_name in data:
+                    encrypted = encrypter.encrypt_token(self.tenant_id, data[field_name] or "")
+                    data[field_name] = encrypted
+
+        return data
+
+    def mask_credentials(self, data: dict[str, Any]) -> dict[str, Any]:
+        """
+        mask credentials
+
+        return a deep copy of credentials with masked values
+        """
+        data = self._deep_copy(data)
+
+        # get fields need to be decrypted
+        fields = dict[str, BasicProviderConfig]()
+        for credential in self.config:
+            fields[credential.name] = credential
+
+        for field_name, field in fields.items():
+            if field.type == BasicProviderConfig.Type.SECRET_INPUT:
+                if field_name in data:
+                    if len(data[field_name]) > 6:
+                        data[field_name] = (
+                            data[field_name][:2] + "*" * (len(data[field_name]) - 4) + data[field_name][-2:]
+                        )
+                    else:
+                        data[field_name] = "*" * len(data[field_name])
+
+        return data
+
+    def mask_tool_credentials(self, data: dict[str, Any]) -> dict[str, Any]:
+        return self.mask_credentials(data)
+
+    def decrypt(self, data: dict[str, str]) -> dict[str, Any]:
+        """
+        decrypt tool credentials with tenant id
+
+        return a deep copy of credentials with decrypted values
+        """
+        cached_credentials = self.provider_config_cache.get()
+        if cached_credentials:
+            return cached_credentials
+
+        data = self._deep_copy(data)
+        # get fields need to be decrypted
+        fields = dict[str, BasicProviderConfig]()
+        for credential in self.config:
+            fields[credential.name] = credential
+
+        for field_name, field in fields.items():
+            if field.type == BasicProviderConfig.Type.SECRET_INPUT:
+                if field_name in data:
+                    with contextlib.suppress(Exception):
+                        # if the value is None or empty string, skip decrypt
+                        if not data[field_name]:
+                            continue
+
+                        data[field_name] = encrypter.decrypt_token(self.tenant_id, data[field_name])
+
+        self.provider_config_cache.set(data)
+        return data
+
+
+def create_provider_encrypter(tenant_id: str, config: list[BasicProviderConfig], cache: ProviderConfigCache):
+    return ProviderConfigEncrypter(tenant_id=tenant_id, config=config, provider_config_cache=cache), cache

api/core/llm_generator/llm_generator.py

@@ -28,9 +28,8 @@ from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
 from core.ops.utils import measure_time
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
 from core.workflow.entities.workflow_node_execution import WorkflowNodeExecutionMetadataKey
-from core.workflow.node_events import AgentLogEvent
-from extensions.ext_database import db
-from models import App, Message, WorkflowNodeExecutionModel
+from core.workflow.graph_engine.entities.event import AgentLogEvent
+from models import App, Message, WorkflowNodeExecutionModel, db
 
 logger = logging.getLogger(__name__)
 

api/core/memory/token_buffer_memory.py

@@ -167,11 +167,11 @@ class TokenBufferMemory:
             else:
                 prompt_messages.append(AssistantPromptMessage(content=message.answer))
 
-            if not prompt_messages:
-                return []
+        if not prompt_messages:
+            return []
 
-            # prune the chat message if it exceeds the max token limit
-            curr_message_tokens = self.model_instance.get_llm_num_tokens(prompt_messages)
+        # prune the chat message if it exceeds the max token limit
+        curr_message_tokens = self.model_instance.get_llm_num_tokens(prompt_messages)
 
         if curr_message_tokens > max_token_limit:
             while curr_message_tokens > max_token_limit and len(prompt_messages) > 1:

api/core/model_manager.py

@@ -158,8 +158,6 @@ class ModelInstance:
         """
         if not isinstance(self.model_type_instance, LargeLanguageModel):
             raise Exception("Model type instance is not LargeLanguageModel")
-
-        self.model_type_instance = cast(LargeLanguageModel, self.model_type_instance)
         return cast(
             Union[LLMResult, Generator],
             self._round_robin_invoke(
@@ -188,8 +186,6 @@ class ModelInstance:
         """
         if not isinstance(self.model_type_instance, LargeLanguageModel):
             raise Exception("Model type instance is not LargeLanguageModel")
-
-        self.model_type_instance = cast(LargeLanguageModel, self.model_type_instance)
         return cast(
             int,
             self._round_robin_invoke(
@@ -214,8 +210,6 @@ class ModelInstance:
         """
         if not isinstance(self.model_type_instance, TextEmbeddingModel):
             raise Exception("Model type instance is not TextEmbeddingModel")
-
-        self.model_type_instance = cast(TextEmbeddingModel, self.model_type_instance)
         return cast(
             TextEmbeddingResult,
             self._round_robin_invoke(
@@ -237,8 +231,6 @@ class ModelInstance:
         """
         if not isinstance(self.model_type_instance, TextEmbeddingModel):
             raise Exception("Model type instance is not TextEmbeddingModel")
-
-        self.model_type_instance = cast(TextEmbeddingModel, self.model_type_instance)
         return cast(
             list[int],
             self._round_robin_invoke(
@@ -269,8 +261,6 @@ class ModelInstance:
         """
         if not isinstance(self.model_type_instance, RerankModel):
             raise Exception("Model type instance is not RerankModel")
-
-        self.model_type_instance = cast(RerankModel, self.model_type_instance)
         return cast(
             RerankResult,
             self._round_robin_invoke(
@@ -295,8 +285,6 @@ class ModelInstance:
         """
         if not isinstance(self.model_type_instance, ModerationModel):
             raise Exception("Model type instance is not ModerationModel")
-
-        self.model_type_instance = cast(ModerationModel, self.model_type_instance)
         return cast(
             bool,
             self._round_robin_invoke(
@@ -318,8 +306,6 @@ class ModelInstance:
         """
         if not isinstance(self.model_type_instance, Speech2TextModel):
             raise Exception("Model type instance is not Speech2TextModel")
-
-        self.model_type_instance = cast(Speech2TextModel, self.model_type_instance)
         return cast(
             str,
             self._round_robin_invoke(
@@ -343,8 +329,6 @@ class ModelInstance:
         """
         if not isinstance(self.model_type_instance, TTSModel):
             raise Exception("Model type instance is not TTSModel")
-
-        self.model_type_instance = cast(TTSModel, self.model_type_instance)
         return cast(
             Iterable[bytes],
             self._round_robin_invoke(
@@ -404,8 +388,6 @@ class ModelInstance:
         """
         if not isinstance(self.model_type_instance, TTSModel):
             raise Exception("Model type instance is not TTSModel")
-
-        self.model_type_instance = cast(TTSModel, self.model_type_instance)
         return self.model_type_instance.get_tts_model_voices(
             model=self.model, credentials=self.credentials, language=language
         )

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

@@ -24,7 +24,8 @@ from core.model_runtime.errors.invoke import (
     InvokeRateLimitError,
     InvokeServerUnavailableError,
 )
-from core.plugin.entities.plugin_daemon import PluginModelProviderEntity
+from core.plugin.entities.plugin_daemon import PluginDaemonInnerError, PluginModelProviderEntity
+from core.plugin.impl.model import PluginModelClient
 
 
 class AIModel(BaseModel):
@@ -52,8 +53,6 @@ class AIModel(BaseModel):
 
         :return: Invoke error mapping
         """
-        from core.plugin.entities.plugin_daemon import PluginDaemonInnerError
-
         return {
             InvokeConnectionError: [InvokeConnectionError],
             InvokeServerUnavailableError: [InvokeServerUnavailableError],
@@ -141,8 +140,6 @@ class AIModel(BaseModel):
         :param credentials: model credentials
         :return: model schema
         """
-        from core.plugin.impl.model import PluginModelClient
-
         plugin_model_manager = PluginModelClient()
         cache_key = f"{self.tenant_id}:{self.plugin_id}:{self.provider_name}:{self.model_type.value}:{model}"
         # sort credentials

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

@@ -22,6 +22,7 @@ from core.model_runtime.entities.model_entities import (
     PriceType,
 )
 from core.model_runtime.model_providers.__base.ai_model import AIModel
+from core.plugin.impl.model import PluginModelClient
 
 logger = logging.getLogger(__name__)
 
@@ -141,8 +142,6 @@ class LargeLanguageModel(AIModel):
         result: Union[LLMResult, Generator[LLMResultChunk, None, None]]
 
         try:
-            from core.plugin.impl.model import PluginModelClient
-
             plugin_model_manager = PluginModelClient()
             result = plugin_model_manager.invoke_llm(
                 tenant_id=self.tenant_id,
@@ -341,8 +340,6 @@ class LargeLanguageModel(AIModel):
         :return:
         """
         if dify_config.PLUGIN_BASED_TOKEN_COUNTING_ENABLED:
-            from core.plugin.impl.model import PluginModelClient
-
             plugin_model_manager = PluginModelClient()
             return plugin_model_manager.get_llm_num_tokens(
                 tenant_id=self.tenant_id,

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

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

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

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

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

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

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

@@ -6,6 +6,7 @@ from core.entities.embedding_type import EmbeddingInputType
 from core.model_runtime.entities.model_entities import ModelPropertyKey, ModelType
 from core.model_runtime.entities.text_embedding_entities import TextEmbeddingResult
 from core.model_runtime.model_providers.__base.ai_model import AIModel
+from core.plugin.impl.model import PluginModelClient
 
 
 class TextEmbeddingModel(AIModel):
@@ -36,8 +37,6 @@ class TextEmbeddingModel(AIModel):
         :param input_type: input type
         :return: embeddings result
         """
-        from core.plugin.impl.model import PluginModelClient
-
         try:
             plugin_model_manager = PluginModelClient()
             return plugin_model_manager.invoke_text_embedding(
@@ -62,8 +61,6 @@ class TextEmbeddingModel(AIModel):
         :param texts: texts to embed
         :return:
         """
-        from core.plugin.impl.model import PluginModelClient
-
         plugin_model_manager = PluginModelClient()
         return plugin_model_manager.get_text_embedding_num_tokens(
             tenant_id=self.tenant_id,

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

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

api/core/model_runtime/model_providers/model_provider_factory.py

@@ -20,8 +20,10 @@ from core.model_runtime.model_providers.__base.text_embedding_model import TextE
 from core.model_runtime.model_providers.__base.tts_model import TTSModel
 from core.model_runtime.schema_validators.model_credential_schema_validator import ModelCredentialSchemaValidator
 from core.model_runtime.schema_validators.provider_credential_schema_validator import ProviderCredentialSchemaValidator
+from core.plugin.entities.plugin import ModelProviderID
 from core.plugin.entities.plugin_daemon import PluginModelProviderEntity
-from models.provider_ids import ModelProviderID
+from core.plugin.impl.asset import PluginAssetManager
+from core.plugin.impl.model import PluginModelClient
 
 logger = logging.getLogger(__name__)
 
@@ -35,8 +37,6 @@ class ModelProviderFactory:
     provider_position_map: dict[str, int]
 
     def __init__(self, tenant_id: str) -> None:
-        from core.plugin.impl.model import PluginModelClient
-
         self.provider_position_map = {}
 
         self.tenant_id = tenant_id
@@ -71,7 +71,7 @@ class ModelProviderFactory:
 
         return [extension.plugin_model_provider_entity.declaration for extension in sorted_extensions.values()]
 
-    def get_plugin_model_providers(self) -> Sequence["PluginModelProviderEntity"]:
+    def get_plugin_model_providers(self) -> Sequence[PluginModelProviderEntity]:
         """
         Get all plugin model providers
         :return: list of plugin model providers
@@ -109,7 +109,7 @@ class ModelProviderFactory:
         plugin_model_provider_entity = self.get_plugin_model_provider(provider=provider)
         return plugin_model_provider_entity.declaration
 
-    def get_plugin_model_provider(self, provider: str) -> "PluginModelProviderEntity":
+    def get_plugin_model_provider(self, provider: str) -> PluginModelProviderEntity:
         """
         Get plugin model provider
         :param provider: provider name
@@ -366,8 +366,6 @@ class ModelProviderFactory:
         mime_type = image_mime_types.get(extension, "image/png")
 
         # get icon bytes from plugin asset manager
-        from core.plugin.impl.asset import PluginAssetManager
-
         plugin_asset_manager = PluginAssetManager()
         return plugin_asset_manager.fetch_asset(tenant_id=self.tenant_id, id=file_name), mime_type
 
@@ -377,6 +375,5 @@ class ModelProviderFactory:
         :param provider: provider name
         :return: plugin id and provider name
         """
-
         provider_id = ModelProviderID(provider)
         return provider_id.plugin_id, provider_id.provider_name

api/core/ops/aliyun_trace/aliyun_trace.py

@@ -54,10 +54,13 @@ from core.ops.entities.trace_entity import (
 )
 from core.rag.models.document import Document
 from core.repositories import SQLAlchemyWorkflowNodeExecutionRepository
-from core.workflow.entities import WorkflowNodeExecution
-from core.workflow.enums import NodeType, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
-from extensions.ext_database import db
-from models import Account, App, EndUser, TenantAccountJoin, WorkflowNodeExecutionTriggeredFrom
+from core.workflow.entities.workflow_node_execution import (
+    WorkflowNodeExecution,
+    WorkflowNodeExecutionMetadataKey,
+    WorkflowNodeExecutionStatus,
+)
+from core.workflow.nodes import NodeType
+from models import Account, App, EndUser, TenantAccountJoin, WorkflowNodeExecutionTriggeredFrom, db
 
 logger = logging.getLogger(__name__)
 

api/core/ops/langfuse_trace/langfuse_trace.py

@@ -29,7 +29,7 @@ from core.ops.langfuse_trace.entities.langfuse_trace_entity import (
 )
 from core.ops.utils import filter_none_values
 from core.repositories import DifyCoreRepositoryFactory
-from core.workflow.enums import NodeType
+from core.workflow.nodes.enums import NodeType
 from extensions.ext_database import db
 from models import EndUser, WorkflowNodeExecutionTriggeredFrom
 from models.enums import MessageStatus

api/core/ops/langsmith_trace/langsmith_trace.py

@@ -28,7 +28,8 @@ from core.ops.langsmith_trace.entities.langsmith_trace_entity import (
 )
 from core.ops.utils import filter_none_values, generate_dotted_order
 from core.repositories import DifyCoreRepositoryFactory
-from core.workflow.enums import NodeType, WorkflowNodeExecutionMetadataKey
+from core.workflow.entities.workflow_node_execution import WorkflowNodeExecutionMetadataKey
+from core.workflow.nodes.enums import NodeType
 from extensions.ext_database import db
 from models import EndUser, MessageFile, WorkflowNodeExecutionTriggeredFrom
 

api/core/ops/opik_trace/opik_trace.py

@@ -22,7 +22,8 @@ from core.ops.entities.trace_entity import (
     WorkflowTraceInfo,
 )
 from core.repositories import DifyCoreRepositoryFactory
-from core.workflow.enums import NodeType, WorkflowNodeExecutionMetadataKey
+from core.workflow.entities.workflow_node_execution import WorkflowNodeExecutionMetadataKey
+from core.workflow.nodes.enums import NodeType
 from extensions.ext_database import db
 from models import EndUser, MessageFile, WorkflowNodeExecutionTriggeredFrom
 

api/core/ops/ops_trace_manager.py

@@ -5,7 +5,7 @@ import queue
 import threading
 import time
 from datetime import timedelta
-from typing import TYPE_CHECKING, Any, Optional, Union
+from typing import Any, Optional, Union
 from uuid import UUID, uuid4
 
 from cachetools import LRUCache
@@ -30,15 +30,13 @@ from core.ops.entities.trace_entity import (
     WorkflowTraceInfo,
 )
 from core.ops.utils import get_message_data
+from core.workflow.entities.workflow_execution import WorkflowExecution
 from extensions.ext_database import db
 from extensions.ext_storage import storage
 from models.model import App, AppModelConfig, Conversation, Message, MessageFile, TraceAppConfig
 from models.workflow import WorkflowAppLog, WorkflowRun
 from tasks.ops_trace_task import process_trace_tasks
 
-if TYPE_CHECKING:
-    from core.workflow.entities import WorkflowExecution
-
 logger = logging.getLogger(__name__)
 
 
@@ -412,7 +410,7 @@ class TraceTask:
         self,
         trace_type: Any,
         message_id: Optional[str] = None,
-        workflow_execution: Optional["WorkflowExecution"] = None,
+        workflow_execution: Optional[WorkflowExecution] = None,
         conversation_id: Optional[str] = None,
         user_id: Optional[str] = None,
         timer: Optional[Any] = None,

api/core/ops/weave_trace/weave_trace.py

@@ -23,7 +23,8 @@ from core.ops.entities.trace_entity import (
 )
 from core.ops.weave_trace.entities.weave_trace_entity import WeaveTraceModel
 from core.repositories import DifyCoreRepositoryFactory
-from core.workflow.enums import NodeType, WorkflowNodeExecutionMetadataKey
+from core.workflow.entities.workflow_node_execution import WorkflowNodeExecutionMetadataKey
+from core.workflow.nodes.enums import NodeType
 from extensions.ext_database import db
 from models import EndUser, MessageFile, WorkflowNodeExecutionTriggeredFrom
 

api/core/plugin/backwards_invocation/app.py

@@ -164,6 +164,7 @@ class PluginAppBackwardsInvocation(BaseBackwardsInvocation):
             invoke_from=InvokeFrom.SERVICE_API,
             streaming=stream,
             call_depth=1,
+            workflow_thread_pool_id=None,
         )
 
     @classmethod

api/core/plugin/backwards_invocation/node.py

@@ -1,5 +1,5 @@
 from core.plugin.backwards_invocation.base import BaseBackwardsInvocation
-from core.workflow.enums import NodeType
+from core.workflow.nodes.enums import NodeType
 from core.workflow.nodes.parameter_extractor.entities import (
     ModelConfig as ParameterExtractorModelConfig,
 )

api/core/plugin/entities/parameters.py

@@ -1,11 +1,11 @@
 import enum
-import json
 from typing import Any, Optional, Union
 
 from pydantic import BaseModel, Field, field_validator
 
 from core.entities.parameter_entities import CommonParameterType
 from core.tools.entities.common_entities import I18nObject
+from core.workflow.nodes.base.entities import NumberType
 
 
 class PluginParameterOption(BaseModel):
@@ -154,7 +154,7 @@ def cast_parameter_value(typ: enum.StrEnum, value: Any, /):
                     raise ValueError("The tools selector must be a list.")
                 return value
             case PluginParameterType.ANY:
-                if value and not isinstance(value, str | dict | list | int | float):
+                if value and not isinstance(value, str | dict | list | NumberType):
                     raise ValueError("The var selector must be a string, dictionary, list or number.")
                 return value
             case PluginParameterType.ARRAY:
@@ -162,6 +162,8 @@ def cast_parameter_value(typ: enum.StrEnum, value: Any, /):
                     # Try to parse JSON string for arrays
                     if isinstance(value, str):
                         try:
+                            import json
+
                             parsed_value = json.loads(value)
                             if isinstance(parsed_value, list):
                                 return parsed_value
@@ -174,6 +176,8 @@ def cast_parameter_value(typ: enum.StrEnum, value: Any, /):
                     # Try to parse JSON string for objects
                     if isinstance(value, str):
                         try:
+                            import json
+
                             parsed_value = json.loads(value)
                             if isinstance(parsed_value, dict):
                                 return parsed_value

api/core/plugin/entities/plugin.py

@@ -1,9 +1,11 @@
 import datetime
 import enum
+import re
 from collections.abc import Mapping
 from typing import Any, Optional
 
 from pydantic import BaseModel, Field, model_validator
+from werkzeug.exceptions import NotFound
 
 from core.agent.plugin_entities import AgentStrategyProviderEntity
 from core.model_runtime.entities.provider_entities import ProviderEntity
@@ -11,6 +13,7 @@ from core.plugin.entities.base import BasePluginEntity
 from core.plugin.entities.endpoint import EndpointProviderDeclaration
 from core.tools.entities.common_entities import I18nObject
 from core.tools.entities.tool_entities import ToolProviderEntity
+from core.trigger.entities.entities import TriggerProviderEntity
 
 
 class PluginInstallationSource(enum.StrEnum):
@@ -60,6 +63,7 @@ class PluginCategory(enum.StrEnum):
     Model = "model"
     Extension = "extension"
     AgentStrategy = "agent-strategy"
+    Trigger = "trigger"
 
 
 class PluginDeclaration(BaseModel):
@@ -67,6 +71,7 @@ class PluginDeclaration(BaseModel):
         tools: Optional[list[str]] = Field(default_factory=list[str])
         models: Optional[list[str]] = Field(default_factory=list[str])
         endpoints: Optional[list[str]] = Field(default_factory=list[str])
+        triggers: Optional[list[str]] = Field(default_factory=list[str])
 
     class Meta(BaseModel):
         minimum_dify_version: Optional[str] = Field(default=None, pattern=r"^\d{1,4}(\.\d{1,4}){1,3}(-\w{1,16})?$")
@@ -87,6 +92,7 @@ class PluginDeclaration(BaseModel):
     repo: Optional[str] = Field(default=None)
     verified: bool = Field(default=False)
     tool: Optional[ToolProviderEntity] = None
+    trigger: Optional[TriggerProviderEntity] = None
     model: Optional[ProviderEntity] = None
     endpoint: Optional[EndpointProviderDeclaration] = None
     agent_strategy: Optional[AgentStrategyProviderEntity] = None
@@ -102,6 +108,8 @@ class PluginDeclaration(BaseModel):
             values["category"] = PluginCategory.Model
         elif values.get("agent_strategy"):
             values["category"] = PluginCategory.AgentStrategy
+        elif values.get("trigger"):
+            values["category"] = PluginCategory.Trigger
         else:
             values["category"] = PluginCategory.Extension
         return values
@@ -133,6 +141,59 @@ class PluginEntity(PluginInstallation):
         return self
 
 
+class GenericProviderID:
+    organization: str
+    plugin_name: str
+    provider_name: str
+    is_hardcoded: bool
+
+    def to_string(self) -> str:
+        return str(self)
+
+    def __str__(self) -> str:
+        return f"{self.organization}/{self.plugin_name}/{self.provider_name}"
+
+    def __init__(self, value: str, is_hardcoded: bool = False) -> None:
+        if not value:
+            raise NotFound("plugin not found, please add plugin")
+        # check if the value is a valid plugin id with format: $organization/$plugin_name/$provider_name
+        if not re.match(r"^[a-z0-9_-]+\/[a-z0-9_-]+\/[a-z0-9_-]+$", value):
+            # check if matches [a-z0-9_-]+, if yes, append with langgenius/$value/$value
+            if re.match(r"^[a-z0-9_-]+$", value):
+                value = f"langgenius/{value}/{value}"
+            else:
+                raise ValueError(f"Invalid plugin id {value}")
+
+        self.organization, self.plugin_name, self.provider_name = value.split("/")
+        self.is_hardcoded = is_hardcoded
+
+    def is_langgenius(self) -> bool:
+        return self.organization == "langgenius"
+
+    @property
+    def plugin_id(self) -> str:
+        return f"{self.organization}/{self.plugin_name}"
+
+
+class ModelProviderID(GenericProviderID):
+    def __init__(self, value: str, is_hardcoded: bool = False) -> None:
+        super().__init__(value, is_hardcoded)
+        if self.organization == "langgenius" and self.provider_name == "google":
+            self.plugin_name = "gemini"
+
+
+class ToolProviderID(GenericProviderID):
+    def __init__(self, value: str, is_hardcoded: bool = False) -> None:
+        super().__init__(value, is_hardcoded)
+        if self.organization == "langgenius":
+            if self.provider_name in ["jina", "siliconflow", "stepfun", "gitee_ai"]:
+                self.plugin_name = f"{self.provider_name}_tool"
+
+
+class TriggerProviderID(GenericProviderID):
+    pass
+
+
 class PluginDependency(BaseModel):
     class Type(enum.StrEnum):
         Github = PluginInstallationSource.Github.value

api/core/plugin/entities/plugin_daemon.py

@@ -1,3 +1,4 @@
+import enum
 from collections.abc import Mapping, Sequence
 from datetime import datetime
 from enum import StrEnum
@@ -13,6 +14,7 @@ from core.plugin.entities.parameters import PluginParameterOption
 from core.plugin.entities.plugin import PluginDeclaration, PluginEntity
 from core.tools.entities.common_entities import I18nObject
 from core.tools.entities.tool_entities import ToolProviderEntityWithPlugin
+from core.trigger.entities.entities import TriggerProviderEntity
 
 T = TypeVar("T", bound=(BaseModel | dict | list | bool | str))
 
@@ -196,3 +198,48 @@ class PluginListResponse(BaseModel):
 
 class PluginDynamicSelectOptionsResponse(BaseModel):
     options: Sequence[PluginParameterOption] = Field(description="The options of the dynamic select.")
+
+
+class PluginTriggerProviderEntity(BaseModel):
+    provider: str
+    plugin_unique_identifier: str
+    plugin_id: str
+    declaration: TriggerProviderEntity
+
+
+class CredentialType(enum.StrEnum):
+    API_KEY = "api-key"
+    OAUTH2 = "oauth2"
+    UNAUTHORIZED = "unauthorized"
+
+    def get_name(self):
+        if self == CredentialType.API_KEY:
+            return "API KEY"
+        elif self == CredentialType.OAUTH2:
+            return "AUTH"
+        elif self == CredentialType.UNAUTHORIZED:
+            return "UNAUTHORIZED"
+        else:
+            return self.value.replace("-", " ").upper()
+
+    def is_editable(self):
+        return self == CredentialType.API_KEY
+
+    def is_validate_allowed(self):
+        return self == CredentialType.API_KEY
+
+    @classmethod
+    def values(cls):
+        return [item.value for item in cls]
+
+    @classmethod
+    def of(cls, credential_type: str) -> "CredentialType":
+        type_name = credential_type.lower()
+        if type_name == "api-key":
+            return cls.API_KEY
+        elif type_name == "oauth2":
+            return cls.OAUTH2
+        elif type_name == "unauthorized":
+            return cls.UNAUTHORIZED
+        else:
+            raise ValueError(f"Invalid credential type: {credential_type}")

api/core/plugin/entities/request.py

@@ -1,5 +1,7 @@
+from collections.abc import Mapping
 from typing import Any, Literal, Optional
 
+from flask import Response
 from pydantic import BaseModel, ConfigDict, Field, field_validator
 
 from core.entities.provider_entities import BasicProviderConfig
@@ -237,3 +239,33 @@ class RequestFetchAppInfo(BaseModel):
     """
 
     app_id: str
+
+
+class Event(BaseModel):
+    variables: Mapping[str, Any]
+
+
+class TriggerInvokeResponse(BaseModel):
+    event: Event
+
+
+class PluginTriggerDispatchResponse(BaseModel):
+    triggers: list[str]
+    raw_http_response: str
+
+
+class TriggerSubscriptionResponse(BaseModel):
+    subscription: dict[str, Any]
+
+
+class TriggerValidateProviderCredentialsResponse(BaseModel):
+    result: bool
+
+
+class TriggerDispatchResponse:
+    triggers: list[str]
+    response: Response
+
+    def __init__(self, triggers: list[str], response: Response):
+        self.triggers = triggers
+        self.response = response

api/core/plugin/impl/agent.py

@@ -2,13 +2,13 @@ from collections.abc import Generator
 from typing import Any, Optional
 
 from core.agent.entities import AgentInvokeMessage
+from core.plugin.entities.plugin import GenericProviderID
 from core.plugin.entities.plugin_daemon import (
     PluginAgentProviderEntity,
 )
 from core.plugin.entities.request import PluginInvokeContext
 from core.plugin.impl.base import BasePluginClient
 from core.plugin.utils.chunk_merger import merge_blob_chunks
-from models.provider_ids import GenericProviderID
 
 
 class PluginAgentClient(BasePluginClient):

api/core/plugin/impl/dynamic_select.py

@@ -1,9 +1,9 @@
 from collections.abc import Mapping
 from typing import Any
 
+from core.plugin.entities.plugin import GenericProviderID
 from core.plugin.entities.plugin_daemon import PluginDynamicSelectOptionsResponse
 from core.plugin.impl.base import BasePluginClient
-from models.provider_ids import GenericProviderID
 
 
 class DynamicSelectClient(BasePluginClient):
@@ -15,6 +15,7 @@ class DynamicSelectClient(BasePluginClient):
         provider: str,
         action: str,
         credentials: Mapping[str, Any],
+        credential_type: str,
         parameter: str,
     ) -> PluginDynamicSelectOptionsResponse:
         """
@@ -29,6 +30,7 @@ class DynamicSelectClient(BasePluginClient):
                 "data": {
                     "provider": GenericProviderID(provider).provider_name,
                     "credentials": credentials,
+                    "credential_type": credential_type,
                     "provider_action": action,
                     "parameter": parameter,
                 },

api/core/plugin/impl/plugin.py

@@ -2,6 +2,7 @@ from collections.abc import Sequence
 
 from core.plugin.entities.bundle import PluginBundleDependency
 from core.plugin.entities.plugin import (
+    GenericProviderID,
     MissingPluginDependency,
     PluginDeclaration,
     PluginEntity,
@@ -15,7 +16,6 @@ from core.plugin.entities.plugin_daemon import (
     PluginListResponse,
 )
 from core.plugin.impl.base import BasePluginClient
-from models.provider_ids import GenericProviderID
 
 
 class PluginInstaller(BasePluginClient):

api/core/plugin/impl/tool.py

@@ -3,11 +3,11 @@ from typing import Any, Optional
 
 from pydantic import BaseModel
 
-from core.plugin.entities.plugin_daemon import PluginBasicBooleanResponse, PluginToolProviderEntity
+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
 from core.plugin.utils.chunk_merger import merge_blob_chunks
-from core.tools.entities.tool_entities import CredentialType, ToolInvokeMessage, ToolParameter
-from models.provider_ids import GenericProviderID, ToolProviderID
+from core.tools.entities.tool_entities import ToolInvokeMessage, ToolParameter
 
 
 class PluginToolManager(BasePluginClient):

api/core/plugin/impl/trigger.py

@@ -0,0 +1,301 @@
+import binascii
+from collections.abc import Mapping
+from typing import Any
+
+from flask import Request
+
+from core.plugin.entities.plugin import GenericProviderID, TriggerProviderID
+from core.plugin.entities.plugin_daemon import CredentialType, PluginTriggerProviderEntity
+from core.plugin.entities.request import (
+    PluginTriggerDispatchResponse,
+    TriggerDispatchResponse,
+    TriggerInvokeResponse,
+    TriggerSubscriptionResponse,
+    TriggerValidateProviderCredentialsResponse,
+)
+from core.plugin.impl.base import BasePluginClient
+from core.plugin.utils.http_parser import deserialize_response, serialize_request
+from core.trigger.entities.entities import Subscription
+
+
+class PluginTriggerManager(BasePluginClient):
+    def fetch_trigger_providers(self, tenant_id: str) -> list[PluginTriggerProviderEntity]:
+        """
+        Fetch trigger providers for the given tenant.
+        """
+
+        def transformer(json_response: dict[str, Any]) -> dict:
+            for provider in json_response.get("data", []):
+                declaration = provider.get("declaration", {}) or {}
+                provider_id = provider.get("plugin_id") + "/" + provider.get("provider")
+                for trigger in declaration.get("triggers", []):
+                    trigger["identity"]["provider"] = provider_id
+
+            return json_response
+
+        response = self._request_with_plugin_daemon_response(
+            "GET",
+            f"plugin/{tenant_id}/management/triggers",
+            list[PluginTriggerProviderEntity],
+            params={"page": 1, "page_size": 256},
+            transformer=transformer,
+        )
+
+        for provider in response:
+            provider.declaration.identity.name = f"{provider.plugin_id}/{provider.declaration.identity.name}"
+
+            # override the provider name for each trigger to plugin_id/provider_name
+            for trigger in provider.declaration.triggers:
+                trigger.identity.provider = provider.declaration.identity.name
+
+        return response
+
+    def fetch_trigger_provider(self, tenant_id: str, provider_id: TriggerProviderID) -> PluginTriggerProviderEntity:
+        """
+        Fetch trigger provider for the given tenant and plugin.
+        """
+
+        def transformer(json_response: dict[str, Any]) -> dict:
+            data = json_response.get("data")
+            if data:
+                for trigger in data.get("declaration", {}).get("triggers", []):
+                    trigger["identity"]["provider"] = str(provider_id)
+
+            return json_response
+
+        response = self._request_with_plugin_daemon_response(
+            "GET",
+            f"plugin/{tenant_id}/management/trigger",
+            PluginTriggerProviderEntity,
+            params={"provider": provider_id.provider_name, "plugin_id": provider_id.plugin_id},
+            transformer=transformer,
+        )
+
+        response.declaration.identity.name = str(provider_id)
+
+        # override the provider name for each trigger to plugin_id/provider_name
+        for trigger in response.declaration.triggers:
+            trigger.identity.provider = str(provider_id)
+
+        return response
+
+    def invoke_trigger(
+        self,
+        tenant_id: str,
+        user_id: str,
+        provider: str,
+        trigger: str,
+        credentials: Mapping[str, str],
+        credential_type: CredentialType,
+        request: Request,
+        parameters: Mapping[str, Any],
+    ) -> TriggerInvokeResponse:
+        """
+        Invoke a trigger with the given parameters.
+        """
+        trigger_provider_id = GenericProviderID(provider)
+
+        response = self._request_with_plugin_daemon_response_stream(
+            "POST",
+            f"plugin/{tenant_id}/dispatch/trigger/invoke",
+            TriggerInvokeResponse,
+            data={
+                "user_id": user_id,
+                "data": {
+                    "provider": trigger_provider_id.provider_name,
+                    "trigger": trigger,
+                    "credentials": credentials,
+                    "credential_type": credential_type,
+                    "raw_http_request": binascii.hexlify(serialize_request(request)).decode(),
+                    "parameters": parameters,
+                },
+            },
+            headers={
+                "X-Plugin-ID": trigger_provider_id.plugin_id,
+                "Content-Type": "application/json",
+            },
+        )
+
+        for resp in response:
+            return TriggerInvokeResponse(event=resp.event)
+
+        raise ValueError("No response received from plugin daemon for invoke trigger")
+
+    def validate_provider_credentials(
+        self, tenant_id: str, user_id: str, provider: str, credentials: Mapping[str, str]
+    ) -> bool:
+        """
+        Validate the credentials of the trigger provider.
+        """
+        trigger_provider_id = GenericProviderID(provider)
+
+        response = self._request_with_plugin_daemon_response_stream(
+            "POST",
+            f"plugin/{tenant_id}/dispatch/trigger/validate_credentials",
+            TriggerValidateProviderCredentialsResponse,
+            data={
+                "user_id": user_id,
+                "data": {
+                    "provider": trigger_provider_id.provider_name,
+                    "credentials": credentials,
+                },
+            },
+            headers={
+                "X-Plugin-ID": trigger_provider_id.plugin_id,
+                "Content-Type": "application/json",
+            },
+        )
+
+        for resp in response:
+            return resp.result
+
+        raise ValueError("No response received from plugin daemon for validate provider credentials")
+
+    def dispatch_event(
+        self,
+        tenant_id: str,
+        user_id: str,
+        provider: str,
+        subscription: Mapping[str, Any],
+        request: Request,
+    ) -> TriggerDispatchResponse:
+        """
+        Dispatch an event to triggers.
+        """
+        trigger_provider_id = GenericProviderID(provider)
+
+        response = self._request_with_plugin_daemon_response_stream(
+            "POST",
+            f"plugin/{tenant_id}/dispatch/trigger/dispatch_event",
+            PluginTriggerDispatchResponse,
+            data={
+                "user_id": user_id,
+                "data": {
+                    "provider": trigger_provider_id.provider_name,
+                    "subscription": subscription,
+                    "raw_http_request": binascii.hexlify(serialize_request(request)).decode(),
+                },
+            },
+            headers={
+                "X-Plugin-ID": trigger_provider_id.plugin_id,
+                "Content-Type": "application/json",
+            },
+        )
+
+        for resp in response:
+            return TriggerDispatchResponse(
+                triggers=resp.triggers,
+                response=deserialize_response(binascii.unhexlify(resp.raw_http_response.encode())),
+            )
+
+        raise ValueError("No response received from plugin daemon for dispatch event")
+
+    def subscribe(
+        self,
+        tenant_id: str,
+        user_id: str,
+        provider: str,
+        credentials: Mapping[str, str],
+        endpoint: str,
+        parameters: Mapping[str, Any],
+    ) -> TriggerSubscriptionResponse:
+        """
+        Subscribe to a trigger.
+        """
+        trigger_provider_id = GenericProviderID(provider)
+
+        response = self._request_with_plugin_daemon_response_stream(
+            "POST",
+            f"plugin/{tenant_id}/dispatch/trigger/subscribe",
+            TriggerSubscriptionResponse,
+            data={
+                "user_id": user_id,
+                "data": {
+                    "provider": trigger_provider_id.provider_name,
+                    "credentials": credentials,
+                    "endpoint": endpoint,
+                    "parameters": parameters,
+                },
+            },
+            headers={
+                "X-Plugin-ID": trigger_provider_id.plugin_id,
+                "Content-Type": "application/json",
+            },
+        )
+
+        for resp in response:
+            return resp
+
+        raise ValueError("No response received from plugin daemon for subscribe")
+
+    def unsubscribe(
+        self,
+        tenant_id: str,
+        user_id: str,
+        provider: str,
+        subscription: Subscription,
+        credentials: Mapping[str, str],
+    ) -> TriggerSubscriptionResponse:
+        """
+        Unsubscribe from a trigger.
+        """
+        trigger_provider_id = GenericProviderID(provider)
+
+        response = self._request_with_plugin_daemon_response_stream(
+            "POST",
+            f"plugin/{tenant_id}/dispatch/trigger/unsubscribe",
+            TriggerSubscriptionResponse,
+            data={
+                "user_id": user_id,
+                "data": {
+                    "provider": trigger_provider_id.provider_name,
+                    "subscription": subscription.model_dump(),
+                    "credentials": credentials,
+                },
+            },
+            headers={
+                "X-Plugin-ID": trigger_provider_id.plugin_id,
+                "Content-Type": "application/json",
+            },
+        )
+
+        for resp in response:
+            return resp
+
+        raise ValueError("No response received from plugin daemon for unsubscribe")
+
+    def refresh(
+        self,
+        tenant_id: str,
+        user_id: str,
+        provider: str,
+        subscription: Subscription,
+        credentials: Mapping[str, str],
+    ) -> TriggerSubscriptionResponse:
+        """
+        Refresh a trigger subscription.
+        """
+        trigger_provider_id = GenericProviderID(provider)
+
+        response = self._request_with_plugin_daemon_response_stream(
+            "POST",
+            f"plugin/{tenant_id}/dispatch/trigger/refresh",
+            TriggerSubscriptionResponse,
+            data={
+                "user_id": user_id,
+                "data": {
+                    "provider": trigger_provider_id.provider_name,
+                    "subscription": subscription.model_dump(),
+                    "credentials": credentials,
+                },
+            },
+            headers={
+                "X-Plugin-ID": trigger_provider_id.plugin_id,
+                "Content-Type": "application/json",
+            },
+        )
+
+        for resp in response:
+            return resp
+
+        raise ValueError("No response received from plugin daemon for refresh")

api/core/plugin/utils/http_parser.py

@@ -0,0 +1,159 @@
+from io import BytesIO
+
+from flask import Request, Response
+from werkzeug.datastructures import Headers
+
+
+def serialize_request(request: Request) -> bytes:
+    method = request.method
+    path = request.full_path.rstrip("?")
+    raw = f"{method} {path} HTTP/1.1\r\n".encode()
+
+    for name, value in request.headers.items():
+        raw += f"{name}: {value}\r\n".encode()
+
+    raw += b"\r\n"
+
+    body = request.get_data(as_text=False)
+    if body:
+        raw += body
+
+    return raw
+
+
+def deserialize_request(raw_data: bytes) -> Request:
+    header_end = raw_data.find(b"\r\n\r\n")
+    if header_end == -1:
+        header_end = raw_data.find(b"\n\n")
+        if header_end == -1:
+            header_data = raw_data
+            body = b""
+        else:
+            header_data = raw_data[:header_end]
+            body = raw_data[header_end + 2 :]
+    else:
+        header_data = raw_data[:header_end]
+        body = raw_data[header_end + 4 :]
+
+    lines = header_data.split(b"\r\n")
+    if len(lines) == 1 and b"\n" in lines[0]:
+        lines = header_data.split(b"\n")
+
+    if not lines or not lines[0]:
+        raise ValueError("Empty HTTP request")
+
+    request_line = lines[0].decode("utf-8", errors="ignore")
+    parts = request_line.split(" ", 2)
+    if len(parts) < 2:
+        raise ValueError(f"Invalid request line: {request_line}")
+
+    method = parts[0]
+    full_path = parts[1]
+    protocol = parts[2] if len(parts) > 2 else "HTTP/1.1"
+
+    if "?" in full_path:
+        path, query_string = full_path.split("?", 1)
+    else:
+        path = full_path
+        query_string = ""
+
+    headers = Headers()
+    for line in lines[1:]:
+        if not line:
+            continue
+        line_str = line.decode("utf-8", errors="ignore")
+        if ":" not in line_str:
+            continue
+        name, value = line_str.split(":", 1)
+        headers.add(name, value.strip())
+
+    host = headers.get("Host", "localhost")
+    if ":" in host:
+        server_name, server_port = host.rsplit(":", 1)
+    else:
+        server_name = host
+        server_port = "80"
+
+    environ = {
+        "REQUEST_METHOD": method,
+        "PATH_INFO": path,
+        "QUERY_STRING": query_string,
+        "SERVER_NAME": server_name,
+        "SERVER_PORT": server_port,
+        "SERVER_PROTOCOL": protocol,
+        "wsgi.input": BytesIO(body),
+        "wsgi.url_scheme": "http",
+    }
+
+    if "Content-Type" in headers:
+        environ["CONTENT_TYPE"] = headers.get("Content-Type")
+
+    if "Content-Length" in headers:
+        environ["CONTENT_LENGTH"] = headers.get("Content-Length")
+    elif body:
+        environ["CONTENT_LENGTH"] = str(len(body))
+
+    for name, value in headers.items():
+        if name.upper() in ("CONTENT-TYPE", "CONTENT-LENGTH"):
+            continue
+        env_name = f"HTTP_{name.upper().replace('-', '_')}"
+        environ[env_name] = value
+
+    return Request(environ)
+
+
+def serialize_response(response: Response) -> bytes:
+    raw = f"HTTP/1.1 {response.status}\r\n".encode()
+
+    for name, value in response.headers.items():
+        raw += f"{name}: {value}\r\n".encode()
+
+    raw += b"\r\n"
+
+    body = response.get_data(as_text=False)
+    if body:
+        raw += body
+
+    return raw
+
+
+def deserialize_response(raw_data: bytes) -> Response:
+    header_end = raw_data.find(b"\r\n\r\n")
+    if header_end == -1:
+        header_end = raw_data.find(b"\n\n")
+        if header_end == -1:
+            header_data = raw_data
+            body = b""
+        else:
+            header_data = raw_data[:header_end]
+            body = raw_data[header_end + 2 :]
+    else:
+        header_data = raw_data[:header_end]
+        body = raw_data[header_end + 4 :]
+
+    lines = header_data.split(b"\r\n")
+    if len(lines) == 1 and b"\n" in lines[0]:
+        lines = header_data.split(b"\n")
+
+    if not lines or not lines[0]:
+        raise ValueError("Empty HTTP response")
+
+    status_line = lines[0].decode("utf-8", errors="ignore")
+    parts = status_line.split(" ", 2)
+    if len(parts) < 2:
+        raise ValueError(f"Invalid status line: {status_line}")
+
+    status_code = int(parts[1])
+
+    response = Response(response=body, status=status_code)
+
+    for line in lines[1:]:
+        if not line:
+            continue
+        line_str = line.decode("utf-8", errors="ignore")
+        if ":" not in line_str:
+            continue
+        name, value = line_str.split(":", 1)
+        response.headers[name] = value.strip()
+
+    return response

api/core/prompt/utils/prompt_message_util.py

@@ -87,7 +87,6 @@ class PromptMessageUtil:
             if isinstance(prompt_message.content, list):
                 for content in prompt_message.content:
                     if content.type == PromptMessageContentType.TEXT:
-                        content = cast(TextPromptMessageContent, content)
                         text += content.data
                     else:
                         content = cast(ImagePromptMessageContent, content)

api/core/provider_manager.py

@@ -2,7 +2,7 @@ import contextlib
 import json
 from collections import defaultdict
 from json import JSONDecodeError
-from typing import Any, Optional, cast
+from typing import Any, Optional
 
 from sqlalchemy import select
 from sqlalchemy.exc import IntegrityError
@@ -34,6 +34,7 @@ from core.model_runtime.entities.provider_entities import (
     ProviderEntity,
 )
 from core.model_runtime.model_providers.model_provider_factory import ModelProviderFactory
+from core.plugin.entities.plugin import ModelProviderID
 from extensions import ext_hosting_provider
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
@@ -48,7 +49,6 @@ from models.provider import (
     TenantDefaultModel,
     TenantPreferredModelProvider,
 )
-from models.provider_ids import ModelProviderID
 from services.feature_service import FeatureService
 
 
@@ -154,8 +154,8 @@ class ProviderManager:
         for provider_entity in provider_entities:
             # handle include, exclude
             if is_filtered(
-                include_set=cast(set[str], dify_config.POSITION_PROVIDER_INCLUDES_SET),
-                exclude_set=cast(set[str], dify_config.POSITION_PROVIDER_EXCLUDES_SET),
+                include_set=dify_config.POSITION_PROVIDER_INCLUDES_SET,
+                exclude_set=dify_config.POSITION_PROVIDER_EXCLUDES_SET,
                 data=provider_entity,
                 name_func=lambda x: x.provider,
             ):

api/core/rag/datasource/retrieval_service.py

@@ -24,7 +24,7 @@ default_retrieval_model = {
     "search_method": RetrievalMethod.SEMANTIC_SEARCH.value,
     "reranking_enable": False,
     "reranking_model": {"reranking_provider_name": "", "reranking_model_name": ""},
-    "top_k": 2,
+    "top_k": 4,
     "score_threshold_enabled": False,
 }
 

api/core/rag/datasource/vdb/couchbase/couchbase_vector.py

@@ -304,7 +304,7 @@ class CouchbaseVector(BaseVector):
         return docs
 
     def search_by_full_text(self, query: str, **kwargs: Any) -> list[Document]:
-        top_k = kwargs.get("top_k", 2)
+        top_k = kwargs.get("top_k", 4)
         try:
             CBrequest = search.SearchRequest.create(search.QueryStringQuery("text:" + query))
             search_iter = self._scope.search(

api/core/rag/datasource/vdb/qdrant/qdrant_vector.py

@@ -3,7 +3,7 @@ import os
 import uuid
 from collections.abc import Generator, Iterable, Sequence
 from itertools import islice
-from typing import TYPE_CHECKING, Any, Optional, Union, cast
+from typing import TYPE_CHECKING, Any, Optional, Union
 
 import qdrant_client
 from flask import current_app
@@ -426,7 +426,6 @@ class QdrantVector(BaseVector):
 
     def _reload_if_needed(self):
         if isinstance(self._client, QdrantLocal):
-            self._client = cast(QdrantLocal, self._client)
             self._client._load()
 
     @classmethod

api/core/rag/extractor/markdown_extractor.py

@@ -2,7 +2,7 @@
 
 import re
 from pathlib import Path
-from typing import Optional, cast
+from typing import Optional
 
 from core.rag.extractor.extractor_base import BaseExtractor
 from core.rag.extractor.helpers import detect_file_encodings
@@ -76,7 +76,7 @@ class MarkdownExtractor(BaseExtractor):
         markdown_tups.append((current_header, current_text))
 
         markdown_tups = [
-            (re.sub(r"#", "", cast(str, key)).strip() if key else None, re.sub(r"<.*?>", "", value))
+            (re.sub(r"#", "", key).strip() if key else None, re.sub(r"<.*?>", "", value))
             for key, value in markdown_tups
         ]
 

api/core/rag/extractor/notion_extractor.py

@@ -385,4 +385,4 @@ class NotionExtractor(BaseExtractor):
                 f"No notion data source binding found for tenant {tenant_id} and notion workspace {notion_workspace_id}"
             )
 
-        return cast(str, data_source_binding.access_token)
+        return data_source_binding.access_token

api/core/rag/extractor/pdf_extractor.py

@@ -2,7 +2,7 @@
 
 import contextlib
 from collections.abc import Iterator
-from typing import Optional, cast
+from typing import Optional
 
 from core.rag.extractor.blob.blob import Blob
 from core.rag.extractor.extractor_base import BaseExtractor
@@ -27,7 +27,7 @@ class PdfExtractor(BaseExtractor):
         plaintext_file_exists = False
         if self._file_cache_key:
             with contextlib.suppress(FileNotFoundError):
-                text = cast(bytes, storage.load(self._file_cache_key)).decode("utf-8")
+                text = storage.load(self._file_cache_key).decode("utf-8")
                 plaintext_file_exists = True
                 return [Document(page_content=text)]
         documents = list(self.load())

api/core/rag/index_processor/index_processor_base.py

@@ -1,9 +1,10 @@
 """Abstract interface for document loader implementations."""
 
 from abc import ABC, abstractmethod
-from typing import TYPE_CHECKING, Optional
+from typing import Optional
 
 from configs import dify_config
+from core.model_manager import ModelInstance
 from core.rag.extractor.entity.extract_setting import ExtractSetting
 from core.rag.models.document import Document
 from core.rag.splitter.fixed_text_splitter import (
@@ -13,9 +14,6 @@ from core.rag.splitter.fixed_text_splitter import (
 from core.rag.splitter.text_splitter import TextSplitter
 from models.dataset import Dataset, DatasetProcessRule
 
-if TYPE_CHECKING:
-    from core.model_manager import ModelInstance
-
 
 class BaseIndexProcessor(ABC):
     """Interface for extract files."""
@@ -53,7 +51,7 @@ class BaseIndexProcessor(ABC):
         max_tokens: int,
         chunk_overlap: int,
         separator: str,
-        embedding_model_instance: Optional["ModelInstance"],
+        embedding_model_instance: Optional[ModelInstance],
     ) -> TextSplitter:
         """
         Get the NodeParser object according to the processing rule.

api/core/rag/retrieval/dataset_retrieval.py

@@ -65,7 +65,7 @@ default_retrieval_model: dict[str, Any] = {
     "search_method": RetrievalMethod.SEMANTIC_SEARCH.value,
     "reranking_enable": False,
     "reranking_model": {"reranking_provider_name": "", "reranking_model_name": ""},
-    "top_k": 2,
+    "top_k": 4,
     "score_threshold_enabled": False,
 }
 
@@ -647,7 +647,7 @@ class DatasetRetrieval:
                             retrieval_method=retrieval_model["search_method"],
                             dataset_id=dataset.id,
                             query=query,
-                            top_k=retrieval_model.get("top_k") or 2,
+                            top_k=retrieval_model.get("top_k") or 4,
                             score_threshold=retrieval_model.get("score_threshold", 0.0)
                             if retrieval_model["score_threshold_enabled"]
                             else 0.0,
@@ -743,7 +743,7 @@ class DatasetRetrieval:
             tool = DatasetMultiRetrieverTool.from_dataset(
                 dataset_ids=[dataset.id for dataset in available_datasets],
                 tenant_id=tenant_id,
-                top_k=retrieve_config.top_k or 2,
+                top_k=retrieve_config.top_k or 4,
                 score_threshold=retrieve_config.score_threshold,
                 hit_callbacks=[hit_callback],
                 return_resource=return_resource,

api/core/repositories/sqlalchemy_workflow_execution_repository.py

@@ -9,8 +9,11 @@ from typing import Optional, Union
 from sqlalchemy.engine import Engine
 from sqlalchemy.orm import sessionmaker
 
-from core.workflow.entities import WorkflowExecution
-from core.workflow.enums import WorkflowExecutionStatus, WorkflowType
+from core.workflow.entities.workflow_execution import (
+    WorkflowExecution,
+    WorkflowExecutionStatus,
+    WorkflowType,
+)
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.workflow_type_encoder import WorkflowRuntimeTypeConverter
 from libs.helper import extract_tenant_id
@@ -200,4 +203,5 @@ class SQLAlchemyWorkflowExecutionRepository(WorkflowExecutionRepository):
             session.commit()
 
             # Update the in-memory cache for faster subsequent lookups
+            logger.debug("Updating cache for execution_id: %s", db_model.id)
             self._execution_cache[db_model.id] = db_model

api/core/repositories/sqlalchemy_workflow_node_execution_repository.py

@@ -12,8 +12,12 @@ from sqlalchemy.engine import Engine
 from sqlalchemy.orm import sessionmaker
 
 from core.model_runtime.utils.encoders import jsonable_encoder
-from core.workflow.entities import WorkflowNodeExecution
-from core.workflow.enums import NodeType, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
+from core.workflow.entities.workflow_node_execution import (
+    WorkflowNodeExecution,
+    WorkflowNodeExecutionMetadataKey,
+    WorkflowNodeExecutionStatus,
+)
+from core.workflow.nodes.enums import NodeType
 from core.workflow.repositories.workflow_node_execution_repository import OrderConfig, WorkflowNodeExecutionRepository
 from core.workflow.workflow_type_encoder import WorkflowRuntimeTypeConverter
 from libs.helper import extract_tenant_id
@@ -211,6 +215,7 @@ class SQLAlchemyWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository)
             # Update the in-memory cache for faster subsequent lookups
             # Only cache if we have a node_execution_id to use as the cache key
             if db_model.node_execution_id:
+                logger.debug("Updating cache for node_execution_id: %s", db_model.node_execution_id)
                 self._node_execution_cache[db_model.node_execution_id] = db_model
 
     def get_db_models_by_workflow_run(

api/core/tools/__base/tool_runtime.py

@@ -4,7 +4,8 @@ from openai import BaseModel
 from pydantic import Field
 
 from core.app.entities.app_invoke_entities import InvokeFrom
-from core.tools.entities.tool_entities import CredentialType, ToolInvokeFrom
+from core.plugin.entities.plugin_daemon import CredentialType
+from core.tools.entities.tool_entities import ToolInvokeFrom
 
 
 class ToolRuntime(BaseModel):

api/core/tools/builtin_tool/provider.py

@@ -4,11 +4,11 @@ from typing import Any
 
 from core.entities.provider_entities import ProviderConfig
 from core.helper.module_import_helper import load_single_subclass_from_source
+from core.plugin.entities.plugin_daemon import CredentialType
 from core.tools.__base.tool_provider import ToolProviderController
 from core.tools.__base.tool_runtime import ToolRuntime
 from core.tools.builtin_tool.tool import BuiltinTool
 from core.tools.entities.tool_entities import (
-    CredentialType,
     OAuthSchema,
     ToolEntity,
     ToolProviderEntity,

api/core/tools/entities/api_entities.py

@@ -4,9 +4,10 @@ from typing import Any, Literal, Optional
 from pydantic import BaseModel, Field, field_validator
 
 from core.model_runtime.utils.encoders import jsonable_encoder
+from core.plugin.entities.plugin_daemon import CredentialType
 from core.tools.__base.tool import ToolParameter
 from core.tools.entities.common_entities import I18nObject
-from core.tools.entities.tool_entities import CredentialType, ToolProviderType
+from core.tools.entities.tool_entities import ToolProviderType
 
 
 class ToolApiEntity(BaseModel):

api/core/tools/entities/tool_entities.py

@@ -476,36 +476,3 @@ class ToolSelector(BaseModel):
 
     def to_plugin_parameter(self) -> dict[str, Any]:
         return self.model_dump()
-
-
-class CredentialType(enum.StrEnum):
-    API_KEY = "api-key"
-    OAUTH2 = "oauth2"
-
-    def get_name(self):
-        if self == CredentialType.API_KEY:
-            return "API KEY"
-        elif self == CredentialType.OAUTH2:
-            return "AUTH"
-        else:
-            return self.value.replace("-", " ").upper()
-
-    def is_editable(self):
-        return self == CredentialType.API_KEY
-
-    def is_validate_allowed(self):
-        return self == CredentialType.API_KEY
-
-    @classmethod
-    def values(cls):
-        return [item.value for item in cls]
-
-    @classmethod
-    def of(cls, credential_type: str) -> "CredentialType":
-        type_name = credential_type.lower()
-        if type_name == "api-key":
-            return cls.API_KEY
-        elif type_name == "oauth2":
-            return cls.OAUTH2
-        else:
-            raise ValueError(f"Invalid credential type: {credential_type}")

api/core/tools/tool_engine.py

@@ -152,6 +152,7 @@ class ToolEngine:
         user_id: str,
         workflow_tool_callback: DifyWorkflowCallbackHandler,
         workflow_call_depth: int,
+        thread_pool_id: Optional[str] = None,
         conversation_id: Optional[str] = None,
         app_id: Optional[str] = None,
         message_id: Optional[str] = None,
@@ -165,6 +166,7 @@ class ToolEngine:
 
             if isinstance(tool, WorkflowTool):
                 tool.workflow_call_depth = workflow_call_depth + 1
+                tool.thread_pool_id = thread_pool_id
 
             if tool.runtime and tool.runtime.runtime_parameters:
                 tool_parameters = {**tool.runtime.runtime_parameters, **tool_parameters}

api/core/tools/tool_manager.py

@@ -13,16 +13,32 @@ from sqlalchemy.orm import Session
 from yarl import URL
 
 import contexts
+from core.helper.provider_cache import ToolProviderCredentialsCache
+from core.plugin.entities.plugin import ToolProviderID
+from core.plugin.impl.oauth import OAuthHandler
+from core.plugin.impl.tool import PluginToolManager
+from core.tools.__base.tool_provider import ToolProviderController
+from core.tools.__base.tool_runtime import ToolRuntime
+from core.tools.mcp_tool.provider import MCPToolProviderController
+from core.tools.mcp_tool.tool import MCPTool
+from core.tools.plugin_tool.provider import PluginToolProviderController
+from core.tools.plugin_tool.tool import PluginTool
+from core.tools.utils.uuid_utils import is_valid_uuid
+from core.tools.workflow_as_tool.provider import WorkflowToolProviderController
+from core.workflow.entities.variable_pool import VariablePool
+from services.tools.mcp_tools_manage_service import MCPToolManageService
+
+if TYPE_CHECKING:
+    from core.workflow.nodes.tool.entities import ToolEntity
+
 from configs import dify_config
 from core.agent.entities import AgentToolEntity
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.helper.module_import_helper import load_single_subclass_from_source
 from core.helper.position_helper import is_filtered
-from core.helper.provider_cache import ToolProviderCredentialsCache
 from core.model_runtime.utils.encoders import jsonable_encoder
+from core.plugin.entities.plugin_daemon import CredentialType
 from core.tools.__base.tool import Tool
-from core.tools.__base.tool_provider import ToolProviderController
-from core.tools.__base.tool_runtime import ToolRuntime
 from core.tools.builtin_tool.provider import BuiltinToolProviderController
 from core.tools.builtin_tool.providers._positions import BuiltinToolProviderSort
 from core.tools.builtin_tool.tool import BuiltinTool
@@ -32,34 +48,21 @@ from core.tools.entities.api_entities import ToolProviderApiEntity, ToolProvider
 from core.tools.entities.common_entities import I18nObject
 from core.tools.entities.tool_entities import (
     ApiProviderAuthType,
-    CredentialType,
     ToolInvokeFrom,
     ToolParameter,
     ToolProviderType,
 )
 from core.tools.errors import ToolProviderNotFoundError
-from core.tools.mcp_tool.provider import MCPToolProviderController
-from core.tools.mcp_tool.tool import MCPTool
-from core.tools.plugin_tool.provider import PluginToolProviderController
-from core.tools.plugin_tool.tool import PluginTool
 from core.tools.tool_label_manager import ToolLabelManager
 from core.tools.utils.configuration import (
     ToolParameterConfigurationManager,
 )
 from core.tools.utils.encryption import create_provider_encrypter, create_tool_provider_encrypter
-from core.tools.utils.uuid_utils import is_valid_uuid
-from core.tools.workflow_as_tool.provider import WorkflowToolProviderController
 from core.tools.workflow_as_tool.tool import WorkflowTool
 from extensions.ext_database import db
-from models.provider_ids import ToolProviderID
 from models.tools import ApiToolProvider, BuiltinToolProvider, MCPToolProvider, WorkflowToolProvider
-from services.tools.mcp_tools_manage_service import MCPToolManageService
 from services.tools.tools_transform_service import ToolTransformService
 
-if TYPE_CHECKING:
-    from core.workflow.entities import VariablePool
-    from core.workflow.nodes.tool.entities import ToolEntity
-
 logger = logging.getLogger(__name__)
 
 
@@ -114,8 +117,6 @@ class ToolManager:
         get the plugin provider
         """
         # check if context is set
-        from core.plugin.impl.tool import PluginToolManager
-
         try:
             contexts.plugin_tool_providers.get()
         except LookupError:
@@ -171,7 +172,6 @@ class ToolManager:
 
         :return: the tool
         """
-
         if provider_type == ToolProviderType.BUILT_IN:
             # check if the builtin tool need credentials
             provider_controller = cls.get_builtin_provider(provider_id, tenant_id)
@@ -216,16 +216,16 @@ class ToolManager:
                 # fallback to the default provider
                 if builtin_provider is None:
                     # use the default provider
-                    with Session(db.engine) as session:
-                        builtin_provider = session.scalar(
-                            sa.select(BuiltinToolProvider)
-                            .where(
-                                BuiltinToolProvider.tenant_id == tenant_id,
-                                (BuiltinToolProvider.provider == str(provider_id_entity))
-                                | (BuiltinToolProvider.provider == provider_id_entity.provider_name),
-                            )
-                            .order_by(BuiltinToolProvider.is_default.desc(), BuiltinToolProvider.created_at.asc())
+                    builtin_provider = (
+                        db.session.query(BuiltinToolProvider)
+                        .where(
+                            BuiltinToolProvider.tenant_id == tenant_id,
+                            (BuiltinToolProvider.provider == str(provider_id_entity))
+                            | (BuiltinToolProvider.provider == provider_id_entity.provider_name),
                         )
+                        .order_by(BuiltinToolProvider.is_default.desc(), BuiltinToolProvider.created_at.asc())
+                        .first()
+                    )
                     if builtin_provider is None:
                         raise ToolProviderNotFoundError(f"no default provider for {provider_id}")
             else:
@@ -256,7 +256,6 @@ class ToolManager:
             # check if the credentials is expired
             if builtin_provider.expires_at != -1 and (builtin_provider.expires_at - 60) < int(time.time()):
                 # TODO: circular import
-                from core.plugin.impl.oauth import OAuthHandler
                 from services.tools.builtin_tools_manage_service import BuiltinToolManageService
 
                 # refresh the credentials
@@ -264,7 +263,6 @@ class ToolManager:
                 provider_name = tool_provider.provider_name
                 redirect_uri = f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{provider_id}/tool/callback"
                 system_credentials = BuiltinToolManageService.get_oauth_client(tenant_id, provider_id)
-
                 oauth_handler = OAuthHandler()
                 # refresh the credentials
                 refreshed_credentials = oauth_handler.refresh_credentials(
@@ -333,16 +331,13 @@ class ToolManager:
             if controller_tools is None or len(controller_tools) == 0:
                 raise ToolProviderNotFoundError(f"workflow provider {provider_id} not found")
 
-            return cast(
-                WorkflowTool,
-                controller.get_tools(tenant_id=workflow_provider.tenant_id)[0].fork_tool_runtime(
-                    runtime=ToolRuntime(
-                        tenant_id=tenant_id,
-                        credentials={},
-                        invoke_from=invoke_from,
-                        tool_invoke_from=tool_invoke_from,
-                    )
-                ),
+            return controller.get_tools(tenant_id=workflow_provider.tenant_id)[0].fork_tool_runtime(
+                runtime=ToolRuntime(
+                    tenant_id=tenant_id,
+                    credentials={},
+                    invoke_from=invoke_from,
+                    tool_invoke_from=tool_invoke_from,
+                )
             )
         elif provider_type == ToolProviderType.APP:
             raise NotImplementedError("app provider not implemented")
@@ -360,7 +355,7 @@ class ToolManager:
         app_id: str,
         agent_tool: AgentToolEntity,
         invoke_from: InvokeFrom = InvokeFrom.DEBUGGER,
-        variable_pool: Optional["VariablePool"] = None,
+        variable_pool: Optional[VariablePool] = None,
     ) -> Tool:
         """
         get the agent tool runtime
@@ -402,7 +397,7 @@ class ToolManager:
         node_id: str,
         workflow_tool: "ToolEntity",
         invoke_from: InvokeFrom = InvokeFrom.DEBUGGER,
-        variable_pool: Optional["VariablePool"] = None,
+        variable_pool: Optional[VariablePool] = None,
     ) -> Tool:
         """
         get the workflow tool runtime
@@ -518,8 +513,6 @@ class ToolManager:
         """
         list all the plugin providers
         """
-        from core.plugin.impl.tool import PluginToolManager
-
         manager = PluginToolManager()
         provider_entities = manager.fetch_tool_providers(tenant_id)
         return [
@@ -652,8 +645,8 @@ class ToolManager:
                 for provider in builtin_providers:
                     # handle include, exclude
                     if is_filtered(
-                        include_set=cast(set[str], dify_config.POSITION_TOOL_INCLUDES_SET),
-                        exclude_set=cast(set[str], dify_config.POSITION_TOOL_EXCLUDES_SET),
+                        include_set=dify_config.POSITION_TOOL_INCLUDES_SET,
+                        exclude_set=dify_config.POSITION_TOOL_EXCLUDES_SET,
                         data=provider,
                         name_func=lambda x: x.identity.name,
                     ):
@@ -981,7 +974,7 @@ class ToolManager:
     def _convert_tool_parameters_type(
         cls,
         parameters: list[ToolParameter],
-        variable_pool: Optional["VariablePool"],
+        variable_pool: Optional[VariablePool],
         tool_configurations: dict[str, Any],
         typ: Literal["agent", "workflow", "tool"] = "workflow",
     ) -> dict[str, Any]:

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

@@ -181,7 +181,7 @@ class DatasetMultiRetrieverTool(DatasetRetrieverBaseTool):
                     retrieval_method="keyword_search",
                     dataset_id=dataset.id,
                     query=query,
-                    top_k=retrieval_model.get("top_k") or 2,
+                    top_k=retrieval_model.get("top_k") or 4,
                 )
                 if documents:
                     all_documents.extend(documents)
@@ -192,7 +192,7 @@ class DatasetMultiRetrieverTool(DatasetRetrieverBaseTool):
                         retrieval_method=retrieval_model["search_method"],
                         dataset_id=dataset.id,
                         query=query,
-                        top_k=retrieval_model.get("top_k") or 2,
+                        top_k=retrieval_model.get("top_k") or 4,
                         score_threshold=retrieval_model.get("score_threshold", 0.0)
                         if retrieval_model["score_threshold_enabled"]
                         else 0.0,