fix: iteration and loop node single step run

Signed-off-by: -LAN- <laipz8200@outlook.com>

.claude/settings.json.example

@@ -0,0 +1,19 @@
+{
+    "permissions": {
+      "allow": [],
+      "deny": []
+    },
+    "env": {
+      "__comment": "Environment variables for MCP servers. Override in .claude/settings.local.json with actual values.",
+      "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
+    },
+    "enabledMcpjsonServers": [
+      "context7",
+      "sequential-thinking",
+      "github",
+      "fetch",
+      "playwright",
+      "ide"
+    ],
+    "enableAllProjectMcpServers": true
+  }

.github/workflows/api-tests.yml

@@ -1,13 +1,7 @@
 name: Run Pytest
 
 on:
-  pull_request:
-    branches:
-      - main
-    paths:
-      - api/**
-      - docker/**
-      - .github/workflows/api-tests.yml
+  workflow_call:
 
 concurrency:
   group: api-tests-${{ github.head_ref || github.run_id }}

.github/workflows/autofix.yml

@@ -1,10 +1,7 @@
 name: autofix.ci
 on:
-  workflow_call:
   pull_request:
-    branches: [ "main" ]
-  push:
-    branches: [ "main" ]
+    branches: ["main"]
 permissions:
   contents: read
 
@@ -18,7 +15,7 @@ jobs:
       # Use uv to ensure we have the same ruff version in CI and locally.
       - uses: astral-sh/setup-uv@v6
         with:
-          python-version: "3.12" 
+          python-version: "3.12"
       - run: |
           cd api
           uv sync --dev
@@ -29,6 +26,7 @@ jobs:
       - name: ast-grep
         run: |
           uvx --from ast-grep-cli sg --pattern 'db.session.query($WHATEVER).filter($HERE)' --rewrite 'db.session.query($WHATEVER).where($HERE)' -l py --update-all
+          uvx --from ast-grep-cli sg --pattern 'session.query($WHATEVER).filter($HERE)' --rewrite 'session.query($WHATEVER).where($HERE)' -l py --update-all
       - name: mdformat
         run: |
           uvx mdformat .

.github/workflows/db-migration-test.yml

@@ -1,13 +1,7 @@
 name: DB Migration Test
 
 on:
-  pull_request:
-    branches:
-      - main
-      - plugins/beta
-    paths:
-      - api/migrations/**
-      - .github/workflows/db-migration-test.yml
+  workflow_call:
 
 concurrency:
   group: db-migration-test-${{ github.ref }}
@@ -33,6 +27,12 @@ jobs:
 
       - name: Install dependencies
         run: uv sync --project api
+      - name: Ensure Offline migration are supported
+        run: |
+          # upgrade
+          uv run --directory api flask db upgrade 'base:head' --sql
+          # downgrade
+          uv run --directory api flask db downgrade 'head:base' --sql
 
       - name: Prepare middleware env
         run: |

.github/workflows/main-ci.yml

@@ -0,0 +1,78 @@
+name: Main CI Pipeline
+
+on:
+  pull_request:
+    branches: ["main"]
+  push:
+    branches: ["main"]
+
+permissions:
+  contents: write
+  pull-requests: write
+  checks: write
+  statuses: write
+
+concurrency:
+  group: main-ci-${{ github.head_ref || github.run_id }}
+  cancel-in-progress: true
+
+jobs:
+  # Check which paths were changed to determine which tests to run
+  check-changes:
+    name: Check Changed Files
+    runs-on: ubuntu-latest
+    outputs:
+      api-changed: ${{ steps.changes.outputs.api }}
+      web-changed: ${{ steps.changes.outputs.web }}
+      vdb-changed: ${{ steps.changes.outputs.vdb }}
+      migration-changed: ${{ steps.changes.outputs.migration }}
+    steps:
+      - uses: actions/checkout@v4
+      - uses: dorny/paths-filter@v3
+        id: changes
+        with:
+          filters: |
+            api:
+              - 'api/**'
+              - 'docker/**'
+              - '.github/workflows/api-tests.yml'
+            web:
+              - 'web/**'
+            vdb:
+              - 'api/core/rag/datasource/**'
+              - 'docker/**'
+              - '.github/workflows/vdb-tests.yml'
+              - 'api/uv.lock'
+              - 'api/pyproject.toml'
+            migration:
+              - 'api/migrations/**'
+              - '.github/workflows/db-migration-test.yml'
+
+  # Run tests in parallel
+  api-tests:
+    name: API Tests
+    needs: check-changes
+    if: needs.check-changes.outputs.api-changed == 'true'
+    uses: ./.github/workflows/api-tests.yml
+
+  web-tests:
+    name: Web Tests
+    needs: check-changes
+    if: needs.check-changes.outputs.web-changed == 'true'
+    uses: ./.github/workflows/web-tests.yml
+
+  style-check:
+    name: Style Check
+    uses: ./.github/workflows/style.yml
+
+  vdb-tests:
+    name: VDB Tests
+    needs: check-changes
+    if: needs.check-changes.outputs.vdb-changed == 'true'
+    uses: ./.github/workflows/vdb-tests.yml
+
+  db-migration-test:
+    name: DB Migration Test
+    needs: check-changes
+    if: needs.check-changes.outputs.migration-changed == 'true'
+    uses: ./.github/workflows/db-migration-test.yml

.github/workflows/style.yml

@@ -1,9 +1,7 @@
 name: Style check
 
 on:
-  pull_request:
-    branches:
-      - main
+  workflow_call:
 
 concurrency:
   group: style-${{ github.head_ref || github.run_id }}
@@ -46,21 +44,10 @@ jobs:
         if: steps.changed-files.outputs.any_changed == 'true'
         run: uv sync --project api --dev
 
-      - name: Ruff check
-        if: steps.changed-files.outputs.any_changed == 'true'
-        run: |
-          uv run --directory api ruff --version
-          uv run --directory api ruff check ./
-          uv run --directory api ruff format --check ./
-
       - name: Dotenv check
         if: steps.changed-files.outputs.any_changed == 'true'
         run: uv run --project api dotenv-linter ./api/.env.example ./web/.env.example
 
-      - name: Lint hints
-        if: failure()
-        run: echo "Please run 'dev/reformat' to fix the fixable linting errors."
-
   web-style:
     name: Web Style
     runs-on: ubuntu-latest

.github/workflows/vdb-tests.yml

@@ -1,15 +1,7 @@
 name: Run VDB Tests
 
 on:
-  pull_request:
-    branches:
-      - main
-    paths:
-      - api/core/rag/datasource/**
-      - docker/**
-      - .github/workflows/vdb-tests.yml
-      - api/uv.lock
-      - api/pyproject.toml
+  workflow_call:
 
 concurrency:
   group: vdb-tests-${{ github.head_ref || github.run_id }}

.github/workflows/web-tests.yml

@@ -1,11 +1,7 @@
 name: Web Tests
 
 on:
-  pull_request:
-    branches:
-      - main
-    paths:
-      - web/**
+  workflow_call:
 
 concurrency:
   group: web-tests-${{ github.head_ref || github.run_id }}

.mcp.json

@@ -0,0 +1,34 @@
+{
+    "mcpServers": {
+      "context7": {
+        "type": "http",
+        "url": "https://mcp.context7.com/mcp"
+      },
+      "sequential-thinking": {
+        "type": "stdio",
+        "command": "npx",
+        "args": ["-y", "@modelcontextprotocol/server-sequential-thinking"],
+        "env": {}
+      },
+      "github": {
+        "type": "stdio",
+        "command": "npx",
+        "args": ["-y", "@modelcontextprotocol/server-github"],
+        "env": {
+          "GITHUB_PERSONAL_ACCESS_TOKEN": "${GITHUB_PERSONAL_ACCESS_TOKEN}"
+        }
+      },
+      "fetch": {
+        "type": "stdio",
+        "command": "uvx",
+        "args": ["mcp-server-fetch"],
+        "env": {}
+      },
+      "playwright": {
+        "type": "stdio",
+        "command": "npx",
+        "args": ["-y", "@playwright/mcp@latest"],
+        "env": {}
+      }
+    }
+  }

CLAUDE.md

@@ -86,3 +86,4 @@ pnpm test                         # Run Jest tests
 ## Project-Specific Conventions
 
 - All async tasks use Celery with Redis as broker
+- **Internationalization**: Frontend supports multiple languages with English (`web/i18n/en-US/`) as the source. All user-facing text must use i18n keys, no hardcoded strings. Edit corresponding module files in `en-US/` directory for translations.

api/.env.example

@@ -460,6 +460,16 @@ 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)

api/.importlinter

@@ -0,0 +1,112 @@
+[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/commands.py

@@ -13,7 +13,6 @@ 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
@@ -31,6 +30,7 @@ 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

api/configs/feature/__init__.py

@@ -529,6 +529,28 @@ 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):
     """

api/controllers/console/__init__.py

@@ -70,7 +70,7 @@ from .app import (
 )
 
 # Import auth controllers
-from .auth import activate, data_source_bearer_auth, data_source_oauth, forgot_password, login, oauth
+from .auth import activate, data_source_bearer_auth, data_source_oauth, forgot_password, login, oauth, oauth_server
 
 # Import billing controllers
 from .billing import billing, compliance

api/controllers/console/app/generator.py

@@ -16,7 +16,10 @@ from core.helper.code_executor.javascript.javascript_code_provider import Javasc
 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):
@@ -135,9 +138,6 @@ class InstructionGenerateApi(Resource):
         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/message.py

@@ -95,18 +95,22 @@ class ChatMessageListApi(Resource):
                 .all()
             )
 
+        # Initialize has_more based on whether we have a full page
         if len(history_messages) == args["limit"]:
             current_page_first_message = history_messages[-1]
-
-        has_more = db.session.scalar(
-            select(
-                exists().where(
-                    Message.conversation_id == conversation.id,
-                    Message.created_at < current_page_first_message.created_at,
-                    Message.id != current_page_first_message.id,
+            # Check if there are more messages before the current page
+            has_more = db.session.scalar(
+                select(
+                    exists().where(
+                        Message.conversation_id == conversation.id,
+                        Message.created_at < current_page_first_message.created_at,
+                        Message.id != current_page_first_message.id,
+                    )
                 )
             )
-        )
+        else:
+            # If we don't have a full page, there are no more messages
+            has_more = False
 
         history_messages = list(reversed(history_messages))
 
@@ -126,7 +130,7 @@ class MessageFeedbackApi(Resource):
 
         message_id = str(args["message_id"])
 
-        message = db.session.query(Message).filter(Message.id == message_id, Message.app_id == app_model.id).first()
+        message = db.session.query(Message).where(Message.id == message_id, Message.app_id == app_model.id).first()
 
         if not message:
             raise NotFound("Message Not Exists.")

api/controllers/console/app/workflow.py

@@ -24,6 +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 extensions.ext_database import db
 from factories import file_factory, variable_factory
 from fields.workflow_fields import workflow_fields, workflow_pagination_fields
@@ -413,7 +414,12 @@ class WorkflowTaskStopApi(Resource):
         if not current_user.is_editor:
             raise Forbidden()
 
-        AppQueueManager.set_stop_flag(task_id, InvokeFrom.DEBUGGER, current_user.id)
+        # 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)
 
         return {"result": "success"}
 

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.entities.workflow_execution import WorkflowExecutionStatus
+from core.workflow.enums 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

api/controllers/console/app/workflow_draft_variable.py

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

api/controllers/console/auth/oauth_server.py

@@ -0,0 +1,189 @@
+from functools import wraps
+from typing import cast
+
+import flask_login
+from flask import request
+from flask_restx import Resource, reqparse
+from werkzeug.exceptions import BadRequest, NotFound
+
+from controllers.console.wraps import account_initialization_required, setup_required
+from core.model_runtime.utils.encoders import jsonable_encoder
+from libs.login import login_required
+from models.account import Account
+from models.model import OAuthProviderApp
+from services.oauth_server import OAUTH_ACCESS_TOKEN_EXPIRES_IN, OAuthGrantType, OAuthServerService
+
+from .. import api
+
+
+def oauth_server_client_id_required(view):
+    @wraps(view)
+    def decorated(*args, **kwargs):
+        parser = reqparse.RequestParser()
+        parser.add_argument("client_id", type=str, required=True, location="json")
+        parsed_args = parser.parse_args()
+        client_id = parsed_args.get("client_id")
+        if not client_id:
+            raise BadRequest("client_id is required")
+
+        oauth_provider_app = OAuthServerService.get_oauth_provider_app(client_id)
+        if not oauth_provider_app:
+            raise NotFound("client_id is invalid")
+
+        kwargs["oauth_provider_app"] = oauth_provider_app
+
+        return view(*args, **kwargs)
+
+    return decorated
+
+
+def oauth_server_access_token_required(view):
+    @wraps(view)
+    def decorated(*args, **kwargs):
+        oauth_provider_app = kwargs.get("oauth_provider_app")
+        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(" ")
+        if len(parts) != 2:
+            raise BadRequest("Invalid Authorization header format")
+
+        token_type = parts[0]
+        if token_type != "Bearer":
+            raise BadRequest("token_type is invalid")
+
+        access_token = parts[1]
+        if not access_token:
+            raise BadRequest("access_token is required")
+
+        account = OAuthServerService.validate_oauth_access_token(oauth_provider_app.client_id, access_token)
+        if not account:
+            raise BadRequest("access_token or client_id is invalid")
+
+        kwargs["account"] = account
+
+        return view(*args, **kwargs)
+
+    return decorated
+
+
+class OAuthServerAppApi(Resource):
+    @setup_required
+    @oauth_server_client_id_required
+    def post(self, oauth_provider_app: OAuthProviderApp):
+        parser = reqparse.RequestParser()
+        parser.add_argument("redirect_uri", type=str, required=True, location="json")
+        parsed_args = parser.parse_args()
+        redirect_uri = parsed_args.get("redirect_uri")
+
+        # check if redirect_uri is valid
+        if redirect_uri not in oauth_provider_app.redirect_uris:
+            raise BadRequest("redirect_uri is invalid")
+
+        return jsonable_encoder(
+            {
+                "app_icon": oauth_provider_app.app_icon,
+                "app_label": oauth_provider_app.app_label,
+                "scope": oauth_provider_app.scope,
+            }
+        )
+
+
+class OAuthServerUserAuthorizeApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @oauth_server_client_id_required
+    def post(self, oauth_provider_app: OAuthProviderApp):
+        account = cast(Account, flask_login.current_user)
+        user_account_id = account.id
+
+        code = OAuthServerService.sign_oauth_authorization_code(oauth_provider_app.client_id, user_account_id)
+        return jsonable_encoder(
+            {
+                "code": code,
+            }
+        )
+
+
+class OAuthServerUserTokenApi(Resource):
+    @setup_required
+    @oauth_server_client_id_required
+    def post(self, oauth_provider_app: OAuthProviderApp):
+        parser = reqparse.RequestParser()
+        parser.add_argument("grant_type", type=str, required=True, location="json")
+        parser.add_argument("code", type=str, required=False, location="json")
+        parser.add_argument("client_secret", type=str, required=False, location="json")
+        parser.add_argument("redirect_uri", type=str, required=False, location="json")
+        parser.add_argument("refresh_token", type=str, required=False, location="json")
+        parsed_args = parser.parse_args()
+
+        grant_type = OAuthGrantType(parsed_args["grant_type"])
+
+        if grant_type == OAuthGrantType.AUTHORIZATION_CODE:
+            if not parsed_args["code"]:
+                raise BadRequest("code is required")
+
+            if parsed_args["client_secret"] != oauth_provider_app.client_secret:
+                raise BadRequest("client_secret is invalid")
+
+            if parsed_args["redirect_uri"] not in oauth_provider_app.redirect_uris:
+                raise BadRequest("redirect_uri is invalid")
+
+            access_token, refresh_token = OAuthServerService.sign_oauth_access_token(
+                grant_type, code=parsed_args["code"], client_id=oauth_provider_app.client_id
+            )
+            return jsonable_encoder(
+                {
+                    "access_token": access_token,
+                    "token_type": "Bearer",
+                    "expires_in": OAUTH_ACCESS_TOKEN_EXPIRES_IN,
+                    "refresh_token": refresh_token,
+                }
+            )
+        elif grant_type == OAuthGrantType.REFRESH_TOKEN:
+            if not parsed_args["refresh_token"]:
+                raise BadRequest("refresh_token is required")
+
+            access_token, refresh_token = OAuthServerService.sign_oauth_access_token(
+                grant_type, refresh_token=parsed_args["refresh_token"], client_id=oauth_provider_app.client_id
+            )
+            return jsonable_encoder(
+                {
+                    "access_token": access_token,
+                    "token_type": "Bearer",
+                    "expires_in": OAUTH_ACCESS_TOKEN_EXPIRES_IN,
+                    "refresh_token": refresh_token,
+                }
+            )
+        else:
+            raise BadRequest("invalid grant_type")
+
+
+class OAuthServerUserAccountApi(Resource):
+    @setup_required
+    @oauth_server_client_id_required
+    @oauth_server_access_token_required
+    def post(self, oauth_provider_app: OAuthProviderApp, account: Account):
+        return jsonable_encoder(
+            {
+                "name": account.name,
+                "email": account.email,
+                "avatar": account.avatar,
+                "interface_language": account.interface_language,
+                "timezone": account.timezone,
+            }
+        )
+
+
+api.add_resource(OAuthServerAppApi, "/oauth/provider")
+api.add_resource(OAuthServerUserAuthorizeApi, "/oauth/provider/authorize")
+api.add_resource(OAuthServerUserTokenApi, "/oauth/provider/token")
+api.add_resource(OAuthServerUserAccountApi, "/oauth/provider/account")

api/controllers/console/datasets/datasets.py

@@ -19,7 +19,6 @@ 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
@@ -31,6 +30,7 @@ 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,6 +20,7 @@ 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
@@ -78,6 +79,11 @@ class InstalledAppWorkflowTaskStopApi(InstalledAppResource):
             raise NotWorkflowAppError()
         assert current_user is not None
 
-        AppQueueManager.set_stop_flag(task_id, InvokeFrom.EXPLORE, current_user.id)
+        # 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)
 
         return {"result": "success"}

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.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/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 models import db as global_db
+from extensions.ext_database import db as global_db
 
 
 @files_ns.route("/tools/<uuid:file_id>.<string:extension>")

api/controllers/mcp/mcp.py

@@ -1,18 +1,27 @@
 from typing import Optional, Union
 
+from flask import Response
 from flask_restx import Resource, reqparse
 from pydantic import ValidationError
+from sqlalchemy.orm import Session
 
 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
-from core.mcp.server.streamable_http import MCPServerStreamableHTTPRequestHandler
-from core.mcp.types import ClientNotification, ClientRequest
-from core.mcp.utils import create_mcp_error_response
+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
+from models.model import App, AppMCPServer, AppMode, EndUser
+
+
+class MCPRequestError(Exception):
+    """Custom exception for MCP request processing errors"""
+
+    def __init__(self, error_code: int, message: str):
+        self.error_code = error_code
+        self.message = message
+        super().__init__(message)
 
 
 def int_or_str(value):
@@ -63,77 +72,173 @@ class MCPAppApi(Resource):
         Raises:
             ValidationError: Invalid request format or parameters
         """
-        # Parse and validate all arguments
         args = mcp_request_parser.parse_args()
-
         request_id: Optional[Union[int, str]] = args.get("id")
+        mcp_request = self._parse_mcp_request(args)
 
-        server = db.session.query(AppMCPServer).where(AppMCPServer.server_code == server_code).first()
-        if not server:
-            return helper.compact_generate_response(
-                create_mcp_error_response(request_id, types.INVALID_REQUEST, "Server Not Found")
-            )
+        with Session(db.engine, expire_on_commit=False) as session:
+            # Get MCP server and app
+            mcp_server, app = self._get_mcp_server_and_app(server_code, session)
+            self._validate_server_status(mcp_server)
 
-        if server.status != AppMCPServerStatus.ACTIVE:
-            return helper.compact_generate_response(
-                create_mcp_error_response(request_id, types.INVALID_REQUEST, "Server is not active")
-            )
+            # Get user input form
+            user_input_form = self._get_user_input_form(app)
 
-        app = db.session.query(App).where(App.id == server.app_id).first()
+            # Handle notification vs request differently
+            return self._process_mcp_message(mcp_request, request_id, app, mcp_server, user_input_form, session)
+
+    def _get_mcp_server_and_app(self, server_code: str, session: Session) -> tuple[AppMCPServer, App]:
+        """Get and validate MCP server and app in one query session"""
+        mcp_server = session.query(AppMCPServer).where(AppMCPServer.server_code == server_code).first()
+        if not mcp_server:
+            raise MCPRequestError(mcp_types.INVALID_REQUEST, "Server Not Found")
+
+        app = session.query(App).where(App.id == mcp_server.app_id).first()
         if not app:
-            return helper.compact_generate_response(
-                create_mcp_error_response(request_id, types.INVALID_REQUEST, "App Not Found")
-            )
+            raise MCPRequestError(mcp_types.INVALID_REQUEST, "App Not Found")
 
-        if app.mode in {AppMode.ADVANCED_CHAT.value, AppMode.WORKFLOW.value}:
-            workflow = app.workflow
-            if workflow is None:
-                return helper.compact_generate_response(
-                    create_mcp_error_response(request_id, types.INVALID_REQUEST, "App is unavailable")
-                )
+        return mcp_server, app
 
-            user_input_form = workflow.user_input_form(to_old_structure=True)
+    def _validate_server_status(self, mcp_server: AppMCPServer) -> None:
+        """Validate MCP server status"""
+        if mcp_server.status != AppMCPServerStatus.ACTIVE:
+            raise MCPRequestError(mcp_types.INVALID_REQUEST, "Server is not active")
+
+    def _process_mcp_message(
+        self,
+        mcp_request: mcp_types.ClientRequest | mcp_types.ClientNotification,
+        request_id: Optional[Union[int, str]],
+        app: App,
+        mcp_server: AppMCPServer,
+        user_input_form: list[VariableEntity],
+        session: Session,
+    ) -> Response:
+        """Process MCP message (notification or request)"""
+        if isinstance(mcp_request, mcp_types.ClientNotification):
+            return self._handle_notification(mcp_request)
         else:
-            app_model_config = app.app_model_config
-            if app_model_config is None:
-                return helper.compact_generate_response(
-                    create_mcp_error_response(request_id, types.INVALID_REQUEST, "App is unavailable")
-                )
+            return self._handle_request(mcp_request, request_id, app, mcp_server, user_input_form, session)
 
-            features_dict = app_model_config.to_dict()
-            user_input_form = features_dict.get("user_input_form", [])
-        converted_user_input_form: list[VariableEntity] = []
-        try:
-            for item in user_input_form:
-                variable_type = item.get("type", "") or list(item.keys())[0]
-                variable = item[variable_type]
-                converted_user_input_form.append(
-                    VariableEntity(
-                        type=variable_type,
-                        variable=variable.get("variable"),
-                        description=variable.get("description") or "",
-                        label=variable.get("label"),
-                        required=variable.get("required", False),
-                        max_length=variable.get("max_length"),
-                        options=variable.get("options") or [],
-                    )
-                )
-        except ValidationError as e:
-            return helper.compact_generate_response(
-                create_mcp_error_response(request_id, types.INVALID_PARAMS, f"Invalid user_input_form: {str(e)}")
-            )
+    def _handle_notification(self, mcp_request: mcp_types.ClientNotification) -> Response:
+        """Handle MCP notification"""
+        # For notifications, only support init notification
+        if mcp_request.root.method != "notifications/initialized":
+            raise MCPRequestError(mcp_types.INVALID_REQUEST, "Invalid notification method")
+        # Return HTTP 202 Accepted for notifications (no response body)
+        return Response("", status=202, content_type="application/json")
 
+    def _handle_request(
+        self,
+        mcp_request: mcp_types.ClientRequest,
+        request_id: Optional[Union[int, str]],
+        app: App,
+        mcp_server: AppMCPServer,
+        user_input_form: list[VariableEntity],
+        session: Session,
+    ) -> Response:
+        """Handle MCP request"""
+        if request_id is None:
+            raise MCPRequestError(mcp_types.INVALID_REQUEST, "Request ID is required")
+
+        result = self._handle_mcp_request(app, mcp_server, mcp_request, user_input_form, session, request_id)
+        if result is None:
+            # This shouldn't happen for requests, but handle gracefully
+            raise MCPRequestError(mcp_types.INTERNAL_ERROR, "No response generated for request")
+
+        return helper.compact_generate_response(result.model_dump(by_alias=True, mode="json", exclude_none=True))
+
+    def _get_user_input_form(self, app: App) -> list[VariableEntity]:
+        """Get and convert user input form"""
+        # Get raw user input form based on app mode
+        if app.mode in {AppMode.ADVANCED_CHAT.value, AppMode.WORKFLOW.value}:
+            if not app.workflow:
+                raise MCPRequestError(mcp_types.INVALID_REQUEST, "App is unavailable")
+            raw_user_input_form = app.workflow.user_input_form(to_old_structure=True)
+        else:
+            if not app.app_model_config:
+                raise MCPRequestError(mcp_types.INVALID_REQUEST, "App is unavailable")
+            features_dict = app.app_model_config.to_dict()
+            raw_user_input_form = features_dict.get("user_input_form", [])
+
+        # Convert to VariableEntity objects
         try:
-            request: ClientRequest | ClientNotification = ClientRequest.model_validate(args)
+            return self._convert_user_input_form(raw_user_input_form)
         except ValidationError as e:
+            raise MCPRequestError(mcp_types.INVALID_PARAMS, f"Invalid user_input_form: {str(e)}")
+
+    def _convert_user_input_form(self, raw_form: list[dict]) -> list[VariableEntity]:
+        """Convert raw user input form to VariableEntity objects"""
+        return [self._create_variable_entity(item) for item in raw_form]
+
+    def _create_variable_entity(self, item: dict) -> VariableEntity:
+        """Create a single VariableEntity from raw form item"""
+        variable_type = item.get("type", "") or list(item.keys())[0]
+        variable = item[variable_type]
+
+        return VariableEntity(
+            type=variable_type,
+            variable=variable.get("variable"),
+            description=variable.get("description") or "",
+            label=variable.get("label"),
+            required=variable.get("required", False),
+            max_length=variable.get("max_length"),
+            options=variable.get("options") or [],
+        )
+
+    def _parse_mcp_request(self, args: dict) -> mcp_types.ClientRequest | mcp_types.ClientNotification:
+        """Parse and validate MCP request"""
+        try:
+            return mcp_types.ClientRequest.model_validate(args)
+        except ValidationError:
             try:
-                notification = ClientNotification.model_validate(args)
-                request = notification
+                return mcp_types.ClientNotification.model_validate(args)
             except ValidationError as e:
-                return helper.compact_generate_response(
-                    create_mcp_error_response(request_id, types.INVALID_PARAMS, f"Invalid MCP request: {str(e)}")
-                )
+                raise MCPRequestError(mcp_types.INVALID_PARAMS, f"Invalid MCP request: {str(e)}")
 
-        mcp_server_handler = MCPServerStreamableHTTPRequestHandler(app, request, converted_user_input_form)
-        response = mcp_server_handler.handle()
-        return helper.compact_generate_response(response)
+    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)

api/controllers/service_api/app/workflow.py

@@ -26,7 +26,8 @@ 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.entities.workflow_execution import WorkflowExecutionStatus
+from core.workflow.enums import WorkflowExecutionStatus
+from core.workflow.graph_engine.manager import GraphEngineManager
 from extensions.ext_database import db
 from fields.workflow_app_log_fields import build_workflow_app_log_pagination_model
 from libs import helper
@@ -262,7 +263,12 @@ class WorkflowTaskStopApi(Resource):
         if app_mode != AppMode.WORKFLOW:
             raise NotWorkflowAppError()
 
-        AppQueueManager.set_stop_flag(task_id, InvokeFrom.SERVICE_API, end_user.id)
+        # 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)
 
         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
@@ -318,10 +318,6 @@ class DatasetApi(DatasetApiResource):
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
         data = marshal(dataset, dataset_detail_fields)
-        if data.get("permission") == "partial_members":
-            part_users_list = DatasetPermissionService.get_dataset_partial_member_list(dataset_id_str)
-            data.update({"partial_member_list": part_users_list})
-
         # check embedding setting
         provider_manager = ProviderManager()
         assert isinstance(current_user, Account)

api/controllers/service_api/dataset/metadata.py

@@ -1,6 +1,6 @@
 from typing import Literal
 
-from flask_login import current_user  # type: ignore
+from flask_login import current_user
 from flask_restx import marshal, reqparse
 from werkzeug.exceptions import NotFound
 

api/controllers/service_api/wraps.py

@@ -6,7 +6,7 @@ from functools import wraps
 from typing import Optional
 
 from flask import current_app, request
-from flask_login import user_logged_in  # type: ignore
+from flask_login import user_logged_in
 from flask_restx import Resource
 from pydantic import BaseModel
 from sqlalchemy import select, update

api/controllers/web/login.py

@@ -1,5 +1,5 @@
 from flask_restx import Resource, reqparse
-from jwt import InvalidTokenError  # type: ignore
+from jwt import InvalidTokenError
 
 import services
 from controllers.console.auth.error import (

api/controllers/web/workflow.py

@@ -21,6 +21,7 @@ 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
@@ -110,7 +111,12 @@ class WorkflowTaskStopApi(WebApiResource):
         if app_mode != AppMode.WORKFLOW:
             raise NotWorkflowAppError()
 
-        AppQueueManager.set_stop_flag(task_id, InvokeFrom.WEB_APP, end_user.id)
+        # 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)
 
         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/app_config/features/more_like_this/manager.py

@@ -1,3 +1,16 @@
+from pydantic import BaseModel, ConfigDict, Field, ValidationError
+
+
+class MoreLikeThisConfig(BaseModel):
+    enabled: bool = False
+    model_config = ConfigDict(extra="allow")
+
+
+class AppConfigModel(BaseModel):
+    more_like_this: MoreLikeThisConfig = Field(default_factory=MoreLikeThisConfig)
+    model_config = ConfigDict(extra="allow")
+
+
 class MoreLikeThisConfigManager:
     @classmethod
     def convert(cls, config: dict) -> bool:
@@ -6,31 +19,14 @@ class MoreLikeThisConfigManager:
 
         :param config: model config args
         """
-        more_like_this = False
-        more_like_this_dict = config.get("more_like_this")
-        if more_like_this_dict:
-            if more_like_this_dict.get("enabled"):
-                more_like_this = True
-
-        return more_like_this
+        validated_config, _ = cls.validate_and_set_defaults(config)
+        return AppConfigModel.model_validate(validated_config).more_like_this.enabled
 
     @classmethod
     def validate_and_set_defaults(cls, config: dict) -> tuple[dict, list[str]]:
-        """
-        Validate and set defaults for more like this feature
-
-        :param config: app model config args
-        """
-        if not config.get("more_like_this"):
-            config["more_like_this"] = {"enabled": False}
-
-        if not isinstance(config["more_like_this"], dict):
-            raise ValueError("more_like_this must be of dict type")
-
-        if "enabled" not in config["more_like_this"] or not config["more_like_this"]["enabled"]:
-            config["more_like_this"]["enabled"] = False
-
-        if not isinstance(config["more_like_this"]["enabled"], bool):
-            raise ValueError("enabled in more_like_this must be of boolean type")
-
-        return config, ["more_like_this"]
+        try:
+            return AppConfigModel.model_validate(config).model_dump(), ["more_like_this"]
+        except ValidationError as e:
+            raise ValueError(
+                "more_like_this must be of dict type and enabled in more_like_this must be of boolean type"
+            )

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,16 +23,17 @@ 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.callbacks import WorkflowCallback, WorkflowLoggingCallback
-from core.workflow.entities.variable_pool import VariablePool
+from core.workflow.entities import GraphRuntimeState, VariablePool
+from core.workflow.graph_engine.command_channels.redis_channel import RedisChannel
 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, WorkflowType
+from models.workflow import ConversationVariable
 
 logger = logging.getLogger(__name__)
 
@@ -76,24 +77,32 @@ 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
@@ -144,16 +153,27 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
             )
 
             # init graph
-            graph = self._init_graph(graph_config=self._workflow.graph_dict)
+            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,
+            )
 
         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,
@@ -164,12 +184,11 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
             ),
             invoke_from=self.application_generate_entity.invoke_from,
             call_depth=self.application_generate_entity.call_depth,
-            variable_pool=variable_pool,
+            graph_runtime_state=graph_runtime_state,
+            command_channel=command_channel,
         )
 
-        generator = workflow_entry.run(
-            callbacks=workflow_callbacks,
-        )
+        generator = workflow_entry.run()
 
         for event in generator:
             self._handle_event(workflow_entry, event)

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

@@ -1,4 +1,5 @@
 import logging
+import re
 import time
 from collections.abc import Callable, Generator, Mapping
 from contextlib import contextmanager
@@ -30,14 +31,9 @@ from core.app.entities.queue_entities import (
     QueueMessageReplaceEvent,
     QueueNodeExceptionEvent,
     QueueNodeFailedEvent,
-    QueueNodeInIterationFailedEvent,
-    QueueNodeInLoopFailedEvent,
     QueueNodeRetryEvent,
     QueueNodeStartedEvent,
     QueueNodeSucceededEvent,
-    QueueParallelBranchRunFailedEvent,
-    QueueParallelBranchRunStartedEvent,
-    QueueParallelBranchRunSucceededEvent,
     QueuePingEvent,
     QueueRetrieverResourcesEvent,
     QueueStopEvent,
@@ -64,8 +60,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.workflow_execution import WorkflowExecutionStatus, WorkflowType
-from core.workflow.graph_engine.entities.graph_runtime_state import GraphRuntimeState
+from core.workflow.entities import GraphRuntimeState
+from core.workflow.enums import WorkflowExecutionStatus, WorkflowType
 from core.workflow.nodes import NodeType
 from core.workflow.repositories.draft_variable_repository import DraftVariableSaverFactory
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
@@ -143,6 +139,7 @@ class AdvancedChatAppGenerateTaskPipeline:
 
         self._workflow_response_converter = WorkflowResponseConverter(
             application_generate_entity=application_generate_entity,
+            user=user,
         )
 
         self._task_state = WorkflowTaskState()
@@ -373,7 +370,7 @@ class AdvancedChatAppGenerateTaskPipeline:
     ) -> Generator[StreamResponse, None, None]:
         """Handle node succeeded events."""
         # Record files if it's an answer node or end node
-        if event.node_type in [NodeType.ANSWER, NodeType.END]:
+        if event.node_type in [NodeType.ANSWER, NodeType.END, NodeType.LLM]:
             self._recorded_files.extend(
                 self._workflow_response_converter.fetch_files_from_node_outputs(event.outputs or {})
             )
@@ -393,9 +390,7 @@ class AdvancedChatAppGenerateTaskPipeline:
 
     def _handle_node_failed_events(
         self,
-        event: Union[
-            QueueNodeFailedEvent, QueueNodeInIterationFailedEvent, QueueNodeInLoopFailedEvent, QueueNodeExceptionEvent
-        ],
+        event: Union[QueueNodeFailedEvent, QueueNodeExceptionEvent],
         **kwargs,
     ) -> Generator[StreamResponse, None, None]:
         """Handle various node failure events."""
@@ -440,32 +435,6 @@ 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]:
@@ -757,8 +726,6 @@ 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,
@@ -806,8 +773,6 @@ class AdvancedChatAppGenerateTaskPipeline:
             event,
             (
                 QueueNodeFailedEvent,
-                QueueNodeInIterationFailedEvent,
-                QueueNodeInLoopFailedEvent,
                 QueueNodeExceptionEvent,
             ),
         ):
@@ -820,17 +785,6 @@ 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
 
@@ -854,11 +808,6 @@ 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
@@ -896,7 +845,14 @@ class AdvancedChatAppGenerateTaskPipeline:
 
     def _save_message(self, *, session: Session, graph_runtime_state: Optional[GraphRuntimeState] = None) -> None:
         message = self._get_message(session=session)
-        message.answer = self._task_state.answer
+
+        # If there are assistant files, remove markdown image links from answer
+        answer_text = self._task_state.answer
+        if self._recorded_files:
+            # Remove markdown image links since we're storing files separately
+            answer_text = re.sub(r"!\[.*?\]\(.*?\)", "", answer_text).strip()
+
+        message.answer = answer_text
         message.updated_at = naive_utc_now()
         message.provider_response_latency = time.perf_counter() - self._base_task_pipeline._start_at
         message.message_metadata = self._task_state.metadata.model_dump_json()

api/core/app/apps/base_app_generator.py

@@ -1,4 +1,3 @@
-import json
 from collections.abc import Generator, Mapping, Sequence
 from typing import TYPE_CHECKING, Any, Optional, Union, final
 
@@ -7,13 +6,14 @@ 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.nodes.enums import NodeType
+from core.workflow.enums import NodeType
 from core.workflow.repositories.draft_variable_repository import (
     DraftVariableSaver,
     DraftVariableSaverFactory,
     NoopDraftVariableSaver,
 )
 from factories import file_factory
+from libs.orjson import orjson_dumps
 from services.workflow_draft_variable_service import DraftVariableSaver as DraftVariableSaverImpl
 
 if TYPE_CHECKING:
@@ -174,7 +174,7 @@ class BaseAppGenerator:
             def gen():
                 for message in generator:
                     if isinstance(message, Mapping | dict):
-                        yield f"data: {json.dumps(message)}\n\n"
+                        yield f"data: {orjson_dumps(message)}\n\n"
                     else:
                         yield f"event: {message}\n\n"
 

api/core/app/apps/base_app_queue_manager.py

@@ -126,6 +126,21 @@ 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,9 +1,8 @@
 import time
 from collections.abc import Mapping, Sequence
 from datetime import UTC, datetime
-from typing import Any, Optional, Union, cast
+from typing import Any, Optional, Union
 
-from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, WorkflowAppGenerateEntity
@@ -17,14 +16,9 @@ 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,
@@ -37,25 +31,21 @@ 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.workflow_execution import WorkflowExecution
-from core.workflow.entities.workflow_node_execution import WorkflowNodeExecution, WorkflowNodeExecutionStatus
+from core.workflow.entities import WorkflowExecution, WorkflowNodeExecution
+from core.workflow.enums import 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 (
     Account,
-    CreatorUserRole,
     EndUser,
-    WorkflowRun,
 )
 
 
@@ -64,8 +54,10 @@ class WorkflowResponseConverter:
         self,
         *,
         application_generate_entity: Union[AdvancedChatAppGenerateEntity, WorkflowAppGenerateEntity],
+        user: Union[Account, EndUser],
     ) -> None:
         self._application_generate_entity = application_generate_entity
+        self._user = user
 
     def workflow_start_to_stream_response(
         self,
@@ -92,27 +84,21 @@ class WorkflowResponseConverter:
         workflow_execution: WorkflowExecution,
     ) -> WorkflowFinishStreamResponse:
         created_by = None
-        workflow_run = session.scalar(select(WorkflowRun).where(WorkflowRun.id == workflow_execution.id_))
-        assert workflow_run is not None
-        if workflow_run.created_by_role == CreatorUserRole.ACCOUNT:
-            stmt = select(Account).where(Account.id == workflow_run.created_by)
-            account = session.scalar(stmt)
-            if account:
-                created_by = {
-                    "id": account.id,
-                    "name": account.name,
-                    "email": account.email,
-                }
-        elif workflow_run.created_by_role == CreatorUserRole.END_USER:
-            stmt = select(EndUser).where(EndUser.id == workflow_run.created_by)
-            end_user = session.scalar(stmt)
-            if end_user:
-                created_by = {
-                    "id": end_user.id,
-                    "user": end_user.session_id,
-                }
+
+        user = self._user
+        if isinstance(user, Account):
+            created_by = {
+                "id": user.id,
+                "name": user.name,
+                "email": user.email,
+            }
+        elif isinstance(user, EndUser):
+            created_by = {
+                "id": user.id,
+                "user": user.session_id,
+            }
         else:
-            raise NotImplementedError(f"unknown created_by_role: {workflow_run.created_by_role}")
+            raise NotImplementedError(f"User type not supported: {type(user)}")
 
         # Handle the case where finished_at is None by using current time as default
         finished_at_timestamp = (
@@ -178,11 +164,10 @@ 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=node_data.provider_type,
-                provider_id=node_data.provider_id,
+                provider_type=ToolProviderType(event.provider_type),
+                provider_id=event.provider_id,
             )
 
         return response
@@ -190,11 +175,7 @@ class WorkflowResponseConverter:
     def workflow_node_finish_to_stream_response(
         self,
         *,
-        event: QueueNodeSucceededEvent
-        | QueueNodeFailedEvent
-        | QueueNodeInIterationFailedEvent
-        | QueueNodeInLoopFailedEvent
-        | QueueNodeExceptionEvent,
+        event: QueueNodeSucceededEvent | QueueNodeFailedEvent | QueueNodeExceptionEvent,
         task_id: str,
         workflow_node_execution: WorkflowNodeExecution,
     ) -> Optional[NodeFinishStreamResponse]:
@@ -228,9 +209,6 @@ 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,
             ),
@@ -282,50 +260,6 @@ 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,
         *,
@@ -340,13 +274,11 @@ class WorkflowResponseConverter:
                 id=event.node_id,
                 node_id=event.node_id,
                 node_type=event.node_type.value,
-                title=event.node_data.title,
+                title=event.node_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,
             ),
         )
 
@@ -364,15 +296,10 @@ class WorkflowResponseConverter:
                 id=event.node_id,
                 node_id=event.node_id,
                 node_type=event.node_type.value,
-                title=event.node_data.title,
+                title=event.node_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,
             ),
         )
 
@@ -391,7 +318,7 @@ class WorkflowResponseConverter:
                 id=event.node_id,
                 node_id=event.node_id,
                 node_type=event.node_type.value,
-                title=event.node_data.title,
+                title=event.node_title,
                 outputs=json_converter.to_json_encodable(event.outputs),
                 created_at=int(time.time()),
                 extras={},
@@ -405,8 +332,6 @@ 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,
             ),
         )
 
@@ -420,7 +345,7 @@ class WorkflowResponseConverter:
                 id=event.node_id,
                 node_id=event.node_id,
                 node_type=event.node_type.value,
-                title=event.node_data.title,
+                title=event.node_title,
                 created_at=int(time.time()),
                 extras={},
                 inputs=event.inputs or {},
@@ -444,7 +369,7 @@ class WorkflowResponseConverter:
                 id=event.node_id,
                 node_id=event.node_id,
                 node_type=event.node_type.value,
-                title=event.node_data.title,
+                title=event.node_title,
                 index=event.index,
                 pre_loop_output=event.output,
                 created_at=int(time.time()),
@@ -452,7 +377,6 @@ 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,
             ),
         )
 
@@ -470,7 +394,7 @@ class WorkflowResponseConverter:
                 id=event.node_id,
                 node_id=event.node_id,
                 node_type=event.node_type.value,
-                title=event.node_data.title,
+                title=event.node_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, Optional, Union, overload
+from typing import Any, Literal, Union, overload
 
 from flask import Flask, current_app
 from pydantic import ValidationError
@@ -53,7 +53,6 @@ class WorkflowAppGenerator(BaseAppGenerator):
         invoke_from: InvokeFrom,
         streaming: Literal[True],
         call_depth: int,
-        workflow_thread_pool_id: Optional[str],
     ) -> Generator[Mapping | str, None, None]: ...
 
     @overload
@@ -67,7 +66,6 @@ class WorkflowAppGenerator(BaseAppGenerator):
         invoke_from: InvokeFrom,
         streaming: Literal[False],
         call_depth: int,
-        workflow_thread_pool_id: Optional[str],
     ) -> Mapping[str, Any]: ...
 
     @overload
@@ -81,7 +79,6 @@ class WorkflowAppGenerator(BaseAppGenerator):
         invoke_from: InvokeFrom,
         streaming: bool,
         call_depth: int,
-        workflow_thread_pool_id: Optional[str],
     ) -> Union[Mapping[str, Any], Generator[Mapping | str, None, None]]: ...
 
     def generate(
@@ -94,7 +91,6 @@ class WorkflowAppGenerator(BaseAppGenerator):
         invoke_from: InvokeFrom,
         streaming: bool = True,
         call_depth: int = 0,
-        workflow_thread_pool_id: Optional[str] = None,
     ) -> Union[Mapping[str, Any], Generator[Mapping | str, None, None]]:
         files: Sequence[Mapping[str, Any]] = args.get("files") or []
 
@@ -186,7 +182,6 @@ 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,
         )
 
     def _generate(
@@ -200,7 +195,6 @@ 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,
     ) -> Union[Mapping[str, Any], Generator[str | Mapping[str, Any], None, None]]:
         """
@@ -214,7 +208,6 @@ 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(
@@ -237,7 +230,6 @@ 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,
             },
         )
@@ -434,17 +426,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
         queue_manager: AppQueueManager,
         context: contextvars.Context,
         variable_loader: VariableLoader,
-        workflow_thread_pool_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(
@@ -474,7 +456,6 @@ 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,

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

@@ -1,7 +1,7 @@
 import logging
-from typing import Optional, cast
+import time
+from typing import 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,13 +9,14 @@ from core.app.entities.app_invoke_entities import (
     InvokeFrom,
     WorkflowAppGenerateEntity,
 )
-from core.workflow.callbacks import WorkflowCallback, WorkflowLoggingCallback
-from core.workflow.entities.variable_pool import VariablePool
+from core.workflow.entities import GraphRuntimeState, VariablePool
+from core.workflow.graph_engine.command_channels.redis_channel import RedisChannel
 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, WorkflowType
+from models.workflow import Workflow
 
 logger = logging.getLogger(__name__)
 
@@ -31,7 +32,6 @@ 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,
     ) -> None:
@@ -41,7 +41,6 @@ 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
 
@@ -52,24 +51,30 @@ 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
@@ -92,15 +97,26 @@ 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 = 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,
+            )
 
         # 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,
@@ -111,11 +127,11 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
             ),
             invoke_from=self.application_generate_entity.invoke_from,
             call_depth=self.application_generate_entity.call_depth,
-            variable_pool=variable_pool,
-            thread_pool_id=self.workflow_thread_pool_id,
+            graph_runtime_state=graph_runtime_state,
+            command_channel=command_channel,
         )
 
-        generator = workflow_entry.run(callbacks=workflow_callbacks)
+        generator = workflow_entry.run()
 
         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 Any, Optional, Union
+from typing import Optional, Union
 
 from sqlalchemy.orm import Session
 
@@ -14,6 +14,7 @@ from core.app.entities.app_invoke_entities import (
     WorkflowAppGenerateEntity,
 )
 from core.app.entities.queue_entities import (
+    AppQueueEvent,
     MessageQueueMessage,
     QueueAgentLogEvent,
     QueueErrorEvent,
@@ -25,14 +26,9 @@ from core.app.entities.queue_entities import (
     QueueLoopStartEvent,
     QueueNodeExceptionEvent,
     QueueNodeFailedEvent,
-    QueueNodeInIterationFailedEvent,
-    QueueNodeInLoopFailedEvent,
     QueueNodeRetryEvent,
     QueueNodeStartedEvent,
     QueueNodeSucceededEvent,
-    QueueParallelBranchRunFailedEvent,
-    QueueParallelBranchRunStartedEvent,
-    QueueParallelBranchRunSucceededEvent,
     QueuePingEvent,
     QueueStopEvent,
     QueueTextChunkEvent,
@@ -57,8 +53,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.workflow_execution import WorkflowExecution, WorkflowExecutionStatus, WorkflowType
-from core.workflow.graph_engine.entities.graph_runtime_state import GraphRuntimeState
+from core.workflow.entities import GraphRuntimeState, WorkflowExecution
+from core.workflow.enums import WorkflowExecutionStatus, WorkflowType
 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
@@ -131,6 +127,7 @@ class WorkflowAppGenerateTaskPipeline:
 
         self._workflow_response_converter = WorkflowResponseConverter(
             application_generate_entity=application_generate_entity,
+            user=user,
         )
 
         self._application_generate_entity = application_generate_entity
@@ -349,9 +346,7 @@ class WorkflowAppGenerateTaskPipeline:
 
     def _handle_node_failed_events(
         self,
-        event: Union[
-            QueueNodeFailedEvent, QueueNodeInIterationFailedEvent, QueueNodeInLoopFailedEvent, QueueNodeExceptionEvent
-        ],
+        event: Union[QueueNodeFailedEvent, QueueNodeExceptionEvent],
         **kwargs,
     ) -> Generator[StreamResponse, None, None]:
         """Handle various node failure events."""
@@ -370,32 +365,6 @@ 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]:
@@ -617,8 +586,6 @@ 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,
@@ -633,7 +600,7 @@ class WorkflowAppGenerateTaskPipeline:
 
     def _dispatch_event(
         self,
-        event: Any,
+        event: AppQueueEvent,
         *,
         graph_runtime_state: Optional[GraphRuntimeState] = None,
         tts_publisher: Optional[AppGeneratorTTSPublisher] = None,
@@ -660,8 +627,6 @@ class WorkflowAppGenerateTaskPipeline:
             event,
             (
                 QueueNodeFailedEvent,
-                QueueNodeInIterationFailedEvent,
-                QueueNodeInLoopFailedEvent,
                 QueueNodeExceptionEvent,
             ),
         ):
@@ -674,17 +639,6 @@ 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

@@ -2,6 +2,7 @@ from collections.abc import Mapping
 from typing import Any, 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,
@@ -13,14 +14,9 @@ from core.app.entities.queue_entities import (
     QueueLoopStartEvent,
     QueueNodeExceptionEvent,
     QueueNodeFailedEvent,
-    QueueNodeInIterationFailedEvent,
-    QueueNodeInLoopFailedEvent,
     QueueNodeRetryEvent,
     QueueNodeStartedEvent,
     QueueNodeSucceededEvent,
-    QueueParallelBranchRunFailedEvent,
-    QueueParallelBranchRunStartedEvent,
-    QueueParallelBranchRunSucceededEvent,
     QueueRetrieverResourcesEvent,
     QueueTextChunkEvent,
     QueueWorkflowFailedEvent,
@@ -28,42 +24,39 @@ from core.app.entities.queue_entities import (
     QueueWorkflowStartedEvent,
     QueueWorkflowSucceededEvent,
 )
-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,
+from core.workflow.entities import GraphInitParams, GraphRuntimeState, VariablePool
+from core.workflow.graph import Graph
+from core.workflow.graph_events import (
     GraphEngineEvent,
     GraphRunFailedEvent,
     GraphRunPartialSucceededEvent,
     GraphRunStartedEvent,
     GraphRunSucceededEvent,
-    IterationRunFailedEvent,
-    IterationRunNextEvent,
-    IterationRunStartedEvent,
-    IterationRunSucceededEvent,
-    LoopRunFailedEvent,
-    LoopRunNextEvent,
-    LoopRunStartedEvent,
-    LoopRunSucceededEvent,
-    NodeInIterationFailedEvent,
-    NodeInLoopFailedEvent,
+    NodeRunAgentLogEvent,
     NodeRunExceptionEvent,
     NodeRunFailedEvent,
+    NodeRunIterationFailedEvent,
+    NodeRunIterationNextEvent,
+    NodeRunIterationStartedEvent,
+    NodeRunIterationSucceededEvent,
+    NodeRunLoopFailedEvent,
+    NodeRunLoopNextEvent,
+    NodeRunLoopStartedEvent,
+    NodeRunLoopSucceededEvent,
     NodeRunRetrieverResourceEvent,
     NodeRunRetryEvent,
     NodeRunStartedEvent,
     NodeRunStreamChunkEvent,
     NodeRunSucceededEvent,
-    ParallelBranchRunFailedEvent,
-    ParallelBranchRunStartedEvent,
-    ParallelBranchRunSucceededEvent,
 )
-from core.workflow.graph_engine.entities.graph import Graph
+from core.workflow.graph_events.graph import GraphRunAbortedEvent
 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
 
 
@@ -79,7 +72,14 @@ class WorkflowBasedAppRunner:
         self._variable_loader = variable_loader
         self._app_id = app_id
 
-    def _init_graph(self, graph_config: Mapping[str, Any]) -> Graph:
+    def _init_graph(
+        self,
+        graph_config: Mapping[str, Any],
+        graph_runtime_state: GraphRuntimeState,
+        workflow_id: str = "",
+        tenant_id: str = "",
+        user_id: str = "",
+    ) -> Graph:
         """
         Init graph
         """
@@ -91,8 +91,28 @@ 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)
+        graph = Graph.init(graph_config=graph_config, node_factory=node_factory)
 
         if not graph:
             raise ValueError("graph not found in workflow")
@@ -104,6 +124,7 @@ class WorkflowBasedAppRunner:
         workflow: Workflow,
         node_id: str,
         user_inputs: dict,
+        graph_runtime_state: GraphRuntimeState,
     ) -> tuple[Graph, VariablePool]:
         """
         Get variable pool of single iteration
@@ -128,7 +149,9 @@ 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
+            if node.get("id") == node_id
+            or node.get("data", {}).get("iteration_id", "") == node_id
+            or node.get("id") == f"{node_id}start"
         ]
 
         graph_config["nodes"] = node_configs
@@ -145,8 +168,25 @@ 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, root_node_id=node_id)
+        graph = Graph.init(graph_config=graph_config, node_factory=node_factory, root_node_id=node_id)
 
         if not graph:
             raise ValueError("graph not found in workflow")
@@ -201,6 +241,7 @@ class WorkflowBasedAppRunner:
         workflow: Workflow,
         node_id: str,
         user_inputs: dict,
+        graph_runtime_state: GraphRuntimeState,
     ) -> tuple[Graph, VariablePool]:
         """
         Get variable pool of single loop
@@ -225,7 +266,9 @@ 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
+            if node.get("id") == node_id
+            or node.get("data", {}).get("loop_id", "") == node_id
+            or node.get("id") == f"{node_id}start"
         ]
 
         graph_config["nodes"] = node_configs
@@ -242,8 +285,25 @@ 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, root_node_id=node_id)
+        graph = Graph.init(graph_config=graph_config, node_factory=node_factory, root_node_id=node_id)
 
         if not graph:
             raise ValueError("graph not found in workflow")
@@ -310,29 +370,21 @@ 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.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
+            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
             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,
@@ -343,6 +395,8 @@ 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):
@@ -350,44 +404,30 @@ class WorkflowBasedAppRunner:
                 QueueNodeStartedEvent(
                     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.route_node_state.start_at,
-                    node_run_index=event.route_node_state.index,
+                    start_at=event.start_at,
                     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.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 = {}
+            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
             self._publish_event(
                 QueueNodeSucceededEvent(
                     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,
+                    start_at=event.start_at,
                     inputs=inputs,
                     process_data=process_data,
                     outputs=outputs,
@@ -396,34 +436,18 @@ 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,
-                    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 {},
+                    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,
                     in_iteration_id=event.in_iteration_id,
                     in_loop_id=event.in_loop_id,
                 )
@@ -434,93 +458,21 @@ class WorkflowBasedAppRunner:
                     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
-                    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 {},
+                    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,
                     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_content,
-                    from_variable_selector=event.from_variable_selector,
+                    text=event.chunk,
+                    from_variable_selector=list(event.selector),
                     in_iteration_id=event.in_iteration_id,
                     in_loop_id=event.in_loop_id,
                 )
@@ -533,10 +485,10 @@ class WorkflowBasedAppRunner:
                     in_loop_id=event.in_loop_id,
                 )
             )
-        elif isinstance(event, AgentLogEvent):
+        elif isinstance(event, NodeRunAgentLogEvent):
             self._publish_event(
                 QueueAgentLogEvent(
-                    id=event.id,
+                    id=event.message_id,
                     label=event.label,
                     node_execution_id=event.node_execution_id,
                     parent_id=event.parent_id,
@@ -547,51 +499,13 @@ class WorkflowBasedAppRunner:
                     node_id=event.node_id,
                 )
             )
-        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):
+        elif isinstance(event, NodeRunIterationStartedEvent):
             self._publish_event(
                 QueueIterationStartEvent(
-                    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,
+                    node_execution_id=event.id,
+                    node_id=event.node_id,
+                    node_type=event.node_type,
+                    node_title=event.node_title,
                     start_at=event.start_at,
                     node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
                     inputs=event.inputs,
@@ -599,55 +513,41 @@ class WorkflowBasedAppRunner:
                     metadata=event.metadata,
                 )
             )
-        elif isinstance(event, IterationRunNextEvent):
+        elif isinstance(event, NodeRunIterationNextEvent):
             self._publish_event(
                 QueueIterationNextEvent(
-                    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,
+                    node_execution_id=event.id,
+                    node_id=event.node_id,
+                    node_type=event.node_type,
+                    node_title=event.node_title,
                     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, (IterationRunSucceededEvent | IterationRunFailedEvent)):
+        elif isinstance(event, (NodeRunIterationSucceededEvent | NodeRunIterationFailedEvent)):
             self._publish_event(
                 QueueIterationCompletedEvent(
-                    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,
+                    node_execution_id=event.id,
+                    node_id=event.node_id,
+                    node_type=event.node_type,
+                    node_title=event.node_title,
                     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, IterationRunFailedEvent) else None,
+                    error=event.error if isinstance(event, NodeRunIterationFailedEvent) else None,
                 )
             )
-        elif isinstance(event, LoopRunStartedEvent):
+        elif isinstance(event, NodeRunLoopStartedEvent):
             self._publish_event(
                 QueueLoopStartEvent(
-                    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,
+                    node_execution_id=event.id,
+                    node_id=event.node_id,
+                    node_type=event.node_type,
+                    node_title=event.node_title,
                     start_at=event.start_at,
                     node_run_index=workflow_entry.graph_engine.graph_runtime_state.node_run_steps,
                     inputs=event.inputs,
@@ -655,42 +555,32 @@ class WorkflowBasedAppRunner:
                     metadata=event.metadata,
                 )
             )
-        elif isinstance(event, LoopRunNextEvent):
+        elif isinstance(event, NodeRunLoopNextEvent):
             self._publish_event(
                 QueueLoopNextEvent(
-                    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,
+                    node_execution_id=event.id,
+                    node_id=event.node_id,
+                    node_type=event.node_type,
+                    node_title=event.node_title,
                     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, (LoopRunSucceededEvent | LoopRunFailedEvent)):
+        elif isinstance(event, (NodeRunLoopSucceededEvent | NodeRunLoopFailedEvent)):
             self._publish_event(
                 QueueLoopCompletedEvent(
-                    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,
+                    node_execution_id=event.id,
+                    node_id=event.node_id,
+                    node_type=event.node_type,
+                    node_title=event.node_title,
                     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, LoopRunFailedEvent) else None,
+                    error=event.error if isinstance(event, NodeRunLoopFailedEvent) else None,
                 )
             )
 

api/core/app/entities/app_invoke_entities.py

@@ -1,5 +1,5 @@
 from collections.abc import Mapping, Sequence
-from enum import Enum
+from enum import StrEnum
 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(Enum):
+class InvokeFrom(StrEnum):
     """
     Invoke From.
     """

api/core/app/entities/queue_entities.py

@@ -7,11 +7,9 @@ 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.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.entities import AgentNodeStrategyInit, GraphRuntimeState
+from core.workflow.enums import WorkflowNodeExecutionMetadataKey
 from core.workflow.nodes import NodeType
-from core.workflow.nodes.base import BaseNodeData
 
 
 class QueueEvent(StrEnum):
@@ -43,9 +41,6 @@ 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"
@@ -80,15 +75,7 @@ class QueueIterationStartEvent(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"""
+    node_title: str
     start_at: datetime
 
     node_run_index: int
@@ -108,20 +95,9 @@ class QueueIterationNextEvent(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"""
-    parallel_mode_run_id: Optional[str] = None
-    """iteratoin run in parallel mode run id"""
+    node_title: str
     node_run_index: int
     output: Optional[Any] = None  # output for the current iteration
-    duration: Optional[float] = None
 
 
 class QueueIterationCompletedEvent(AppQueueEvent):
@@ -134,15 +110,7 @@ class QueueIterationCompletedEvent(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"""
+    node_title: str
     start_at: datetime
 
     node_run_index: int
@@ -163,7 +131,7 @@ class QueueLoopStartEvent(AppQueueEvent):
     node_execution_id: str
     node_id: str
     node_type: NodeType
-    node_data: BaseNodeData
+    node_title: str
     parallel_id: Optional[str] = None
     """parallel id if node is in parallel"""
     parallel_start_node_id: Optional[str] = None
@@ -191,7 +159,7 @@ class QueueLoopNextEvent(AppQueueEvent):
     node_execution_id: str
     node_id: str
     node_type: NodeType
-    node_data: BaseNodeData
+    node_title: str
     parallel_id: Optional[str] = None
     """parallel id if node is in parallel"""
     parallel_start_node_id: Optional[str] = None
@@ -201,10 +169,9 @@ class QueueLoopNextEvent(AppQueueEvent):
     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
-    """iteratoin run in parallel mode run id"""
+    """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):
@@ -217,7 +184,7 @@ class QueueLoopCompletedEvent(AppQueueEvent):
     node_execution_id: str
     node_id: str
     node_type: NodeType
-    node_data: BaseNodeData
+    node_title: str
     parallel_id: Optional[str] = None
     """parallel id if node is in parallel"""
     parallel_start_node_id: Optional[str] = None
@@ -364,27 +331,24 @@ class QueueNodeStartedEvent(AppQueueEvent):
 
     node_execution_id: str
     node_id: str
+    node_title: str
     node_type: NodeType
-    node_data: BaseNodeData
-    node_run_index: int = 1
+    node_run_index: int = 1  # FIXME(-LAN-): may not used
     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
-    """iteratoin 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):
     """
@@ -396,7 +360,6 @@ 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
@@ -417,10 +380,6 @@ 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):
@@ -454,72 +413,6 @@ 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
@@ -530,7 +423,6 @@ 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
@@ -563,15 +455,7 @@ 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
@@ -678,61 +562,3 @@ 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.node_entities import AgentNodeStrategyInit
-from core.workflow.entities.workflow_node_execution import WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
+from core.workflow.entities import AgentNodeStrategyInit
+from core.workflow.enums import WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
 
 
 class AnnotationReplyAccount(BaseModel):
@@ -71,8 +71,6 @@ 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"
@@ -440,54 +438,6 @@ 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
@@ -506,8 +456,6 @@ 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
@@ -530,12 +478,7 @@ 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
@@ -567,8 +510,6 @@ 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
@@ -622,7 +563,6 @@ 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/exc.py

@@ -1,8 +1,8 @@
-class TaskPipilineError(ValueError):
+class TaskPipelineError(ValueError):
     pass
 
 
-class RecordNotFoundError(TaskPipilineError):
+class RecordNotFoundError(TaskPipelineError):
     def __init__(self, record_name: str, record_id: str):
         super().__init__(f"{record_name} with id {record_id} not found")
 

api/core/entities/provider_configuration.py

@@ -28,7 +28,6 @@ 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 (
@@ -41,6 +40,7 @@ from models.provider import (
     ProviderType,
     TenantPreferredModelProvider,
 )
+from models.provider_ids import ModelProviderID
 
 logger = logging.getLogger(__name__)
 
@@ -627,6 +627,7 @@ 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():
@@ -1124,6 +1125,7 @@ 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():
@@ -1207,6 +1209,7 @@ 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/file/file_manager.py

@@ -88,6 +88,7 @@ def to_prompt_message_content(
         "url": _to_url(f) if dify_config.MULTIMODAL_SEND_FORMAT == "url" else "",
         "format": f.extension.removeprefix("."),
         "mime_type": f.mime_type,
+        "filename": f.filename or "",
     }
     if f.type == FileType.IMAGE:
         params["detail"] = image_detail_config or ImagePromptMessageContent.DETAIL.LOW

api/core/helper/encrypter.py

@@ -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/llm_generator/llm_generator.py

@@ -28,8 +28,9 @@ 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.graph_engine.entities.event import AgentLogEvent
-from models import App, Message, WorkflowNodeExecutionModel, db
+from core.workflow.node_events import AgentLogEvent
+from extensions.ext_database import db
+from models import App, Message, WorkflowNodeExecutionModel
 
 logger = logging.getLogger(__name__)
 

api/core/mcp/server/streamable_http.py

@@ -4,224 +4,259 @@ from collections.abc import Mapping
 from typing import Any, cast
 
 from configs import dify_config
-from controllers.web.passport import generate_session_id
 from core.app.app_config.entities import VariableEntity, VariableEntityType
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.app.features.rate_limiting.rate_limit import RateLimitGenerator
-from core.mcp import types
-from core.mcp.types import INTERNAL_ERROR, INVALID_PARAMS, METHOD_NOT_FOUND
-from core.mcp.utils import create_mcp_error_response
-from core.model_runtime.utils.encoders import jsonable_encoder
-from extensions.ext_database import db
+from core.mcp import types as mcp_types
 from models.model import App, AppMCPServer, AppMode, EndUser
 from services.app_generate_service import AppGenerateService
 
 logger = logging.getLogger(__name__)
 
 
-class MCPServerStreamableHTTPRequestHandler:
+def handle_mcp_request(
+    app: App,
+    request: mcp_types.ClientRequest,
+    user_input_form: list[VariableEntity],
+    mcp_server: AppMCPServer,
+    end_user: EndUser | None = None,
+    request_id: int | str = 1,
+) -> mcp_types.JSONRPCResponse | mcp_types.JSONRPCError:
     """
-    Apply to MCP HTTP streamable server with stateless http
+    Handle MCP request and return JSON-RPC response
+
+    Args:
+        app: The Dify app instance
+        request: The JSON-RPC request message
+        user_input_form: List of variable entities for the app
+        mcp_server: The MCP server configuration
+        end_user: Optional end user
+        request_id: The request ID
+
+    Returns:
+        JSON-RPC response or error
     """
 
-    def __init__(
-        self, app: App, request: types.ClientRequest | types.ClientNotification, user_input_form: list[VariableEntity]
-    ):
-        self.app = app
-        self.request = request
-        mcp_server = db.session.query(AppMCPServer).where(AppMCPServer.app_id == self.app.id).first()
-        if not mcp_server:
-            raise ValueError("MCP server not found")
-        self.mcp_server: AppMCPServer = mcp_server
-        self.end_user = self.retrieve_end_user()
-        self.user_input_form = user_input_form
+    request_type = type(request.root)
 
-    @property
-    def request_type(self):
-        return type(self.request.root)
+    def create_success_response(result_data: mcp_types.Result) -> mcp_types.JSONRPCResponse:
+        """Create success response with business result data"""
+        return mcp_types.JSONRPCResponse(
+            jsonrpc="2.0",
+            id=request_id,
+            result=result_data.model_dump(by_alias=True, mode="json", exclude_none=True),
+        )
 
-    @property
-    def parameter_schema(self):
-        parameters, required = self._convert_input_form_to_parameters(self.user_input_form)
-        if self.app.mode in {AppMode.COMPLETION.value, AppMode.WORKFLOW.value}:
-            return {
-                "type": "object",
-                "properties": parameters,
-                "required": required,
-            }
+    def create_error_response(code: int, message: str) -> mcp_types.JSONRPCError:
+        """Create error response with error code and message"""
+        from core.mcp.types import ErrorData
+
+        error_data = ErrorData(code=code, message=message)
+        return mcp_types.JSONRPCError(
+            jsonrpc="2.0",
+            id=request_id,
+            error=error_data,
+        )
+
+    # Request handler mapping using functional approach
+    request_handlers = {
+        mcp_types.InitializeRequest: lambda: handle_initialize(mcp_server.description),
+        mcp_types.ListToolsRequest: lambda: handle_list_tools(
+            app.name, app.mode, user_input_form, mcp_server.description, mcp_server.parameters_dict
+        ),
+        mcp_types.CallToolRequest: lambda: handle_call_tool(app, request, user_input_form, end_user),
+        mcp_types.PingRequest: lambda: handle_ping(),
+    }
+
+    try:
+        # Dispatch request to appropriate handler
+        handler = request_handlers.get(request_type)
+        if handler:
+            return create_success_response(handler())
+        else:
+            return create_error_response(mcp_types.METHOD_NOT_FOUND, f"Method not found: {request_type.__name__}")
+
+    except ValueError as e:
+        logger.exception("Invalid params")
+        return create_error_response(mcp_types.INVALID_PARAMS, str(e))
+    except Exception as e:
+        logger.exception("Internal server error")
+        return create_error_response(mcp_types.INTERNAL_ERROR, "Internal server error: " + str(e))
+
+
+def handle_ping() -> mcp_types.EmptyResult:
+    """Handle ping request"""
+    return mcp_types.EmptyResult()
+
+
+def handle_initialize(description: str) -> mcp_types.InitializeResult:
+    """Handle initialize request"""
+    capabilities = mcp_types.ServerCapabilities(
+        tools=mcp_types.ToolsCapability(listChanged=False),
+    )
+
+    return mcp_types.InitializeResult(
+        protocolVersion=mcp_types.SERVER_LATEST_PROTOCOL_VERSION,
+        capabilities=capabilities,
+        serverInfo=mcp_types.Implementation(name="Dify", version=dify_config.project.version),
+        instructions=description,
+    )
+
+
+def handle_list_tools(
+    app_name: str,
+    app_mode: str,
+    user_input_form: list[VariableEntity],
+    description: str,
+    parameters_dict: dict[str, str],
+) -> mcp_types.ListToolsResult:
+    """Handle list tools request"""
+    parameter_schema = build_parameter_schema(app_mode, user_input_form, parameters_dict)
+
+    return mcp_types.ListToolsResult(
+        tools=[
+            mcp_types.Tool(
+                name=app_name,
+                description=description,
+                inputSchema=parameter_schema,
+            )
+        ],
+    )
+
+
+def handle_call_tool(
+    app: App,
+    request: mcp_types.ClientRequest,
+    user_input_form: list[VariableEntity],
+    end_user: EndUser | None,
+) -> mcp_types.CallToolResult:
+    """Handle call tool request"""
+    request_obj = cast(mcp_types.CallToolRequest, request.root)
+    args = prepare_tool_arguments(app, request_obj.params.arguments or {})
+
+    if not end_user:
+        raise ValueError("End user not found")
+
+    response = AppGenerateService.generate(
+        app,
+        end_user,
+        args,
+        InvokeFrom.SERVICE_API,
+        streaming=app.mode == AppMode.AGENT_CHAT.value,
+    )
+
+    answer = extract_answer_from_response(app, response)
+    return mcp_types.CallToolResult(content=[mcp_types.TextContent(text=answer, type="text")])
+
+
+def build_parameter_schema(
+    app_mode: str,
+    user_input_form: list[VariableEntity],
+    parameters_dict: dict[str, str],
+) -> dict[str, Any]:
+    """Build parameter schema for the tool"""
+    parameters, required = convert_input_form_to_parameters(user_input_form, parameters_dict)
+
+    if app_mode in {AppMode.COMPLETION.value, AppMode.WORKFLOW.value}:
         return {
             "type": "object",
-            "properties": {
-                "query": {"type": "string", "description": "User Input/Question content"},
-                **parameters,
-            },
-            "required": ["query", *required],
+            "properties": parameters,
+            "required": required,
         }
+    return {
+        "type": "object",
+        "properties": {
+            "query": {"type": "string", "description": "User Input/Question content"},
+            **parameters,
+        },
+        "required": ["query", *required],
+    }
 
-    @property
-    def capabilities(self):
-        return types.ServerCapabilities(
-            tools=types.ToolsCapability(listChanged=False),
-        )
 
-    def response(self, response: types.Result | str):
-        if isinstance(response, str):
-            sse_content = f"event: ping\ndata: {response}\n\n".encode()
-            yield sse_content
-            return
-        json_response = types.JSONRPCResponse(
-            jsonrpc="2.0",
-            id=(self.request.root.model_extra or {}).get("id", 1),
-            result=response.model_dump(by_alias=True, mode="json", exclude_none=True),
-        )
-        json_data = json.dumps(jsonable_encoder(json_response))
+def prepare_tool_arguments(app: App, arguments: dict[str, Any]) -> dict[str, Any]:
+    """Prepare arguments based on app mode"""
+    if app.mode == AppMode.WORKFLOW.value:
+        return {"inputs": arguments}
+    elif app.mode == AppMode.COMPLETION.value:
+        return {"query": "", "inputs": arguments}
+    else:
+        # Chat modes - create a copy to avoid modifying original dict
+        args_copy = arguments.copy()
+        query = args_copy.pop("query", "")
+        return {"query": query, "inputs": args_copy}
 
-        sse_content = f"event: message\ndata: {json_data}\n\n".encode()
 
-        yield sse_content
+def extract_answer_from_response(app: App, response: Any) -> str:
+    """Extract answer from app generate response"""
+    answer = ""
 
-    def error_response(self, code: int, message: str, data=None):
-        request_id = (self.request.root.model_extra or {}).get("id", 1) or 1
-        return create_mcp_error_response(request_id, code, message, data)
+    if isinstance(response, RateLimitGenerator):
+        answer = process_streaming_response(response)
+    elif isinstance(response, Mapping):
+        answer = process_mapping_response(app, response)
+    else:
+        logger.warning("Unexpected response type: %s", type(response))
 
-    def handle(self):
-        handle_map = {
-            types.InitializeRequest: self.initialize,
-            types.ListToolsRequest: self.list_tools,
-            types.CallToolRequest: self.invoke_tool,
-            types.InitializedNotification: self.handle_notification,
-            types.PingRequest: self.handle_ping,
-        }
-        try:
-            if self.request_type in handle_map:
-                return self.response(handle_map[self.request_type]())
-            else:
-                return self.error_response(METHOD_NOT_FOUND, f"Method not found: {self.request_type}")
-        except ValueError as e:
-            logger.exception("Invalid params")
-            return self.error_response(INVALID_PARAMS, str(e))
-        except Exception as e:
-            logger.exception("Internal server error")
-            return self.error_response(INTERNAL_ERROR, f"Internal server error: {str(e)}")
+    return answer
 
-    def handle_notification(self):
-        return "ping"
 
-    def handle_ping(self):
-        return types.EmptyResult()
-
-    def initialize(self):
-        request = cast(types.InitializeRequest, self.request.root)
-        client_info = request.params.clientInfo
-        client_name = f"{client_info.name}@{client_info.version}"
-        if not self.end_user:
-            end_user = EndUser(
-                tenant_id=self.app.tenant_id,
-                app_id=self.app.id,
-                type="mcp",
-                name=client_name,
-                session_id=generate_session_id(),
-                external_user_id=self.mcp_server.id,
-            )
-            db.session.add(end_user)
-            db.session.commit()
-        return types.InitializeResult(
-            protocolVersion=types.SERVER_LATEST_PROTOCOL_VERSION,
-            capabilities=self.capabilities,
-            serverInfo=types.Implementation(name="Dify", version=dify_config.project.version),
-            instructions=self.mcp_server.description,
-        )
-
-    def list_tools(self):
-        if not self.end_user:
-            raise ValueError("User not found")
-        return types.ListToolsResult(
-            tools=[
-                types.Tool(
-                    name=self.app.name,
-                    description=self.mcp_server.description,
-                    inputSchema=self.parameter_schema,
-                )
-            ],
-        )
-
-    def invoke_tool(self):
-        if not self.end_user:
-            raise ValueError("User not found")
-        request = cast(types.CallToolRequest, self.request.root)
-        args = request.params.arguments or {}
-        if self.app.mode in {AppMode.WORKFLOW.value}:
-            args = {"inputs": args}
-        elif self.app.mode in {AppMode.COMPLETION.value}:
-            args = {"query": "", "inputs": args}
-        else:
-            args = {"query": args["query"], "inputs": {k: v for k, v in args.items() if k != "query"}}
-        response = AppGenerateService.generate(
-            self.app,
-            self.end_user,
-            args,
-            InvokeFrom.SERVICE_API,
-            streaming=self.app.mode == AppMode.AGENT_CHAT.value,
-        )
-        answer = ""
-        if isinstance(response, RateLimitGenerator):
-            for item in response.generator:
-                data = item
-                if isinstance(data, str) and data.startswith("data: "):
-                    try:
-                        json_str = data[6:].strip()
-                        parsed_data = json.loads(json_str)
-                        if parsed_data.get("event") == "agent_thought":
-                            answer += parsed_data.get("thought", "")
-                    except json.JSONDecodeError:
-                        continue
-        if isinstance(response, Mapping):
-            if self.app.mode in {
-                AppMode.ADVANCED_CHAT.value,
-                AppMode.COMPLETION.value,
-                AppMode.CHAT.value,
-                AppMode.AGENT_CHAT.value,
-            }:
-                answer = response["answer"]
-            elif self.app.mode in {AppMode.WORKFLOW.value}:
-                answer = json.dumps(response["data"]["outputs"], ensure_ascii=False)
-            else:
-                raise ValueError("Invalid app mode")
-            # Not support image yet
-        return types.CallToolResult(content=[types.TextContent(text=answer, type="text")])
-
-    def retrieve_end_user(self):
-        return (
-            db.session.query(EndUser)
-            .where(EndUser.external_user_id == self.mcp_server.id, EndUser.type == "mcp")
-            .first()
-        )
-
-    def _convert_input_form_to_parameters(self, user_input_form: list[VariableEntity]):
-        parameters: dict[str, dict[str, Any]] = {}
-        required = []
-        for item in user_input_form:
-            parameters[item.variable] = {}
-            if item.type in (
-                VariableEntityType.FILE,
-                VariableEntityType.FILE_LIST,
-                VariableEntityType.EXTERNAL_DATA_TOOL,
-            ):
-                continue
-            if item.required:
-                required.append(item.variable)
-            # if the workflow republished, the parameters not changed
-            # we should not raise error here
+def process_streaming_response(response: RateLimitGenerator) -> str:
+    """Process streaming response for agent chat mode"""
+    answer = ""
+    for item in response.generator:
+        if isinstance(item, str) and item.startswith("data: "):
             try:
-                description = self.mcp_server.parameters_dict[item.variable]
-            except KeyError:
-                description = ""
-            parameters[item.variable]["description"] = description
-            if item.type in (VariableEntityType.TEXT_INPUT, VariableEntityType.PARAGRAPH):
-                parameters[item.variable]["type"] = "string"
-            elif item.type == VariableEntityType.SELECT:
-                parameters[item.variable]["type"] = "string"
-                parameters[item.variable]["enum"] = item.options
-            elif item.type == VariableEntityType.NUMBER:
-                parameters[item.variable]["type"] = "float"
-        return parameters, required
+                json_str = item[6:].strip()
+                parsed_data = json.loads(json_str)
+                if parsed_data.get("event") == "agent_thought":
+                    answer += parsed_data.get("thought", "")
+            except json.JSONDecodeError:
+                continue
+    return answer
+
+
+def process_mapping_response(app: App, response: Mapping) -> str:
+    """Process mapping response based on app mode"""
+    if app.mode in {
+        AppMode.ADVANCED_CHAT.value,
+        AppMode.COMPLETION.value,
+        AppMode.CHAT.value,
+        AppMode.AGENT_CHAT.value,
+    }:
+        return response.get("answer", "")
+    elif app.mode == AppMode.WORKFLOW.value:
+        return json.dumps(response["data"]["outputs"], ensure_ascii=False)
+    else:
+        raise ValueError("Invalid app mode: " + str(app.mode))
+
+
+def convert_input_form_to_parameters(
+    user_input_form: list[VariableEntity],
+    parameters_dict: dict[str, str],
+) -> tuple[dict[str, dict[str, Any]], list[str]]:
+    """Convert user input form to parameter schema"""
+    parameters: dict[str, dict[str, Any]] = {}
+    required = []
+
+    for item in user_input_form:
+        if item.type in (
+            VariableEntityType.FILE,
+            VariableEntityType.FILE_LIST,
+            VariableEntityType.EXTERNAL_DATA_TOOL,
+        ):
+            continue
+        parameters[item.variable] = {}
+        if item.required:
+            required.append(item.variable)
+        # if the workflow republished, the parameters not changed
+        # we should not raise error here
+        description = parameters_dict.get(item.variable, "")
+        parameters[item.variable]["description"] = description
+        if item.type in (VariableEntityType.TEXT_INPUT, VariableEntityType.PARAGRAPH):
+            parameters[item.variable]["type"] = "string"
+        elif item.type == VariableEntityType.SELECT:
+            parameters[item.variable]["type"] = "string"
+            parameters[item.variable]["enum"] = item.options
+        elif item.type == VariableEntityType.NUMBER:
+            parameters[item.variable]["type"] = "float"
+    return parameters, required

api/core/mcp/utils.py

@@ -138,5 +138,5 @@ def create_mcp_error_response(
         error=error_data,
     )
     json_data = json.dumps(jsonable_encoder(json_response))
-    sse_content = f"event: message\ndata: {json_data}\n\n".encode()
+    sse_content = json_data.encode()
     yield sse_content

api/core/memory/token_buffer_memory.py

@@ -31,6 +31,65 @@ class TokenBufferMemory:
         self.conversation = conversation
         self.model_instance = model_instance
 
+    def _build_prompt_message_with_files(
+        self, message_files: list[MessageFile], text_content: str, message: Message, app_record, is_user_message: bool
+    ) -> PromptMessage:
+        """
+        Build prompt message with files.
+        :param message_files: list of MessageFile objects
+        :param text_content: text content of the message
+        :param message: Message object
+        :param app_record: app record
+        :param is_user_message: whether this is a user message
+        :return: PromptMessage
+        """
+        if self.conversation.mode in {AppMode.AGENT_CHAT, AppMode.COMPLETION, AppMode.CHAT}:
+            file_extra_config = FileUploadConfigManager.convert(self.conversation.model_config)
+        elif self.conversation.mode in {AppMode.ADVANCED_CHAT, AppMode.WORKFLOW}:
+            workflow_run = db.session.scalar(select(WorkflowRun).where(WorkflowRun.id == message.workflow_run_id))
+            if not workflow_run:
+                raise ValueError(f"Workflow run not found: {message.workflow_run_id}")
+            workflow = db.session.scalar(select(Workflow).where(Workflow.id == workflow_run.workflow_id))
+            if not workflow:
+                raise ValueError(f"Workflow not found: {workflow_run.workflow_id}")
+            file_extra_config = FileUploadConfigManager.convert(workflow.features_dict, is_vision=False)
+        else:
+            raise AssertionError(f"Invalid app mode: {self.conversation.mode}")
+
+        detail = ImagePromptMessageContent.DETAIL.HIGH
+        if file_extra_config and app_record:
+            # Build files directly without filtering by belongs_to
+            file_objs = [
+                file_factory.build_from_message_file(
+                    message_file=message_file, tenant_id=app_record.tenant_id, config=file_extra_config
+                )
+                for message_file in message_files
+            ]
+            if file_extra_config.image_config and file_extra_config.image_config.detail:
+                detail = file_extra_config.image_config.detail
+        else:
+            file_objs = []
+
+        if not file_objs:
+            if is_user_message:
+                return UserPromptMessage(content=text_content)
+            else:
+                return AssistantPromptMessage(content=text_content)
+        else:
+            prompt_message_contents: list[PromptMessageContentUnionTypes] = []
+            for file in file_objs:
+                prompt_message = file_manager.to_prompt_message_content(
+                    file,
+                    image_detail_config=detail,
+                )
+                prompt_message_contents.append(prompt_message)
+            prompt_message_contents.append(TextPromptMessageContent(data=text_content))
+
+            if is_user_message:
+                return UserPromptMessage(content=prompt_message_contents)
+            else:
+                return AssistantPromptMessage(content=prompt_message_contents)
+
     def get_history_prompt_messages(
         self, max_token_limit: int = 2000, message_limit: Optional[int] = None
     ) -> Sequence[PromptMessage]:
@@ -67,58 +126,52 @@ class TokenBufferMemory:
 
         prompt_messages: list[PromptMessage] = []
         for message in messages:
-            files = db.session.query(MessageFile).where(MessageFile.message_id == message.id).all()
-            if files:
-                file_extra_config = None
-                if self.conversation.mode in {AppMode.AGENT_CHAT, AppMode.COMPLETION, AppMode.CHAT}:
-                    file_extra_config = FileUploadConfigManager.convert(self.conversation.model_config)
-                elif self.conversation.mode in {AppMode.ADVANCED_CHAT, AppMode.WORKFLOW}:
-                    workflow_run = db.session.scalar(
-                        select(WorkflowRun).where(WorkflowRun.id == message.workflow_run_id)
-                    )
-                    if not workflow_run:
-                        raise ValueError(f"Workflow run not found: {message.workflow_run_id}")
-                    workflow = db.session.scalar(select(Workflow).where(Workflow.id == workflow_run.workflow_id))
-                    if not workflow:
-                        raise ValueError(f"Workflow not found: {workflow_run.workflow_id}")
-                    file_extra_config = FileUploadConfigManager.convert(workflow.features_dict, is_vision=False)
-                else:
-                    raise AssertionError(f"Invalid app mode: {self.conversation.mode}")
-
-                detail = ImagePromptMessageContent.DETAIL.LOW
-                if file_extra_config and app_record:
-                    file_objs = file_factory.build_from_message_files(
-                        message_files=files, tenant_id=app_record.tenant_id, config=file_extra_config
-                    )
-                    if file_extra_config.image_config and file_extra_config.image_config.detail:
-                        detail = file_extra_config.image_config.detail
-                else:
-                    file_objs = []
-
-                if not file_objs:
-                    prompt_messages.append(UserPromptMessage(content=message.query))
-                else:
-                    prompt_message_contents: list[PromptMessageContentUnionTypes] = []
-                    for file in file_objs:
-                        prompt_message = file_manager.to_prompt_message_content(
-                            file,
-                            image_detail_config=detail,
-                        )
-                        prompt_message_contents.append(prompt_message)
-                    prompt_message_contents.append(TextPromptMessageContent(data=message.query))
-
-                    prompt_messages.append(UserPromptMessage(content=prompt_message_contents))
+            # Process user message with files
+            user_files = (
+                db.session.query(MessageFile)
+                .where(
+                    MessageFile.message_id == message.id,
+                    (MessageFile.belongs_to == "user") | (MessageFile.belongs_to.is_(None)),
+                )
+                .all()
+            )
 
+            if user_files:
+                user_prompt_message = self._build_prompt_message_with_files(
+                    message_files=user_files,
+                    text_content=message.query,
+                    message=message,
+                    app_record=app_record,
+                    is_user_message=True,
+                )
+                prompt_messages.append(user_prompt_message)
             else:
                 prompt_messages.append(UserPromptMessage(content=message.query))
 
-            prompt_messages.append(AssistantPromptMessage(content=message.answer))
+            # Process assistant message with files
+            assistant_files = (
+                db.session.query(MessageFile)
+                .where(MessageFile.message_id == message.id, MessageFile.belongs_to == "assistant")
+                .all()
+            )
 
-        if not prompt_messages:
-            return []
+            if assistant_files:
+                assistant_prompt_message = self._build_prompt_message_with_files(
+                    message_files=assistant_files,
+                    text_content=message.answer,
+                    message=message,
+                    app_record=app_record,
+                    is_user_message=False,
+                )
+                prompt_messages.append(assistant_prompt_message)
+            else:
+                prompt_messages.append(AssistantPromptMessage(content=message.answer))
 
-        # prune the chat message if it exceeds the max token limit
-        curr_message_tokens = self.model_instance.get_llm_num_tokens(prompt_messages)
+            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)
 
         if curr_message_tokens > max_token_limit:
             while curr_message_tokens > max_token_limit and len(prompt_messages) > 1:

api/core/model_runtime/entities/message_entities.py

@@ -87,6 +87,7 @@ class MultiModalPromptMessageContent(PromptMessageContent):
     base64_data: str = Field(default="", description="the base64 data of multi-modal file")
     url: str = Field(default="", description="the url of multi-modal file")
     mime_type: str = Field(default=..., description="the mime type of multi-modal file")
+    filename: str = Field(default="", description="the filename of multi-modal file")
 
     @property
     def data(self):

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

@@ -24,8 +24,7 @@ from core.model_runtime.errors.invoke import (
     InvokeRateLimitError,
     InvokeServerUnavailableError,
 )
-from core.plugin.entities.plugin_daemon import PluginDaemonInnerError, PluginModelProviderEntity
-from core.plugin.impl.model import PluginModelClient
+from core.plugin.entities.plugin_daemon import PluginModelProviderEntity
 
 
 class AIModel(BaseModel):
@@ -53,6 +52,8 @@ class AIModel(BaseModel):
 
         :return: Invoke error mapping
         """
+        from core.plugin.entities.plugin_daemon import PluginDaemonInnerError
+
         return {
             InvokeConnectionError: [InvokeConnectionError],
             InvokeServerUnavailableError: [InvokeServerUnavailableError],
@@ -140,6 +141,8 @@ 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,7 +22,6 @@ 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__)
 
@@ -142,6 +141,8 @@ 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,
@@ -340,6 +341,8 @@ 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,7 +5,6 @@ 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):
@@ -31,6 +30,8 @@ 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,7 +3,6 @@ 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):
@@ -36,6 +35,8 @@ 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,7 +4,6 @@ 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):
@@ -28,6 +27,8 @@ 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,7 +6,6 @@ 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):
@@ -37,6 +36,8 @@ 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(
@@ -61,6 +62,8 @@ 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/tokenizers/gpt2_tokenizer.py

@@ -43,7 +43,7 @@ class GPT2Tokenizer:
                 except Exception:
                     from os.path import abspath, dirname, join
 
-                    from transformers import GPT2Tokenizer as TransformerGPT2Tokenizer  # type: ignore
+                    from transformers import GPT2Tokenizer as TransformerGPT2Tokenizer
 
                     base_path = abspath(__file__)
                     gpt2_tokenizer_path = join(dirname(base_path), "gpt2")

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

@@ -6,7 +6,6 @@ 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__)
 
@@ -42,6 +41,8 @@ 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,
@@ -65,6 +66,8 @@ 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,10 +20,8 @@ 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 core.plugin.impl.asset import PluginAssetManager
-from core.plugin.impl.model import PluginModelClient
+from models.provider_ids import ModelProviderID
 
 logger = logging.getLogger(__name__)
 
@@ -37,6 +35,8 @@ 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,6 +366,8 @@ 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
 
@@ -375,5 +377,6 @@ 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,13 +54,10 @@ from core.ops.entities.trace_entity import (
 )
 from core.rag.models.document import Document
 from core.repositories import SQLAlchemyWorkflowNodeExecutionRepository
-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
+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
 
 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.nodes.enums import NodeType
+from core.workflow.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,8 +28,7 @@ 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.entities.workflow_node_execution import WorkflowNodeExecutionMetadataKey
-from core.workflow.nodes.enums import NodeType
+from core.workflow.enums import NodeType, WorkflowNodeExecutionMetadataKey
 from extensions.ext_database import db
 from models import EndUser, MessageFile, WorkflowNodeExecutionTriggeredFrom
 

api/core/ops/opik_trace/opik_trace.py

@@ -22,8 +22,7 @@ from core.ops.entities.trace_entity import (
     WorkflowTraceInfo,
 )
 from core.repositories import DifyCoreRepositoryFactory
-from core.workflow.entities.workflow_node_execution import WorkflowNodeExecutionMetadataKey
-from core.workflow.nodes.enums import NodeType
+from core.workflow.enums import NodeType, WorkflowNodeExecutionMetadataKey
 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 Any, Optional, Union
+from typing import TYPE_CHECKING, Any, Optional, Union
 from uuid import UUID, uuid4
 
 from cachetools import LRUCache
@@ -30,13 +30,15 @@ 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__)
 
 
@@ -330,7 +332,7 @@ class OpsTraceManager:
             except KeyError:
                 raise ValueError(f"Invalid tracing provider: {tracing_provider}")
         else:
-            if tracing_provider is not None:
+            if tracing_provider is None:
                 raise ValueError(f"Invalid tracing provider: {tracing_provider}")
 
         app_config: Optional[App] = db.session.query(App).where(App.id == app_id).first()
@@ -410,7 +412,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,8 +23,7 @@ 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.entities.workflow_node_execution import WorkflowNodeExecutionMetadataKey
-from core.workflow.nodes.enums import NodeType
+from core.workflow.enums import NodeType, WorkflowNodeExecutionMetadataKey
 from extensions.ext_database import db
 from models import EndUser, MessageFile, WorkflowNodeExecutionTriggeredFrom
 

api/core/plugin/backwards_invocation/app.py

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

api/core/plugin/backwards_invocation/model.py

@@ -375,16 +375,16 @@ Here is the extra instruction you need to follow:
 
         # merge lines into messages with max tokens
         messages: list[str] = []
-        for i in new_lines:  # type: ignore
+        for line in new_lines:
             if len(messages) == 0:
-                messages.append(i)  # type: ignore
+                messages.append(line)
             else:
-                if len(messages[-1]) + len(i) < max_tokens * 0.5:  # type: ignore
-                    messages[-1] += i  # type: ignore
-                if get_prompt_tokens(messages[-1] + i) > max_tokens * 0.7:  # type: ignore
-                    messages.append(i)  # type: ignore
+                if len(messages[-1]) + len(line) < max_tokens * 0.5:
+                    messages[-1] += line
+                if get_prompt_tokens(messages[-1] + line) > max_tokens * 0.7:
+                    messages.append(line)
                 else:
-                    messages[-1] += i  # type: ignore
+                    messages[-1] += line
 
         summaries = []
         for i in range(len(messages)):

api/core/plugin/backwards_invocation/node.py

@@ -1,5 +1,5 @@
 from core.plugin.backwards_invocation.base import BaseBackwardsInvocation
-from core.workflow.nodes.enums import NodeType
+from core.workflow.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 | NumberType):
+                if value and not isinstance(value, str | dict | list | int | float):
                     raise ValueError("The var selector must be a string, dictionary, list or number.")
                 return value
             case PluginParameterType.ARRAY:
@@ -162,8 +162,6 @@ 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
@@ -176,8 +174,6 @@ 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,11 +1,9 @@
 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
@@ -135,55 +133,6 @@ 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 PluginDependency(BaseModel):
     class Type(enum.StrEnum):
         Github = PluginInstallationSource.Github.value

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):

api/core/plugin/impl/plugin.py

@@ -2,7 +2,6 @@ from collections.abc import Sequence
 
 from core.plugin.entities.bundle import PluginBundleDependency
 from core.plugin.entities.plugin import (
-    GenericProviderID,
     MissingPluginDependency,
     PluginDeclaration,
     PluginEntity,
@@ -16,6 +15,7 @@ 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 import GenericProviderID, ToolProviderID
 from core.plugin.entities.plugin_daemon import 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
 
 
 class PluginToolManager(BasePluginClient):

api/core/provider_manager.py

@@ -34,7 +34,6 @@ 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
@@ -49,6 +48,7 @@ from models.provider import (
     TenantDefaultModel,
     TenantPreferredModelProvider,
 )
+from models.provider_ids import ModelProviderID
 from services.feature_service import FeatureService
 
 

api/core/rag/datasource/vdb/analyticdb/analyticdb_vector_sql.py

@@ -3,8 +3,8 @@ import uuid
 from contextlib import contextmanager
 from typing import Any
 
-import psycopg2.extras  # type: ignore
-import psycopg2.pool  # type: ignore
+import psycopg2.extras
+import psycopg2.pool
 from pydantic import BaseModel, model_validator
 
 from core.rag.models.document import Document

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

@@ -3,8 +3,8 @@ import uuid
 from contextlib import contextmanager
 from typing import Any
 
-import psycopg2.extras  # type: ignore
-import psycopg2.pool  # type: ignore
+import psycopg2.extras
+import psycopg2.pool
 from pydantic import BaseModel, model_validator
 
 from configs import dify_config

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

@@ -48,7 +48,7 @@ class OpenSearchConfig(BaseModel):
         return values
 
     def create_aws_managed_iam_auth(self) -> Urllib3AWSV4SignerAuth:
-        import boto3  # type: ignore
+        import boto3
 
         return Urllib3AWSV4SignerAuth(
             credentials=boto3.Session().get_credentials(),

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

@@ -6,8 +6,8 @@ from contextlib import contextmanager
 from typing import Any
 
 import psycopg2.errors
-import psycopg2.extras  # type: ignore
-import psycopg2.pool  # type: ignore
+import psycopg2.extras
+import psycopg2.pool
 from pydantic import BaseModel, model_validator
 
 from configs import dify_config

api/core/rag/datasource/vdb/pyvastbase/vastbase_vector.py

@@ -3,8 +3,8 @@ import uuid
 from contextlib import contextmanager
 from typing import Any
 
-import psycopg2.extras  # type: ignore
-import psycopg2.pool  # type: ignore
+import psycopg2.extras
+import psycopg2.pool
 from pydantic import BaseModel, model_validator
 
 from configs import dify_config

api/core/rag/extractor/excel_extractor.py

@@ -4,7 +4,7 @@ import os
 from typing import Optional, cast
 
 import pandas as pd
-from openpyxl import load_workbook  # type: ignore
+from openpyxl import load_workbook
 
 from core.rag.extractor.extractor_base import BaseExtractor
 from core.rag.models.document import Document

api/core/rag/extractor/html_extractor.py

@@ -1,6 +1,6 @@
 """Abstract interface for document loader implementations."""
 
-from bs4 import BeautifulSoup  # type: ignore
+from bs4 import BeautifulSoup
 
 from core.rag.extractor.extractor_base import BaseExtractor
 from core.rag.models.document import Document

api/core/rag/extractor/unstructured/unstructured_eml_extractor.py

@@ -3,7 +3,7 @@ import contextlib
 import logging
 from typing import Optional
 
-from bs4 import BeautifulSoup  # type: ignore
+from bs4 import BeautifulSoup
 
 from core.rag.extractor.extractor_base import BaseExtractor
 from core.rag.models.document import Document

api/core/rag/index_processor/index_processor_base.py

@@ -1,10 +1,9 @@
 """Abstract interface for document loader implementations."""
 
 from abc import ABC, abstractmethod
-from typing import Optional
+from typing import TYPE_CHECKING, 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 (
@@ -14,6 +13,9 @@ 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."""
@@ -51,7 +53,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/splitter/text_splitter.py

@@ -144,7 +144,7 @@ class TextSplitter(BaseDocumentTransformer, ABC):
     def from_huggingface_tokenizer(cls, tokenizer: Any, **kwargs: Any) -> TextSplitter:
         """Text splitter that uses HuggingFace tokenizer to count length."""
         try:
-            from transformers import PreTrainedTokenizerBase  # type: ignore
+            from transformers import PreTrainedTokenizerBase
 
             if not isinstance(tokenizer, PreTrainedTokenizerBase):
                 raise ValueError("Tokenizer received was not an instance of PreTrainedTokenizerBase")

api/core/repositories/sqlalchemy_workflow_execution_repository.py

@@ -9,11 +9,8 @@ from typing import Optional, Union
 from sqlalchemy.engine import Engine
 from sqlalchemy.orm import sessionmaker
 
-from core.workflow.entities.workflow_execution import (
-    WorkflowExecution,
-    WorkflowExecutionStatus,
-    WorkflowType,
-)
+from core.workflow.entities import WorkflowExecution
+from core.workflow.enums import 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
@@ -203,5 +200,4 @@ 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,12 +12,8 @@ from sqlalchemy.engine import Engine
 from sqlalchemy.orm import sessionmaker
 
 from core.model_runtime.utils.encoders import jsonable_encoder
-from core.workflow.entities.workflow_node_execution import (
-    WorkflowNodeExecution,
-    WorkflowNodeExecutionMetadataKey,
-    WorkflowNodeExecutionStatus,
-)
-from core.workflow.nodes.enums import NodeType
+from core.workflow.entities import WorkflowNodeExecution
+from core.workflow.enums import NodeType, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
 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
@@ -215,7 +211,6 @@ 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/tool_engine.py

@@ -152,7 +152,6 @@ 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,
@@ -166,7 +165,6 @@ 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,31 +13,16 @@ 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.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
@@ -53,16 +38,28 @@ from core.tools.entities.tool_entities import (
     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__)
 
 
@@ -117,6 +114,8 @@ 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:
@@ -172,6 +171,7 @@ 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
-                    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),
+                    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())
                         )
-                        .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,6 +256,7 @@ 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
@@ -263,6 +264,7 @@ 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(
@@ -358,7 +360,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
@@ -400,7 +402,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
@@ -516,6 +518,8 @@ 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 [
@@ -977,7 +981,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/parser.py

@@ -6,7 +6,7 @@ from typing import Optional
 
 from flask import request
 from requests import get
-from yaml import YAMLError, safe_load  # type: ignore
+from yaml import YAMLError, safe_load
 
 from core.tools.entities.common_entities import I18nObject
 from core.tools.entities.tool_bundle import ApiToolBundle

api/core/tools/workflow_as_tool/tool.py

@@ -37,14 +37,12 @@ class WorkflowTool(Tool):
         entity: ToolEntity,
         runtime: ToolRuntime,
         label: str = "Workflow",
-        thread_pool_id: Optional[str] = None,
     ):
         self.workflow_app_id = workflow_app_id
         self.workflow_as_tool_id = workflow_as_tool_id
         self.version = version
         self.workflow_entities = workflow_entities
         self.workflow_call_depth = workflow_call_depth
-        self.thread_pool_id = thread_pool_id
         self.label = label
 
         super().__init__(entity=entity, runtime=runtime)
@@ -88,7 +86,6 @@ class WorkflowTool(Tool):
             invoke_from=self.runtime.invoke_from,
             streaming=False,
             call_depth=self.workflow_call_depth + 1,
-            workflow_thread_pool_id=self.thread_pool_id,
         )
         assert isinstance(result, dict)
         data = result.get("data", {})

api/core/variables/segments.py

@@ -130,7 +130,7 @@ class ArraySegment(Segment):
     def markdown(self) -> str:
         items = []
         for item in self.value:
-            items.append(str(item))
+            items.append(f"- {item}")
         return "\n".join(items)
 
 

api/core/workflow/callbacks/__init__.py

@@ -1,7 +0,0 @@
-from .base_workflow_callback import WorkflowCallback
-from .workflow_logging_callback import WorkflowLoggingCallback
-
-__all__ = [
-    "WorkflowCallback",
-    "WorkflowLoggingCallback",
-]

api/core/workflow/callbacks/base_workflow_callback.py

@@ -1,12 +0,0 @@
-from abc import ABC, abstractmethod
-
-from core.workflow.graph_engine.entities.event import GraphEngineEvent
-
-
-class WorkflowCallback(ABC):
-    @abstractmethod
-    def on_event(self, event: GraphEngineEvent) -> None:
-        """
-        Published event
-        """
-        raise NotImplementedError

api/core/workflow/callbacks/workflow_logging_callback.py

@@ -1,263 +0,0 @@
-from typing import Optional
-
-from core.model_runtime.utils.encoders import jsonable_encoder
-from core.workflow.graph_engine.entities.event import (
-    GraphEngineEvent,
-    GraphRunFailedEvent,
-    GraphRunPartialSucceededEvent,
-    GraphRunStartedEvent,
-    GraphRunSucceededEvent,
-    IterationRunFailedEvent,
-    IterationRunNextEvent,
-    IterationRunStartedEvent,
-    IterationRunSucceededEvent,
-    LoopRunFailedEvent,
-    LoopRunNextEvent,
-    LoopRunStartedEvent,
-    LoopRunSucceededEvent,
-    NodeRunFailedEvent,
-    NodeRunStartedEvent,
-    NodeRunStreamChunkEvent,
-    NodeRunSucceededEvent,
-    ParallelBranchRunFailedEvent,
-    ParallelBranchRunStartedEvent,
-    ParallelBranchRunSucceededEvent,
-)
-
-from .base_workflow_callback import WorkflowCallback
-
-_TEXT_COLOR_MAPPING = {
-    "blue": "36;1",
-    "yellow": "33;1",
-    "pink": "38;5;200",
-    "green": "32;1",
-    "red": "31;1",
-}
-
-
-class WorkflowLoggingCallback(WorkflowCallback):
-    def __init__(self) -> None:
-        self.current_node_id: Optional[str] = None
-
-    def on_event(self, event: GraphEngineEvent) -> None:
-        if isinstance(event, GraphRunStartedEvent):
-            self.print_text("\n[GraphRunStartedEvent]", color="pink")
-        elif isinstance(event, GraphRunSucceededEvent):
-            self.print_text("\n[GraphRunSucceededEvent]", color="green")
-        elif isinstance(event, GraphRunPartialSucceededEvent):
-            self.print_text("\n[GraphRunPartialSucceededEvent]", color="pink")
-        elif isinstance(event, GraphRunFailedEvent):
-            self.print_text(f"\n[GraphRunFailedEvent] reason: {event.error}", color="red")
-        elif isinstance(event, NodeRunStartedEvent):
-            self.on_workflow_node_execute_started(event=event)
-        elif isinstance(event, NodeRunSucceededEvent):
-            self.on_workflow_node_execute_succeeded(event=event)
-        elif isinstance(event, NodeRunFailedEvent):
-            self.on_workflow_node_execute_failed(event=event)
-        elif isinstance(event, NodeRunStreamChunkEvent):
-            self.on_node_text_chunk(event=event)
-        elif isinstance(event, ParallelBranchRunStartedEvent):
-            self.on_workflow_parallel_started(event=event)
-        elif isinstance(event, ParallelBranchRunSucceededEvent | ParallelBranchRunFailedEvent):
-            self.on_workflow_parallel_completed(event=event)
-        elif isinstance(event, IterationRunStartedEvent):
-            self.on_workflow_iteration_started(event=event)
-        elif isinstance(event, IterationRunNextEvent):
-            self.on_workflow_iteration_next(event=event)
-        elif isinstance(event, IterationRunSucceededEvent | IterationRunFailedEvent):
-            self.on_workflow_iteration_completed(event=event)
-        elif isinstance(event, LoopRunStartedEvent):
-            self.on_workflow_loop_started(event=event)
-        elif isinstance(event, LoopRunNextEvent):
-            self.on_workflow_loop_next(event=event)
-        elif isinstance(event, LoopRunSucceededEvent | LoopRunFailedEvent):
-            self.on_workflow_loop_completed(event=event)
-        else:
-            self.print_text(f"\n[{event.__class__.__name__}]", color="blue")
-
-    def on_workflow_node_execute_started(self, event: NodeRunStartedEvent) -> None:
-        """
-        Workflow node execute started
-        """
-        self.print_text("\n[NodeRunStartedEvent]", color="yellow")
-        self.print_text(f"Node ID: {event.node_id}", color="yellow")
-        self.print_text(f"Node Title: {event.node_data.title}", color="yellow")
-        self.print_text(f"Type: {event.node_type.value}", color="yellow")
-
-    def on_workflow_node_execute_succeeded(self, event: NodeRunSucceededEvent) -> None:
-        """
-        Workflow node execute succeeded
-        """
-        route_node_state = event.route_node_state
-
-        self.print_text("\n[NodeRunSucceededEvent]", color="green")
-        self.print_text(f"Node ID: {event.node_id}", color="green")
-        self.print_text(f"Node Title: {event.node_data.title}", color="green")
-        self.print_text(f"Type: {event.node_type.value}", color="green")
-
-        if route_node_state.node_run_result:
-            node_run_result = route_node_state.node_run_result
-            self.print_text(
-                f"Inputs: {jsonable_encoder(node_run_result.inputs) if node_run_result.inputs else ''}",
-                color="green",
-            )
-            self.print_text(
-                f"Process Data: "
-                f"{jsonable_encoder(node_run_result.process_data) if node_run_result.process_data else ''}",
-                color="green",
-            )
-            self.print_text(
-                f"Outputs: {jsonable_encoder(node_run_result.outputs) if node_run_result.outputs else ''}",
-                color="green",
-            )
-            self.print_text(
-                f"Metadata: {jsonable_encoder(node_run_result.metadata) if node_run_result.metadata else ''}",
-                color="green",
-            )
-
-    def on_workflow_node_execute_failed(self, event: NodeRunFailedEvent) -> None:
-        """
-        Workflow node execute failed
-        """
-        route_node_state = event.route_node_state
-
-        self.print_text("\n[NodeRunFailedEvent]", color="red")
-        self.print_text(f"Node ID: {event.node_id}", color="red")
-        self.print_text(f"Node Title: {event.node_data.title}", color="red")
-        self.print_text(f"Type: {event.node_type.value}", color="red")
-
-        if route_node_state.node_run_result:
-            node_run_result = route_node_state.node_run_result
-            self.print_text(f"Error: {node_run_result.error}", color="red")
-            self.print_text(
-                f"Inputs: {jsonable_encoder(node_run_result.inputs) if node_run_result.inputs else ''}",
-                color="red",
-            )
-            self.print_text(
-                f"Process Data: "
-                f"{jsonable_encoder(node_run_result.process_data) if node_run_result.process_data else ''}",
-                color="red",
-            )
-            self.print_text(
-                f"Outputs: {jsonable_encoder(node_run_result.outputs) if node_run_result.outputs else ''}",
-                color="red",
-            )
-
-    def on_node_text_chunk(self, event: NodeRunStreamChunkEvent) -> None:
-        """
-        Publish text chunk
-        """
-        route_node_state = event.route_node_state
-        if not self.current_node_id or self.current_node_id != route_node_state.node_id:
-            self.current_node_id = route_node_state.node_id
-            self.print_text("\n[NodeRunStreamChunkEvent]")
-            self.print_text(f"Node ID: {route_node_state.node_id}")
-
-            node_run_result = route_node_state.node_run_result
-            if node_run_result:
-                self.print_text(
-                    f"Metadata: {jsonable_encoder(node_run_result.metadata) if node_run_result.metadata else ''}"
-                )
-
-        self.print_text(event.chunk_content, color="pink", end="")
-
-    def on_workflow_parallel_started(self, event: ParallelBranchRunStartedEvent) -> None:
-        """
-        Publish parallel started
-        """
-        self.print_text("\n[ParallelBranchRunStartedEvent]", color="blue")
-        self.print_text(f"Parallel ID: {event.parallel_id}", color="blue")
-        self.print_text(f"Branch ID: {event.parallel_start_node_id}", color="blue")
-        if event.in_iteration_id:
-            self.print_text(f"Iteration ID: {event.in_iteration_id}", color="blue")
-        if event.in_loop_id:
-            self.print_text(f"Loop ID: {event.in_loop_id}", color="blue")
-
-    def on_workflow_parallel_completed(
-        self, event: ParallelBranchRunSucceededEvent | ParallelBranchRunFailedEvent
-    ) -> None:
-        """
-        Publish parallel completed
-        """
-        if isinstance(event, ParallelBranchRunSucceededEvent):
-            color = "blue"
-        elif isinstance(event, ParallelBranchRunFailedEvent):
-            color = "red"
-
-        self.print_text(
-            "\n[ParallelBranchRunSucceededEvent]"
-            if isinstance(event, ParallelBranchRunSucceededEvent)
-            else "\n[ParallelBranchRunFailedEvent]",
-            color=color,
-        )
-        self.print_text(f"Parallel ID: {event.parallel_id}", color=color)
-        self.print_text(f"Branch ID: {event.parallel_start_node_id}", color=color)
-        if event.in_iteration_id:
-            self.print_text(f"Iteration ID: {event.in_iteration_id}", color=color)
-        if event.in_loop_id:
-            self.print_text(f"Loop ID: {event.in_loop_id}", color=color)
-
-        if isinstance(event, ParallelBranchRunFailedEvent):
-            self.print_text(f"Error: {event.error}", color=color)
-
-    def on_workflow_iteration_started(self, event: IterationRunStartedEvent) -> None:
-        """
-        Publish iteration started
-        """
-        self.print_text("\n[IterationRunStartedEvent]", color="blue")
-        self.print_text(f"Iteration Node ID: {event.iteration_id}", color="blue")
-
-    def on_workflow_iteration_next(self, event: IterationRunNextEvent) -> None:
-        """
-        Publish iteration next
-        """
-        self.print_text("\n[IterationRunNextEvent]", color="blue")
-        self.print_text(f"Iteration Node ID: {event.iteration_id}", color="blue")
-        self.print_text(f"Iteration Index: {event.index}", color="blue")
-
-    def on_workflow_iteration_completed(self, event: IterationRunSucceededEvent | IterationRunFailedEvent) -> None:
-        """
-        Publish iteration completed
-        """
-        self.print_text(
-            "\n[IterationRunSucceededEvent]"
-            if isinstance(event, IterationRunSucceededEvent)
-            else "\n[IterationRunFailedEvent]",
-            color="blue",
-        )
-        self.print_text(f"Node ID: {event.iteration_id}", color="blue")
-
-    def on_workflow_loop_started(self, event: LoopRunStartedEvent) -> None:
-        """
-        Publish loop started
-        """
-        self.print_text("\n[LoopRunStartedEvent]", color="blue")
-        self.print_text(f"Loop Node ID: {event.loop_node_id}", color="blue")
-
-    def on_workflow_loop_next(self, event: LoopRunNextEvent) -> None:
-        """
-        Publish loop next
-        """
-        self.print_text("\n[LoopRunNextEvent]", color="blue")
-        self.print_text(f"Loop Node ID: {event.loop_node_id}", color="blue")
-        self.print_text(f"Loop Index: {event.index}", color="blue")
-
-    def on_workflow_loop_completed(self, event: LoopRunSucceededEvent | LoopRunFailedEvent) -> None:
-        """
-        Publish loop completed
-        """
-        self.print_text(
-            "\n[LoopRunSucceededEvent]" if isinstance(event, LoopRunSucceededEvent) else "\n[LoopRunFailedEvent]",
-            color="blue",
-        )
-        self.print_text(f"Loop Node ID: {event.loop_node_id}", color="blue")
-
-    def print_text(self, text: str, color: Optional[str] = None, end: str = "\n") -> None:
-        """Print text with highlighting and no end characters."""
-        text_to_print = self._get_colored_text(text, color) if color else text
-        print(f"{text_to_print}", end=end)
-
-    def _get_colored_text(self, text: str, color: str) -> str:
-        """Get colored text."""
-        color_str = _TEXT_COLOR_MAPPING[color]
-        return f"\u001b[{color_str}m\033[1;3m{text}\u001b[0m"