chore: fix test cases

.devcontainer/post_create_command.sh

@@ -6,7 +6,7 @@ cd web && pnpm install
 pipx install uv
 
 echo "alias start-api=\"cd $WORKSPACE_ROOT/api && uv run python -m flask run --host 0.0.0.0 --port=5001 --debug\"" >> ~/.bashrc
-echo "alias start-worker=\"cd $WORKSPACE_ROOT/api && uv run python -m celery -A app.celery worker -P threads -c 1 --loglevel INFO -Q dataset,generation,mail,ops_trace,app_deletion,plugin,workflow_storage\"" >> ~/.bashrc
+echo "alias start-worker=\"cd $WORKSPACE_ROOT/api && uv run python -m celery -A app.celery worker -P gevent -c 1 --loglevel INFO -Q dataset,generation,mail,ops_trace,app_deletion,plugin,workflow_storage\"" >> ~/.bashrc
 echo "alias start-web=\"cd $WORKSPACE_ROOT/web && pnpm dev\"" >> ~/.bashrc
 echo "alias start-web-prod=\"cd $WORKSPACE_ROOT/web && pnpm build && pnpm start\"" >> ~/.bashrc
 echo "alias start-containers=\"cd $WORKSPACE_ROOT/docker && docker-compose -f docker-compose.middleware.yaml -p dify --env-file middleware.env up -d\"" >> ~/.bashrc

.github/workflows/autofix.yml

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

.gitignore

@@ -6,9 +6,6 @@ __pycache__/
 # C extensions
 *.so
 
-# *db files
-*.db
-
 # Distribution / packaging
 .Python
 build/
@@ -238,7 +235,4 @@ scripts/stress-test/reports/
 
 # mcp
 .playwright-mcp/
-.serena/
-
-# settings
-*.local.json
+.serena/

.vscode/launch.json.template

@@ -8,7 +8,8 @@
             "module": "flask",
             "env": {
                 "FLASK_APP": "app.py",
-                "FLASK_ENV": "development"
+                "FLASK_ENV": "development",
+                "GEVENT_SUPPORT": "True"
             },
             "args": [
                 "run",
@@ -27,7 +28,9 @@
             "type": "debugpy",
             "request": "launch",
             "module": "celery",
-            "env": {},
+            "env": {
+                "GEVENT_SUPPORT": "True"
+            },
             "args": [
                 "-A",
                 "app.celery",

README.md

@@ -117,7 +117,7 @@ All of Dify's offerings come with corresponding APIs, so you could effortlessly
   Use our [documentation](https://docs.dify.ai) for further references and more in-depth instructions.
 
 - **Dify for enterprise / organizations<br/>**
-  We provide additional enterprise-centric features. [Send us an email](mailto:business@dify.ai?subject=%5BGitHub%5DBusiness%20License%20Inquiry) to discuss your enterprise needs. <br/>
+  We provide additional enterprise-centric features. [Log your questions for us through this chatbot](https://udify.app/chat/22L1zSxg6yW1cWQg) or [send us an email](mailto:business@dify.ai?subject=%5BGitHub%5DBusiness%20License%20Inquiry) to discuss enterprise needs. <br/>
 
   > For startups and small businesses using AWS, check out [Dify Premium on AWS Marketplace](https://aws.amazon.com/marketplace/pp/prodview-t22mebxzwjhu6) and deploy it to your own AWS VPC with one click. It's an affordable AMI offering with the option to create apps with custom logo and branding.
 

api/.env.example

@@ -27,9 +27,6 @@ FILES_URL=http://localhost:5001
 # Example: INTERNAL_FILES_URL=http://api:5001
 INTERNAL_FILES_URL=http://127.0.0.1:5001
 
-# TRIGGER URL
-TRIGGER_URL=http://localhost:5001
-
 # The time in seconds after the signature is rejected
 FILES_ACCESS_TIMEOUT=300
 
@@ -374,12 +371,6 @@ UPLOAD_IMAGE_FILE_SIZE_LIMIT=10
 UPLOAD_VIDEO_FILE_SIZE_LIMIT=100
 UPLOAD_AUDIO_FILE_SIZE_LIMIT=50
 
-# Comma-separated list of file extensions blocked from upload for security reasons.
-# Extensions should be lowercase without dots (e.g., exe,bat,sh,dll).
-# Empty by default to allow all file types.
-# Recommended: exe,bat,cmd,com,scr,vbs,ps1,msi,dll
-UPLOAD_FILE_EXTENSION_BLACKLIST=
-
 # Model configuration
 MULTIMODAL_SEND_FORMAT=base64
 PROMPT_GENERATION_MAX_TOKENS=512
@@ -469,9 +460,6 @@ HTTP_REQUEST_NODE_MAX_BINARY_SIZE=10485760
 HTTP_REQUEST_NODE_MAX_TEXT_SIZE=1048576
 HTTP_REQUEST_NODE_SSL_VERIFY=True
 
-# Webhook request configuration
-WEBHOOK_REQUEST_BODY_MAX_SIZE=10485760
-
 # Respect X-* headers to redirect clients
 RESPECT_XFORWARD_HEADERS_ENABLED=false
 
@@ -527,7 +515,7 @@ API_WORKFLOW_NODE_EXECUTION_REPOSITORY=repositories.sqlalchemy_api_workflow_node
 API_WORKFLOW_RUN_REPOSITORY=repositories.sqlalchemy_api_workflow_run_repository.DifyAPISQLAlchemyWorkflowRunRepository
 # Workflow log cleanup configuration
 # Enable automatic cleanup of workflow run logs to manage database size
-WORKFLOW_LOG_CLEANUP_ENABLED=false
+WORKFLOW_LOG_CLEANUP_ENABLED=true
 # Number of days to retain workflow run logs (default: 30 days)
 WORKFLOW_LOG_RETENTION_DAYS=30
 # Batch size for workflow log cleanup operations (default: 100)
@@ -549,12 +537,6 @@ ENABLE_CLEAN_MESSAGES=false
 ENABLE_MAIL_CLEAN_DOCUMENT_NOTIFY_TASK=false
 ENABLE_DATASETS_QUEUE_MONITOR=false
 ENABLE_CHECK_UPGRADABLE_PLUGIN_TASK=true
-ENABLE_WORKFLOW_SCHEDULE_POLLER_TASK=true
-# Interval time in minutes for polling scheduled workflows(default: 1 min)
-WORKFLOW_SCHEDULE_POLLER_INTERVAL=1
-WORKFLOW_SCHEDULE_POLLER_BATCH_SIZE=100
-# Maximum number of scheduled workflows to dispatch per tick (0 for unlimited)
-WORKFLOW_SCHEDULE_MAX_DISPATCH_PER_TICK=0
 
 # Position configuration
 POSITION_TOOL_PINS=
@@ -626,9 +608,3 @@ SWAGGER_UI_PATH=/swagger-ui.html
 # Whether to encrypt dataset IDs when exporting DSL files (default: true)
 # Set to false to export dataset IDs as plain text for easier cross-environment import
 DSL_EXPORT_ENCRYPT_DATASET_ID=true
-
-# Tenant isolated task queue configuration
-TENANT_ISOLATED_TASK_CONCURRENCY=1
-
-# Maximum number of segments for dataset segments API (0 for unlimited)
-DATASET_MAX_SEGMENTS_PER_REQUEST=0

api/.vscode/launch.json.example

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

api/AGENTS.md

@@ -1,62 +0,0 @@
-# Agent Skill Index
-
-Start with the section that best matches your need. Each entry lists the problems it solves plus key files/concepts so you know what to expect before opening it.
-
-______________________________________________________________________
-
-## Platform Foundations
-
-- **[Infrastructure Overview](agent_skills/infra.md)**\
-  When to read this:
-
-  - You need to understand where a feature belongs in the architecture.
-  - You’re wiring storage, Redis, vector stores, or OTEL.
-  - You’re about to add CLI commands or async jobs.\
-    What it covers: configuration stack (`configs/app_config.py`, remote settings), storage entry points (`extensions/ext_storage.py`, `core/file/file_manager.py`), Redis conventions (`extensions/ext_redis.py`), plugin runtime topology, vector-store factory (`core/rag/datasource/vdb/*`), observability hooks, SSRF proxy usage, and core CLI commands.
-
-- **[Coding Style](agent_skills/coding_style.md)**\
-  When to read this:
-
-  - You’re writing or reviewing backend code and need the authoritative checklist.
-  - You’re unsure about Pydantic validators, SQLAlchemy session usage, or logging patterns.
-  - You want the exact lint/type/test commands used in PRs.\
-    Includes: Ruff & BasedPyright commands, no-annotation policy, session examples (`with Session(db.engine, ...)`), `@field_validator` usage, logging expectations, and the rule set for file size, helpers, and package management.
-
-______________________________________________________________________
-
-## Plugin & Extension Development
-
-- **[Plugin Systems](agent_skills/plugin.md)**\
-  When to read this:
-
-  - You’re building or debugging a marketplace plugin.
-  - You need to know how manifests, providers, daemons, and migrations fit together.\
-    What it covers: plugin manifests (`core/plugin/entities/plugin.py`), installation/upgrade flows (`services/plugin/plugin_service.py`, CLI commands), runtime adapters (`core/plugin/impl/*` for tool/model/datasource/trigger/endpoint/agent), daemon coordination (`core/plugin/entities/plugin_daemon.py`), and how provider registries surface capabilities to the rest of the platform.
-
-- **[Plugin OAuth](agent_skills/plugin_oauth.md)**\
-  When to read this:
-
-  - You must integrate OAuth for a plugin or datasource.
-  - You’re handling credential encryption or refresh flows.\
-    Topics: credential storage, encryption helpers (`core/helper/provider_encryption.py`), OAuth client bootstrap (`services/plugin/oauth_service.py`, `services/plugin/plugin_parameter_service.py`), and how console/API layers expose the flows.
-
-______________________________________________________________________
-
-## Workflow Entry & Execution
-
-- **[Trigger Concepts](agent_skills/trigger.md)**\
-  When to read this:
-  - You’re debugging why a workflow didn’t start.
-  - You’re adding a new trigger type or hook.
-  - You need to trace async execution, draft debugging, or webhook/schedule pipelines.\
-    Details: Start-node taxonomy, webhook & schedule internals (`core/workflow/nodes/trigger_*`, `services/trigger/*`), async orchestration (`services/async_workflow_service.py`, Celery queues), debug event bus, and storage/logging interactions.
-
-______________________________________________________________________
-
-## Additional Notes for Agents
-
-- All skill docs assume you follow the coding style guide—run Ruff/BasedPyright/tests listed there before submitting changes.
-- When you cannot find an answer in these briefs, search the codebase using the paths referenced (e.g., `core/plugin/impl/tool.py`, `services/dataset_service.py`).
-- If you run into cross-cutting concerns (tenancy, configuration, storage), check the infrastructure guide first; it links to most supporting modules.
-- Keep multi-tenancy and configuration central: everything flows through `configs.dify_config` and `tenant_id`.
-- When touching plugins or triggers, consult both the system overview and the specialised doc to ensure you adjust lifecycle, storage, and observability consistently.

api/Dockerfile

@@ -15,11 +15,7 @@ FROM base AS packages
 # RUN sed -i 's@deb.debian.org@mirrors.aliyun.com@g' /etc/apt/sources.list.d/debian.sources
 
 RUN apt-get update \
-    && apt-get install -y --no-install-recommends \
-          # basic environment
-          g++ \
-          # for building gmpy2
-          libmpfr-dev libmpc-dev
+    && apt-get install -y --no-install-recommends gcc g++ libc-dev libffi-dev libgmp-dev libmpfr-dev libmpc-dev
 
 # Install Python dependencies
 COPY pyproject.toml uv.lock ./
@@ -53,9 +49,7 @@ RUN \
     # Install dependencies
     && apt-get install -y --no-install-recommends \
         # basic environment
-        curl nodejs \
-        # for gmpy2 \
-        libgmp-dev libmpfr-dev libmpc-dev \
+        curl nodejs libgmp-dev libmpfr-dev libmpc-dev \
         # For Security
         expat libldap-2.5-0 perl libsqlite3-0 zlib1g \
         # install fonts to support the use of tools like pypdfium2

api/README.md

@@ -80,7 +80,7 @@
 1. If you need to handle and debug the async tasks (e.g. dataset importing and documents indexing), please start the worker service.
 
 ```bash
-uv run celery -A app.celery worker -P threads -c 2 --loglevel INFO -Q dataset,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,priority_pipeline,pipeline
+uv run celery -A app.celery worker -P gevent -c 2 --loglevel INFO -Q dataset,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,priority_pipeline,pipeline
 ```
 
 Additionally, if you want to debug the celery scheduled tasks, you can run the following command in another terminal to start the beat service:

api/agent_skills/coding_style.md

@@ -1,115 +0,0 @@
-## Linter
-
-- Always follow `.ruff.toml`.
-- Run `uv run ruff check --fix --unsafe-fixes`.
-- Keep each line under 100 characters (including spaces).
-
-## Code Style
-
-- `snake_case` for variables and functions.
-- `PascalCase` for classes.
-- `UPPER_CASE` for constants.
-
-## Rules
-
-- Use Pydantic v2 standard.
-- Use `uv` for package management.
-- Do not override dunder methods like `__init__`, `__iadd__`, etc.
-- Never launch services (`uv run app.py`, `flask run`, etc.); running tests under `tests/` is allowed.
-- Prefer simple functions over classes for lightweight helpers.
-- Keep files below 800 lines; split when necessary.
-- Keep code readable—no clever hacks.
-- Never use `print`; log with `logger = logging.getLogger(__name__)`.
-
-## Guiding Principles
-
-- Mirror the project’s layered architecture: controller → service → core/domain.
-- Reuse existing helpers in `core/`, `services/`, and `libs/` before creating new abstractions.
-- Optimise for observability: deterministic control flow, clear logging, actionable errors.
-
-## SQLAlchemy Patterns
-
-- Models inherit from `models.base.Base`; never create ad-hoc metadata or engines.
-
-- Open sessions with context managers:
-
-  ```python
-  from sqlalchemy.orm import Session
-
-  with Session(db.engine, expire_on_commit=False) as session:
-      stmt = select(Workflow).where(
-          Workflow.id == workflow_id,
-          Workflow.tenant_id == tenant_id,
-      )
-      workflow = session.execute(stmt).scalar_one_or_none()
-  ```
-
-- Use SQLAlchemy expressions; avoid raw SQL unless necessary.
-
-- Introduce repository abstractions only for very large tables (e.g., workflow executions) to support alternative storage strategies.
-
-- Always scope queries by `tenant_id` and protect write paths with safeguards (`FOR UPDATE`, row counts, etc.).
-
-## Storage & External IO
-
-- Access storage via `extensions.ext_storage.storage`.
-- Use `core.helper.ssrf_proxy` for outbound HTTP fetches.
-- Background tasks that touch storage must be idempotent and log the relevant object identifiers.
-
-## Pydantic Usage
-
-- Define DTOs with Pydantic v2 models and forbid extras by default.
-
-- Use `@field_validator` / `@model_validator` for domain rules.
-
-- Example:
-
-  ```python
-  from pydantic import BaseModel, ConfigDict, HttpUrl, field_validator
-
-  class TriggerConfig(BaseModel):
-      endpoint: HttpUrl
-      secret: str
-
-      model_config = ConfigDict(extra="forbid")
-
-      @field_validator("secret")
-      def ensure_secret_prefix(cls, value: str) -> str:
-          if not value.startswith("dify_"):
-              raise ValueError("secret must start with dify_")
-          return value
-  ```
-
-## Generics & Protocols
-
-- Use `typing.Protocol` to define behavioural contracts (e.g., cache interfaces).
-- Apply generics (`TypeVar`, `Generic`) for reusable utilities like caches or providers.
-- Validate dynamic inputs at runtime when generics cannot enforce safety alone.
-
-## Error Handling & Logging
-
-- Raise domain-specific exceptions (`services/errors`, `core/errors`) and translate to HTTP responses in controllers.
-- Declare `logger = logging.getLogger(__name__)` at module top.
-- Include tenant/app/workflow identifiers in log context.
-- Log retryable events at `warning`, terminal failures at `error`.
-
-## Tooling & Checks
-
-- Format/lint: `uv run --project api --dev ruff format ./api` and `uv run --project api --dev ruff check --fix --unsafe-fixes ./api`.
-- Type checks: `uv run --directory api --dev basedpyright`.
-- Tests: `uv run --project api --dev dev/pytest/pytest_unit_tests.sh`.
-- Run all of the above before submitting your work.
-
-## Controllers & Services
-
-- Controllers: parse input via Pydantic, invoke services, return serialised responses; no business logic.
-- Services: coordinate repositories, providers, background tasks; keep side effects explicit.
-- Avoid repositories unless necessary; direct SQLAlchemy usage is preferred for typical tables.
-- Document non-obvious behaviour with concise comments.
-
-## Miscellaneous
-
-- Use `configs.dify_config` for configuration—never read environment variables directly.
-- Maintain tenant awareness end-to-end; `tenant_id` must flow through every layer touching shared resources.
-- Queue async work through `services/async_workflow_service`; implement tasks under `tasks/` with explicit queue selection.
-- Keep experimental scripts under `dev/`; do not ship them in production builds.

api/agent_skills/infra.md

@@ -1,96 +0,0 @@
-## Configuration
-
-- Import `configs.dify_config` for every runtime toggle. Do not read environment variables directly.
-- Add new settings to the proper mixin inside `configs/` (deployment, feature, middleware, etc.) so they load through `DifyConfig`.
-- Remote overrides come from the optional providers in `configs/remote_settings_sources`; keep defaults in code safe when the value is missing.
-- Example: logging pulls targets from `extensions/ext_logging.py`, and model provider URLs are assembled in `services/entities/model_provider_entities.py`.
-
-## Dependencies
-
-- Runtime dependencies live in `[project].dependencies` inside `pyproject.toml`. Optional clients go into the `storage`, `tools`, or `vdb` groups under `[dependency-groups]`.
-- Always pin versions and keep the list alphabetised. Shared tooling (lint, typing, pytest) belongs in the `dev` group.
-- When code needs a new package, explain why in the PR and run `uv lock` so the lockfile stays current.
-
-## Storage & Files
-
-- Use `extensions.ext_storage.storage` for all blob IO; it already respects the configured backend.
-- Convert files for workflows with helpers in `core/file/file_manager.py`; they handle signed URLs and multimodal payloads.
-- When writing controller logic, delegate upload quotas and metadata to `services/file_service.py` instead of touching storage directly.
-- All outbound HTTP fetches (webhooks, remote files) must go through the SSRF-safe client in `core/helper/ssrf_proxy.py`; it wraps `httpx` with the allow/deny rules configured for the platform.
-
-## Redis & Shared State
-
-- Access Redis through `extensions.ext_redis.redis_client`. For locking, reuse `redis_client.lock`.
-- Prefer higher-level helpers when available: rate limits use `libs.helper.RateLimiter`, provider metadata uses caches in `core/helper/provider_cache.py`.
-
-## Models
-
-- SQLAlchemy models sit in `models/` and inherit from the shared declarative `Base` defined in `models/base.py` (metadata configured via `models/engine.py`).
-- `models/__init__.py` exposes grouped aggregates: account/tenant models, app and conversation tables, datasets, providers, workflow runs, triggers, etc. Import from there to avoid deep path churn.
-- Follow the DDD boundary: persistence objects live in `models/`, repositories under `repositories/` translate them into domain entities, and services consume those repositories.
-- When adding a table, create the model class, register it in `models/__init__.py`, wire a repository if needed, and generate an Alembic migration as described below.
-
-## Vector Stores
-
-- Vector client implementations live in `core/rag/datasource/vdb/<provider>`, with a common factory in `core/rag/datasource/vdb/vector_factory.py` and enums in `core/rag/datasource/vdb/vector_type.py`.
-- Retrieval pipelines call these providers through `core/rag/datasource/retrieval_service.py` and dataset ingestion flows in `services/dataset_service.py`.
-- The CLI helper `flask vdb-migrate` orchestrates bulk migrations using routines in `commands.py`; reuse that pattern when adding new backend transitions.
-- To add another store, mirror the provider layout, register it with the factory, and include any schema changes in Alembic migrations.
-
-## Observability & OTEL
-
-- OpenTelemetry settings live under the observability mixin in `configs/observability`. Toggle exporters and sampling via `dify_config`, not ad-hoc env reads.
-- HTTP, Celery, Redis, SQLAlchemy, and httpx instrumentation is initialised in `extensions/ext_app_metrics.py` and `extensions/ext_request_logging.py`; reuse these hooks when adding new workers or entrypoints.
-- When creating background tasks or external calls, propagate tracing context with helpers in the existing instrumented clients (e.g. use the shared `httpx` session from `core/helper/http_client_pooling.py`).
-- If you add a new external integration, ensure spans and metrics are emitted by wiring the appropriate OTEL instrumentation package in `pyproject.toml` and configuring it in `extensions/`.
-
-## Ops Integrations
-
-- Langfuse support and other tracing bridges live under `core/ops/opik_trace`. Config toggles sit in `configs/observability`, while exporters are initialised in the OTEL extensions mentioned above.
-- External monitoring services should follow this pattern: keep client code in `core/ops`, expose switches via `dify_config`, and hook initialisation in `extensions/ext_app_metrics.py` or sibling modules.
-- Before instrumenting new code paths, check whether existing context helpers (e.g. `extensions/ext_request_logging.py`) already capture the necessary metadata.
-
-## Controllers, Services, Core
-
-- Controllers only parse HTTP input and call a service method. Keep business rules in `services/`.
-- Services enforce tenant rules, quotas, and orchestration, then call into `core/` engines (workflow execution, tools, LLMs).
-- When adding a new endpoint, search for an existing service to extend before introducing a new layer. Example: workflow APIs pipe through `services/workflow_service.py` into `core/workflow`.
-
-## Plugins, Tools, Providers
-
-- In Dify a plugin is a tenant-installable bundle that declares one or more providers (tool, model, datasource, trigger, endpoint, agent strategy) plus its resource needs and version metadata. The manifest (`core/plugin/entities/plugin.py`) mirrors what you see in the marketplace documentation.
-- Installation, upgrades, and migrations are orchestrated by `services/plugin/plugin_service.py` together with helpers such as `services/plugin/plugin_migration.py`.
-- Runtime loading happens through the implementations under `core/plugin/impl/*` (tool/model/datasource/trigger/endpoint/agent). These modules normalise plugin providers so that downstream systems (`core/tools/tool_manager.py`, `services/model_provider_service.py`, `services/trigger/*`) can treat builtin and plugin capabilities the same way.
-- For remote execution, plugin daemons (`core/plugin/entities/plugin_daemon.py`, `core/plugin/impl/plugin.py`) manage lifecycle hooks, credential forwarding, and background workers that keep plugin processes in sync with the main application.
-- Acquire tool implementations through `core/tools/tool_manager.py`; it resolves builtin, plugin, and workflow-as-tool providers uniformly, injecting the right context (tenant, credentials, runtime config).
-- To add a new plugin capability, extend the relevant `core/plugin/entities` schema and register the implementation in the matching `core/plugin/impl` module rather than importing the provider directly.
-
-## Async Workloads
-
-see `agent_skills/trigger.md` for more detailed documentation.
-
-- Enqueue background work through `services/async_workflow_service.py`. It routes jobs to the tiered Celery queues defined in `tasks/`.
-- Workers boot from `celery_entrypoint.py` and execute functions in `tasks/workflow_execution_tasks.py`, `tasks/trigger_processing_tasks.py`, etc.
-- Scheduled workflows poll from `schedule/workflow_schedule_tasks.py`. Follow the same pattern if you need new periodic jobs.
-
-## Database & Migrations
-
-- SQLAlchemy models live under `models/` and map directly to migration files in `migrations/versions`.
-- Generate migrations with `uv run --project api flask db revision --autogenerate -m "<summary>"`, then review the diff; never hand-edit the database outside Alembic.
-- Apply migrations locally using `uv run --project api flask db upgrade`; production deploys expect the same history.
-- If you add tenant-scoped data, confirm the upgrade includes tenant filters or defaults consistent with the service logic touching those tables.
-
-## CLI Commands
-
-- Maintenance commands from `commands.py` are registered on the Flask CLI. Run them via `uv run --project api flask <command>`.
-- Use the built-in `db` commands from Flask-Migrate for schema operations (`flask db upgrade`, `flask db stamp`, etc.). Only fall back to custom helpers if you need their extra behaviour.
-- Custom entries such as `flask reset-password`, `flask reset-email`, and `flask vdb-migrate` handle self-hosted account recovery and vector database migrations.
-- Before adding a new command, check whether an existing service can be reused and ensure the command guards edition-specific behaviour (many enforce `SELF_HOSTED`). Document any additions in the PR.
-- Ruff helpers are run directly with `uv`: `uv run --project api --dev ruff format ./api` for formatting and `uv run --project api --dev ruff check ./api` (add `--fix` if you want automatic fixes).
-
-## When You Add Features
-
-- Check for an existing helper or service before writing a new util.
-- Uphold tenancy: every service method should receive the tenant ID from controller wrappers such as `controllers/console/wraps.py`.
-- Update or create tests alongside behaviour changes (`tests/unit_tests` for fast coverage, `tests/integration_tests` when touching orchestrations).
-- Run `uv run --project api --dev ruff check ./api`, `uv run --directory api --dev basedpyright`, and `uv run --project api --dev dev/pytest/pytest_unit_tests.sh` before submitting changes.

api/agent_skills/plugin.md

@@ -1 +0,0 @@
-// TBD

api/agent_skills/plugin_oauth.md

@@ -1 +0,0 @@
-// TBD

api/agent_skills/trigger.md

@@ -1,53 +0,0 @@
-## Overview
-
-Trigger is a collection of nodes that we called `Start` nodes, also, the concept of `Start` is the same as `RootNode` in the workflow engine `core/workflow/graph_engine`, On the other hand, `Start` node is the entry point of workflows, every workflow run always starts from a `Start` node.
-
-## Trigger nodes
-
-- `UserInput`
-- `Trigger Webhook`
-- `Trigger Schedule`
-- `Trigger Plugin`
-
-### UserInput
-
-Before `Trigger` concept is introduced, it's what we called `Start` node, but now, to avoid confusion, it was renamed to `UserInput` node, has a strong relation with `ServiceAPI` in `controllers/service_api/app`
-
-1. `UserInput` node introduces a list of arguments that need to be provided by the user, finally it will be converted into variables in the workflow variable pool.
-1. `ServiceAPI` accept those arguments, and pass through them into `UserInput` node.
-1. For its detailed implementation, please refer to `core/workflow/nodes/start`
-
-### Trigger Webhook
-
-Inside Webhook Node, Dify provided a UI panel that allows user define a HTTP manifest `core/workflow/nodes/trigger_webhook/entities.py`.`WebhookData`, also, Dify generates a random webhook id for each `Trigger Webhook` node, the implementation was implemented in `core/trigger/utils/endpoint.py`, as you can see, `webhook-debug` is a debug mode for webhook, you may find it in `controllers/trigger/webhook.py`.
-
-Finally, requests to `webhook` endpoint will be converted into variables in workflow variable pool during workflow execution.
-
-### Trigger Schedule
-
-`Trigger Schedule` node is a node that allows user define a schedule to trigger the workflow, detailed manifest is here `core/workflow/nodes/trigger_schedule/entities.py`, we have a poller and executor to handle millions of schedules, see `docker/entrypoint.sh` / `schedule/workflow_schedule_task.py` for help.
-
-To Achieve this, a `WorkflowSchedulePlan` model was introduced in `models/trigger.py`, and a `events/event_handlers/sync_workflow_schedule_when_app_published.py` was used to sync workflow schedule plans when app is published.
-
-### Trigger Plugin
-
-`Trigger Plugin` node allows user define there own distributed trigger plugin, whenever a request was received, Dify forwards it to the plugin and wait for parsed variables from it.
-
-1. Requests were saved in storage by `services/trigger/trigger_request_service.py`, referenced by `services/trigger/trigger_service.py`.`TriggerService`.`process_endpoint`
-1. Plugins accept those requests and parse variables from it, see `core/plugin/impl/trigger.py` for details.
-
-A `subscription` concept was out here by Dify, it means an endpoint address from Dify was bound to thirdparty webhook service like `Github` `Slack` `Linear` `GoogleDrive` `Gmail` etc. Once a subscription was created, Dify continually receives requests from the platforms and handle them one by one.
-
-## Worker Pool / Async Task
-
-All the events that triggered a new workflow run is always in async mode, a unified entrypoint can be found here `services/async_workflow_service.py`.`AsyncWorkflowService`.`trigger_workflow_async`.
-
-The infrastructure we used is `celery`, we've already configured it in `docker/entrypoint.sh`, and the consumers are in `tasks/async_workflow_tasks.py`, 3 queues were used to handle different tiers of users, `PROFESSIONAL_QUEUE` `TEAM_QUEUE` `SANDBOX_QUEUE`.
-
-## Debug Strategy
-
-Dify divided users into 2 groups: builders / end users.
-
-Builders are the users who create workflows, in this stage, debugging a workflow becomes a critical part of the workflow development process, as the start node in workflows, trigger nodes can `listen` to the events from `WebhookDebug` `Schedule` `Plugin`, debugging process was created in `controllers/console/app/workflow.py`.`DraftWorkflowTriggerNodeApi`.
-
-A polling process can be considered as combine of few single `poll` operations, each `poll` operation fetches events cached in `Redis`, returns `None` if no event was found, more detailed implemented: `core/trigger/debug/event_bus.py` was used to handle the polling process, and `core/trigger/debug/event_selectors.py` was used to select the event poller based on the trigger type.

api/app.py

@@ -1,7 +1,7 @@
 import sys
 
 
-def is_db_command() -> bool:
+def is_db_command():
     if len(sys.argv) > 1 and sys.argv[0].endswith("flask") and sys.argv[1] == "db":
         return True
     return False
@@ -13,12 +13,23 @@ if is_db_command():
 
     app = create_migrations_app()
 else:
-    # Gunicorn and Celery handle monkey patching automatically in production by
-    # specifying the `gevent` worker class. Manual monkey patching is not required here.
+    # It seems that JetBrains Python debugger does not work well with gevent,
+    # so we need to disable gevent in debug mode.
+    # If you are using debugpy and set GEVENT_SUPPORT=True, you can debug with gevent.
+    # if (flask_debug := os.environ.get("FLASK_DEBUG", "0")) and flask_debug.lower() in {"false", "0", "no"}:
+    # from gevent import monkey
     #
-    # See `api/docker/entrypoint.sh` (lines 33 and 47) for details.
+    # # gevent
+    # monkey.patch_all()
     #
-    # For third-party library patching, refer to `gunicorn.conf.py` and `celery_entrypoint.py`.
+    # from grpc.experimental import gevent as grpc_gevent  # type: ignore
+    #
+    # # grpc gevent
+    # grpc_gevent.init_gevent()
+
+    # import psycogreen.gevent  # type: ignore
+    #
+    # psycogreen.gevent.patch_psycopg()
 
     from app_factory import create_app
 

api/commands.py

@@ -15,12 +15,12 @@ from sqlalchemy.orm import sessionmaker
 from configs import dify_config
 from constants.languages import languages
 from core.helper import encrypter
-from core.plugin.entities.plugin_daemon import CredentialType
 from core.plugin.impl.plugin import PluginInstaller
 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
 from core.rag.models.document import Document
+from core.tools.entities.tool_entities import CredentialType
 from core.tools.utils.system_oauth_encryption import encrypt_system_oauth_params
 from events.app_event import app_was_created
 from extensions.ext_database import db
@@ -1229,55 +1229,6 @@ def setup_system_tool_oauth_client(provider, client_params):
     click.echo(click.style(f"OAuth client params setup successfully. id: {oauth_client.id}", fg="green"))
 
 
-@click.command("setup-system-trigger-oauth-client", help="Setup system trigger oauth client.")
-@click.option("--provider", prompt=True, help="Provider name")
-@click.option("--client-params", prompt=True, help="Client Params")
-def setup_system_trigger_oauth_client(provider, client_params):
-    """
-    Setup system trigger oauth client
-    """
-    from models.provider_ids import TriggerProviderID
-    from models.trigger import TriggerOAuthSystemClient
-
-    provider_id = TriggerProviderID(provider)
-    provider_name = provider_id.provider_name
-    plugin_id = provider_id.plugin_id
-
-    try:
-        # json validate
-        click.echo(click.style(f"Validating client params: {client_params}", fg="yellow"))
-        client_params_dict = TypeAdapter(dict[str, Any]).validate_json(client_params)
-        click.echo(click.style("Client params validated successfully.", fg="green"))
-
-        click.echo(click.style(f"Encrypting client params: {client_params}", fg="yellow"))
-        click.echo(click.style(f"Using SECRET_KEY: `{dify_config.SECRET_KEY}`", fg="yellow"))
-        oauth_client_params = encrypt_system_oauth_params(client_params_dict)
-        click.echo(click.style("Client params encrypted successfully.", fg="green"))
-    except Exception as e:
-        click.echo(click.style(f"Error parsing client params: {str(e)}", fg="red"))
-        return
-
-    deleted_count = (
-        db.session.query(TriggerOAuthSystemClient)
-        .filter_by(
-            provider=provider_name,
-            plugin_id=plugin_id,
-        )
-        .delete()
-    )
-    if deleted_count > 0:
-        click.echo(click.style(f"Deleted {deleted_count} existing oauth client params.", fg="yellow"))
-
-    oauth_client = TriggerOAuthSystemClient(
-        provider=provider_name,
-        plugin_id=plugin_id,
-        encrypted_oauth_params=oauth_client_params,
-    )
-    db.session.add(oauth_client)
-    db.session.commit()
-    click.echo(click.style(f"OAuth client params setup successfully. id: {oauth_client.id}", fg="green"))
-
-
 def _find_orphaned_draft_variables(batch_size: int = 1000) -> list[str]:
     """
     Find draft variables that reference non-existent apps.
@@ -1471,10 +1422,7 @@ def setup_datasource_oauth_client(provider, client_params):
 
 
 @click.command("transform-datasource-credentials", help="Transform datasource credentials.")
-@click.option(
-    "--environment", prompt=True, help="the environment to transform datasource credentials", default="online"
-)
-def transform_datasource_credentials(environment: str):
+def transform_datasource_credentials():
     """
     Transform datasource credentials
     """
@@ -1485,14 +1433,9 @@ def transform_datasource_credentials(environment: str):
         notion_plugin_id = "langgenius/notion_datasource"
         firecrawl_plugin_id = "langgenius/firecrawl_datasource"
         jina_plugin_id = "langgenius/jina_datasource"
-        if environment == "online":
-            notion_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(notion_plugin_id)  # pyright: ignore[reportPrivateUsage]
-            firecrawl_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(firecrawl_plugin_id)  # pyright: ignore[reportPrivateUsage]
-            jina_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(jina_plugin_id)  # pyright: ignore[reportPrivateUsage]
-        else:
-            notion_plugin_unique_identifier = None
-            firecrawl_plugin_unique_identifier = None
-            jina_plugin_unique_identifier = None
+        notion_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(notion_plugin_id)  # pyright: ignore[reportPrivateUsage]
+        firecrawl_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(firecrawl_plugin_id)  # pyright: ignore[reportPrivateUsage]
+        jina_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(jina_plugin_id)  # pyright: ignore[reportPrivateUsage]
         oauth_credential_type = CredentialType.OAUTH2
         api_key_credential_type = CredentialType.API_KEY
 
@@ -1658,7 +1601,7 @@ def transform_datasource_credentials(environment: str):
                             "integration_secret": api_key,
                         }
                         datasource_provider = DatasourceProvider(
-                            provider="jinareader",
+                            provider="jina",
                             tenant_id=tenant_id,
                             plugin_id=jina_plugin_id,
                             auth_type=api_key_credential_type.value,

api/configs/feature/__init__.py

@@ -174,33 +174,6 @@ class CodeExecutionSandboxConfig(BaseSettings):
     )
 
 
-class TriggerConfig(BaseSettings):
-    """
-    Configuration for trigger
-    """
-
-    WEBHOOK_REQUEST_BODY_MAX_SIZE: PositiveInt = Field(
-        description="Maximum allowed size for webhook request bodies in bytes",
-        default=10485760,
-    )
-
-
-class AsyncWorkflowConfig(BaseSettings):
-    """
-    Configuration for async workflow
-    """
-
-    ASYNC_WORKFLOW_SCHEDULER_GRANULARITY: int = Field(
-        description="Granularity for async workflow scheduler, "
-        "sometime, few users could block the queue due to some time-consuming tasks, "
-        "to avoid this, workflow can be suspended if needed, to achieve"
-        "this, a time-based checker is required, every granularity seconds, "
-        "the checker will check the workflow queue and suspend the workflow",
-        default=120,
-        ge=1,
-    )
-
-
 class PluginConfig(BaseSettings):
     """
     Plugin configs
@@ -290,8 +263,6 @@ class EndpointConfig(BaseSettings):
         description="Template url for endpoint plugin", default="http://localhost:5002/e/{hook_id}"
     )
 
-    TRIGGER_URL: str = Field(description="Template url for triggers", default="http://localhost:5001")
-
 
 class FileAccessConfig(BaseSettings):
     """
@@ -360,31 +331,6 @@ class FileUploadConfig(BaseSettings):
         default=10,
     )
 
-    inner_UPLOAD_FILE_EXTENSION_BLACKLIST: str = Field(
-        description=(
-            "Comma-separated list of file extensions that are blocked from upload. "
-            "Extensions should be lowercase without dots (e.g., 'exe,bat,sh,dll'). "
-            "Empty by default to allow all file types."
-        ),
-        validation_alias=AliasChoices("UPLOAD_FILE_EXTENSION_BLACKLIST"),
-        default="",
-    )
-
-    @computed_field  # type: ignore[misc]
-    @property
-    def UPLOAD_FILE_EXTENSION_BLACKLIST(self) -> set[str]:
-        """
-        Parse and return the blacklist as a set of lowercase extensions.
-        Returns an empty set if no blacklist is configured.
-        """
-        if not self.inner_UPLOAD_FILE_EXTENSION_BLACKLIST:
-            return set()
-        return {
-            ext.strip().lower().strip(".")
-            for ext in self.inner_UPLOAD_FILE_EXTENSION_BLACKLIST.split(",")
-            if ext.strip()
-        }
-
 
 class HttpConfig(BaseSettings):
     """
@@ -974,11 +920,6 @@ class DataSetConfig(BaseSettings):
         default=True,
     )
 
-    DATASET_MAX_SEGMENTS_PER_REQUEST: NonNegativeInt = Field(
-        description="Maximum number of segments for dataset segments API (0 for unlimited)",
-        default=0,
-    )
-
 
 class WorkspaceConfig(BaseSettings):
     """
@@ -1054,44 +995,6 @@ class CeleryScheduleTasksConfig(BaseSettings):
         description="Enable check upgradable plugin task",
         default=True,
     )
-    ENABLE_WORKFLOW_SCHEDULE_POLLER_TASK: bool = Field(
-        description="Enable workflow schedule poller task",
-        default=True,
-    )
-    WORKFLOW_SCHEDULE_POLLER_INTERVAL: int = Field(
-        description="Workflow schedule poller interval in minutes",
-        default=1,
-    )
-    WORKFLOW_SCHEDULE_POLLER_BATCH_SIZE: int = Field(
-        description="Maximum number of schedules to process in each poll batch",
-        default=100,
-    )
-    WORKFLOW_SCHEDULE_MAX_DISPATCH_PER_TICK: int = Field(
-        description="Maximum schedules to dispatch per tick (0=unlimited, circuit breaker)",
-        default=0,
-    )
-
-    # Trigger provider refresh (simple version)
-    ENABLE_TRIGGER_PROVIDER_REFRESH_TASK: bool = Field(
-        description="Enable trigger provider refresh poller",
-        default=True,
-    )
-    TRIGGER_PROVIDER_REFRESH_INTERVAL: int = Field(
-        description="Trigger provider refresh poller interval in minutes",
-        default=1,
-    )
-    TRIGGER_PROVIDER_REFRESH_BATCH_SIZE: int = Field(
-        description="Max trigger subscriptions to process per tick",
-        default=200,
-    )
-    TRIGGER_PROVIDER_CREDENTIAL_THRESHOLD_SECONDS: int = Field(
-        description="Proactive credential refresh threshold in seconds",
-        default=180,
-    )
-    TRIGGER_PROVIDER_SUBSCRIPTION_THRESHOLD_SECONDS: int = Field(
-        description="Proactive subscription refresh threshold in seconds",
-        default=60 * 60,
-    )
 
 
 class PositionConfig(BaseSettings):
@@ -1190,7 +1093,7 @@ class AccountConfig(BaseSettings):
 
 
 class WorkflowLogConfig(BaseSettings):
-    WORKFLOW_LOG_CLEANUP_ENABLED: bool = Field(default=False, description="Enable workflow run log cleanup")
+    WORKFLOW_LOG_CLEANUP_ENABLED: bool = Field(default=True, description="Enable workflow run log cleanup")
     WORKFLOW_LOG_RETENTION_DAYS: int = Field(default=30, description="Retention days for workflow run logs")
     WORKFLOW_LOG_CLEANUP_BATCH_SIZE: int = Field(
         default=100, description="Batch size for workflow run log cleanup operations"
@@ -1209,21 +1112,12 @@ class SwaggerUIConfig(BaseSettings):
     )
 
 
-class TenantIsolatedTaskQueueConfig(BaseSettings):
-    TENANT_ISOLATED_TASK_CONCURRENCY: int = Field(
-        description="Number of tasks allowed to be delivered concurrently from isolated queue per tenant",
-        default=1,
-    )
-
-
 class FeatureConfig(
     # place the configs in alphabet order
     AppExecutionConfig,
     AuthConfig,  # Changed from OAuthConfig to AuthConfig
     BillingConfig,
     CodeExecutionSandboxConfig,
-    TriggerConfig,
-    AsyncWorkflowConfig,
     PluginConfig,
     MarketplaceConfig,
     DataSetConfig,
@@ -1242,7 +1136,6 @@ class FeatureConfig(
     RagEtlConfig,
     RepositoryConfig,
     SecurityConfig,
-    TenantIsolatedTaskQueueConfig,
     ToolConfig,
     UpdateConfig,
     WorkflowConfig,

api/configs/middleware/vdb/weaviate_config.py

@@ -22,11 +22,6 @@ class WeaviateConfig(BaseSettings):
         default=True,
     )
 
-    WEAVIATE_GRPC_ENDPOINT: str | None = Field(
-        description="URL of the Weaviate gRPC server (e.g., 'grpc://localhost:50051' or 'grpcs://weaviate.example.com:443')",
-        default=None,
-    )
-
     WEAVIATE_BATCH_SIZE: PositiveInt = Field(
         description="Number of objects to be processed in a single batch operation (default is 100)",
         default=100,

api/contexts/__init__.py

@@ -9,7 +9,6 @@ if TYPE_CHECKING:
     from core.model_runtime.entities.model_entities import AIModelEntity
     from core.plugin.entities.plugin_daemon import PluginModelProviderEntity
     from core.tools.plugin_tool.provider import PluginToolProviderController
-    from core.trigger.provider import PluginTriggerProviderController
 
 
 """
@@ -42,11 +41,3 @@ datasource_plugin_providers: RecyclableContextVar[dict[str, "DatasourcePluginPro
 datasource_plugin_providers_lock: RecyclableContextVar[Lock] = RecyclableContextVar(
     ContextVar("datasource_plugin_providers_lock")
 )
-
-plugin_trigger_providers: RecyclableContextVar[dict[str, "PluginTriggerProviderController"]] = RecyclableContextVar(
-    ContextVar("plugin_trigger_providers")
-)
-
-plugin_trigger_providers_lock: RecyclableContextVar[Lock] = RecyclableContextVar(
-    ContextVar("plugin_trigger_providers_lock")
-)

api/controllers/common/errors.py

@@ -25,12 +25,6 @@ class UnsupportedFileTypeError(BaseHTTPException):
     code = 415
 
 
-class BlockedFileExtensionError(BaseHTTPException):
-    error_code = "file_extension_blocked"
-    description = "The file extension is blocked for security reasons."
-    code = 400
-
-
 class TooManyFilesError(BaseHTTPException):
     error_code = "too_many_files"
     description = "Only one file is allowed."

api/controllers/console/__init__.py

@@ -66,7 +66,6 @@ from .app import (
     workflow_draft_variable,
     workflow_run,
     workflow_statistic,
-    workflow_trigger,
 )
 
 # Import auth controllers
@@ -127,7 +126,6 @@ from .workspace import (
     models,
     plugin,
     tool_providers,
-    trigger_providers,
     workspace,
 )
 
@@ -198,7 +196,6 @@ __all__ = [
     "statistic",
     "tags",
     "tool_providers",
-    "trigger_providers",
     "version",
     "website",
     "workflow",
@@ -206,6 +203,5 @@ __all__ = [
     "workflow_draft_variable",
     "workflow_run",
     "workflow_statistic",
-    "workflow_trigger",
     "workspace",
 ]

api/controllers/console/app/annotation.py

@@ -16,7 +16,6 @@ from fields.annotation_fields import (
     annotation_fields,
     annotation_hit_history_fields,
 )
-from libs.helper import uuid_value
 from libs.login import login_required
 from services.annotation_service import AppAnnotationService
 
@@ -176,10 +175,8 @@ class AnnotationApi(Resource):
         api.model(
             "CreateAnnotationRequest",
             {
-                "message_id": fields.String(description="Message ID (optional)"),
-                "question": fields.String(description="Question text (required when message_id not provided)"),
-                "answer": fields.String(description="Answer text (use 'answer' or 'content')"),
-                "content": fields.String(description="Content text (use 'answer' or 'content')"),
+                "question": fields.String(required=True, description="Question text"),
+                "answer": fields.String(required=True, description="Answer text"),
                 "annotation_reply": fields.Raw(description="Annotation reply data"),
             },
         )
@@ -196,14 +193,11 @@ class AnnotationApi(Resource):
         app_id = str(app_id)
         parser = (
             reqparse.RequestParser()
-            .add_argument("message_id", required=False, type=uuid_value, location="json")
-            .add_argument("question", required=False, type=str, location="json")
-            .add_argument("answer", required=False, type=str, location="json")
-            .add_argument("content", required=False, type=str, location="json")
-            .add_argument("annotation_reply", required=False, type=dict, location="json")
+            .add_argument("question", required=True, type=str, location="json")
+            .add_argument("answer", required=True, type=str, location="json")
         )
         args = parser.parse_args()
-        annotation = AppAnnotationService.up_insert_app_annotation_from_message(args, app_id)
+        annotation = AppAnnotationService.insert_app_annotation_directly(args, app_id)
         return annotation
 
     @setup_required

api/controllers/console/app/conversation.py

@@ -1,5 +1,7 @@
+from datetime import datetime
+
+import pytz
 import sqlalchemy as sa
-from flask import abort
 from flask_restx import Resource, marshal_with, reqparse
 from flask_restx.inputs import int_range
 from sqlalchemy import func, or_
@@ -17,7 +19,7 @@ from fields.conversation_fields import (
     conversation_pagination_fields,
     conversation_with_summary_pagination_fields,
 )
-from libs.datetime_utils import naive_utc_now, parse_time_range
+from libs.datetime_utils import naive_utc_now
 from libs.helper import DatetimeString
 from libs.login import current_account_with_tenant, login_required
 from models import Conversation, EndUser, Message, MessageAnnotation
@@ -88,17 +90,25 @@ class CompletionConversationApi(Resource):
 
         account = current_user
         assert account.timezone is not None
+        timezone = pytz.timezone(account.timezone)
+        utc_timezone = pytz.utc
 
-        try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
-        except ValueError as e:
-            abort(400, description=str(e))
+        if args["start"]:
+            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
+            start_datetime = start_datetime.replace(second=0)
+
+            start_datetime_timezone = timezone.localize(start_datetime)
+            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
 
-        if start_datetime_utc:
             query = query.where(Conversation.created_at >= start_datetime_utc)
 
-        if end_datetime_utc:
-            end_datetime_utc = end_datetime_utc.replace(second=59)
+        if args["end"]:
+            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
+            end_datetime = end_datetime.replace(second=59)
+
+            end_datetime_timezone = timezone.localize(end_datetime)
+            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
+
             query = query.where(Conversation.created_at < end_datetime_utc)
 
         # FIXME, the type ignore in this file
@@ -260,21 +270,29 @@ class ChatConversationApi(Resource):
 
         account = current_user
         assert account.timezone is not None
+        timezone = pytz.timezone(account.timezone)
+        utc_timezone = pytz.utc
 
-        try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
-        except ValueError as e:
-            abort(400, description=str(e))
+        if args["start"]:
+            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
+            start_datetime = start_datetime.replace(second=0)
+
+            start_datetime_timezone = timezone.localize(start_datetime)
+            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
 
-        if start_datetime_utc:
             match args["sort_by"]:
                 case "updated_at" | "-updated_at":
                     query = query.where(Conversation.updated_at >= start_datetime_utc)
                 case "created_at" | "-created_at" | _:
                     query = query.where(Conversation.created_at >= start_datetime_utc)
 
-        if end_datetime_utc:
-            end_datetime_utc = end_datetime_utc.replace(second=59)
+        if args["end"]:
+            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
+            end_datetime = end_datetime.replace(second=59)
+
+            end_datetime_timezone = timezone.localize(end_datetime)
+            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
+
             match args["sort_by"]:
                 case "updated_at" | "-updated_at":
                     query = query.where(Conversation.updated_at <= end_datetime_utc)

api/controllers/console/app/generator.py

@@ -11,7 +11,6 @@ from controllers.console.app.error import (
 )
 from controllers.console.wraps import account_initialization_required, setup_required
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
-from core.helper.code_executor.code_node_provider import CodeNodeProvider
 from core.helper.code_executor.javascript.javascript_code_provider import JavascriptCodeProvider
 from core.helper.code_executor.python3.python3_code_provider import Python3CodeProvider
 from core.llm_generator.llm_generator import LLMGenerator
@@ -207,11 +206,13 @@ class InstructionGenerateApi(Resource):
         )
         args = parser.parse_args()
         _, current_tenant_id = current_account_with_tenant()
-        providers: list[type[CodeNodeProvider]] = [Python3CodeProvider, JavascriptCodeProvider]
-        code_provider: type[CodeNodeProvider] | None = next(
-            (p for p in providers if p.is_accept_language(args["language"])), None
+        code_template = (
+            Python3CodeProvider.get_default_code()
+            if args["language"] == "python"
+            else (JavascriptCodeProvider.get_default_code())
+            if args["language"] == "javascript"
+            else ""
         )
-        code_template = code_provider.get_default_code() if code_provider else ""
         try:
             # Generate from nothing for a workflow node
             if (args["current"] == code_template or args["current"] == "") and args["node_id"] != "":

api/controllers/console/app/message.py

@@ -16,6 +16,7 @@ from controllers.console.app.wraps import get_app_model
 from controllers.console.explore.error import AppSuggestedQuestionsAfterAnswerDisabledError
 from controllers.console.wraps import (
     account_initialization_required,
+    cloud_edition_billing_resource_check,
     edit_permission_required,
     setup_required,
 )
@@ -23,11 +24,12 @@ from core.app.entities.app_invoke_entities import InvokeFrom
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
 from core.model_runtime.errors.invoke import InvokeError
 from extensions.ext_database import db
-from fields.conversation_fields import message_detail_fields
+from fields.conversation_fields import annotation_fields, message_detail_fields
 from libs.helper import uuid_value
 from libs.infinite_scroll_pagination import InfiniteScrollPagination
 from libs.login import current_account_with_tenant, login_required
 from models.model import AppMode, Conversation, Message, MessageAnnotation, MessageFeedback
+from services.annotation_service import AppAnnotationService
 from services.errors.conversation import ConversationNotExistsError
 from services.errors.message import MessageNotExistsError, SuggestedQuestionsAfterAnswerDisabledError
 from services.message_service import MessageService
@@ -192,6 +194,45 @@ class MessageFeedbackApi(Resource):
         return {"result": "success"}
 
 
+@console_ns.route("/apps/<uuid:app_id>/annotations")
+class MessageAnnotationApi(Resource):
+    @api.doc("create_message_annotation")
+    @api.doc(description="Create message annotation")
+    @api.doc(params={"app_id": "Application ID"})
+    @api.expect(
+        api.model(
+            "MessageAnnotationRequest",
+            {
+                "message_id": fields.String(description="Message ID"),
+                "question": fields.String(required=True, description="Question text"),
+                "answer": fields.String(required=True, description="Answer text"),
+                "annotation_reply": fields.Raw(description="Annotation reply"),
+            },
+        )
+    )
+    @api.response(200, "Annotation created successfully", annotation_fields)
+    @api.response(403, "Insufficient permissions")
+    @marshal_with(annotation_fields)
+    @get_app_model
+    @setup_required
+    @login_required
+    @cloud_edition_billing_resource_check("annotation")
+    @account_initialization_required
+    @edit_permission_required
+    def post(self, app_model):
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("message_id", required=False, type=uuid_value, location="json")
+            .add_argument("question", required=True, type=str, location="json")
+            .add_argument("answer", required=True, type=str, location="json")
+            .add_argument("annotation_reply", required=False, type=dict, location="json")
+        )
+        args = parser.parse_args()
+        annotation = AppAnnotationService.up_insert_app_annotation_from_message(args, app_model.id)
+
+        return annotation
+
+
 @console_ns.route("/apps/<uuid:app_id>/annotations/count")
 class MessageAnnotationCountApi(Resource):
     @api.doc("get_annotation_count")

api/controllers/console/app/statistic.py

@@ -1,7 +1,9 @@
+from datetime import datetime
 from decimal import Decimal
 
+import pytz
 import sqlalchemy as sa
-from flask import abort, jsonify
+from flask import jsonify
 from flask_restx import Resource, fields, reqparse
 
 from controllers.console import api, console_ns
@@ -9,7 +11,6 @@ from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
 from core.app.entities.app_invoke_entities import InvokeFrom
 from extensions.ext_database import db
-from libs.datetime_utils import parse_time_range
 from libs.helper import DatetimeString
 from libs.login import current_account_with_tenant, login_required
 from models import AppMode, Message
@@ -55,16 +56,26 @@ WHERE
         arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
         assert account.timezone is not None
 
-        try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
-        except ValueError as e:
-            abort(400, description=str(e))
+        timezone = pytz.timezone(account.timezone)
+        utc_timezone = pytz.utc
+
+        if args["start"]:
+            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
+            start_datetime = start_datetime.replace(second=0)
+
+            start_datetime_timezone = timezone.localize(start_datetime)
+            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
 
-        if start_datetime_utc:
             sql_query += " AND created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
-        if end_datetime_utc:
+        if args["end"]:
+            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
+            end_datetime = end_datetime.replace(second=0)
+
+            end_datetime_timezone = timezone.localize(end_datetime)
+            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
+
             sql_query += " AND created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
@@ -109,11 +120,8 @@ class DailyConversationStatistic(Resource):
         )
         args = parser.parse_args()
         assert account.timezone is not None
-
-        try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
-        except ValueError as e:
-            abort(400, description=str(e))
+        timezone = pytz.timezone(account.timezone)
+        utc_timezone = pytz.utc
 
         stmt = (
             sa.select(
@@ -126,10 +134,18 @@ class DailyConversationStatistic(Resource):
             .where(Message.app_id == app_model.id, Message.invoke_from != InvokeFrom.DEBUGGER)
         )
 
-        if start_datetime_utc:
+        if args["start"]:
+            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
+            start_datetime = start_datetime.replace(second=0)
+            start_datetime_timezone = timezone.localize(start_datetime)
+            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
             stmt = stmt.where(Message.created_at >= start_datetime_utc)
 
-        if end_datetime_utc:
+        if args["end"]:
+            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
+            end_datetime = end_datetime.replace(second=0)
+            end_datetime_timezone = timezone.localize(end_datetime)
+            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
             stmt = stmt.where(Message.created_at < end_datetime_utc)
 
         stmt = stmt.group_by("date").order_by("date")
@@ -182,17 +198,26 @@ WHERE
     AND invoke_from != :invoke_from"""
         arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
         assert account.timezone is not None
+        timezone = pytz.timezone(account.timezone)
+        utc_timezone = pytz.utc
 
-        try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
-        except ValueError as e:
-            abort(400, description=str(e))
+        if args["start"]:
+            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
+            start_datetime = start_datetime.replace(second=0)
+
+            start_datetime_timezone = timezone.localize(start_datetime)
+            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
 
-        if start_datetime_utc:
             sql_query += " AND created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
-        if end_datetime_utc:
+        if args["end"]:
+            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
+            end_datetime = end_datetime.replace(second=0)
+
+            end_datetime_timezone = timezone.localize(end_datetime)
+            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
+
             sql_query += " AND created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
@@ -248,17 +273,26 @@ WHERE
     AND invoke_from != :invoke_from"""
         arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
         assert account.timezone is not None
+        timezone = pytz.timezone(account.timezone)
+        utc_timezone = pytz.utc
 
-        try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
-        except ValueError as e:
-            abort(400, description=str(e))
+        if args["start"]:
+            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
+            start_datetime = start_datetime.replace(second=0)
+
+            start_datetime_timezone = timezone.localize(start_datetime)
+            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
 
-        if start_datetime_utc:
             sql_query += " AND created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
-        if end_datetime_utc:
+        if args["end"]:
+            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
+            end_datetime = end_datetime.replace(second=0)
+
+            end_datetime_timezone = timezone.localize(end_datetime)
+            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
+
             sql_query += " AND created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
@@ -323,17 +357,26 @@ FROM
             AND m.invoke_from != :invoke_from"""
         arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
         assert account.timezone is not None
+        timezone = pytz.timezone(account.timezone)
+        utc_timezone = pytz.utc
 
-        try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
-        except ValueError as e:
-            abort(400, description=str(e))
+        if args["start"]:
+            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
+            start_datetime = start_datetime.replace(second=0)
+
+            start_datetime_timezone = timezone.localize(start_datetime)
+            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
 
-        if start_datetime_utc:
             sql_query += " AND c.created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
-        if end_datetime_utc:
+        if args["end"]:
+            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
+            end_datetime = end_datetime.replace(second=0)
+
+            end_datetime_timezone = timezone.localize(end_datetime)
+            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
+
             sql_query += " AND c.created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
@@ -403,17 +446,26 @@ WHERE
     AND m.invoke_from != :invoke_from"""
         arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
         assert account.timezone is not None
+        timezone = pytz.timezone(account.timezone)
+        utc_timezone = pytz.utc
 
-        try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
-        except ValueError as e:
-            abort(400, description=str(e))
+        if args["start"]:
+            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
+            start_datetime = start_datetime.replace(second=0)
+
+            start_datetime_timezone = timezone.localize(start_datetime)
+            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
 
-        if start_datetime_utc:
             sql_query += " AND m.created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
-        if end_datetime_utc:
+        if args["end"]:
+            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
+            end_datetime = end_datetime.replace(second=0)
+
+            end_datetime_timezone = timezone.localize(end_datetime)
+            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
+
             sql_query += " AND m.created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
@@ -473,17 +525,26 @@ WHERE
     AND invoke_from != :invoke_from"""
         arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
         assert account.timezone is not None
+        timezone = pytz.timezone(account.timezone)
+        utc_timezone = pytz.utc
 
-        try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
-        except ValueError as e:
-            abort(400, description=str(e))
+        if args["start"]:
+            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
+            start_datetime = start_datetime.replace(second=0)
+
+            start_datetime_timezone = timezone.localize(start_datetime)
+            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
 
-        if start_datetime_utc:
             sql_query += " AND created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
-        if end_datetime_utc:
+        if args["end"]:
+            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
+            end_datetime = end_datetime.replace(second=0)
+
+            end_datetime_timezone = timezone.localize(end_datetime)
+            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
+
             sql_query += " AND created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
@@ -541,17 +602,26 @@ WHERE
     AND invoke_from != :invoke_from"""
         arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
         assert account.timezone is not None
+        timezone = pytz.timezone(account.timezone)
+        utc_timezone = pytz.utc
 
-        try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
-        except ValueError as e:
-            abort(400, description=str(e))
+        if args["start"]:
+            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
+            start_datetime = start_datetime.replace(second=0)
+
+            start_datetime_timezone = timezone.localize(start_datetime)
+            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
 
-        if start_datetime_utc:
             sql_query += " AND created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
-        if end_datetime_utc:
+        if args["end"]:
+            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
+            end_datetime = end_datetime.replace(second=0)
+
+            end_datetime_timezone = timezone.localize(end_datetime)
+            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
+
             sql_query += " AND created_at < :end"
             arg_dict["end"] = end_datetime_utc
 

api/controllers/console/app/workflow.py

@@ -16,19 +16,9 @@ from controllers.console.wraps import account_initialization_required, edit_perm
 from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
 from core.app.app_config.features.file_upload.manager import FileUploadConfigManager
 from core.app.apps.base_app_queue_manager import AppQueueManager
-from core.app.apps.workflow.app_generator import SKIP_PREPARE_USER_INPUTS_KEY
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.file.models import File
 from core.helper.trace_id_helper import get_external_trace_id
-from core.model_runtime.utils.encoders import jsonable_encoder
-from core.plugin.impl.exc import PluginInvokeError
-from core.trigger.debug.event_selectors import (
-    TriggerDebugEvent,
-    TriggerDebugEventPoller,
-    create_event_poller,
-    select_trigger_debug_events,
-)
-from core.workflow.enums import NodeType
 from core.workflow.graph_engine.manager import GraphEngineManager
 from extensions.ext_database import db
 from factories import file_factory, variable_factory
@@ -47,7 +37,6 @@ from services.errors.llm import InvokeRateLimitError
 from services.workflow_service import DraftWorkflowDeletionError, WorkflowInUseError, WorkflowService
 
 logger = logging.getLogger(__name__)
-LISTENING_RETRY_IN = 2000
 
 
 # TODO(QuantumGhost): Refactor existing node run API to handle file parameter parsing
@@ -113,18 +102,7 @@ class DraftWorkflowApi(Resource):
             },
         )
     )
-    @api.response(
-        200,
-        "Draft workflow synced successfully",
-        api.model(
-            "SyncDraftWorkflowResponse",
-            {
-                "result": fields.String,
-                "hash": fields.String,
-                "updated_at": fields.String,
-            },
-        ),
-    )
+    @api.response(200, "Draft workflow synced successfully", workflow_fields)
     @api.response(400, "Invalid workflow configuration")
     @api.response(403, "Permission denied")
     @edit_permission_required
@@ -937,234 +915,3 @@ class DraftWorkflowNodeLastRunApi(Resource):
         if node_exec is None:
             raise NotFound("last run not found")
         return node_exec
-
-
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/trigger/run")
-class DraftWorkflowTriggerRunApi(Resource):
-    """
-    Full workflow debug - Polling API for trigger events
-    Path: /apps/<uuid:app_id>/workflows/draft/trigger/run
-    """
-
-    @api.doc("poll_draft_workflow_trigger_run")
-    @api.doc(description="Poll for trigger events and execute full workflow when event arrives")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.expect(
-        api.model(
-            "DraftWorkflowTriggerRunRequest",
-            {
-                "node_id": fields.String(required=True, description="Node ID"),
-            },
-        )
-    )
-    @api.response(200, "Trigger event received and workflow executed successfully")
-    @api.response(403, "Permission denied")
-    @api.response(500, "Internal server error")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.WORKFLOW])
-    @edit_permission_required
-    def post(self, app_model: App):
-        """
-        Poll for trigger events and execute full workflow when event arrives
-        """
-        current_user, _ = current_account_with_tenant()
-        parser = reqparse.RequestParser()
-        parser.add_argument("node_id", type=str, required=True, location="json", nullable=False)
-        args = parser.parse_args()
-        node_id = args["node_id"]
-        workflow_service = WorkflowService()
-        draft_workflow = workflow_service.get_draft_workflow(app_model)
-        if not draft_workflow:
-            raise ValueError("Workflow not found")
-
-        poller: TriggerDebugEventPoller = create_event_poller(
-            draft_workflow=draft_workflow,
-            tenant_id=app_model.tenant_id,
-            user_id=current_user.id,
-            app_id=app_model.id,
-            node_id=node_id,
-        )
-        event: TriggerDebugEvent | None = None
-        try:
-            event = poller.poll()
-            if not event:
-                return jsonable_encoder({"status": "waiting", "retry_in": LISTENING_RETRY_IN})
-            workflow_args = dict(event.workflow_args)
-            workflow_args[SKIP_PREPARE_USER_INPUTS_KEY] = True
-            return helper.compact_generate_response(
-                AppGenerateService.generate(
-                    app_model=app_model,
-                    user=current_user,
-                    args=workflow_args,
-                    invoke_from=InvokeFrom.DEBUGGER,
-                    streaming=True,
-                    root_node_id=node_id,
-                )
-            )
-        except InvokeRateLimitError as ex:
-            raise InvokeRateLimitHttpError(ex.description)
-        except PluginInvokeError as e:
-            return jsonable_encoder({"status": "error", "error": e.to_user_friendly_error()}), 400
-        except Exception as e:
-            logger.exception("Error polling trigger debug event")
-            raise e
-
-
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/trigger/run")
-class DraftWorkflowTriggerNodeApi(Resource):
-    """
-    Single node debug - Polling API for trigger events
-    Path: /apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/trigger/run
-    """
-
-    @api.doc("poll_draft_workflow_trigger_node")
-    @api.doc(description="Poll for trigger events and execute single node when event arrives")
-    @api.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
-    @api.response(200, "Trigger event received and node executed successfully")
-    @api.response(403, "Permission denied")
-    @api.response(500, "Internal server error")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.WORKFLOW])
-    @edit_permission_required
-    def post(self, app_model: App, node_id: str):
-        """
-        Poll for trigger events and execute single node when event arrives
-        """
-        current_user, _ = current_account_with_tenant()
-
-        workflow_service = WorkflowService()
-        draft_workflow = workflow_service.get_draft_workflow(app_model)
-        if not draft_workflow:
-            raise ValueError("Workflow not found")
-
-        node_config = draft_workflow.get_node_config_by_id(node_id=node_id)
-        if not node_config:
-            raise ValueError("Node data not found for node %s", node_id)
-        node_type: NodeType = draft_workflow.get_node_type_from_node_config(node_config)
-        event: TriggerDebugEvent | None = None
-        # for schedule trigger, when run single node, just execute directly
-        if node_type == NodeType.TRIGGER_SCHEDULE:
-            event = TriggerDebugEvent(
-                workflow_args={},
-                node_id=node_id,
-            )
-        # for other trigger types, poll for the event
-        else:
-            try:
-                poller: TriggerDebugEventPoller = create_event_poller(
-                    draft_workflow=draft_workflow,
-                    tenant_id=app_model.tenant_id,
-                    user_id=current_user.id,
-                    app_id=app_model.id,
-                    node_id=node_id,
-                )
-                event = poller.poll()
-            except PluginInvokeError as e:
-                return jsonable_encoder({"status": "error", "error": e.to_user_friendly_error()}), 400
-            except Exception as e:
-                logger.exception("Error polling trigger debug event")
-                raise e
-        if not event:
-            return jsonable_encoder({"status": "waiting", "retry_in": LISTENING_RETRY_IN})
-
-        raw_files = event.workflow_args.get("files")
-        files = _parse_file(draft_workflow, raw_files if isinstance(raw_files, list) else None)
-        try:
-            node_execution = workflow_service.run_draft_workflow_node(
-                app_model=app_model,
-                draft_workflow=draft_workflow,
-                node_id=node_id,
-                user_inputs=event.workflow_args.get("inputs") or {},
-                account=current_user,
-                query="",
-                files=files,
-            )
-            return jsonable_encoder(node_execution)
-        except Exception as e:
-            logger.exception("Error running draft workflow trigger node")
-            return jsonable_encoder(
-                {"status": "error", "error": "An unexpected error occurred while running the node."}
-            ), 400
-
-
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/trigger/run-all")
-class DraftWorkflowTriggerRunAllApi(Resource):
-    """
-    Full workflow debug - Polling API for trigger events
-    Path: /apps/<uuid:app_id>/workflows/draft/trigger/run-all
-    """
-
-    @api.doc("draft_workflow_trigger_run_all")
-    @api.doc(description="Full workflow debug when the start node is a trigger")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.expect(
-        api.model(
-            "DraftWorkflowTriggerRunAllRequest",
-            {
-                "node_ids": fields.List(fields.String, required=True, description="Node IDs"),
-            },
-        )
-    )
-    @api.response(200, "Workflow executed successfully")
-    @api.response(403, "Permission denied")
-    @api.response(500, "Internal server error")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.WORKFLOW])
-    @edit_permission_required
-    def post(self, app_model: App):
-        """
-        Full workflow debug when the start node is a trigger
-        """
-        current_user, _ = current_account_with_tenant()
-
-        parser = reqparse.RequestParser()
-        parser.add_argument("node_ids", type=list, required=True, location="json", nullable=False)
-        args = parser.parse_args()
-        node_ids = args["node_ids"]
-        workflow_service = WorkflowService()
-        draft_workflow = workflow_service.get_draft_workflow(app_model)
-        if not draft_workflow:
-            raise ValueError("Workflow not found")
-
-        try:
-            trigger_debug_event: TriggerDebugEvent | None = select_trigger_debug_events(
-                draft_workflow=draft_workflow,
-                app_model=app_model,
-                user_id=current_user.id,
-                node_ids=node_ids,
-            )
-        except PluginInvokeError as e:
-            return jsonable_encoder({"status": "error", "error": e.to_user_friendly_error()}), 400
-        except Exception as e:
-            logger.exception("Error polling trigger debug event")
-            raise e
-        if trigger_debug_event is None:
-            return jsonable_encoder({"status": "waiting", "retry_in": LISTENING_RETRY_IN})
-
-        try:
-            workflow_args = dict(trigger_debug_event.workflow_args)
-            workflow_args[SKIP_PREPARE_USER_INPUTS_KEY] = True
-            response = AppGenerateService.generate(
-                app_model=app_model,
-                user=current_user,
-                args=workflow_args,
-                invoke_from=InvokeFrom.DEBUGGER,
-                streaming=True,
-                root_node_id=trigger_debug_event.node_id,
-            )
-            return helper.compact_generate_response(response)
-        except InvokeRateLimitError as ex:
-            raise InvokeRateLimitHttpError(ex.description)
-        except Exception:
-            logger.exception("Error running draft workflow trigger run-all")
-            return jsonable_encoder(
-                {
-                    "status": "error",
-                }
-            ), 400

api/controllers/console/app/workflow_app_log.py

@@ -28,7 +28,6 @@ class WorkflowAppLogApi(Resource):
             "created_at__after": "Filter logs created after this timestamp",
             "created_by_end_user_session_id": "Filter by end user session ID",
             "created_by_account": "Filter by account",
-            "detail": "Whether to return detailed logs",
             "page": "Page number (1-99999)",
             "limit": "Number of items per page (1-100)",
         }
@@ -69,7 +68,6 @@ class WorkflowAppLogApi(Resource):
                 required=False,
                 default=None,
             )
-            .add_argument("detail", type=bool, location="args", required=False, default=False)
             .add_argument("page", type=int_range(1, 99999), default=1, location="args")
             .add_argument("limit", type=int_range(1, 100), default=20, location="args")
         )
@@ -94,7 +92,6 @@ class WorkflowAppLogApi(Resource):
                 created_at_after=args.created_at__after,
                 page=args.page,
                 limit=args.limit,
-                detail=args.detail,
                 created_by_end_user_session_id=args.created_by_end_user_session_id,
                 created_by_account=args.created_by_account,
             )

api/controllers/console/app/workflow_statistic.py

@@ -1,4 +1,7 @@
-from flask import abort, jsonify
+from datetime import datetime
+
+import pytz
+from flask import jsonify
 from flask_restx import Resource, reqparse
 from sqlalchemy.orm import sessionmaker
 
@@ -6,7 +9,6 @@ from controllers.console import api, console_ns
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
 from extensions.ext_database import db
-from libs.datetime_utils import parse_time_range
 from libs.helper import DatetimeString
 from libs.login import current_account_with_tenant, login_required
 from models.enums import WorkflowRunTriggeredFrom
@@ -41,11 +43,23 @@ class WorkflowDailyRunsStatistic(Resource):
         args = parser.parse_args()
 
         assert account.timezone is not None
+        timezone = pytz.timezone(account.timezone)
+        utc_timezone = pytz.utc
 
-        try:
-            start_date, end_date = parse_time_range(args["start"], args["end"], account.timezone)
-        except ValueError as e:
-            abort(400, description=str(e))
+        start_date = None
+        end_date = None
+
+        if args["start"]:
+            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
+            start_datetime = start_datetime.replace(second=0)
+            start_datetime_timezone = timezone.localize(start_datetime)
+            start_date = start_datetime_timezone.astimezone(utc_timezone)
+
+        if args["end"]:
+            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
+            end_datetime = end_datetime.replace(second=0)
+            end_datetime_timezone = timezone.localize(end_datetime)
+            end_date = end_datetime_timezone.astimezone(utc_timezone)
 
         response_data = self._workflow_run_repo.get_daily_runs_statistics(
             tenant_id=app_model.tenant_id,
@@ -86,11 +100,23 @@ class WorkflowDailyTerminalsStatistic(Resource):
         args = parser.parse_args()
 
         assert account.timezone is not None
+        timezone = pytz.timezone(account.timezone)
+        utc_timezone = pytz.utc
 
-        try:
-            start_date, end_date = parse_time_range(args["start"], args["end"], account.timezone)
-        except ValueError as e:
-            abort(400, description=str(e))
+        start_date = None
+        end_date = None
+
+        if args["start"]:
+            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
+            start_datetime = start_datetime.replace(second=0)
+            start_datetime_timezone = timezone.localize(start_datetime)
+            start_date = start_datetime_timezone.astimezone(utc_timezone)
+
+        if args["end"]:
+            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
+            end_datetime = end_datetime.replace(second=0)
+            end_datetime_timezone = timezone.localize(end_datetime)
+            end_date = end_datetime_timezone.astimezone(utc_timezone)
 
         response_data = self._workflow_run_repo.get_daily_terminals_statistics(
             tenant_id=app_model.tenant_id,
@@ -131,11 +157,23 @@ class WorkflowDailyTokenCostStatistic(Resource):
         args = parser.parse_args()
 
         assert account.timezone is not None
+        timezone = pytz.timezone(account.timezone)
+        utc_timezone = pytz.utc
 
-        try:
-            start_date, end_date = parse_time_range(args["start"], args["end"], account.timezone)
-        except ValueError as e:
-            abort(400, description=str(e))
+        start_date = None
+        end_date = None
+
+        if args["start"]:
+            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
+            start_datetime = start_datetime.replace(second=0)
+            start_datetime_timezone = timezone.localize(start_datetime)
+            start_date = start_datetime_timezone.astimezone(utc_timezone)
+
+        if args["end"]:
+            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
+            end_datetime = end_datetime.replace(second=0)
+            end_datetime_timezone = timezone.localize(end_datetime)
+            end_date = end_datetime_timezone.astimezone(utc_timezone)
 
         response_data = self._workflow_run_repo.get_daily_token_cost_statistics(
             tenant_id=app_model.tenant_id,
@@ -176,11 +214,23 @@ class WorkflowAverageAppInteractionStatistic(Resource):
         args = parser.parse_args()
 
         assert account.timezone is not None
+        timezone = pytz.timezone(account.timezone)
+        utc_timezone = pytz.utc
 
-        try:
-            start_date, end_date = parse_time_range(args["start"], args["end"], account.timezone)
-        except ValueError as e:
-            abort(400, description=str(e))
+        start_date = None
+        end_date = None
+
+        if args["start"]:
+            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
+            start_datetime = start_datetime.replace(second=0)
+            start_datetime_timezone = timezone.localize(start_datetime)
+            start_date = start_datetime_timezone.astimezone(utc_timezone)
+
+        if args["end"]:
+            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
+            end_datetime = end_datetime.replace(second=0)
+            end_datetime_timezone = timezone.localize(end_datetime)
+            end_date = end_datetime_timezone.astimezone(utc_timezone)
 
         response_data = self._workflow_run_repo.get_average_app_interaction_statistics(
             tenant_id=app_model.tenant_id,

api/controllers/console/app/workflow_trigger.py

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

api/controllers/console/billing/billing.py

@@ -2,7 +2,6 @@ from flask_restx import Resource, reqparse
 
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, only_edition_cloud, setup_required
-from enums.cloud_plan import CloudPlan
 from libs.login import current_account_with_tenant, login_required
 from services.billing_service import BillingService
 
@@ -17,13 +16,7 @@ class Subscription(Resource):
         current_user, current_tenant_id = current_account_with_tenant()
         parser = (
             reqparse.RequestParser()
-            .add_argument(
-                "plan",
-                type=str,
-                required=True,
-                location="args",
-                choices=[CloudPlan.PROFESSIONAL, CloudPlan.TEAM],
-            )
+            .add_argument("plan", type=str, required=True, location="args", choices=["professional", "team"])
             .add_argument("interval", type=str, required=True, location="args", choices=["month", "year"])
         )
         args = parser.parse_args()

api/controllers/console/datasets/datasets_document.py

@@ -746,7 +746,7 @@ class DocumentApi(DocumentResource):
                 "name": document.name,
                 "created_from": document.created_from,
                 "created_by": document.created_by,
-                "created_at": int(document.created_at.timestamp()),
+                "created_at": document.created_at.timestamp(),
                 "tokens": document.tokens,
                 "indexing_status": document.indexing_status,
                 "completed_at": int(document.completed_at.timestamp()) if document.completed_at else None,
@@ -779,7 +779,7 @@ class DocumentApi(DocumentResource):
                 "name": document.name,
                 "created_from": document.created_from,
                 "created_by": document.created_by,
-                "created_at": int(document.created_at.timestamp()),
+                "created_at": document.created_at.timestamp(),
                 "tokens": document.tokens,
                 "indexing_status": document.indexing_status,
                 "completed_at": int(document.completed_at.timestamp()) if document.completed_at else None,

api/controllers/console/files.py

@@ -8,7 +8,6 @@ import services
 from configs import dify_config
 from constants import DOCUMENT_EXTENSIONS
 from controllers.common.errors import (
-    BlockedFileExtensionError,
     FilenameNotExistsError,
     FileTooLargeError,
     NoFileUploadedError,
@@ -84,8 +83,6 @@ class FileApi(Resource):
             raise FileTooLargeError(file_too_large_error.description)
         except services.errors.file.UnsupportedFileTypeError:
             raise UnsupportedFileTypeError()
-        except services.errors.file.BlockedFileExtensionError as blocked_extension_error:
-            raise BlockedFileExtensionError(blocked_extension_error.description)
 
         return upload_file, 201
 

api/controllers/console/workspace/plugin.py

@@ -114,25 +114,6 @@ class PluginIconApi(Resource):
         return send_file(io.BytesIO(icon_bytes), mimetype=mimetype, max_age=icon_cache_max_age)
 
 
-@console_ns.route("/workspaces/current/plugin/asset")
-class PluginAssetApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def get(self):
-        req = reqparse.RequestParser()
-        req.add_argument("plugin_unique_identifier", type=str, required=True, location="args")
-        req.add_argument("file_name", type=str, required=True, location="args")
-        args = req.parse_args()
-
-        _, tenant_id = current_account_with_tenant()
-        try:
-            binary = PluginService.extract_asset(tenant_id, args["plugin_unique_identifier"], args["file_name"])
-            return send_file(io.BytesIO(binary), mimetype="application/octet-stream")
-        except PluginDaemonClientSideError as e:
-            raise ValueError(e)
-
-
 @console_ns.route("/workspaces/current/plugin/upload/pkg")
 class PluginUploadFromPkgApi(Resource):
     @setup_required
@@ -577,21 +558,19 @@ class PluginFetchDynamicSelectOptionsApi(Resource):
             .add_argument("provider", type=str, required=True, location="args")
             .add_argument("action", type=str, required=True, location="args")
             .add_argument("parameter", type=str, required=True, location="args")
-            .add_argument("credential_id", type=str, required=False, location="args")
             .add_argument("provider_type", type=str, required=True, location="args")
         )
         args = parser.parse_args()
 
         try:
             options = PluginParameterService.get_dynamic_select_options(
-                tenant_id=tenant_id,
-                user_id=user_id,
-                plugin_id=args["plugin_id"],
-                provider=args["provider"],
-                action=args["action"],
-                parameter=args["parameter"],
-                credential_id=args["credential_id"],
-                provider_type=args["provider_type"],
+                tenant_id,
+                user_id,
+                args["plugin_id"],
+                args["provider"],
+                args["action"],
+                args["parameter"],
+                args["provider_type"],
             )
         except PluginDaemonClientSideError as e:
             raise ValueError(e)
@@ -707,23 +686,3 @@ class PluginAutoUpgradeExcludePluginApi(Resource):
         args = req.parse_args()
 
         return jsonable_encoder({"success": PluginAutoUpgradeService.exclude_plugin(tenant_id, args["plugin_id"])})
-
-
-@console_ns.route("/workspaces/current/plugin/readme")
-class PluginReadmeApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def get(self):
-        _, tenant_id = current_account_with_tenant()
-        parser = reqparse.RequestParser()
-        parser.add_argument("plugin_unique_identifier", type=str, required=True, location="args")
-        parser.add_argument("language", type=str, required=False, location="args")
-        args = parser.parse_args()
-        return jsonable_encoder(
-            {
-                "readme": PluginService.fetch_plugin_readme(
-                    tenant_id, args["plugin_unique_identifier"], args.get("language", "en-US")
-                )
-            }
-        )

api/controllers/console/workspace/tool_providers.py

@@ -21,14 +21,12 @@ from core.mcp.auth.auth_flow import auth, handle_callback
 from core.mcp.error import MCPAuthError, MCPError, MCPRefreshTokenError
 from core.mcp.mcp_client import MCPClient
 from core.model_runtime.utils.encoders import jsonable_encoder
-from core.plugin.entities.plugin_daemon import CredentialType
 from core.plugin.impl.oauth import OAuthHandler
+from core.tools.entities.tool_entities import CredentialType
 from extensions.ext_database import db
 from libs.helper import StrLen, alphanumeric, uuid_value
 from libs.login import current_account_with_tenant, login_required
 from models.provider_ids import ToolProviderID
-
-# from models.provider_ids import ToolProviderID
 from services.plugin.oauth_service import OAuthProxyService
 from services.tools.api_tools_manage_service import ApiToolManageService
 from services.tools.builtin_tools_manage_service import BuiltinToolManageService

api/controllers/console/workspace/trigger_providers.py

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

api/controllers/console/workspace/workspace.py

@@ -21,7 +21,6 @@ from controllers.console.wraps import (
     cloud_edition_billing_resource_check,
     setup_required,
 )
-from enums.cloud_plan import CloudPlan
 from extensions.ext_database import db
 from libs.helper import TimestampField
 from libs.login import current_account_with_tenant, login_required
@@ -84,7 +83,7 @@ class TenantListApi(Resource):
                 "name": tenant.name,
                 "status": tenant.status,
                 "created_at": tenant.created_at,
-                "plan": features.billing.subscription.plan if features.billing.enabled else CloudPlan.SANDBOX,
+                "plan": features.billing.subscription.plan if features.billing.enabled else "sandbox",
                 "current": tenant.id == current_tenant_id if current_tenant_id else False,
             }
 

api/controllers/console/wraps.py

@@ -10,7 +10,6 @@ from flask import abort, request
 
 from configs import dify_config
 from controllers.console.workspace.error import AccountNotInitializedError
-from enums.cloud_plan import CloudPlan
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from libs.login import current_account_with_tenant
@@ -134,7 +133,7 @@ def cloud_edition_billing_knowledge_limit_check(resource: str):
             features = FeatureService.get_features(current_tenant_id)
             if features.billing.enabled:
                 if resource == "add_segment":
-                    if features.billing.subscription.plan == CloudPlan.SANDBOX:
+                    if features.billing.subscription.plan == "sandbox":
                         abort(
                             403,
                             "To unlock this feature and elevate your Dify experience, please upgrade to a paid plan.",

api/controllers/service_api/dataset/document.py

@@ -592,7 +592,7 @@ class DocumentApi(DatasetApiResource):
                 "name": document.name,
                 "created_from": document.created_from,
                 "created_by": document.created_by,
-                "created_at": int(document.created_at.timestamp()),
+                "created_at": document.created_at.timestamp(),
                 "tokens": document.tokens,
                 "indexing_status": document.indexing_status,
                 "completed_at": int(document.completed_at.timestamp()) if document.completed_at else None,
@@ -625,7 +625,7 @@ class DocumentApi(DatasetApiResource):
                 "name": document.name,
                 "created_from": document.created_from,
                 "created_by": document.created_by,
-                "created_at": int(document.created_at.timestamp()),
+                "created_at": document.created_at.timestamp(),
                 "tokens": document.tokens,
                 "indexing_status": document.indexing_status,
                 "completed_at": int(document.completed_at.timestamp()) if document.completed_at else None,

api/controllers/service_api/dataset/segment.py

@@ -2,7 +2,6 @@ from flask import request
 from flask_restx import marshal, reqparse
 from werkzeug.exceptions import NotFound
 
-from configs import dify_config
 from controllers.service_api import service_api_ns
 from controllers.service_api.app.error import ProviderNotInitializeError
 from controllers.service_api.wraps import (
@@ -108,10 +107,6 @@ class SegmentApi(DatasetApiResource):
         # validate args
         args = segment_create_parser.parse_args()
         if args["segments"] is not None:
-            segments_limit = dify_config.DATASET_MAX_SEGMENTS_PER_REQUEST
-            if segments_limit > 0 and len(args["segments"]) > segments_limit:
-                raise ValueError(f"Exceeded maximum segments limit of {segments_limit}.")
-
             for args_item in args["segments"]:
                 SegmentService.segment_create_args_validate(args_item, document)
             segments = SegmentService.multi_create_segment(args["segments"], document, dataset)

api/controllers/service_api/wraps.py

@@ -13,15 +13,13 @@ from sqlalchemy import select, update
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden, NotFound, Unauthorized
 
-from enums.cloud_plan import CloudPlan
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from libs.datetime_utils import naive_utc_now
 from libs.login import current_user
 from models import Account, Tenant, TenantAccountJoin, TenantStatus
 from models.dataset import Dataset, RateLimitLog
-from models.model import ApiToken, App
-from services.end_user_service import EndUserService
+from models.model import ApiToken, App, DefaultEndUserSessionID, EndUser
 from services.feature_service import FeatureService
 
 P = ParamSpec("P")
@@ -68,7 +66,6 @@ def validate_app_token(view: Callable[P, R] | None = None, *, fetch_user_arg: Fe
 
             kwargs["app_model"] = app_model
 
-            # If caller needs end-user context, attach EndUser to current_user
             if fetch_user_arg:
                 if fetch_user_arg.fetch_from == WhereisUserArg.QUERY:
                     user_id = request.args.get("user")
@@ -77,6 +74,7 @@ def validate_app_token(view: Callable[P, R] | None = None, *, fetch_user_arg: Fe
                 elif fetch_user_arg.fetch_from == WhereisUserArg.FORM:
                     user_id = request.form.get("user")
                 else:
+                    # use default-user
                     user_id = None
 
                 if not user_id and fetch_user_arg.required:
@@ -85,34 +83,12 @@ def validate_app_token(view: Callable[P, R] | None = None, *, fetch_user_arg: Fe
                 if user_id:
                     user_id = str(user_id)
 
-                end_user = EndUserService.get_or_create_end_user(app_model, user_id)
+                end_user = create_or_update_end_user_for_user_id(app_model, user_id)
                 kwargs["end_user"] = end_user
 
                 # Set EndUser as current logged-in user for flask_login.current_user
                 current_app.login_manager._update_request_context_with_user(end_user)  # type: ignore
                 user_logged_in.send(current_app._get_current_object(), user=end_user)  # type: ignore
-            else:
-                # For service API without end-user context, ensure an Account is logged in
-                # so services relying on current_account_with_tenant() work correctly.
-                tenant_owner_info = (
-                    db.session.query(Tenant, Account)
-                    .join(TenantAccountJoin, Tenant.id == TenantAccountJoin.tenant_id)
-                    .join(Account, TenantAccountJoin.account_id == Account.id)
-                    .where(
-                        Tenant.id == app_model.tenant_id,
-                        TenantAccountJoin.role == "owner",
-                        Tenant.status == TenantStatus.NORMAL,
-                    )
-                    .one_or_none()
-                )
-
-                if tenant_owner_info:
-                    tenant_model, account = tenant_owner_info
-                    account.current_tenant = tenant_model
-                    current_app.login_manager._update_request_context_with_user(account)  # type: ignore
-                    user_logged_in.send(current_app._get_current_object(), user=current_user)  # type: ignore
-                else:
-                    raise Unauthorized("Tenant owner account not found or tenant is not active.")
 
             return view_func(*args, **kwargs)
 
@@ -162,7 +138,7 @@ def cloud_edition_billing_knowledge_limit_check(resource: str, api_token_type: s
             features = FeatureService.get_features(api_token.tenant_id)
             if features.billing.enabled:
                 if resource == "add_segment":
-                    if features.billing.subscription.plan == CloudPlan.SANDBOX:
+                    if features.billing.subscription.plan == "sandbox":
                         raise Forbidden(
                             "To unlock this feature and elevate your Dify experience, please upgrade to a paid plan."
                         )
@@ -332,6 +308,39 @@ def validate_and_get_api_token(scope: str | None = None):
     return api_token
 
 
+def create_or_update_end_user_for_user_id(app_model: App, user_id: str | None = None) -> EndUser:
+    """
+    Create or update session terminal based on user ID.
+    """
+    if not user_id:
+        user_id = DefaultEndUserSessionID.DEFAULT_SESSION_ID
+
+    with Session(db.engine, expire_on_commit=False) as session:
+        end_user = (
+            session.query(EndUser)
+            .where(
+                EndUser.tenant_id == app_model.tenant_id,
+                EndUser.app_id == app_model.id,
+                EndUser.session_id == user_id,
+                EndUser.type == "service_api",
+            )
+            .first()
+        )
+
+        if end_user is None:
+            end_user = EndUser(
+                tenant_id=app_model.tenant_id,
+                app_id=app_model.id,
+                type="service_api",
+                is_anonymous=user_id == DefaultEndUserSessionID.DEFAULT_SESSION_ID,
+                session_id=user_id,
+            )
+            session.add(end_user)
+            session.commit()
+
+    return end_user
+
+
 class DatasetApiResource(Resource):
     method_decorators = [validate_dataset_token]
 

api/controllers/trigger/__init__.py

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

api/controllers/trigger/trigger.py

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

api/controllers/trigger/webhook.py

@@ -1,105 +0,0 @@
-import logging
-import time
-
-from flask import jsonify
-from werkzeug.exceptions import NotFound, RequestEntityTooLarge
-
-from controllers.trigger import bp
-from core.trigger.debug.event_bus import TriggerDebugEventBus
-from core.trigger.debug.events import WebhookDebugEvent, build_webhook_pool_key
-from services.trigger.webhook_service import WebhookService
-
-logger = logging.getLogger(__name__)
-
-
-def _prepare_webhook_execution(webhook_id: str, is_debug: bool = False):
-    """Fetch trigger context, extract request data, and validate payload using unified processing.
-
-    Args:
-        webhook_id: The webhook ID to process
-        is_debug: If True, skip status validation for debug mode
-    """
-    webhook_trigger, workflow, node_config = WebhookService.get_webhook_trigger_and_workflow(
-        webhook_id, is_debug=is_debug
-    )
-
-    try:
-        # Use new unified extraction and validation
-        webhook_data = WebhookService.extract_and_validate_webhook_data(webhook_trigger, node_config)
-        return webhook_trigger, workflow, node_config, webhook_data, None
-    except ValueError as e:
-        # Fall back to raw extraction for error reporting
-        webhook_data = WebhookService.extract_webhook_data(webhook_trigger)
-        return webhook_trigger, workflow, node_config, webhook_data, str(e)
-
-
-@bp.route("/webhook/<string:webhook_id>", methods=["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"])
-def handle_webhook(webhook_id: str):
-    """
-    Handle webhook trigger calls.
-
-    This endpoint receives webhook calls and processes them according to the
-    configured webhook trigger settings.
-    """
-    try:
-        webhook_trigger, workflow, node_config, webhook_data, error = _prepare_webhook_execution(webhook_id)
-        if error:
-            return jsonify({"error": "Bad Request", "message": error}), 400
-
-        # Process webhook call (send to Celery)
-        WebhookService.trigger_workflow_execution(webhook_trigger, webhook_data, workflow)
-
-        # Return configured response
-        response_data, status_code = WebhookService.generate_webhook_response(node_config)
-        return jsonify(response_data), status_code
-
-    except ValueError as e:
-        raise NotFound(str(e))
-    except RequestEntityTooLarge:
-        raise
-    except Exception as e:
-        logger.exception("Webhook processing failed for %s", webhook_id)
-        return jsonify({"error": "Internal server error", "message": str(e)}), 500
-
-
-@bp.route("/webhook-debug/<string:webhook_id>", methods=["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"])
-def handle_webhook_debug(webhook_id: str):
-    """Handle webhook debug calls without triggering production workflow execution."""
-    try:
-        webhook_trigger, _, node_config, webhook_data, error = _prepare_webhook_execution(webhook_id, is_debug=True)
-        if error:
-            return jsonify({"error": "Bad Request", "message": error}), 400
-
-        workflow_inputs = WebhookService.build_workflow_inputs(webhook_data)
-
-        # Generate pool key and dispatch debug event
-        pool_key: str = build_webhook_pool_key(
-            tenant_id=webhook_trigger.tenant_id,
-            app_id=webhook_trigger.app_id,
-            node_id=webhook_trigger.node_id,
-        )
-        event = WebhookDebugEvent(
-            request_id=f"webhook_debug_{webhook_trigger.webhook_id}_{int(time.time() * 1000)}",
-            timestamp=int(time.time()),
-            node_id=webhook_trigger.node_id,
-            payload={
-                "inputs": workflow_inputs,
-                "webhook_data": webhook_data,
-                "method": webhook_data.get("method"),
-            },
-        )
-        TriggerDebugEventBus.dispatch(
-            tenant_id=webhook_trigger.tenant_id,
-            event=event,
-            pool_key=pool_key,
-        )
-        response_data, status_code = WebhookService.generate_webhook_response(node_config)
-        return jsonify(response_data), status_code
-
-    except ValueError as e:
-        raise NotFound(str(e))
-    except RequestEntityTooLarge:
-        raise
-    except Exception as e:
-        logger.exception("Webhook debug processing failed for %s", webhook_id)
-        return jsonify({"error": "Internal server error", "message": "An internal error has occurred."}), 500

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

@@ -1,6 +1,6 @@
 import logging
 import time
-from collections.abc import Mapping, Sequence
+from collections.abc import Mapping
 from typing import Any, cast
 
 from sqlalchemy import select
@@ -25,7 +25,6 @@ from core.moderation.input_moderation import InputModeration
 from core.variables.variables import VariableUnion
 from core.workflow.enums import WorkflowType
 from core.workflow.graph_engine.command_channels.redis_channel import RedisChannel
-from core.workflow.graph_engine.layers.base import GraphEngineLayer
 from core.workflow.graph_engine.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
@@ -62,13 +61,11 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
         app: App,
         workflow_execution_repository: WorkflowExecutionRepository,
         workflow_node_execution_repository: WorkflowNodeExecutionRepository,
-        graph_engine_layers: Sequence[GraphEngineLayer] = (),
     ):
         super().__init__(
             queue_manager=queue_manager,
             variable_loader=variable_loader,
             app_id=application_generate_entity.app_config.app_id,
-            graph_engine_layers=graph_engine_layers,
         )
         self.application_generate_entity = application_generate_entity
         self.conversation = conversation
@@ -198,8 +195,6 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
         )
 
         workflow_entry.graph_engine.layer(persistence_layer)
-        for layer in self._graph_engine_layers:
-            workflow_entry.graph_engine.layer(layer)
 
         generator = workflow_entry.run()
 

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

@@ -1,4 +1,3 @@
-import json
 import logging
 import re
 import time
@@ -61,7 +60,6 @@ from core.app.task_pipeline.based_generate_task_pipeline import BasedGenerateTas
 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.model_runtime.utils.encoders import jsonable_encoder
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.workflow.enums import WorkflowExecutionStatus
 from core.workflow.nodes import NodeType
@@ -393,14 +391,6 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         if should_direct_answer:
             return
 
-        current_time = time.perf_counter()
-        if self._task_state.first_token_time is None and delta_text.strip():
-            self._task_state.first_token_time = current_time
-            self._task_state.is_streaming_response = True
-
-        if delta_text.strip():
-            self._task_state.last_token_time = current_time
-
         # Only publish tts message at text chunk streaming
         if tts_publisher and queue_message:
             tts_publisher.publish(queue_message)
@@ -782,33 +772,7 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         message.answer = answer_text
         message.updated_at = naive_utc_now()
         message.provider_response_latency = time.perf_counter() - self._base_task_pipeline.start_at
-
-        # Set usage first before dumping metadata
-        if graph_runtime_state and graph_runtime_state.llm_usage:
-            usage = graph_runtime_state.llm_usage
-            message.message_tokens = usage.prompt_tokens
-            message.message_unit_price = usage.prompt_unit_price
-            message.message_price_unit = usage.prompt_price_unit
-            message.answer_tokens = usage.completion_tokens
-            message.answer_unit_price = usage.completion_unit_price
-            message.answer_price_unit = usage.completion_price_unit
-            message.total_price = usage.total_price
-            message.currency = usage.currency
-            self._task_state.metadata.usage = usage
-        else:
-            usage = LLMUsage.empty_usage()
-            self._task_state.metadata.usage = usage
-
-        # Add streaming metrics to usage if available
-        if self._task_state.is_streaming_response and self._task_state.first_token_time:
-            start_time = self._base_task_pipeline.start_at
-            first_token_time = self._task_state.first_token_time
-            last_token_time = self._task_state.last_token_time or first_token_time
-            usage.time_to_first_token = round(first_token_time - start_time, 3)
-            usage.time_to_generate = round(last_token_time - first_token_time, 3)
-
-        metadata = self._task_state.metadata.model_dump()
-        message.message_metadata = json.dumps(jsonable_encoder(metadata))
+        message.message_metadata = self._task_state.metadata.model_dump_json()
         message_files = [
             MessageFile(
                 message_id=message.id,
@@ -826,6 +790,20 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         ]
         session.add_all(message_files)
 
+        if graph_runtime_state and graph_runtime_state.llm_usage:
+            usage = graph_runtime_state.llm_usage
+            message.message_tokens = usage.prompt_tokens
+            message.message_unit_price = usage.prompt_unit_price
+            message.message_price_unit = usage.prompt_price_unit
+            message.answer_tokens = usage.completion_tokens
+            message.answer_unit_price = usage.completion_unit_price
+            message.answer_price_unit = usage.completion_price_unit
+            message.total_price = usage.total_price
+            message.currency = usage.currency
+            self._task_state.metadata.usage = usage
+        else:
+            self._task_state.metadata.usage = LLMUsage.empty_usage()
+
     def _seed_graph_runtime_state_from_queue_manager(self) -> None:
         """Bootstrap the cached runtime state from the queue manager when present."""
         candidate = self._base_task_pipeline.queue_manager.graph_runtime_state

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

@@ -144,7 +144,7 @@ class AgentChatAppRunner(AppRunner):
             prompt_template_entity=app_config.prompt_template,
             inputs=dict(inputs),
             files=list(files),
-            query=query,
+            query=query or "",
             memory=memory,
         )
 
@@ -172,7 +172,7 @@ class AgentChatAppRunner(AppRunner):
             prompt_template_entity=app_config.prompt_template,
             inputs=dict(inputs),
             files=list(files),
-            query=query,
+            query=query or "",
             memory=memory,
         )
 

api/core/app/apps/base_app_runner.py

@@ -79,7 +79,7 @@ class AppRunner:
         prompt_template_entity: PromptTemplateEntity,
         inputs: Mapping[str, str],
         files: Sequence["File"],
-        query: str = "",
+        query: str | None = None,
         context: str | None = None,
         memory: TokenBufferMemory | None = None,
         image_detail_config: ImagePromptMessageContent.DETAIL | None = None,
@@ -105,7 +105,7 @@ class AppRunner:
                 app_mode=AppMode.value_of(app_record.mode),
                 prompt_template_entity=prompt_template_entity,
                 inputs=inputs,
-                query=query,
+                query=query or "",
                 files=files,
                 context=context,
                 memory=memory,

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

@@ -37,7 +37,6 @@ from core.file import FILE_MODEL_IDENTITY, File
 from core.plugin.impl.datasource import PluginDatasourceManager
 from core.tools.entities.tool_entities import ToolProviderType
 from core.tools.tool_manager import ToolManager
-from core.trigger.trigger_manager import TriggerManager
 from core.variables.segments import ArrayFileSegment, FileSegment, Segment
 from core.workflow.enums import (
     NodeType,
@@ -304,11 +303,6 @@ class WorkflowResponseConverter:
             response.data.extras["icon"] = provider_entity.declaration.identity.generate_datasource_icon_url(
                 self._application_generate_entity.app_config.tenant_id
             )
-        elif event.node_type == NodeType.TRIGGER_PLUGIN:
-            response.data.extras["icon"] = TriggerManager.get_trigger_plugin_icon(
-                self._application_generate_entity.app_config.tenant_id,
-                event.provider_id,
-            )
 
         return response
 

api/core/app/apps/message_based_app_generator.py

@@ -190,7 +190,7 @@ class MessageBasedAppGenerator(BaseAppGenerator):
             override_model_configs=json.dumps(override_model_configs) if override_model_configs else None,
             conversation_id=conversation.id,
             inputs=application_generate_entity.inputs,
-            query=application_generate_entity.query,
+            query=application_generate_entity.query or "",
             message="",
             message_tokens=0,
             message_unit_price=0,

api/core/app/apps/pipeline/pipeline_generator.py

@@ -41,14 +41,18 @@ from core.workflow.repositories.workflow_execution_repository import WorkflowExe
 from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 from core.workflow.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader
 from extensions.ext_database import db
+from extensions.ext_redis import redis_client
 from libs.flask_utils import preserve_flask_contexts
 from models import Account, EndUser, Workflow, WorkflowNodeExecutionTriggeredFrom
 from models.dataset import Document, DocumentPipelineExecutionLog, Pipeline
 from models.enums import WorkflowRunTriggeredFrom
 from models.model import AppMode
 from services.datasource_provider_service import DatasourceProviderService
-from services.rag_pipeline.rag_pipeline_task_proxy import RagPipelineTaskProxy
+from services.feature_service import FeatureService
+from services.file_service import FileService
 from services.workflow_draft_variable_service import DraftVarLoader, WorkflowDraftVariableService
+from tasks.rag_pipeline.priority_rag_pipeline_run_task import priority_rag_pipeline_run_task
+from tasks.rag_pipeline.rag_pipeline_run_task import rag_pipeline_run_task
 
 logger = logging.getLogger(__name__)
 
@@ -244,7 +248,34 @@ class PipelineGenerator(BaseAppGenerator):
                 )
 
         if rag_pipeline_invoke_entities:
-            RagPipelineTaskProxy(dataset.tenant_id, user.id, rag_pipeline_invoke_entities).delay()
+            # store the rag_pipeline_invoke_entities to object storage
+            text = [item.model_dump() for item in rag_pipeline_invoke_entities]
+            name = "rag_pipeline_invoke_entities.json"
+            # Convert list to proper JSON string
+            json_text = json.dumps(text)
+            upload_file = FileService(db.engine).upload_text(json_text, name, user.id, dataset.tenant_id)
+            features = FeatureService.get_features(dataset.tenant_id)
+            if features.billing.enabled and features.billing.subscription.plan == "sandbox":
+                tenant_pipeline_task_key = f"tenant_pipeline_task:{dataset.tenant_id}"
+                tenant_self_pipeline_task_queue = f"tenant_self_pipeline_task_queue:{dataset.tenant_id}"
+
+                if redis_client.get(tenant_pipeline_task_key):
+                    # Add to waiting queue using List operations (lpush)
+                    redis_client.lpush(tenant_self_pipeline_task_queue, upload_file.id)
+                else:
+                    # Set flag and execute task
+                    redis_client.set(tenant_pipeline_task_key, 1, ex=60 * 60)
+                    rag_pipeline_run_task.delay(  # type: ignore
+                        rag_pipeline_invoke_entities_file_id=upload_file.id,
+                        tenant_id=dataset.tenant_id,
+                    )
+
+            else:
+                priority_rag_pipeline_run_task.delay(  # type: ignore
+                    rag_pipeline_invoke_entities_file_id=upload_file.id,
+                    tenant_id=dataset.tenant_id,
+                )
+
         # return batch, dataset, documents
         return {
             "batch": batch,

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

@@ -27,7 +27,6 @@ from core.helper.trace_id_helper import extract_external_trace_id_from_args
 from core.model_runtime.errors.invoke import InvokeAuthorizationError
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.repositories import DifyCoreRepositoryFactory
-from core.workflow.graph_engine.layers.base import GraphEngineLayer
 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
@@ -39,16 +38,10 @@ from models import Account, App, EndUser, Workflow, WorkflowNodeExecutionTrigger
 from models.enums import WorkflowRunTriggeredFrom
 from services.workflow_draft_variable_service import DraftVarLoader, WorkflowDraftVariableService
 
-SKIP_PREPARE_USER_INPUTS_KEY = "_skip_prepare_user_inputs"
-
 logger = logging.getLogger(__name__)
 
 
 class WorkflowAppGenerator(BaseAppGenerator):
-    @staticmethod
-    def _should_prepare_user_inputs(args: Mapping[str, Any]) -> bool:
-        return not bool(args.get(SKIP_PREPARE_USER_INPUTS_KEY))
-
     @overload
     def generate(
         self,
@@ -60,10 +53,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
         invoke_from: InvokeFrom,
         streaming: Literal[True],
         call_depth: int,
-        triggered_from: WorkflowRunTriggeredFrom | None = None,
-        root_node_id: str | None = None,
-        graph_engine_layers: Sequence[GraphEngineLayer] = (),
-    ) -> Generator[Mapping[str, Any] | str, None, None]: ...
+    ) -> Generator[Mapping | str, None, None]: ...
 
     @overload
     def generate(
@@ -76,9 +66,6 @@ class WorkflowAppGenerator(BaseAppGenerator):
         invoke_from: InvokeFrom,
         streaming: Literal[False],
         call_depth: int,
-        triggered_from: WorkflowRunTriggeredFrom | None = None,
-        root_node_id: str | None = None,
-        graph_engine_layers: Sequence[GraphEngineLayer] = (),
     ) -> Mapping[str, Any]: ...
 
     @overload
@@ -92,10 +79,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
         invoke_from: InvokeFrom,
         streaming: bool,
         call_depth: int,
-        triggered_from: WorkflowRunTriggeredFrom | None = None,
-        root_node_id: str | None = None,
-        graph_engine_layers: Sequence[GraphEngineLayer] = (),
-    ) -> Union[Mapping[str, Any], Generator[Mapping[str, Any] | str, None, None]]: ...
+    ) -> Union[Mapping[str, Any], Generator[Mapping | str, None, None]]: ...
 
     def generate(
         self,
@@ -107,10 +91,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
         invoke_from: InvokeFrom,
         streaming: bool = True,
         call_depth: int = 0,
-        triggered_from: WorkflowRunTriggeredFrom | None = None,
-        root_node_id: str | None = None,
-        graph_engine_layers: Sequence[GraphEngineLayer] = (),
-    ) -> Union[Mapping[str, Any], Generator[Mapping[str, Any] | str, None, None]]:
+    ) -> Union[Mapping[str, Any], Generator[Mapping | str, None, None]]:
         files: Sequence[Mapping[str, Any]] = args.get("files") or []
 
         # parse files
@@ -145,20 +126,17 @@ class WorkflowAppGenerator(BaseAppGenerator):
             **extract_external_trace_id_from_args(args),
         }
         workflow_run_id = str(uuid.uuid4())
-        # for trigger debug run, not prepare user inputs
-        if self._should_prepare_user_inputs(args):
-            inputs = self._prepare_user_inputs(
-                user_inputs=inputs,
-                variables=app_config.variables,
-                tenant_id=app_model.tenant_id,
-                strict_type_validation=True if invoke_from == InvokeFrom.SERVICE_API else False,
-            )
         # init application generate entity
         application_generate_entity = WorkflowAppGenerateEntity(
             task_id=str(uuid.uuid4()),
             app_config=app_config,
             file_upload_config=file_extra_config,
-            inputs=inputs,
+            inputs=self._prepare_user_inputs(
+                user_inputs=inputs,
+                variables=app_config.variables,
+                tenant_id=app_model.tenant_id,
+                strict_type_validation=True if invoke_from == InvokeFrom.SERVICE_API else False,
+            ),
             files=list(system_files),
             user_id=user.id,
             stream=streaming,
@@ -177,10 +155,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
         # Create session factory
         session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
         # Create workflow execution(aka workflow run) repository
-        if triggered_from is not None:
-            # Use explicitly provided triggered_from (for async triggers)
-            workflow_triggered_from = triggered_from
-        elif invoke_from == InvokeFrom.DEBUGGER:
+        if invoke_from == InvokeFrom.DEBUGGER:
             workflow_triggered_from = WorkflowRunTriggeredFrom.DEBUGGING
         else:
             workflow_triggered_from = WorkflowRunTriggeredFrom.APP_RUN
@@ -207,16 +182,8 @@ class WorkflowAppGenerator(BaseAppGenerator):
             workflow_execution_repository=workflow_execution_repository,
             workflow_node_execution_repository=workflow_node_execution_repository,
             streaming=streaming,
-            root_node_id=root_node_id,
-            graph_engine_layers=graph_engine_layers,
         )
 
-    def resume(self, *, workflow_run_id: str) -> None:
-        """
-        @TBD
-        """
-        pass
-
     def _generate(
         self,
         *,
@@ -229,8 +196,6 @@ class WorkflowAppGenerator(BaseAppGenerator):
         workflow_node_execution_repository: WorkflowNodeExecutionRepository,
         streaming: bool = True,
         variable_loader: VariableLoader = DUMMY_VARIABLE_LOADER,
-        root_node_id: str | None = None,
-        graph_engine_layers: Sequence[GraphEngineLayer] = (),
     ) -> Union[Mapping[str, Any], Generator[str | Mapping[str, Any], None, None]]:
         """
         Generate App response.
@@ -266,10 +231,8 @@ class WorkflowAppGenerator(BaseAppGenerator):
                 "queue_manager": queue_manager,
                 "context": context,
                 "variable_loader": variable_loader,
-                "root_node_id": root_node_id,
                 "workflow_execution_repository": workflow_execution_repository,
                 "workflow_node_execution_repository": workflow_node_execution_repository,
-                "graph_engine_layers": graph_engine_layers,
             },
         )
 
@@ -463,8 +426,6 @@ class WorkflowAppGenerator(BaseAppGenerator):
         variable_loader: VariableLoader,
         workflow_execution_repository: WorkflowExecutionRepository,
         workflow_node_execution_repository: WorkflowNodeExecutionRepository,
-        root_node_id: str | None = None,
-        graph_engine_layers: Sequence[GraphEngineLayer] = (),
     ) -> None:
         """
         Generate worker in a new thread.
@@ -508,8 +469,6 @@ class WorkflowAppGenerator(BaseAppGenerator):
                 system_user_id=system_user_id,
                 workflow_execution_repository=workflow_execution_repository,
                 workflow_node_execution_repository=workflow_node_execution_repository,
-                root_node_id=root_node_id,
-                graph_engine_layers=graph_engine_layers,
             )
 
             try:

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

@@ -1,6 +1,5 @@
 import logging
 import time
-from collections.abc import Sequence
 from typing import cast
 
 from core.app.apps.base_app_queue_manager import AppQueueManager
@@ -9,7 +8,6 @@ from core.app.apps.workflow_app_runner import WorkflowBasedAppRunner
 from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity
 from core.workflow.enums import WorkflowType
 from core.workflow.graph_engine.command_channels.redis_channel import RedisChannel
-from core.workflow.graph_engine.layers.base import GraphEngineLayer
 from core.workflow.graph_engine.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
@@ -18,7 +16,6 @@ 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 libs.datetime_utils import naive_utc_now
 from models.enums import UserFrom
 from models.workflow import Workflow
 
@@ -38,21 +35,17 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
         variable_loader: VariableLoader,
         workflow: Workflow,
         system_user_id: str,
-        root_node_id: str | None = None,
         workflow_execution_repository: WorkflowExecutionRepository,
         workflow_node_execution_repository: WorkflowNodeExecutionRepository,
-        graph_engine_layers: Sequence[GraphEngineLayer] = (),
     ):
         super().__init__(
             queue_manager=queue_manager,
             variable_loader=variable_loader,
             app_id=application_generate_entity.app_config.app_id,
-            graph_engine_layers=graph_engine_layers,
         )
         self.application_generate_entity = application_generate_entity
         self._workflow = workflow
         self._sys_user_id = system_user_id
-        self._root_node_id = root_node_id
         self._workflow_execution_repository = workflow_execution_repository
         self._workflow_node_execution_repository = workflow_node_execution_repository
 
@@ -67,7 +60,6 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
             files=self.application_generate_entity.files,
             user_id=self._sys_user_id,
             app_id=app_config.app_id,
-            timestamp=int(naive_utc_now().timestamp()),
             workflow_id=app_config.workflow_id,
             workflow_execution_id=self.application_generate_entity.workflow_execution_id,
         )
@@ -100,7 +92,6 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
                 workflow_id=self._workflow.id,
                 tenant_id=self._workflow.tenant_id,
                 user_id=self.application_generate_entity.user_id,
-                root_node_id=self._root_node_id,
             )
 
         # RUN WORKFLOW
@@ -144,8 +135,6 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
         )
 
         workflow_entry.graph_engine.layer(persistence_layer)
-        for layer in self._graph_engine_layers:
-            workflow_entry.graph_engine.layer(layer)
 
         generator = workflow_entry.run()
 

api/core/app/apps/workflow_app_runner.py

@@ -1,5 +1,5 @@
 import time
-from collections.abc import Mapping, Sequence
+from collections.abc import Mapping
 from typing import Any, cast
 
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
@@ -27,7 +27,6 @@ from core.app.entities.queue_entities import (
 )
 from core.workflow.entities import GraphInitParams
 from core.workflow.graph import Graph
-from core.workflow.graph_engine.layers.base import GraphEngineLayer
 from core.workflow.graph_events import (
     GraphEngineEvent,
     GraphRunFailedEvent,
@@ -70,12 +69,10 @@ class WorkflowBasedAppRunner:
         queue_manager: AppQueueManager,
         variable_loader: VariableLoader = DUMMY_VARIABLE_LOADER,
         app_id: str,
-        graph_engine_layers: Sequence[GraphEngineLayer] = (),
     ):
         self._queue_manager = queue_manager
         self._variable_loader = variable_loader
         self._app_id = app_id
-        self._graph_engine_layers = graph_engine_layers
 
     def _init_graph(
         self,
@@ -84,7 +81,6 @@ class WorkflowBasedAppRunner:
         workflow_id: str = "",
         tenant_id: str = "",
         user_id: str = "",
-        root_node_id: str | None = None,
     ) -> Graph:
         """
         Init graph
@@ -118,7 +114,7 @@ class WorkflowBasedAppRunner:
         )
 
         # init graph
-        graph = Graph.init(graph_config=graph_config, node_factory=node_factory, root_node_id=root_node_id)
+        graph = Graph.init(graph_config=graph_config, node_factory=node_factory)
 
         if not graph:
             raise ValueError("graph not found in workflow")

api/core/app/entities/app_invoke_entities.py

@@ -32,10 +32,6 @@ class InvokeFrom(StrEnum):
     # https://docs.dify.ai/en/guides/application-publishing/launch-your-webapp-quickly/README
     WEB_APP = "web-app"
 
-    # TRIGGER indicates that this invocation is from a trigger.
-    # this is used for plugin trigger and webhook trigger.
-    TRIGGER = "trigger"
-
     # EXPLORE indicates that this invocation is from
     # the workflow (or chatflow) explore page.
     EXPLORE = "explore"
@@ -44,9 +40,6 @@ class InvokeFrom(StrEnum):
     DEBUGGER = "debugger"
     PUBLISHED = "published"
 
-    # VALIDATION indicates that this invocation is from validation.
-    VALIDATION = "validation"
-
     @classmethod
     def value_of(cls, value: str):
         """
@@ -72,8 +65,6 @@ class InvokeFrom(StrEnum):
             return "dev"
         elif self == InvokeFrom.EXPLORE:
             return "explore_app"
-        elif self == InvokeFrom.TRIGGER:
-            return "trigger"
         elif self == InvokeFrom.SERVICE_API:
             return "api"
 
@@ -113,11 +104,6 @@ class AppGenerateEntity(BaseModel):
 
     inputs: Mapping[str, Any]
     files: Sequence[File]
-
-    # Unique identifier of the user initiating the execution.
-    # This corresponds to `Account.id` for platform users or `EndUser.id` for end users.
-    #
-    # Note: The `user_id` field does not indicate whether the user is a platform user or an end user.
     user_id: str
 
     # extras
@@ -143,7 +129,7 @@ class EasyUIBasedAppGenerateEntity(AppGenerateEntity):
     app_config: EasyUIBasedAppConfig = None  # type: ignore
     model_conf: ModelConfigWithCredentialsEntity
 
-    query: str = ""
+    query: str | None = None
 
     # pydantic configs
     model_config = ConfigDict(protected_namespaces=())

api/core/app/entities/task_entities.py

@@ -48,9 +48,6 @@ class WorkflowTaskState(TaskState):
     """
 
     answer: str = ""
-    first_token_time: float | None = None
-    last_token_time: float | None = None
-    is_streaming_response: bool = False
 
 
 class StreamEvent(StrEnum):

api/core/app/layers/pause_state_persist_layer.py

@@ -1,133 +0,0 @@
-from typing import Annotated, Literal, Self, TypeAlias
-
-from pydantic import BaseModel, Field
-from sqlalchemy import Engine
-from sqlalchemy.orm import Session, sessionmaker
-
-from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, WorkflowAppGenerateEntity
-from core.workflow.graph_engine.layers.base import GraphEngineLayer
-from core.workflow.graph_events.base import GraphEngineEvent
-from core.workflow.graph_events.graph import GraphRunPausedEvent
-from models.model import AppMode
-from repositories.api_workflow_run_repository import APIWorkflowRunRepository
-from repositories.factory import DifyAPIRepositoryFactory
-
-
-# Wrapper types for `WorkflowAppGenerateEntity` and
-# `AdvancedChatAppGenerateEntity`. These wrappers enable type discrimination
-# and correct reconstruction of the entity field during (de)serialization.
-class _WorkflowGenerateEntityWrapper(BaseModel):
-    type: Literal[AppMode.WORKFLOW] = AppMode.WORKFLOW
-    entity: WorkflowAppGenerateEntity
-
-
-class _AdvancedChatAppGenerateEntityWrapper(BaseModel):
-    type: Literal[AppMode.ADVANCED_CHAT] = AppMode.ADVANCED_CHAT
-    entity: AdvancedChatAppGenerateEntity
-
-
-_GenerateEntityUnion: TypeAlias = Annotated[
-    _WorkflowGenerateEntityWrapper | _AdvancedChatAppGenerateEntityWrapper,
-    Field(discriminator="type"),
-]
-
-
-class WorkflowResumptionContext(BaseModel):
-    """WorkflowResumptionContext captures all state necessary for resumption."""
-
-    version: Literal["1"] = "1"
-
-    # Only workflow / chatflow could be paused.
-    generate_entity: _GenerateEntityUnion
-    serialized_graph_runtime_state: str
-
-    def dumps(self) -> str:
-        return self.model_dump_json()
-
-    @classmethod
-    def loads(cls, value: str) -> Self:
-        return cls.model_validate_json(value)
-
-    def get_generate_entity(self) -> WorkflowAppGenerateEntity | AdvancedChatAppGenerateEntity:
-        return self.generate_entity.entity
-
-
-class PauseStatePersistenceLayer(GraphEngineLayer):
-    def __init__(
-        self,
-        session_factory: Engine | sessionmaker[Session],
-        generate_entity: WorkflowAppGenerateEntity | AdvancedChatAppGenerateEntity,
-        state_owner_user_id: str,
-    ):
-        """Create a PauseStatePersistenceLayer.
-
-        The `state_owner_user_id` is used when creating state file for pause.
-        It generally should id of the creator of workflow.
-        """
-        if isinstance(session_factory, Engine):
-            session_factory = sessionmaker(session_factory)
-        self._session_maker = session_factory
-        self._state_owner_user_id = state_owner_user_id
-        self._generate_entity = generate_entity
-
-    def _get_repo(self) -> APIWorkflowRunRepository:
-        return DifyAPIRepositoryFactory.create_api_workflow_run_repository(self._session_maker)
-
-    def on_graph_start(self) -> None:
-        """
-        Called when graph execution starts.
-
-        This is called after the engine has been initialized but before any nodes
-        are executed. Layers can use this to set up resources or log start information.
-        """
-        pass
-
-    def on_event(self, event: GraphEngineEvent) -> None:
-        """
-        Called for every event emitted by the engine.
-
-        This method receives all events generated during graph execution, including:
-        - Graph lifecycle events (start, success, failure)
-        - Node execution events (start, success, failure, retry)
-        - Stream events for response nodes
-        - Container events (iteration, loop)
-
-        Args:
-            event: The event emitted by the engine
-        """
-        if not isinstance(event, GraphRunPausedEvent):
-            return
-
-        assert self.graph_runtime_state is not None
-
-        entity_wrapper: _GenerateEntityUnion
-        if isinstance(self._generate_entity, WorkflowAppGenerateEntity):
-            entity_wrapper = _WorkflowGenerateEntityWrapper(entity=self._generate_entity)
-        else:
-            entity_wrapper = _AdvancedChatAppGenerateEntityWrapper(entity=self._generate_entity)
-
-        state = WorkflowResumptionContext(
-            serialized_graph_runtime_state=self.graph_runtime_state.dumps(),
-            generate_entity=entity_wrapper,
-        )
-
-        workflow_run_id: str | None = self.graph_runtime_state.system_variable.workflow_execution_id
-        assert workflow_run_id is not None
-        repo = self._get_repo()
-        repo.create_workflow_pause(
-            workflow_run_id=workflow_run_id,
-            state_owner_user_id=self._state_owner_user_id,
-            state=state.dumps(),
-        )
-
-    def on_graph_end(self, error: Exception | None) -> None:
-        """
-        Called when graph execution ends.
-
-        This is called after all nodes have been executed or when execution is
-        aborted. Layers can use this to clean up resources or log final state.
-
-        Args:
-            error: The exception that caused execution to fail, or None if successful
-        """
-        pass

api/core/app/layers/suspend_layer.py

@@ -1,21 +0,0 @@
-from core.workflow.graph_engine.layers.base import GraphEngineLayer
-from core.workflow.graph_events.base import GraphEngineEvent
-from core.workflow.graph_events.graph import GraphRunPausedEvent
-
-
-class SuspendLayer(GraphEngineLayer):
-    """ """
-
-    def on_graph_start(self):
-        pass
-
-    def on_event(self, event: GraphEngineEvent):
-        """
-        Handle the paused event, stash runtime state into storage and wait for resume.
-        """
-        if isinstance(event, GraphRunPausedEvent):
-            pass
-
-    def on_graph_end(self, error: Exception | None):
-        """ """
-        pass

api/core/app/layers/timeslice_layer.py

@@ -1,88 +0,0 @@
-import logging
-import uuid
-from typing import ClassVar
-
-from apscheduler.schedulers.background import BackgroundScheduler  # type: ignore
-
-from core.workflow.graph_engine.entities.commands import CommandType, GraphEngineCommand
-from core.workflow.graph_engine.layers.base import GraphEngineLayer
-from core.workflow.graph_events.base import GraphEngineEvent
-from services.workflow.entities import WorkflowScheduleCFSPlanEntity
-from services.workflow.scheduler import CFSPlanScheduler, SchedulerCommand
-
-logger = logging.getLogger(__name__)
-
-
-class TimeSliceLayer(GraphEngineLayer):
-    """
-    CFS plan scheduler to control the timeslice of the workflow.
-    """
-
-    scheduler: ClassVar[BackgroundScheduler] = BackgroundScheduler()
-
-    def __init__(self, cfs_plan_scheduler: CFSPlanScheduler) -> None:
-        """
-        CFS plan scheduler allows to control the timeslice of the workflow.
-        """
-
-        if not TimeSliceLayer.scheduler.running:
-            TimeSliceLayer.scheduler.start()
-
-        super().__init__()
-        self.cfs_plan_scheduler = cfs_plan_scheduler
-        self.stopped = False
-        self.schedule_id = ""
-
-    def _checker_job(self, schedule_id: str):
-        """
-        Check if the workflow need to be suspended.
-        """
-        try:
-            if self.stopped:
-                self.scheduler.remove_job(schedule_id)
-                return
-
-            if self.cfs_plan_scheduler.can_schedule() == SchedulerCommand.RESOURCE_LIMIT_REACHED:
-                # remove the job
-                self.scheduler.remove_job(schedule_id)
-
-                if not self.command_channel:
-                    logger.exception("No command channel to stop the workflow")
-                    return
-
-                # send command to pause the workflow
-                self.command_channel.send_command(
-                    GraphEngineCommand(
-                        command_type=CommandType.PAUSE,
-                        payload={
-                            "reason": SchedulerCommand.RESOURCE_LIMIT_REACHED,
-                        },
-                    )
-                )
-
-        except Exception:
-            logger.exception("scheduler error during check if the workflow need to be suspended")
-
-    def on_graph_start(self):
-        """
-        Start timer to check if the workflow need to be suspended.
-        """
-
-        if self.cfs_plan_scheduler.plan.schedule_strategy == WorkflowScheduleCFSPlanEntity.Strategy.TimeSlice:
-            self.schedule_id = uuid.uuid4().hex
-
-            self.scheduler.add_job(
-                lambda: self._checker_job(self.schedule_id),
-                "interval",
-                seconds=self.cfs_plan_scheduler.plan.granularity,
-                id=self.schedule_id,
-            )
-
-    def on_event(self, event: GraphEngineEvent):
-        pass
-
-    def on_graph_end(self, error: Exception | None) -> None:
-        self.stopped = True
-        # remove the scheduler
-        if self.schedule_id:
-            self.scheduler.remove_job(self.schedule_id)

api/core/app/layers/trigger_post_layer.py

@@ -1,88 +0,0 @@
-import logging
-from datetime import UTC, datetime
-from typing import Any, ClassVar
-
-from pydantic import TypeAdapter
-from sqlalchemy.orm import Session, sessionmaker
-
-from core.workflow.graph_engine.layers.base import GraphEngineLayer
-from core.workflow.graph_events.base import GraphEngineEvent
-from core.workflow.graph_events.graph import GraphRunFailedEvent, GraphRunPausedEvent, GraphRunSucceededEvent
-from models.enums import WorkflowTriggerStatus
-from repositories.sqlalchemy_workflow_trigger_log_repository import SQLAlchemyWorkflowTriggerLogRepository
-from tasks.workflow_cfs_scheduler.cfs_scheduler import AsyncWorkflowCFSPlanEntity
-
-logger = logging.getLogger(__name__)
-
-
-class TriggerPostLayer(GraphEngineLayer):
-    """
-    Trigger post layer.
-    """
-
-    _STATUS_MAP: ClassVar[dict[type[GraphEngineEvent], WorkflowTriggerStatus]] = {
-        GraphRunSucceededEvent: WorkflowTriggerStatus.SUCCEEDED,
-        GraphRunFailedEvent: WorkflowTriggerStatus.FAILED,
-        GraphRunPausedEvent: WorkflowTriggerStatus.PAUSED,
-    }
-
-    def __init__(
-        self,
-        cfs_plan_scheduler_entity: AsyncWorkflowCFSPlanEntity,
-        start_time: datetime,
-        trigger_log_id: str,
-        session_maker: sessionmaker[Session],
-    ):
-        self.trigger_log_id = trigger_log_id
-        self.start_time = start_time
-        self.cfs_plan_scheduler_entity = cfs_plan_scheduler_entity
-        self.session_maker = session_maker
-
-    def on_graph_start(self):
-        pass
-
-    def on_event(self, event: GraphEngineEvent):
-        """
-        Update trigger log with success or failure.
-        """
-        if isinstance(event, tuple(self._STATUS_MAP.keys())):
-            with self.session_maker() as session:
-                repo = SQLAlchemyWorkflowTriggerLogRepository(session)
-                trigger_log = repo.get_by_id(self.trigger_log_id)
-                if not trigger_log:
-                    logger.exception("Trigger log not found: %s", self.trigger_log_id)
-                    return
-
-                # Calculate elapsed time
-                elapsed_time = (datetime.now(UTC) - self.start_time).total_seconds()
-
-                # Extract relevant data from result
-                if not self.graph_runtime_state:
-                    logger.exception("Graph runtime state is not set")
-                    return
-
-                outputs = self.graph_runtime_state.outputs
-
-                # BASICLY, workflow_execution_id is the same as workflow_run_id
-                workflow_run_id = self.graph_runtime_state.system_variable.workflow_execution_id
-                assert workflow_run_id, "Workflow run id is not set"
-
-                total_tokens = self.graph_runtime_state.total_tokens
-
-                # Update trigger log with success
-                trigger_log.status = self._STATUS_MAP[type(event)]
-                trigger_log.workflow_run_id = workflow_run_id
-                trigger_log.outputs = TypeAdapter(dict[str, Any]).dump_json(outputs).decode()
-
-                if trigger_log.elapsed_time is None:
-                    trigger_log.elapsed_time = elapsed_time
-                else:
-                    trigger_log.elapsed_time += elapsed_time
-
-                trigger_log.total_tokens = total_tokens
-                trigger_log.finished_at = datetime.now(UTC)
-                repo.update(trigger_log)
-                session.commit()
-
-    def on_graph_end(self, error: Exception | None) -> None:
-        pass

api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py

@@ -121,7 +121,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
         if self._application_generate_entity.app_config.app_mode != AppMode.COMPLETION:
             # start generate conversation name thread
             self._conversation_name_generate_thread = self._message_cycle_manager.generate_conversation_name(
-                conversation_id=self._conversation_id, query=self._application_generate_entity.query
+                conversation_id=self._conversation_id, query=self._application_generate_entity.query or ""
             )
 
         generator = self._wrapper_process_stream_response(trace_manager=self._application_generate_entity.trace_manager)

api/core/app/task_pipeline/message_cycle_manager.py

@@ -140,27 +140,7 @@ class MessageCycleManager:
         if not self._application_generate_entity.app_config.additional_features:
             raise ValueError("Additional features not found")
         if self._application_generate_entity.app_config.additional_features.show_retrieve_source:
-            merged_resources = [r for r in self._task_state.metadata.retriever_resources or [] if r]
-            existing_ids = {(r.dataset_id, r.document_id) for r in merged_resources if r.dataset_id and r.document_id}
-
-            # Add new unique resources from the event
-            for resource in event.retriever_resources or []:
-                if not resource:
-                    continue
-
-                is_duplicate = (
-                    resource.dataset_id
-                    and resource.document_id
-                    and (resource.dataset_id, resource.document_id) in existing_ids
-                )
-
-                if not is_duplicate:
-                    merged_resources.append(resource)
-
-            for i, resource in enumerate(merged_resources, 1):
-                resource.position = i
-
-            self._task_state.metadata.retriever_resources = merged_resources
+            self._task_state.metadata.retriever_resources = event.retriever_resources
 
     def message_file_to_stream_response(self, event: QueueMessageFileEvent) -> MessageFileStreamResponse | None:
         """

api/core/entities/document_task.py

@@ -1,15 +0,0 @@
-from collections.abc import Sequence
-from dataclasses import dataclass
-
-
-@dataclass
-class DocumentTask:
-    """Document task entity for document indexing operations.
-
-    This class represents a document indexing task that can be queued
-    and processed by the document indexing system.
-    """
-
-    tenant_id: str
-    dataset_id: str
-    document_ids: Sequence[str]

api/core/entities/mcp_provider.py

@@ -271,6 +271,7 @@ class MCPProviderEntity(BaseModel):
 
     def _decrypt_dict(self, data: dict[str, Any]) -> dict[str, Any]:
         """Generic method to decrypt dictionary fields"""
+        # Lazy import to avoid circular dependency
         from core.tools.utils.encryption import create_provider_encrypter
 
         if not data:

api/core/entities/parameter_entities.py

@@ -14,7 +14,6 @@ class CommonParameterType(StrEnum):
     APP_SELECTOR = "app-selector"
     MODEL_SELECTOR = "model-selector"
     TOOLS_SELECTOR = "array[tools]"
-    CHECKBOX = "checkbox"
     ANY = auto()
 
     # Dynamic select parameter

api/core/entities/provider_configuration.py

@@ -1533,9 +1533,6 @@ class ProviderConfiguration(BaseModel):
             # Return composite sort key: (model_type value, model position index)
             return (model.model_type.value, position_index)
 
-        # Deduplicate
-        provider_models = list({(m.model, m.model_type, m.fetch_from): m for m in provider_models}.values())
-
         # Sort using the composite sort key
         return sorted(provider_models, key=get_sort_key)
 

api/core/entities/provider_entities.py

@@ -107,7 +107,7 @@ class CustomModelConfiguration(BaseModel):
 
     model: str
     model_type: ModelType
-    credentials: dict | None
+    credentials: dict | None = None
     current_credential_id: str | None = None
     current_credential_name: str | None = None
     available_model_credentials: list[CredentialConfiguration] = []
@@ -207,7 +207,6 @@ class ProviderConfig(BasicProviderConfig):
     required: bool = False
     default: Union[int, str, float, bool] | None = None
     options: list[Option] | None = None
-    multiple: bool | None = False
     label: I18nObject | None = None
     help: I18nObject | None = None
     url: str | None = None

api/core/file/models.py

@@ -74,10 +74,6 @@ class File(BaseModel):
         storage_key: str | None = None,
         dify_model_identity: str | None = FILE_MODEL_IDENTITY,
         url: str | None = None,
-        # Legacy compatibility fields - explicitly handle known extra fields
-        tool_file_id: str | None = None,
-        upload_file_id: str | None = None,
-        datasource_file_id: str | None = None,
     ):
         super().__init__(
             id=id,

api/core/helper/code_executor/javascript/javascript_transformer.py

@@ -6,7 +6,10 @@ from core.helper.code_executor.template_transformer import TemplateTransformer
 class NodeJsTemplateTransformer(TemplateTransformer):
     @classmethod
     def get_runner_script(cls) -> str:
-        runner_script = dedent(f"""            {cls._code_placeholder}
+        runner_script = dedent(
+            f"""
+            // declare main function
+            {cls._code_placeholder}
 
             // decode and prepare input object
             var inputs_obj = JSON.parse(Buffer.from('{cls._inputs_placeholder}', 'base64').toString('utf-8'))
@@ -18,5 +21,6 @@ class NodeJsTemplateTransformer(TemplateTransformer):
             var output_json = JSON.stringify(output_obj)
             var result = `<<RESULT>>${{output_json}}<<RESULT>>`
             console.log(result)
-            """)
+            """
+        )
         return runner_script

api/core/helper/code_executor/python3/python3_transformer.py

@@ -6,7 +6,9 @@ from core.helper.code_executor.template_transformer import TemplateTransformer
 class Python3TemplateTransformer(TemplateTransformer):
     @classmethod
     def get_runner_script(cls) -> str:
-        runner_script = dedent(f"""            {cls._code_placeholder}
+        runner_script = dedent(f"""
+            # declare main function
+            {cls._code_placeholder}
 
             import json
             from base64 import b64decode

api/core/helper/marketplace.py

@@ -29,18 +29,6 @@ def batch_fetch_plugin_manifests(plugin_ids: list[str]) -> Sequence[MarketplaceP
     return [MarketplacePluginDeclaration.model_validate(plugin) for plugin in response.json()["data"]["plugins"]]
 
 
-def batch_fetch_plugin_by_ids(plugin_ids: list[str]) -> list[dict]:
-    if not plugin_ids:
-        return []
-
-    url = str(marketplace_api_url / "api/v1/plugins/batch")
-    response = httpx.post(url, json={"plugin_ids": plugin_ids}, headers={"X-Dify-Version": dify_config.project.version})
-    response.raise_for_status()
-
-    data = response.json()
-    return data.get("data", {}).get("plugins", [])
-
-
 def batch_fetch_plugin_manifests_ignore_deserialization_error(
     plugin_ids: list[str],
 ) -> Sequence[MarketplacePluginDeclaration]:

api/core/helper/name_generator.py

@@ -3,7 +3,7 @@ import re
 from collections.abc import Sequence
 from typing import Any
 
-from core.plugin.entities.plugin_daemon import CredentialType
+from core.tools.entities.tool_entities import CredentialType
 
 logger = logging.getLogger(__name__)
 

api/core/helper/provider_encryption.py

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

api/core/model_runtime/entities/llm_entities.py

@@ -38,8 +38,6 @@ class LLMUsageMetadata(TypedDict, total=False):
     prompt_price: Union[float, str]
     completion_price: Union[float, str]
     latency: float
-    time_to_first_token: float
-    time_to_generate: float
 
 
 class LLMUsage(ModelUsage):
@@ -59,8 +57,6 @@ class LLMUsage(ModelUsage):
     total_price: Decimal
     currency: str
     latency: float
-    time_to_first_token: float | None = None
-    time_to_generate: float | None = None
 
     @classmethod
     def empty_usage(cls):
@@ -77,8 +73,6 @@ class LLMUsage(ModelUsage):
             total_price=Decimal("0.0"),
             currency="USD",
             latency=0.0,
-            time_to_first_token=None,
-            time_to_generate=None,
         )
 
     @classmethod
@@ -114,8 +108,6 @@ class LLMUsage(ModelUsage):
             prompt_price=Decimal(str(metadata.get("prompt_price", 0))),
             completion_price=Decimal(str(metadata.get("completion_price", 0))),
             latency=metadata.get("latency", 0.0),
-            time_to_first_token=metadata.get("time_to_first_token"),
-            time_to_generate=metadata.get("time_to_generate"),
         )
 
     def plus(self, other: LLMUsage) -> LLMUsage:
@@ -141,8 +133,6 @@ class LLMUsage(ModelUsage):
                 total_price=self.total_price + other.total_price,
                 currency=other.currency,
                 latency=self.latency + other.latency,
-                time_to_first_token=other.time_to_first_token,
-                time_to_generate=other.time_to_generate,
             )
 
     def __add__(self, other: LLMUsage) -> LLMUsage:

api/core/ops/arize_phoenix_trace/arize_phoenix_trace.py

@@ -1,22 +1,21 @@
+import hashlib
 import json
 import logging
 import os
-import traceback
 from datetime import datetime, timedelta
 from typing import Any, Union, cast
 from urllib.parse import urlparse
 
-from openinference.semconv.trace import OpenInferenceMimeTypeValues, OpenInferenceSpanKindValues, SpanAttributes
+from openinference.semconv.trace import OpenInferenceSpanKindValues, SpanAttributes
+from opentelemetry import trace
 from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter as GrpcOTLPSpanExporter
 from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter as HttpOTLPSpanExporter
 from opentelemetry.sdk import trace as trace_sdk
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.sdk.trace.export import SimpleSpanProcessor
-from opentelemetry.semconv.trace import SpanAttributes as OTELSpanAttributes
-from opentelemetry.trace import Span, Status, StatusCode, set_span_in_context, use_span
-from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
-from opentelemetry.util.types import AttributeValue
-from sqlalchemy.orm import sessionmaker
+from opentelemetry.sdk.trace.id_generator import RandomIdGenerator
+from opentelemetry.trace import SpanContext, TraceFlags, TraceState
+from sqlalchemy import select
 
 from core.ops.base_trace_instance import BaseTraceInstance
 from core.ops.entities.config_entity import ArizeConfig, PhoenixConfig
@@ -31,10 +30,9 @@ from core.ops.entities.trace_entity import (
     TraceTaskName,
     WorkflowTraceInfo,
 )
-from core.repositories import DifyCoreRepositoryFactory
 from extensions.ext_database import db
 from models.model import EndUser, MessageFile
-from models.workflow import WorkflowNodeExecutionTriggeredFrom
+from models.workflow import WorkflowNodeExecutionModel
 
 logger = logging.getLogger(__name__)
 
@@ -101,45 +99,22 @@ def datetime_to_nanos(dt: datetime | None) -> int:
     return int(dt.timestamp() * 1_000_000_000)
 
 
-def error_to_string(error: Exception | str | None) -> str:
-    """Convert an error to a string with traceback information."""
-    error_message = "Empty Stack Trace"
-    if error:
-        if isinstance(error, Exception):
-            string_stacktrace = "".join(traceback.format_exception(error))
-            error_message = f"{error.__class__.__name__}: {error}\n\n{string_stacktrace}"
-        else:
-            error_message = str(error)
-    return error_message
+def string_to_trace_id128(string: str | None) -> int:
+    """
+    Convert any input string into a stable 128-bit integer trace ID.
 
+    This uses SHA-256 hashing and takes the first 16 bytes (128 bits) of the digest.
+    It's suitable for generating consistent, unique identifiers from strings.
+    """
+    if string is None:
+        string = ""
+    hash_object = hashlib.sha256(string.encode())
 
-def set_span_status(current_span: Span, error: Exception | str | None = None):
-    """Set the status of the current span based on the presence of an error."""
-    if error:
-        error_string = error_to_string(error)
-        current_span.set_status(Status(StatusCode.ERROR, error_string))
+    # Take the first 16 bytes (128 bits) of the hash digest
+    digest = hash_object.digest()[:16]
 
-        if isinstance(error, Exception):
-            current_span.record_exception(error)
-        else:
-            exception_type = error.__class__.__name__
-            exception_message = str(error)
-            if not exception_message:
-                exception_message = repr(error)
-            attributes: dict[str, AttributeValue] = {
-                OTELSpanAttributes.EXCEPTION_TYPE: exception_type,
-                OTELSpanAttributes.EXCEPTION_MESSAGE: exception_message,
-                OTELSpanAttributes.EXCEPTION_ESCAPED: False,
-                OTELSpanAttributes.EXCEPTION_STACKTRACE: error_string,
-            }
-            current_span.add_event(name="exception", attributes=attributes)
-    else:
-        current_span.set_status(Status(StatusCode.OK))
-
-
-def safe_json_dumps(obj: Any) -> str:
-    """A convenience wrapper around `json.dumps` that ensures that any object can be safely encoded."""
-    return json.dumps(obj, default=str, ensure_ascii=False)
+    # Convert to a 128-bit integer
+    return int.from_bytes(digest, byteorder="big")
 
 
 class ArizePhoenixDataTrace(BaseTraceInstance):
@@ -156,12 +131,9 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
         self.tracer, self.processor = setup_tracer(arize_phoenix_config)
         self.project = arize_phoenix_config.project
         self.file_base_url = os.getenv("FILES_URL", "http://127.0.0.1:5001")
-        self.propagator = TraceContextTextMapPropagator()
-        self.dify_trace_ids: set[str] = set()
 
     def trace(self, trace_info: BaseTraceInfo):
-        logger.info("[Arize/Phoenix] Trace Entity Info: %s", trace_info)
-        logger.info("[Arize/Phoenix] Trace Entity Type: %s", type(trace_info))
+        logger.info("[Arize/Phoenix] Trace: %s", trace_info)
         try:
             if isinstance(trace_info, WorkflowTraceInfo):
                 self.workflow_trace(trace_info)
@@ -179,7 +151,7 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
                 self.generate_name_trace(trace_info)
 
         except Exception as e:
-            logger.error("[Arize/Phoenix] Trace Entity Error: %s", str(e), exc_info=True)
+            logger.error("[Arize/Phoenix] Error in the trace: %s", str(e), exc_info=True)
             raise
 
     def workflow_trace(self, trace_info: WorkflowTraceInfo):
@@ -194,9 +166,15 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
         }
         workflow_metadata.update(trace_info.metadata)
 
-        dify_trace_id = trace_info.trace_id or trace_info.message_id or trace_info.workflow_run_id
-        self.ensure_root_span(dify_trace_id)
-        root_span_context = self.propagator.extract(carrier=self.carrier)
+        trace_id = string_to_trace_id128(trace_info.trace_id or trace_info.workflow_run_id)
+        span_id = RandomIdGenerator().generate_span_id()
+        context = SpanContext(
+            trace_id=trace_id,
+            span_id=span_id,
+            is_remote=False,
+            trace_flags=TraceFlags(TraceFlags.SAMPLED),
+            trace_state=TraceState(),
+        )
 
         workflow_span = self.tracer.start_span(
             name=TraceTaskName.WORKFLOW_TRACE.value,
@@ -208,58 +186,31 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
                 SpanAttributes.SESSION_ID: trace_info.conversation_id or "",
             },
             start_time=datetime_to_nanos(trace_info.start_time),
-            context=root_span_context,
-        )
-
-        # Through workflow_run_id, get all_nodes_execution using repository
-        session_factory = sessionmaker(bind=db.engine)
-
-        # Find the app's creator account
-        app_id = trace_info.metadata.get("app_id")
-        if not app_id:
-            raise ValueError("No app_id found in trace_info metadata")
-
-        service_account = self.get_service_account_with_tenant(app_id)
-
-        workflow_node_execution_repository = DifyCoreRepositoryFactory.create_workflow_node_execution_repository(
-            session_factory=session_factory,
-            user=service_account,
-            app_id=app_id,
-            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
-        )
-
-        # Get all executions for this workflow run
-        workflow_node_executions = workflow_node_execution_repository.get_by_workflow_run(
-            workflow_run_id=trace_info.workflow_run_id
+            context=trace.set_span_in_context(trace.NonRecordingSpan(context)),
         )
 
         try:
-            for node_execution in workflow_node_executions:
-                tenant_id = trace_info.tenant_id  # Use from trace_info instead
-                app_id = trace_info.metadata.get("app_id")  # Use from trace_info instead
-                inputs_value = node_execution.inputs or {}
-                outputs_value = node_execution.outputs or {}
-
+            # Process workflow nodes
+            for node_execution in self._get_workflow_nodes(trace_info.workflow_run_id):
                 created_at = node_execution.created_at or datetime.now()
                 elapsed_time = node_execution.elapsed_time
                 finished_at = created_at + timedelta(seconds=elapsed_time)
 
-                process_data = node_execution.process_data or {}
-                execution_metadata = node_execution.metadata or {}
-                node_metadata = {str(k): v for k, v in execution_metadata.items()}
+                process_data = json.loads(node_execution.process_data) if node_execution.process_data else {}
 
-                node_metadata.update(
-                    {
-                        "node_id": node_execution.id,
-                        "node_type": node_execution.node_type,
-                        "node_status": node_execution.status,
-                        "tenant_id": tenant_id,
-                        "app_id": app_id,
-                        "app_name": node_execution.title,
-                        "status": node_execution.status,
-                        "level": "ERROR" if node_execution.status == "failed" else "DEFAULT",
-                    }
-                )
+                node_metadata = {
+                    "node_id": node_execution.id,
+                    "node_type": node_execution.node_type,
+                    "node_status": node_execution.status,
+                    "tenant_id": node_execution.tenant_id,
+                    "app_id": node_execution.app_id,
+                    "app_name": node_execution.title,
+                    "status": node_execution.status,
+                    "level": "ERROR" if node_execution.status != "succeeded" else "DEFAULT",
+                }
+
+                if node_execution.execution_metadata:
+                    node_metadata.update(json.loads(node_execution.execution_metadata))
 
                 # Determine the correct span kind based on node type
                 span_kind = OpenInferenceSpanKindValues.CHAIN
@@ -272,9 +223,8 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
                     if model:
                         node_metadata["ls_model_name"] = model
 
-                    usage_data = (
-                        process_data.get("usage", {}) if "usage" in process_data else outputs_value.get("usage", {})
-                    )
+                    outputs = json.loads(node_execution.outputs).get("usage", {}) if "outputs" in node_execution else {}
+                    usage_data = process_data.get("usage", {}) if "usage" in process_data else outputs.get("usage", {})
                     if usage_data:
                         node_metadata["total_tokens"] = usage_data.get("total_tokens", 0)
                         node_metadata["prompt_tokens"] = usage_data.get("prompt_tokens", 0)
@@ -286,20 +236,17 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
                 else:
                     span_kind = OpenInferenceSpanKindValues.CHAIN
 
-                workflow_span_context = set_span_in_context(workflow_span)
                 node_span = self.tracer.start_span(
                     name=node_execution.node_type,
                     attributes={
-                        SpanAttributes.INPUT_VALUE: safe_json_dumps(inputs_value),
-                        SpanAttributes.INPUT_MIME_TYPE: OpenInferenceMimeTypeValues.JSON.value,
-                        SpanAttributes.OUTPUT_VALUE: safe_json_dumps(outputs_value),
-                        SpanAttributes.OUTPUT_MIME_TYPE: OpenInferenceMimeTypeValues.JSON.value,
+                        SpanAttributes.INPUT_VALUE: node_execution.inputs or "{}",
+                        SpanAttributes.OUTPUT_VALUE: node_execution.outputs or "{}",
                         SpanAttributes.OPENINFERENCE_SPAN_KIND: span_kind.value,
-                        SpanAttributes.METADATA: safe_json_dumps(node_metadata),
+                        SpanAttributes.METADATA: json.dumps(node_metadata, ensure_ascii=False),
                         SpanAttributes.SESSION_ID: trace_info.conversation_id or "",
                     },
                     start_time=datetime_to_nanos(created_at),
-                    context=workflow_span_context,
+                    context=trace.set_span_in_context(trace.NonRecordingSpan(context)),
                 )
 
                 try:
@@ -313,8 +260,11 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
                             llm_attributes[SpanAttributes.LLM_PROVIDER] = provider
                         if model:
                             llm_attributes[SpanAttributes.LLM_MODEL_NAME] = model
+                        outputs = (
+                            json.loads(node_execution.outputs).get("usage", {}) if "outputs" in node_execution else {}
+                        )
                         usage_data = (
-                            process_data.get("usage", {}) if "usage" in process_data else outputs_value.get("usage", {})
+                            process_data.get("usage", {}) if "usage" in process_data else outputs.get("usage", {})
                         )
                         if usage_data:
                             llm_attributes[SpanAttributes.LLM_TOKEN_COUNT_TOTAL] = usage_data.get("total_tokens", 0)
@@ -325,16 +275,8 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
                         llm_attributes.update(self._construct_llm_attributes(process_data.get("prompts", [])))
                         node_span.set_attributes(llm_attributes)
                 finally:
-                    if node_execution.status == "failed":
-                        set_span_status(node_span, node_execution.error)
-                    else:
-                        set_span_status(node_span)
                     node_span.end(end_time=datetime_to_nanos(finished_at))
         finally:
-            if trace_info.error:
-                set_span_status(workflow_span, trace_info.error)
-            else:
-                set_span_status(workflow_span)
             workflow_span.end(end_time=datetime_to_nanos(trace_info.end_time))
 
     def message_trace(self, trace_info: MessageTraceInfo):
@@ -380,18 +322,34 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
             SpanAttributes.SESSION_ID: trace_info.message_data.conversation_id,
         }
 
-        dify_trace_id = trace_info.trace_id or trace_info.message_id
-        self.ensure_root_span(dify_trace_id)
-        root_span_context = self.propagator.extract(carrier=self.carrier)
+        trace_id = string_to_trace_id128(trace_info.trace_id or trace_info.message_id)
+        message_span_id = RandomIdGenerator().generate_span_id()
+        span_context = SpanContext(
+            trace_id=trace_id,
+            span_id=message_span_id,
+            is_remote=False,
+            trace_flags=TraceFlags(TraceFlags.SAMPLED),
+            trace_state=TraceState(),
+        )
 
         message_span = self.tracer.start_span(
             name=TraceTaskName.MESSAGE_TRACE.value,
             attributes=attributes,
             start_time=datetime_to_nanos(trace_info.start_time),
-            context=root_span_context,
+            context=trace.set_span_in_context(trace.NonRecordingSpan(span_context)),
         )
 
         try:
+            if trace_info.error:
+                message_span.add_event(
+                    "exception",
+                    attributes={
+                        "exception.message": trace_info.error,
+                        "exception.type": "Error",
+                        "exception.stacktrace": trace_info.error,
+                    },
+                )
+
             # Convert outputs to string based on type
             if isinstance(trace_info.outputs, dict | list):
                 outputs_str = json.dumps(trace_info.outputs, ensure_ascii=False)
@@ -425,26 +383,26 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
                 if model_params := metadata_dict.get("model_parameters"):
                     llm_attributes[SpanAttributes.LLM_INVOCATION_PARAMETERS] = json.dumps(model_params)
 
-            message_span_context = set_span_in_context(message_span)
             llm_span = self.tracer.start_span(
                 name="llm",
                 attributes=llm_attributes,
                 start_time=datetime_to_nanos(trace_info.start_time),
-                context=message_span_context,
+                context=trace.set_span_in_context(trace.NonRecordingSpan(span_context)),
             )
 
             try:
-                if trace_info.message_data.error:
-                    set_span_status(llm_span, trace_info.message_data.error)
-                else:
-                    set_span_status(llm_span)
+                if trace_info.error:
+                    llm_span.add_event(
+                        "exception",
+                        attributes={
+                            "exception.message": trace_info.error,
+                            "exception.type": "Error",
+                            "exception.stacktrace": trace_info.error,
+                        },
+                    )
             finally:
                 llm_span.end(end_time=datetime_to_nanos(trace_info.end_time))
         finally:
-            if trace_info.error:
-                set_span_status(message_span, trace_info.error)
-            else:
-                set_span_status(message_span)
             message_span.end(end_time=datetime_to_nanos(trace_info.end_time))
 
     def moderation_trace(self, trace_info: ModerationTraceInfo):
@@ -460,9 +418,15 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
         }
         metadata.update(trace_info.metadata)
 
-        dify_trace_id = trace_info.trace_id or trace_info.message_id
-        self.ensure_root_span(dify_trace_id)
-        root_span_context = self.propagator.extract(carrier=self.carrier)
+        trace_id = string_to_trace_id128(trace_info.message_id)
+        span_id = RandomIdGenerator().generate_span_id()
+        context = SpanContext(
+            trace_id=trace_id,
+            span_id=span_id,
+            is_remote=False,
+            trace_flags=TraceFlags(TraceFlags.SAMPLED),
+            trace_state=TraceState(),
+        )
 
         span = self.tracer.start_span(
             name=TraceTaskName.MODERATION_TRACE.value,
@@ -481,14 +445,19 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
                 SpanAttributes.METADATA: json.dumps(metadata, ensure_ascii=False),
             },
             start_time=datetime_to_nanos(trace_info.start_time),
-            context=root_span_context,
+            context=trace.set_span_in_context(trace.NonRecordingSpan(context)),
         )
 
         try:
             if trace_info.message_data.error:
-                set_span_status(span, trace_info.message_data.error)
-            else:
-                set_span_status(span)
+                span.add_event(
+                    "exception",
+                    attributes={
+                        "exception.message": trace_info.message_data.error,
+                        "exception.type": "Error",
+                        "exception.stacktrace": trace_info.message_data.error,
+                    },
+                )
         finally:
             span.end(end_time=datetime_to_nanos(trace_info.end_time))
 
@@ -511,9 +480,15 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
         }
         metadata.update(trace_info.metadata)
 
-        dify_trace_id = trace_info.trace_id or trace_info.message_id
-        self.ensure_root_span(dify_trace_id)
-        root_span_context = self.propagator.extract(carrier=self.carrier)
+        trace_id = string_to_trace_id128(trace_info.message_id)
+        span_id = RandomIdGenerator().generate_span_id()
+        context = SpanContext(
+            trace_id=trace_id,
+            span_id=span_id,
+            is_remote=False,
+            trace_flags=TraceFlags(TraceFlags.SAMPLED),
+            trace_state=TraceState(),
+        )
 
         span = self.tracer.start_span(
             name=TraceTaskName.SUGGESTED_QUESTION_TRACE.value,
@@ -524,14 +499,19 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
                 SpanAttributes.METADATA: json.dumps(metadata, ensure_ascii=False),
             },
             start_time=datetime_to_nanos(start_time),
-            context=root_span_context,
+            context=trace.set_span_in_context(trace.NonRecordingSpan(context)),
         )
 
         try:
             if trace_info.error:
-                set_span_status(span, trace_info.error)
-            else:
-                set_span_status(span)
+                span.add_event(
+                    "exception",
+                    attributes={
+                        "exception.message": trace_info.error,
+                        "exception.type": "Error",
+                        "exception.stacktrace": trace_info.error,
+                    },
+                )
         finally:
             span.end(end_time=datetime_to_nanos(end_time))
 
@@ -553,9 +533,15 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
         }
         metadata.update(trace_info.metadata)
 
-        dify_trace_id = trace_info.trace_id or trace_info.message_id
-        self.ensure_root_span(dify_trace_id)
-        root_span_context = self.propagator.extract(carrier=self.carrier)
+        trace_id = string_to_trace_id128(trace_info.message_id)
+        span_id = RandomIdGenerator().generate_span_id()
+        context = SpanContext(
+            trace_id=trace_id,
+            span_id=span_id,
+            is_remote=False,
+            trace_flags=TraceFlags(TraceFlags.SAMPLED),
+            trace_state=TraceState(),
+        )
 
         span = self.tracer.start_span(
             name=TraceTaskName.DATASET_RETRIEVAL_TRACE.value,
@@ -568,14 +554,19 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
                 "end_time": end_time.isoformat() if end_time else "",
             },
             start_time=datetime_to_nanos(start_time),
-            context=root_span_context,
+            context=trace.set_span_in_context(trace.NonRecordingSpan(context)),
         )
 
         try:
             if trace_info.message_data.error:
-                set_span_status(span, trace_info.message_data.error)
-            else:
-                set_span_status(span)
+                span.add_event(
+                    "exception",
+                    attributes={
+                        "exception.message": trace_info.message_data.error,
+                        "exception.type": "Error",
+                        "exception.stacktrace": trace_info.message_data.error,
+                    },
+                )
         finally:
             span.end(end_time=datetime_to_nanos(end_time))
 
@@ -589,9 +580,20 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
             "tool_config": json.dumps(trace_info.tool_config, ensure_ascii=False),
         }
 
-        dify_trace_id = trace_info.trace_id or trace_info.message_id
-        self.ensure_root_span(dify_trace_id)
-        root_span_context = self.propagator.extract(carrier=self.carrier)
+        trace_id = string_to_trace_id128(trace_info.message_id)
+        tool_span_id = RandomIdGenerator().generate_span_id()
+        logger.info("[Arize/Phoenix] Creating tool trace with trace_id: %s, span_id: %s", trace_id, tool_span_id)
+
+        # Create span context with the same trace_id as the parent
+        # todo: Create with the appropriate parent span context, so that the tool span is
+        # a child of the appropriate span (e.g. message span)
+        span_context = SpanContext(
+            trace_id=trace_id,
+            span_id=tool_span_id,
+            is_remote=False,
+            trace_flags=TraceFlags(TraceFlags.SAMPLED),
+            trace_state=TraceState(),
+        )
 
         tool_params_str = (
             json.dumps(trace_info.tool_parameters, ensure_ascii=False)
@@ -610,14 +612,19 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
                 SpanAttributes.TOOL_PARAMETERS: tool_params_str,
             },
             start_time=datetime_to_nanos(trace_info.start_time),
-            context=root_span_context,
+            context=trace.set_span_in_context(trace.NonRecordingSpan(span_context)),
         )
 
         try:
             if trace_info.error:
-                set_span_status(span, trace_info.error)
-            else:
-                set_span_status(span)
+                span.add_event(
+                    "exception",
+                    attributes={
+                        "exception.message": trace_info.error,
+                        "exception.type": "Error",
+                        "exception.stacktrace": trace_info.error,
+                    },
+                )
         finally:
             span.end(end_time=datetime_to_nanos(trace_info.end_time))
 
@@ -634,9 +641,15 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
         }
         metadata.update(trace_info.metadata)
 
-        dify_trace_id = trace_info.trace_id or trace_info.message_id or trace_info.conversation_id
-        self.ensure_root_span(dify_trace_id)
-        root_span_context = self.propagator.extract(carrier=self.carrier)
+        trace_id = string_to_trace_id128(trace_info.message_id)
+        span_id = RandomIdGenerator().generate_span_id()
+        context = SpanContext(
+            trace_id=trace_id,
+            span_id=span_id,
+            is_remote=False,
+            trace_flags=TraceFlags(TraceFlags.SAMPLED),
+            trace_state=TraceState(),
+        )
 
         span = self.tracer.start_span(
             name=TraceTaskName.GENERATE_NAME_TRACE.value,
@@ -650,34 +663,22 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
                 "end_time": trace_info.end_time.isoformat() if trace_info.end_time else "",
             },
             start_time=datetime_to_nanos(trace_info.start_time),
-            context=root_span_context,
+            context=trace.set_span_in_context(trace.NonRecordingSpan(context)),
         )
 
         try:
             if trace_info.message_data.error:
-                set_span_status(span, trace_info.message_data.error)
-            else:
-                set_span_status(span)
+                span.add_event(
+                    "exception",
+                    attributes={
+                        "exception.message": trace_info.message_data.error,
+                        "exception.type": "Error",
+                        "exception.stacktrace": trace_info.message_data.error,
+                    },
+                )
         finally:
             span.end(end_time=datetime_to_nanos(trace_info.end_time))
 
-    def ensure_root_span(self, dify_trace_id: str | None):
-        """Ensure a unique root span exists for the given Dify trace ID."""
-        if str(dify_trace_id) not in self.dify_trace_ids:
-            self.carrier: dict[str, str] = {}
-
-            root_span = self.tracer.start_span(name="Dify")
-            root_span.set_attribute(SpanAttributes.OPENINFERENCE_SPAN_KIND, OpenInferenceSpanKindValues.CHAIN.value)
-            root_span.set_attribute("dify_project_name", str(self.project))
-            root_span.set_attribute("dify_trace_id", str(dify_trace_id))
-
-            with use_span(root_span, end_on_exit=False):
-                self.propagator.inject(carrier=self.carrier)
-
-            set_span_status(root_span)
-            root_span.end()
-            self.dify_trace_ids.add(str(dify_trace_id))
-
     def api_check(self):
         try:
             with self.tracer.start_span("api_check") as span:
@@ -697,6 +698,26 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
             logger.info("[Arize/Phoenix] Get run url failed: %s", str(e), exc_info=True)
             raise ValueError(f"[Arize/Phoenix] Get run url failed: {str(e)}")
 
+    def _get_workflow_nodes(self, workflow_run_id: str):
+        """Helper method to get workflow nodes"""
+        workflow_nodes = db.session.scalars(
+            select(
+                WorkflowNodeExecutionModel.id,
+                WorkflowNodeExecutionModel.tenant_id,
+                WorkflowNodeExecutionModel.app_id,
+                WorkflowNodeExecutionModel.title,
+                WorkflowNodeExecutionModel.node_type,
+                WorkflowNodeExecutionModel.status,
+                WorkflowNodeExecutionModel.inputs,
+                WorkflowNodeExecutionModel.outputs,
+                WorkflowNodeExecutionModel.created_at,
+                WorkflowNodeExecutionModel.elapsed_time,
+                WorkflowNodeExecutionModel.process_data,
+                WorkflowNodeExecutionModel.execution_metadata,
+            ).where(WorkflowNodeExecutionModel.workflow_run_id == workflow_run_id)
+        ).all()
+        return workflow_nodes
+
     def _construct_llm_attributes(self, prompts: dict | list | str | None) -> dict[str, str]:
         """Helper method to construct LLM attributes with passed prompts."""
         attributes = {}

api/core/ops/entities/trace_entity.py

@@ -62,9 +62,6 @@ class MessageTraceInfo(BaseTraceInfo):
     file_list: Union[str, dict[str, Any], list] | None = None
     message_file_data: Any | None = None
     conversation_mode: str
-    gen_ai_server_time_to_first_token: float | None = None
-    llm_streaming_time_to_generate: float | None = None
-    is_streaming_request: bool = False
 
 
 class ModerationTraceInfo(BaseTraceInfo):

api/core/ops/ops_trace_manager.py

@@ -14,7 +14,7 @@ from flask import current_app
 from sqlalchemy import select
 from sqlalchemy.orm import Session, sessionmaker
 
-from core.helper.encrypter import batch_decrypt_token, encrypt_token, obfuscated_token
+from core.helper.encrypter import decrypt_token, encrypt_token, obfuscated_token
 from core.ops.entities.config_entity import (
     OPS_FILE_PATH,
     TracingProviderEnum,
@@ -141,8 +141,6 @@ provider_config_map = OpsTraceProviderConfigMap()
 
 class OpsTraceManager:
     ops_trace_instances_cache: LRUCache = LRUCache(maxsize=128)
-    decrypted_configs_cache: LRUCache = LRUCache(maxsize=128)
-    _decryption_cache_lock = threading.RLock()
 
     @classmethod
     def encrypt_tracing_config(
@@ -163,7 +161,7 @@ class OpsTraceManager:
             provider_config_map[tracing_provider]["other_keys"],
         )
 
-        new_config: dict[str, Any] = {}
+        new_config = {}
         # Encrypt necessary keys
         for key in secret_keys:
             if key in tracing_config:
@@ -193,41 +191,20 @@ class OpsTraceManager:
         :param tracing_config: tracing config
         :return:
         """
-        config_json = json.dumps(tracing_config, sort_keys=True)
-        decrypted_config_key = (
-            tenant_id,
-            tracing_provider,
-            config_json,
+        config_class, secret_keys, other_keys = (
+            provider_config_map[tracing_provider]["config_class"],
+            provider_config_map[tracing_provider]["secret_keys"],
+            provider_config_map[tracing_provider]["other_keys"],
         )
+        new_config = {}
+        for key in secret_keys:
+            if key in tracing_config:
+                new_config[key] = decrypt_token(tenant_id, tracing_config[key])
 
-        # First check without lock for performance
-        cached_config = cls.decrypted_configs_cache.get(decrypted_config_key)
-        if cached_config is not None:
-            return dict(cached_config)
+        for key in other_keys:
+            new_config[key] = tracing_config.get(key, "")
 
-        with cls._decryption_cache_lock:
-            # Second check (double-checked locking) to prevent race conditions
-            cached_config = cls.decrypted_configs_cache.get(decrypted_config_key)
-            if cached_config is not None:
-                return dict(cached_config)
-
-            config_class, secret_keys, other_keys = (
-                provider_config_map[tracing_provider]["config_class"],
-                provider_config_map[tracing_provider]["secret_keys"],
-                provider_config_map[tracing_provider]["other_keys"],
-            )
-            new_config: dict[str, Any] = {}
-            keys_to_decrypt = [key for key in secret_keys if key in tracing_config]
-            if keys_to_decrypt:
-                decrypted_values = batch_decrypt_token(tenant_id, [tracing_config[key] for key in keys_to_decrypt])
-                new_config.update(zip(keys_to_decrypt, decrypted_values))
-
-            for key in other_keys:
-                new_config[key] = tracing_config.get(key, "")
-
-            decrypted_config = config_class(**new_config).model_dump()
-            cls.decrypted_configs_cache[decrypted_config_key] = decrypted_config
-            return dict(decrypted_config)
+        return config_class(**new_config).model_dump()
 
     @classmethod
     def obfuscated_decrypt_token(cls, tracing_provider: str, decrypt_tracing_config: dict):
@@ -242,7 +219,7 @@ class OpsTraceManager:
             provider_config_map[tracing_provider]["secret_keys"],
             provider_config_map[tracing_provider]["other_keys"],
         )
-        new_config: dict[str, Any] = {}
+        new_config = {}
         for key in secret_keys:
             if key in decrypt_tracing_config:
                 new_config[key] = obfuscated_token(decrypt_tracing_config[key])
@@ -619,8 +596,6 @@ class TraceTask:
             file_url = f"{self.file_base_url}/{message_file_data.url}" if message_file_data else ""
             file_list.append(file_url)
 
-        streaming_metrics = self._extract_streaming_metrics(message_data)
-
         metadata = {
             "conversation_id": message_data.conversation_id,
             "ls_provider": message_data.model_provider,
@@ -653,9 +628,6 @@ class TraceTask:
             metadata=metadata,
             message_file_data=message_file_data,
             conversation_mode=conversation_mode,
-            gen_ai_server_time_to_first_token=streaming_metrics.get("gen_ai_server_time_to_first_token"),
-            llm_streaming_time_to_generate=streaming_metrics.get("llm_streaming_time_to_generate"),
-            is_streaming_request=streaming_metrics.get("is_streaming_request", False),
         )
 
         return message_trace_info
@@ -881,24 +853,6 @@ class TraceTask:
 
         return generate_name_trace_info
 
-    def _extract_streaming_metrics(self, message_data) -> dict:
-        if not message_data.message_metadata:
-            return {}
-
-        try:
-            metadata = json.loads(message_data.message_metadata)
-            usage = metadata.get("usage", {})
-            time_to_first_token = usage.get("time_to_first_token")
-            time_to_generate = usage.get("time_to_generate")
-
-            return {
-                "gen_ai_server_time_to_first_token": time_to_first_token,
-                "llm_streaming_time_to_generate": time_to_generate,
-                "is_streaming_request": time_to_first_token is not None,
-            }
-        except (json.JSONDecodeError, AttributeError):
-            return {}
-
 
 trace_manager_timer: threading.Timer | None = None
 trace_manager_queue: queue.Queue = queue.Queue()

api/core/ops/tencent_trace/client.py

@@ -5,18 +5,12 @@ Tencent APM Trace Client - handles network operations, metrics, and API communic
 from __future__ import annotations
 
 import importlib
-import json
 import logging
 import os
 import socket
 from typing import TYPE_CHECKING
 from urllib.parse import urlparse
 
-try:
-    from importlib.metadata import version
-except ImportError:
-    from importlib_metadata import version  # type: ignore[import-not-found]
-
 if TYPE_CHECKING:
     from opentelemetry.metrics import Meter
     from opentelemetry.metrics._internal.instrument import Histogram
@@ -33,27 +27,12 @@ from opentelemetry.util.types import AttributeValue
 
 from configs import dify_config
 
-from .entities.semconv import (
-    GEN_AI_SERVER_TIME_TO_FIRST_TOKEN,
-    GEN_AI_STREAMING_TIME_TO_GENERATE,
-    GEN_AI_TOKEN_USAGE,
-    GEN_AI_TRACE_DURATION,
-    LLM_OPERATION_DURATION,
-)
+from .entities.tencent_semconv import LLM_OPERATION_DURATION
 from .entities.tencent_trace_entity import SpanData
 
 logger = logging.getLogger(__name__)
 
 
-def _get_opentelemetry_sdk_version() -> str:
-    """Get OpenTelemetry SDK version dynamically."""
-    try:
-        return version("opentelemetry-sdk")
-    except Exception:
-        logger.debug("Failed to get opentelemetry-sdk version, using default")
-        return "1.27.0"  # fallback version
-
-
 class TencentTraceClient:
     """Tencent APM trace client using OpenTelemetry OTLP exporter"""
 
@@ -78,9 +57,6 @@ class TencentTraceClient:
                 ResourceAttributes.SERVICE_VERSION: f"dify-{dify_config.project.version}-{dify_config.COMMIT_SHA}",
                 ResourceAttributes.DEPLOYMENT_ENVIRONMENT: f"{dify_config.DEPLOY_ENV}-{dify_config.EDITION}",
                 ResourceAttributes.HOST_NAME: socket.gethostname(),
-                ResourceAttributes.TELEMETRY_SDK_LANGUAGE: "python",
-                ResourceAttributes.TELEMETRY_SDK_NAME: "opentelemetry",
-                ResourceAttributes.TELEMETRY_SDK_VERSION: _get_opentelemetry_sdk_version(),
             }
         )
         # Prepare gRPC endpoint/metadata
@@ -104,23 +80,18 @@ class TencentTraceClient:
         )
         self.tracer_provider.add_span_processor(self.span_processor)
 
-        # use dify api version as tracer version
-        self.tracer = self.tracer_provider.get_tracer("dify-sdk", dify_config.project.version)
+        self.tracer = self.tracer_provider.get_tracer("dify.tencent_apm")
 
         # Store span contexts for parent-child relationships
         self.span_contexts: dict[int, trace_api.SpanContext] = {}
 
         self.meter: Meter | None = None
-        self.meter_provider: MeterProvider | None = None
         self.hist_llm_duration: Histogram | None = None
-        self.hist_token_usage: Histogram | None = None
-        self.hist_time_to_first_token: Histogram | None = None
-        self.hist_time_to_generate: Histogram | None = None
-        self.hist_trace_duration: Histogram | None = None
         self.metric_reader: MetricReader | None = None
 
         # Metrics exporter and instruments
         try:
+            from opentelemetry import metrics
             from opentelemetry.sdk.metrics import Histogram, MeterProvider
             from opentelemetry.sdk.metrics.export import AggregationTemporality, PeriodicExportingMetricReader
 
@@ -128,7 +99,7 @@ class TencentTraceClient:
             use_http_protobuf = protocol in {"http/protobuf", "http-protobuf"}
             use_http_json = protocol in {"http/json", "http-json"}
 
-            # Tencent APM works best with delta aggregation temporality
+            # Set preferred temporality for histograms to DELTA
             preferred_temporality: dict[type, AggregationTemporality] = {Histogram: AggregationTemporality.DELTA}
 
             def _create_metric_exporter(exporter_cls, **kwargs):
@@ -203,66 +174,23 @@ class TencentTraceClient:
                 )
 
             if metric_reader is not None:
-                # Use instance-level MeterProvider instead of global to support config changes
-                # without worker restart. Each TencentTraceClient manages its own MeterProvider.
                 provider = MeterProvider(resource=self.resource, metric_readers=[metric_reader])
-                self.meter_provider = provider
-                self.meter = provider.get_meter("dify-sdk", dify_config.project.version)
-
-                # LLM operation duration histogram
+                metrics.set_meter_provider(provider)
+                self.meter = metrics.get_meter("dify-sdk", dify_config.project.version)
                 self.hist_llm_duration = self.meter.create_histogram(
                     name=LLM_OPERATION_DURATION,
                     unit="s",
                     description="LLM operation duration (seconds)",
                 )
-
-                # Token usage histogram with exponential buckets
-                self.hist_token_usage = self.meter.create_histogram(
-                    name=GEN_AI_TOKEN_USAGE,
-                    unit="token",
-                    description="Number of tokens used in prompt and completions",
-                )
-
-                # Time to first token histogram
-                self.hist_time_to_first_token = self.meter.create_histogram(
-                    name=GEN_AI_SERVER_TIME_TO_FIRST_TOKEN,
-                    unit="s",
-                    description="Time to first token for streaming LLM responses (seconds)",
-                )
-
-                # Time to generate histogram
-                self.hist_time_to_generate = self.meter.create_histogram(
-                    name=GEN_AI_STREAMING_TIME_TO_GENERATE,
-                    unit="s",
-                    description="Total time to generate streaming LLM responses (seconds)",
-                )
-
-                # Trace duration histogram
-                self.hist_trace_duration = self.meter.create_histogram(
-                    name=GEN_AI_TRACE_DURATION,
-                    unit="s",
-                    description="End-to-end GenAI trace duration (seconds)",
-                )
-
                 self.metric_reader = metric_reader
             else:
                 self.meter = None
-                self.meter_provider = None
                 self.hist_llm_duration = None
-                self.hist_token_usage = None
-                self.hist_time_to_first_token = None
-                self.hist_time_to_generate = None
-                self.hist_trace_duration = None
                 self.metric_reader = None
         except Exception:
             logger.exception("[Tencent APM] Metrics initialization failed; metrics disabled")
             self.meter = None
-            self.meter_provider = None
             self.hist_llm_duration = None
-            self.hist_token_usage = None
-            self.hist_time_to_first_token = None
-            self.hist_time_to_generate = None
-            self.hist_trace_duration = None
             self.metric_reader = None
 
     def add_span(self, span_data: SpanData) -> None:
@@ -284,158 +212,10 @@ class TencentTraceClient:
             if attributes:
                 for k, v in attributes.items():
                     attrs[k] = str(v) if not isinstance(v, (str, int, float, bool)) else v  # type: ignore[assignment]
-
-            logger.info(
-                "[Tencent Metrics] Metric: %s | Value: %.4f | Attributes: %s",
-                LLM_OPERATION_DURATION,
-                latency_seconds,
-                json.dumps(attrs, ensure_ascii=False),
-            )
-
             self.hist_llm_duration.record(latency_seconds, attrs)  # type: ignore[attr-defined]
         except Exception:
             logger.debug("[Tencent APM] Failed to record LLM duration", exc_info=True)
 
-    def record_token_usage(
-        self,
-        token_count: int,
-        token_type: str,
-        operation_name: str,
-        request_model: str,
-        response_model: str,
-        server_address: str,
-        provider: str,
-    ) -> None:
-        """Record token usage histogram.
-
-        Args:
-            token_count: Number of tokens used
-            token_type: "input" or "output"
-            operation_name: Operation name (e.g., "chat")
-            request_model: Model used in request
-            response_model: Model used in response
-            server_address: Server address
-            provider: Model provider name
-        """
-        try:
-            if not hasattr(self, "hist_token_usage") or self.hist_token_usage is None:
-                return
-
-            attributes = {
-                "gen_ai.operation.name": operation_name,
-                "gen_ai.request.model": request_model,
-                "gen_ai.response.model": response_model,
-                "gen_ai.system": provider,
-                "gen_ai.token.type": token_type,
-                "server.address": server_address,
-            }
-
-            logger.info(
-                "[Tencent Metrics] Metric: %s | Value: %d | Attributes: %s",
-                GEN_AI_TOKEN_USAGE,
-                token_count,
-                json.dumps(attributes, ensure_ascii=False),
-            )
-
-            self.hist_token_usage.record(token_count, attributes)  # type: ignore[attr-defined]
-        except Exception:
-            logger.debug("[Tencent APM] Failed to record token usage", exc_info=True)
-
-    def record_time_to_first_token(
-        self, ttft_seconds: float, provider: str, model: str, operation_name: str = "chat"
-    ) -> None:
-        """Record time to first token histogram.
-
-        Args:
-            ttft_seconds: Time to first token in seconds
-            provider: Model provider name
-            model: Model name
-            operation_name: Operation name (default: "chat")
-        """
-        try:
-            if not hasattr(self, "hist_time_to_first_token") or self.hist_time_to_first_token is None:
-                return
-
-            attributes = {
-                "gen_ai.operation.name": operation_name,
-                "gen_ai.system": provider,
-                "gen_ai.request.model": model,
-                "gen_ai.response.model": model,
-                "stream": "true",
-            }
-
-            logger.info(
-                "[Tencent Metrics] Metric: %s | Value: %.4f | Attributes: %s",
-                GEN_AI_SERVER_TIME_TO_FIRST_TOKEN,
-                ttft_seconds,
-                json.dumps(attributes, ensure_ascii=False),
-            )
-
-            self.hist_time_to_first_token.record(ttft_seconds, attributes)  # type: ignore[attr-defined]
-        except Exception:
-            logger.debug("[Tencent APM] Failed to record time to first token", exc_info=True)
-
-    def record_time_to_generate(
-        self, ttg_seconds: float, provider: str, model: str, operation_name: str = "chat"
-    ) -> None:
-        """Record time to generate histogram.
-
-        Args:
-            ttg_seconds: Time to generate in seconds
-            provider: Model provider name
-            model: Model name
-            operation_name: Operation name (default: "chat")
-        """
-        try:
-            if not hasattr(self, "hist_time_to_generate") or self.hist_time_to_generate is None:
-                return
-
-            attributes = {
-                "gen_ai.operation.name": operation_name,
-                "gen_ai.system": provider,
-                "gen_ai.request.model": model,
-                "gen_ai.response.model": model,
-                "stream": "true",
-            }
-
-            logger.info(
-                "[Tencent Metrics] Metric: %s | Value: %.4f | Attributes: %s",
-                GEN_AI_STREAMING_TIME_TO_GENERATE,
-                ttg_seconds,
-                json.dumps(attributes, ensure_ascii=False),
-            )
-
-            self.hist_time_to_generate.record(ttg_seconds, attributes)  # type: ignore[attr-defined]
-        except Exception:
-            logger.debug("[Tencent APM] Failed to record time to generate", exc_info=True)
-
-    def record_trace_duration(self, duration_seconds: float, attributes: dict[str, str] | None = None) -> None:
-        """Record end-to-end trace duration histogram in seconds.
-
-        Args:
-            duration_seconds: Trace duration in seconds
-            attributes: Optional attributes (e.g., conversation_mode, app_id)
-        """
-        try:
-            if not hasattr(self, "hist_trace_duration") or self.hist_trace_duration is None:
-                return
-
-            attrs: dict[str, str] = {}
-            if attributes:
-                for k, v in attributes.items():
-                    attrs[k] = str(v) if not isinstance(v, (str, int, float, bool)) else v  # type: ignore[assignment]
-
-            logger.info(
-                "[Tencent Metrics] Metric: %s | Value: %.4f | Attributes: %s",
-                GEN_AI_TRACE_DURATION,
-                duration_seconds,
-                json.dumps(attrs, ensure_ascii=False),
-            )
-
-            self.hist_trace_duration.record(duration_seconds, attrs)  # type: ignore[attr-defined]
-        except Exception:
-            logger.debug("[Tencent APM] Failed to record trace duration", exc_info=True)
-
     def _create_and_export_span(self, span_data: SpanData) -> None:
         """Create span using OpenTelemetry Tracer API"""
         try:
@@ -516,19 +296,11 @@ class TencentTraceClient:
 
             if self.tracer_provider:
                 self.tracer_provider.shutdown()
-
-            # Shutdown instance-level meter provider
-            if self.meter_provider is not None:
-                try:
-                    self.meter_provider.shutdown()  # type: ignore[attr-defined]
-                except Exception:
-                    logger.debug("[Tencent APM] Error shutting down meter provider", exc_info=True)
-
             if self.metric_reader is not None:
                 try:
                     self.metric_reader.shutdown()  # type: ignore[attr-defined]
                 except Exception:
-                    logger.debug("[Tencent APM] Error shutting down metric reader", exc_info=True)
+                    pass
 
         except Exception:
             logger.exception("[Tencent APM] Error during client shutdown")

api/core/ops/tencent_trace/entities/tencent_semconv.py

@@ -47,9 +47,6 @@ GEN_AI_COMPLETION = "gen_ai.completion"
 
 GEN_AI_RESPONSE_FINISH_REASON = "gen_ai.response.finish_reason"
 
-# Streaming Span Attributes
-GEN_AI_IS_STREAMING_REQUEST = "llm.is_streaming"  # Same as OpenLLMetry semconv
-
 # Tool
 TOOL_NAME = "tool.name"
 
@@ -65,19 +62,6 @@ INSTRUMENTATION_LANGUAGE = "python"
 
 # Metrics
 LLM_OPERATION_DURATION = "gen_ai.client.operation.duration"
-GEN_AI_TOKEN_USAGE = "gen_ai.client.token.usage"
-GEN_AI_SERVER_TIME_TO_FIRST_TOKEN = "gen_ai.server.time_to_first_token"
-GEN_AI_STREAMING_TIME_TO_GENERATE = "gen_ai.streaming.time_to_generate"
-# The LLM trace duration which is exclusive to tencent apm
-GEN_AI_TRACE_DURATION = "gen_ai.trace.duration"
-
-# Token Usage Attributes
-GEN_AI_OPERATION_NAME = "gen_ai.operation.name"
-GEN_AI_REQUEST_MODEL = "gen_ai.request.model"
-GEN_AI_RESPONSE_MODEL = "gen_ai.response.model"
-GEN_AI_SYSTEM = "gen_ai.system"
-GEN_AI_TOKEN_TYPE = "gen_ai.token.type"
-SERVER_ADDRESS = "server.address"
 
 
 class GenAISpanKind(Enum):

api/core/ops/tencent_trace/span_builder.py

@@ -14,11 +14,10 @@ from core.ops.entities.trace_entity import (
     ToolTraceInfo,
     WorkflowTraceInfo,
 )
-from core.ops.tencent_trace.entities.semconv import (
+from core.ops.tencent_trace.entities.tencent_semconv import (
     GEN_AI_COMPLETION,
     GEN_AI_FRAMEWORK,
     GEN_AI_IS_ENTRY,
-    GEN_AI_IS_STREAMING_REQUEST,
     GEN_AI_MODEL_NAME,
     GEN_AI_PROMPT,
     GEN_AI_PROVIDER,
@@ -157,25 +156,6 @@ class TencentSpanBuilder:
         outputs = node_execution.outputs or {}
         usage_data = process_data.get("usage", {}) if "usage" in process_data else outputs.get("usage", {})
 
-        attributes = {
-            GEN_AI_SESSION_ID: trace_info.metadata.get("conversation_id", ""),
-            GEN_AI_SPAN_KIND: GenAISpanKind.GENERATION.value,
-            GEN_AI_FRAMEWORK: "dify",
-            GEN_AI_MODEL_NAME: process_data.get("model_name", ""),
-            GEN_AI_PROVIDER: process_data.get("model_provider", ""),
-            GEN_AI_USAGE_INPUT_TOKENS: str(usage_data.get("prompt_tokens", 0)),
-            GEN_AI_USAGE_OUTPUT_TOKENS: str(usage_data.get("completion_tokens", 0)),
-            GEN_AI_USAGE_TOTAL_TOKENS: str(usage_data.get("total_tokens", 0)),
-            GEN_AI_PROMPT: json.dumps(process_data.get("prompts", []), ensure_ascii=False),
-            GEN_AI_COMPLETION: str(outputs.get("text", "")),
-            GEN_AI_RESPONSE_FINISH_REASON: outputs.get("finish_reason", ""),
-            INPUT_VALUE: json.dumps(process_data.get("prompts", []), ensure_ascii=False),
-            OUTPUT_VALUE: str(outputs.get("text", "")),
-        }
-
-        if usage_data.get("time_to_first_token") is not None:
-            attributes[GEN_AI_IS_STREAMING_REQUEST] = "true"
-
         return SpanData(
             trace_id=trace_id,
             parent_span_id=workflow_span_id,
@@ -183,7 +163,21 @@ class TencentSpanBuilder:
             name="GENERATION",
             start_time=TencentSpanBuilder._get_time_nanoseconds(node_execution.created_at),
             end_time=TencentSpanBuilder._get_time_nanoseconds(node_execution.finished_at),
-            attributes=attributes,
+            attributes={
+                GEN_AI_SESSION_ID: trace_info.metadata.get("conversation_id", ""),
+                GEN_AI_SPAN_KIND: GenAISpanKind.GENERATION.value,
+                GEN_AI_FRAMEWORK: "dify",
+                GEN_AI_MODEL_NAME: process_data.get("model_name", ""),
+                GEN_AI_PROVIDER: process_data.get("model_provider", ""),
+                GEN_AI_USAGE_INPUT_TOKENS: str(usage_data.get("prompt_tokens", 0)),
+                GEN_AI_USAGE_OUTPUT_TOKENS: str(usage_data.get("completion_tokens", 0)),
+                GEN_AI_USAGE_TOTAL_TOKENS: str(usage_data.get("total_tokens", 0)),
+                GEN_AI_PROMPT: json.dumps(process_data.get("prompts", []), ensure_ascii=False),
+                GEN_AI_COMPLETION: str(outputs.get("text", "")),
+                GEN_AI_RESPONSE_FINISH_REASON: outputs.get("finish_reason", ""),
+                INPUT_VALUE: json.dumps(process_data.get("prompts", []), ensure_ascii=False),
+                OUTPUT_VALUE: str(outputs.get("text", "")),
+            },
             status=TencentSpanBuilder._get_workflow_node_status(node_execution),
         )
 
@@ -197,19 +191,6 @@ class TencentSpanBuilder:
         if trace_info.error:
             status = Status(StatusCode.ERROR, trace_info.error)
 
-        attributes = {
-            GEN_AI_SESSION_ID: trace_info.metadata.get("conversation_id", ""),
-            GEN_AI_USER_ID: str(user_id),
-            GEN_AI_SPAN_KIND: GenAISpanKind.WORKFLOW.value,
-            GEN_AI_FRAMEWORK: "dify",
-            GEN_AI_IS_ENTRY: "true",
-            INPUT_VALUE: str(trace_info.inputs or ""),
-            OUTPUT_VALUE: str(trace_info.outputs or ""),
-        }
-
-        if trace_info.is_streaming_request:
-            attributes[GEN_AI_IS_STREAMING_REQUEST] = "true"
-
         return SpanData(
             trace_id=trace_id,
             parent_span_id=None,
@@ -217,7 +198,15 @@ class TencentSpanBuilder:
             name="message",
             start_time=TencentSpanBuilder._get_time_nanoseconds(trace_info.start_time),
             end_time=TencentSpanBuilder._get_time_nanoseconds(trace_info.end_time),
-            attributes=attributes,
+            attributes={
+                GEN_AI_SESSION_ID: trace_info.metadata.get("conversation_id", ""),
+                GEN_AI_USER_ID: str(user_id),
+                GEN_AI_SPAN_KIND: GenAISpanKind.WORKFLOW.value,
+                GEN_AI_FRAMEWORK: "dify",
+                GEN_AI_IS_ENTRY: "true",
+                INPUT_VALUE: str(trace_info.inputs or ""),
+                OUTPUT_VALUE: str(trace_info.outputs or ""),
+            },
             status=status,
             links=links,
         )

api/core/ops/tencent_trace/tencent_trace.py

@@ -90,9 +90,6 @@ class TencentDataTrace(BaseTraceInstance):
 
             self._process_workflow_nodes(trace_info, trace_id)
 
-            # Record trace duration for entry span
-            self._record_workflow_trace_duration(trace_info)
-
         except Exception:
             logger.exception("[Tencent APM] Failed to process workflow trace")
 
@@ -110,11 +107,6 @@ class TencentDataTrace(BaseTraceInstance):
 
             self.trace_client.add_span(message_span)
 
-            self._record_message_llm_metrics(trace_info)
-
-            # Record trace duration for entry span
-            self._record_message_trace_duration(trace_info)
-
         except Exception:
             logger.exception("[Tencent APM] Failed to process message trace")
 
@@ -298,219 +290,24 @@ class TencentDataTrace(BaseTraceInstance):
     def _record_llm_metrics(self, node_execution: WorkflowNodeExecution) -> None:
         """Record LLM performance metrics"""
         try:
+            if not hasattr(self.trace_client, "record_llm_duration"):
+                return
+
             process_data = node_execution.process_data or {}
-            outputs = node_execution.outputs or {}
-            usage = process_data.get("usage", {}) if "usage" in process_data else outputs.get("usage", {})
+            usage = process_data.get("usage", {})
+            latency_s = float(usage.get("latency", 0.0))
 
-            model_provider = process_data.get("model_provider", "unknown")
-            model_name = process_data.get("model_name", "unknown")
-            model_mode = process_data.get("model_mode", "chat")
-
-            # Record LLM duration
-            if hasattr(self.trace_client, "record_llm_duration"):
-                latency_s = float(usage.get("latency", 0.0))
-
-                if latency_s > 0:
-                    # Determine if streaming from usage metrics
-                    is_streaming = usage.get("time_to_first_token") is not None
-
-                    attributes = {
-                        "gen_ai.system": model_provider,
-                        "gen_ai.response.model": model_name,
-                        "gen_ai.operation.name": model_mode,
-                        "stream": "true" if is_streaming else "false",
-                    }
-                    self.trace_client.record_llm_duration(latency_s, attributes)
-
-            # Record streaming metrics from usage
-            time_to_first_token = usage.get("time_to_first_token")
-            if time_to_first_token is not None and hasattr(self.trace_client, "record_time_to_first_token"):
-                ttft_seconds = float(time_to_first_token)
-                if ttft_seconds > 0:
-                    self.trace_client.record_time_to_first_token(
-                        ttft_seconds=ttft_seconds, provider=model_provider, model=model_name, operation_name=model_mode
-                    )
-
-            time_to_generate = usage.get("time_to_generate")
-            if time_to_generate is not None and hasattr(self.trace_client, "record_time_to_generate"):
-                ttg_seconds = float(time_to_generate)
-                if ttg_seconds > 0:
-                    self.trace_client.record_time_to_generate(
-                        ttg_seconds=ttg_seconds, provider=model_provider, model=model_name, operation_name=model_mode
-                    )
-
-            # Record token usage
-            if hasattr(self.trace_client, "record_token_usage"):
-                # Extract token counts
-                input_tokens = int(usage.get("prompt_tokens", 0))
-                output_tokens = int(usage.get("completion_tokens", 0))
-
-                if input_tokens > 0 or output_tokens > 0:
-                    server_address = f"{model_provider}"
-
-                    # Record input tokens
-                    if input_tokens > 0:
-                        self.trace_client.record_token_usage(
-                            token_count=input_tokens,
-                            token_type="input",
-                            operation_name=model_mode,
-                            request_model=model_name,
-                            response_model=model_name,
-                            server_address=server_address,
-                            provider=model_provider,
-                        )
-
-                    # Record output tokens
-                    if output_tokens > 0:
-                        self.trace_client.record_token_usage(
-                            token_count=output_tokens,
-                            token_type="output",
-                            operation_name=model_mode,
-                            request_model=model_name,
-                            response_model=model_name,
-                            server_address=server_address,
-                            provider=model_provider,
-                        )
+            if latency_s > 0:
+                attributes = {
+                    "provider": process_data.get("model_provider", ""),
+                    "model": process_data.get("model_name", ""),
+                    "span_kind": "GENERATION",
+                }
+                self.trace_client.record_llm_duration(latency_s, attributes)
 
         except Exception:
             logger.debug("[Tencent APM] Failed to record LLM metrics")
 
-    def _record_message_llm_metrics(self, trace_info: MessageTraceInfo) -> None:
-        """Record LLM metrics for message traces"""
-        try:
-            trace_metadata = trace_info.metadata or {}
-            message_data = trace_info.message_data or {}
-            provider_latency = 0.0
-            if isinstance(message_data, dict):
-                provider_latency = float(message_data.get("provider_response_latency", 0.0) or 0.0)
-            else:
-                provider_latency = float(getattr(message_data, "provider_response_latency", 0.0) or 0.0)
-
-            model_provider = trace_metadata.get("ls_provider") or (
-                message_data.get("model_provider", "") if isinstance(message_data, dict) else ""
-            )
-            model_name = trace_metadata.get("ls_model_name") or (
-                message_data.get("model_id", "") if isinstance(message_data, dict) else ""
-            )
-
-            # Record LLM duration
-            if provider_latency > 0 and hasattr(self.trace_client, "record_llm_duration"):
-                is_streaming = trace_info.is_streaming_request
-
-                duration_attributes = {
-                    "gen_ai.system": model_provider,
-                    "gen_ai.response.model": model_name,
-                    "gen_ai.operation.name": "chat",  # Message traces are always chat
-                    "stream": "true" if is_streaming else "false",
-                }
-                self.trace_client.record_llm_duration(provider_latency, duration_attributes)
-
-            # Record streaming metrics for message traces
-            if trace_info.is_streaming_request:
-                # Record time to first token
-                if trace_info.gen_ai_server_time_to_first_token is not None and hasattr(
-                    self.trace_client, "record_time_to_first_token"
-                ):
-                    ttft_seconds = float(trace_info.gen_ai_server_time_to_first_token)
-                    if ttft_seconds > 0:
-                        self.trace_client.record_time_to_first_token(
-                            ttft_seconds=ttft_seconds, provider=str(model_provider or ""), model=str(model_name or "")
-                        )
-
-                # Record time to generate
-                if trace_info.llm_streaming_time_to_generate is not None and hasattr(
-                    self.trace_client, "record_time_to_generate"
-                ):
-                    ttg_seconds = float(trace_info.llm_streaming_time_to_generate)
-                    if ttg_seconds > 0:
-                        self.trace_client.record_time_to_generate(
-                            ttg_seconds=ttg_seconds, provider=str(model_provider or ""), model=str(model_name or "")
-                        )
-
-            # Record token usage
-            if hasattr(self.trace_client, "record_token_usage"):
-                input_tokens = int(trace_info.message_tokens or 0)
-                output_tokens = int(trace_info.answer_tokens or 0)
-
-                if input_tokens > 0:
-                    self.trace_client.record_token_usage(
-                        token_count=input_tokens,
-                        token_type="input",
-                        operation_name="chat",
-                        request_model=str(model_name or ""),
-                        response_model=str(model_name or ""),
-                        server_address=str(model_provider or ""),
-                        provider=str(model_provider or ""),
-                    )
-
-                if output_tokens > 0:
-                    self.trace_client.record_token_usage(
-                        token_count=output_tokens,
-                        token_type="output",
-                        operation_name="chat",
-                        request_model=str(model_name or ""),
-                        response_model=str(model_name or ""),
-                        server_address=str(model_provider or ""),
-                        provider=str(model_provider or ""),
-                    )
-
-        except Exception:
-            logger.debug("[Tencent APM] Failed to record message LLM metrics")
-
-    def _record_workflow_trace_duration(self, trace_info: WorkflowTraceInfo) -> None:
-        """Record end-to-end workflow trace duration."""
-        try:
-            if not hasattr(self.trace_client, "record_trace_duration"):
-                return
-
-            # Calculate duration from start_time and end_time to match span duration
-            if trace_info.start_time and trace_info.end_time:
-                duration_s = (trace_info.end_time - trace_info.start_time).total_seconds()
-            else:
-                # Fallback to workflow_run_elapsed_time if timestamps not available
-                duration_s = float(trace_info.workflow_run_elapsed_time)
-
-            if duration_s > 0:
-                attributes = {
-                    "conversation_mode": "workflow",
-                    "workflow_status": trace_info.workflow_run_status,
-                }
-
-                # Add conversation_id if available
-                if trace_info.conversation_id:
-                    attributes["has_conversation"] = "true"
-                else:
-                    attributes["has_conversation"] = "false"
-
-                self.trace_client.record_trace_duration(duration_s, attributes)
-
-        except Exception:
-            logger.debug("[Tencent APM] Failed to record workflow trace duration")
-
-    def _record_message_trace_duration(self, trace_info: MessageTraceInfo) -> None:
-        """Record end-to-end message trace duration."""
-        try:
-            if not hasattr(self.trace_client, "record_trace_duration"):
-                return
-
-            # Calculate duration from start_time and end_time
-            if trace_info.start_time and trace_info.end_time:
-                duration = (trace_info.end_time - trace_info.start_time).total_seconds()
-
-                if duration > 0:
-                    attributes = {
-                        "conversation_mode": trace_info.conversation_mode,
-                    }
-
-                    # Add streaming flag if available
-                    if hasattr(trace_info, "is_streaming_request"):
-                        attributes["stream"] = "true" if trace_info.is_streaming_request else "false"
-
-                    self.trace_client.record_trace_duration(duration, attributes)
-
-        except Exception:
-            logger.debug("[Tencent APM] Failed to record message trace duration")
-
     def __del__(self):
         """Ensure proper cleanup on garbage collection."""
         try:

api/core/plugin/backwards_invocation/app.py

@@ -4,6 +4,7 @@ from typing import Union
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
+from controllers.service_api.wraps import create_or_update_end_user_for_user_id
 from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
 from core.app.apps.advanced_chat.app_generator import AdvancedChatAppGenerator
 from core.app.apps.agent_chat.app_generator import AgentChatAppGenerator
@@ -15,7 +16,6 @@ from core.plugin.backwards_invocation.base import BaseBackwardsInvocation
 from extensions.ext_database import db
 from models import Account
 from models.model import App, AppMode, EndUser
-from services.end_user_service import EndUserService
 
 
 class PluginAppBackwardsInvocation(BaseBackwardsInvocation):
@@ -64,7 +64,7 @@ class PluginAppBackwardsInvocation(BaseBackwardsInvocation):
         """
         app = cls._get_app(app_id, tenant_id)
         if not user_id:
-            user = EndUserService.get_or_create_end_user(app)
+            user = create_or_update_end_user_for_user_id(app)
         else:
             user = cls._get_user(user_id)
 

api/core/plugin/entities/parameters.py

@@ -39,7 +39,7 @@ class PluginParameterType(StrEnum):
     TOOLS_SELECTOR = CommonParameterType.TOOLS_SELECTOR
     ANY = CommonParameterType.ANY
     DYNAMIC_SELECT = CommonParameterType.DYNAMIC_SELECT
-    CHECKBOX = CommonParameterType.CHECKBOX
+
     # deprecated, should not use.
     SYSTEM_FILES = CommonParameterType.SYSTEM_FILES
 
@@ -94,7 +94,6 @@ def as_normal_type(typ: StrEnum):
     if typ.value in {
         PluginParameterType.SECRET_INPUT,
         PluginParameterType.SELECT,
-        PluginParameterType.CHECKBOX,
     }:
         return "string"
     return typ.value
@@ -103,13 +102,7 @@ def as_normal_type(typ: StrEnum):
 def cast_parameter_value(typ: StrEnum, value: Any, /):
     try:
         match typ.value:
-            case (
-                PluginParameterType.STRING
-                | PluginParameterType.SECRET_INPUT
-                | PluginParameterType.SELECT
-                | PluginParameterType.CHECKBOX
-                | PluginParameterType.DYNAMIC_SELECT
-            ):
+            case PluginParameterType.STRING | PluginParameterType.SECRET_INPUT | PluginParameterType.SELECT:
                 if value is None:
                     return ""
                 else:

api/core/plugin/entities/plugin.py

@@ -13,7 +13,6 @@ from core.plugin.entities.base import BasePluginEntity
 from core.plugin.entities.endpoint import EndpointProviderDeclaration
 from core.tools.entities.common_entities import I18nObject
 from core.tools.entities.tool_entities import ToolProviderEntity
-from core.trigger.entities.entities import TriggerProviderEntity
 
 
 class PluginInstallationSource(StrEnum):
@@ -64,7 +63,6 @@ class PluginCategory(StrEnum):
     Extension = auto()
     AgentStrategy = "agent-strategy"
     Datasource = "datasource"
-    Trigger = "trigger"
 
 
 class PluginDeclaration(BaseModel):
@@ -73,7 +71,6 @@ class PluginDeclaration(BaseModel):
         models: list[str] | None = Field(default_factory=list[str])
         endpoints: list[str] | None = Field(default_factory=list[str])
         datasources: list[str] | None = Field(default_factory=list[str])
-        triggers: list[str] | None = Field(default_factory=list[str])
 
     class Meta(BaseModel):
         minimum_dify_version: str | None = Field(default=None)
@@ -109,7 +106,6 @@ class PluginDeclaration(BaseModel):
     endpoint: EndpointProviderDeclaration | None = None
     agent_strategy: AgentStrategyProviderEntity | None = None
     datasource: DatasourceProviderEntity | None = None
-    trigger: TriggerProviderEntity | None = None
     meta: Meta
 
     @field_validator("version")
@@ -133,8 +129,6 @@ class PluginDeclaration(BaseModel):
             values["category"] = PluginCategory.Datasource
         elif values.get("agent_strategy"):
             values["category"] = PluginCategory.AgentStrategy
-        elif values.get("trigger"):
-            values["category"] = PluginCategory.Trigger
         else:
             values["category"] = PluginCategory.Extension
         return values

api/core/plugin/entities/plugin_daemon.py

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

api/core/plugin/entities/request.py

@@ -1,9 +1,5 @@
-import binascii
-import json
-from collections.abc import Mapping
 from typing import Any, Literal
 
-from flask import Response
 from pydantic import BaseModel, ConfigDict, Field, field_validator
 
 from core.entities.provider_entities import BasicProviderConfig
@@ -17,7 +13,6 @@ from core.model_runtime.entities.message_entities import (
     UserPromptMessage,
 )
 from core.model_runtime.entities.model_entities import ModelType
-from core.plugin.utils.http_parser import deserialize_response
 from core.workflow.nodes.parameter_extractor.entities import (
     ModelConfig as ParameterExtractorModelConfig,
 )
@@ -242,43 +237,3 @@ class RequestFetchAppInfo(BaseModel):
     """
 
     app_id: str
-
-
-class TriggerInvokeEventResponse(BaseModel):
-    variables: Mapping[str, Any] = Field(default_factory=dict)
-    cancelled: bool = Field(default=False)
-
-    model_config = ConfigDict(protected_namespaces=(), arbitrary_types_allowed=True)
-
-    @field_validator("variables", mode="before")
-    @classmethod
-    def convert_variables(cls, v):
-        if isinstance(v, str):
-            return json.loads(v)
-        else:
-            return v
-
-
-class TriggerSubscriptionResponse(BaseModel):
-    subscription: dict[str, Any]
-
-
-class TriggerValidateProviderCredentialsResponse(BaseModel):
-    result: bool
-
-
-class TriggerDispatchResponse(BaseModel):
-    user_id: str
-    events: list[str]
-    response: Response
-    payload: Mapping[str, Any] = Field(default_factory=dict)
-
-    model_config = ConfigDict(protected_namespaces=(), arbitrary_types_allowed=True)
-
-    @field_validator("response", mode="before")
-    @classmethod
-    def convert_response(cls, v: str):
-        try:
-            return deserialize_response(binascii.unhexlify(v.encode()))
-        except Exception as e:
-            raise ValueError("Failed to deserialize response from hex string") from e

api/core/plugin/impl/asset.py

@@ -10,13 +10,3 @@ class PluginAssetManager(BasePluginClient):
         if response.status_code != 200:
             raise ValueError(f"can not found asset {id}")
         return response.content
-
-    def extract_asset(self, tenant_id: str, plugin_unique_identifier: str, filename: str) -> bytes:
-        response = self._request(
-            method="GET",
-            path=f"plugin/{tenant_id}/extract-asset/",
-            params={"plugin_unique_identifier": plugin_unique_identifier, "file_path": filename},
-        )
-        if response.status_code != 200:
-            raise ValueError(f"can not found asset {plugin_unique_identifier}, {str(response.status_code)}")
-        return response.content

api/core/plugin/impl/base.py

@@ -29,12 +29,6 @@ from core.plugin.impl.exc import (
     PluginPermissionDeniedError,
     PluginUniqueIdentifierError,
 )
-from core.trigger.errors import (
-    EventIgnoreError,
-    TriggerInvokeError,
-    TriggerPluginInvokeError,
-    TriggerProviderCredentialValidationError,
-)
 
 plugin_daemon_inner_api_baseurl = URL(str(dify_config.PLUGIN_DAEMON_URL))
 _plugin_daemon_timeout_config = cast(
@@ -49,7 +43,7 @@ elif isinstance(_plugin_daemon_timeout_config, httpx.Timeout):
 else:
     plugin_daemon_request_timeout = httpx.Timeout(_plugin_daemon_timeout_config)
 
-T = TypeVar("T", bound=(BaseModel | dict[str, Any] | list[Any] | bool | str))
+T = TypeVar("T", bound=(BaseModel | dict | list | bool | str))
 
 logger = logging.getLogger(__name__)
 
@@ -59,10 +53,10 @@ class BasePluginClient:
         self,
         method: str,
         path: str,
-        headers: dict[str, str] | None = None,
-        data: bytes | dict[str, Any] | str | None = None,
-        params: dict[str, Any] | None = None,
-        files: dict[str, Any] | None = None,
+        headers: dict | None = None,
+        data: bytes | dict | str | None = None,
+        params: dict | None = None,
+        files: dict | None = None,
     ) -> httpx.Response:
         """
         Make a request to the plugin daemon inner API.
@@ -93,17 +87,17 @@ class BasePluginClient:
     def _prepare_request(
         self,
         path: str,
-        headers: dict[str, str] | None,
-        data: bytes | dict[str, Any] | str | None,
-        params: dict[str, Any] | None,
-        files: dict[str, Any] | None,
-    ) -> tuple[str, dict[str, str], bytes | dict[str, Any] | str | None, dict[str, Any] | None, dict[str, Any] | None]:
+        headers: dict | None,
+        data: bytes | dict | str | None,
+        params: dict | None,
+        files: dict | None,
+    ) -> tuple[str, dict, bytes | dict | str | None, dict | None, dict | None]:
         url = plugin_daemon_inner_api_baseurl / path
         prepared_headers = dict(headers or {})
         prepared_headers["X-Api-Key"] = dify_config.PLUGIN_DAEMON_KEY
         prepared_headers.setdefault("Accept-Encoding", "gzip, deflate, br")
 
-        prepared_data: bytes | dict[str, Any] | str | None = (
+        prepared_data: bytes | dict | str | None = (
             data if isinstance(data, (bytes, str, dict)) or data is None else None
         )
         if isinstance(data, dict):
@@ -118,10 +112,10 @@ class BasePluginClient:
         self,
         method: str,
         path: str,
-        params: dict[str, Any] | None = None,
-        headers: dict[str, str] | None = None,
-        data: bytes | dict[str, Any] | None = None,
-        files: dict[str, Any] | None = None,
+        params: dict | None = None,
+        headers: dict | None = None,
+        data: bytes | dict | None = None,
+        files: dict | None = None,
     ) -> Generator[str, None, None]:
         """
         Make a stream request to the plugin daemon inner API
@@ -144,7 +138,7 @@ class BasePluginClient:
         try:
             with httpx.stream(**stream_kwargs) as response:
                 for raw_line in response.iter_lines():
-                    if not raw_line:
+                    if raw_line is None:
                         continue
                     line = raw_line.decode("utf-8") if isinstance(raw_line, bytes) else raw_line
                     line = line.strip()
@@ -161,10 +155,10 @@ class BasePluginClient:
         method: str,
         path: str,
         type_: type[T],
-        headers: dict[str, str] | None = None,
-        data: bytes | dict[str, Any] | None = None,
-        params: dict[str, Any] | None = None,
-        files: dict[str, Any] | None = None,
+        headers: dict | None = None,
+        data: bytes | dict | None = None,
+        params: dict | None = None,
+        files: dict | None = None,
     ) -> Generator[T, None, None]:
         """
         Make a stream request to the plugin daemon inner API and yield the response as a model.
@@ -177,10 +171,10 @@ class BasePluginClient:
         method: str,
         path: str,
         type_: type[T],
-        headers: dict[str, str] | None = None,
+        headers: dict | None = None,
         data: bytes | None = None,
-        params: dict[str, Any] | None = None,
-        files: dict[str, Any] | None = None,
+        params: dict | None = None,
+        files: dict | None = None,
     ) -> T:
         """
         Make a request to the plugin daemon inner API and return the response as a model.
@@ -193,11 +187,11 @@ class BasePluginClient:
         method: str,
         path: str,
         type_: type[T],
-        headers: dict[str, str] | None = None,
-        data: bytes | dict[str, Any] | None = None,
-        params: dict[str, Any] | None = None,
-        files: dict[str, Any] | None = None,
-        transformer: Callable[[dict[str, Any]], dict[str, Any]] | None = None,
+        headers: dict | None = None,
+        data: bytes | dict | None = None,
+        params: dict | None = None,
+        files: dict | None = None,
+        transformer: Callable[[dict], dict] | None = None,
     ) -> T:
         """
         Make a request to the plugin daemon inner API and return the response as a model.
@@ -245,10 +239,10 @@ class BasePluginClient:
         method: str,
         path: str,
         type_: type[T],
-        headers: dict[str, str] | None = None,
-        data: bytes | dict[str, Any] | None = None,
-        params: dict[str, Any] | None = None,
-        files: dict[str, Any] | None = None,
+        headers: dict | None = None,
+        data: bytes | dict | None = None,
+        params: dict | None = None,
+        files: dict | None = None,
     ) -> Generator[T, None, None]:
         """
         Make a stream request to the plugin daemon inner API and yield the response as a model.
@@ -308,14 +302,6 @@ class BasePluginClient:
                         raise CredentialsValidateFailedError(error_object.get("message"))
                     case EndpointSetupFailedError.__name__:
                         raise EndpointSetupFailedError(error_object.get("message"))
-                    case TriggerProviderCredentialValidationError.__name__:
-                        raise TriggerProviderCredentialValidationError(error_object.get("message"))
-                    case TriggerPluginInvokeError.__name__:
-                        raise TriggerPluginInvokeError(description=error_object.get("description"))
-                    case TriggerInvokeError.__name__:
-                        raise TriggerInvokeError(error_object.get("message"))
-                    case EventIgnoreError.__name__:
-                        raise EventIgnoreError(description=error_object.get("description"))
                     case _:
                         raise PluginInvokeError(description=message)
             case PluginDaemonInternalServerError.__name__:

api/core/plugin/impl/dynamic_select.py

@@ -15,7 +15,6 @@ class DynamicSelectClient(BasePluginClient):
         provider: str,
         action: str,
         credentials: Mapping[str, Any],
-        credential_type: str,
         parameter: str,
     ) -> PluginDynamicSelectOptionsResponse:
         """
@@ -30,7 +29,6 @@ class DynamicSelectClient(BasePluginClient):
                 "data": {
                     "provider": GenericProviderID(provider).provider_name,
                     "credentials": credentials,
-                    "credential_type": credential_type,
                     "provider_action": action,
                     "parameter": parameter,
                 },

api/core/plugin/impl/exc.py

@@ -58,20 +58,6 @@ class PluginInvokeError(PluginDaemonClientSideError, ValueError):
         except Exception:
             return self.description
 
-    def to_user_friendly_error(self, plugin_name: str = "currently running plugin") -> str:
-        """
-        Convert the error to a user-friendly error message.
-
-        :param plugin_name: The name of the plugin that caused the error.
-        :return: A user-friendly error message.
-        """
-        return (
-            f"An error occurred in the {plugin_name}, "
-            f"please contact the author of {plugin_name} for help, "
-            f"error type: {self.get_error_type()}, "
-            f"error details: {self.get_error_message()}"
-        )
-
 
 class PluginUniqueIdentifierError(PluginDaemonClientSideError):
     description: str = "Unique Identifier Error"

api/core/plugin/impl/plugin.py

@@ -1,7 +1,5 @@
 from collections.abc import Sequence
 
-from requests import HTTPError
-
 from core.plugin.entities.bundle import PluginBundleDependency
 from core.plugin.entities.plugin import (
     MissingPluginDependency,
@@ -15,35 +13,12 @@ from core.plugin.entities.plugin_daemon import (
     PluginInstallTask,
     PluginInstallTaskStartResponse,
     PluginListResponse,
-    PluginReadmeResponse,
 )
 from core.plugin.impl.base import BasePluginClient
 from models.provider_ids import GenericProviderID
 
 
 class PluginInstaller(BasePluginClient):
-    def fetch_plugin_readme(self, tenant_id: str, plugin_unique_identifier: str, language: str) -> str:
-        """
-        Fetch plugin readme
-        """
-        try:
-            response = self._request_with_plugin_daemon_response(
-                "GET",
-                f"plugin/{tenant_id}/management/fetch/readme",
-                PluginReadmeResponse,
-                params={
-                    "tenant_id": tenant_id,
-                    "plugin_unique_identifier": plugin_unique_identifier,
-                    "language": language,
-                },
-            )
-            return response.content
-        except HTTPError as e:
-            message = e.args[0]
-            if "404" in message:
-                return ""
-            raise e
-
     def fetch_plugin_by_identifier(
         self,
         tenant_id: str,

api/core/plugin/impl/tool.py

@@ -3,12 +3,14 @@ from typing import Any
 
 from pydantic import BaseModel
 
-# from core.plugin.entities.plugin import GenericProviderID, ToolProviderID
-from core.plugin.entities.plugin_daemon import CredentialType, PluginBasicBooleanResponse, PluginToolProviderEntity
+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.schemas.resolver import resolve_dify_schema_refs
-from core.tools.entities.tool_entities import ToolInvokeMessage, ToolParameter
+from core.tools.entities.tool_entities import CredentialType, ToolInvokeMessage, ToolParameter
 from models.provider_ids import GenericProviderID, ToolProviderID
 
 

api/core/plugin/impl/trigger.py

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

api/core/plugin/utils/http_parser.py

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

api/core/rag/datasource/vdb/elasticsearch/elasticsearch_vector.py

@@ -147,8 +147,7 @@ class ElasticSearchVector(BaseVector):
 
     def _get_version(self) -> str:
         info = self._client.info()
-        # remove any suffix like "-SNAPSHOT" from the version string
-        return cast(str, info["version"]["number"]).split("-")[0]
+        return cast(str, info["version"]["number"])
 
     def _check_version(self):
         if parse_version(self._version) < parse_version("8.0.0"):

api/core/rag/datasource/vdb/weaviate/weaviate_vector.py

@@ -39,13 +39,11 @@ class WeaviateConfig(BaseModel):
 
     Attributes:
         endpoint: Weaviate server endpoint URL
-        grpc_endpoint: Optional Weaviate gRPC server endpoint URL
         api_key: Optional API key for authentication
         batch_size: Number of objects to batch per insert operation
     """
 
     endpoint: str
-    grpc_endpoint: str | None = None
     api_key: str | None = None
     batch_size: int = 100
 
@@ -90,22 +88,9 @@ class WeaviateVector(BaseVector):
         http_secure = p.scheme == "https"
         http_port = p.port or (443 if http_secure else 80)
 
-        # Parse gRPC configuration
-        if config.grpc_endpoint:
-            # Urls without scheme won't be parsed correctly in some python versions,
-            # see https://bugs.python.org/issue27657
-            grpc_endpoint_with_scheme = (
-                config.grpc_endpoint if "://" in config.grpc_endpoint else f"grpc://{config.grpc_endpoint}"
-            )
-            grpc_p = urlparse(grpc_endpoint_with_scheme)
-            grpc_host = grpc_p.hostname or "localhost"
-            grpc_port = grpc_p.port or (443 if grpc_p.scheme == "grpcs" else 50051)
-            grpc_secure = grpc_p.scheme == "grpcs"
-        else:
-            # Infer from HTTP endpoint as fallback
-            grpc_host = host
-            grpc_secure = http_secure
-            grpc_port = 443 if grpc_secure else 50051
+        grpc_host = host
+        grpc_secure = http_secure
+        grpc_port = 443 if grpc_secure else 50051
 
         client = weaviate.connect_to_custom(
             http_host=host,
@@ -115,7 +100,6 @@ class WeaviateVector(BaseVector):
             grpc_port=grpc_port,
             grpc_secure=grpc_secure,
             auth_credentials=Auth.api_key(config.api_key) if config.api_key else None,
-            skip_init_checks=True,  # Skip PyPI version check to avoid unnecessary HTTP requests
         )
 
         if not client.is_ready():
@@ -447,7 +431,6 @@ class WeaviateVectorFactory(AbstractVectorFactory):
             collection_name=collection_name,
             config=WeaviateConfig(
                 endpoint=dify_config.WEAVIATE_ENDPOINT or "",
-                grpc_endpoint=dify_config.WEAVIATE_GRPC_ENDPOINT or "",
                 api_key=dify_config.WEAVIATE_API_KEY,
                 batch_size=dify_config.WEAVIATE_BATCH_SIZE,
             ),

api/core/rag/pipeline/queue.py

@@ -1,79 +0,0 @@
-import json
-from collections.abc import Sequence
-from typing import Any
-
-from pydantic import BaseModel, ValidationError
-
-from extensions.ext_redis import redis_client
-
-_DEFAULT_TASK_TTL = 60 * 60  # 1 hour
-
-
-class TaskWrapper(BaseModel):
-    data: Any
-
-    def serialize(self) -> str:
-        return self.model_dump_json()
-
-    @classmethod
-    def deserialize(cls, serialized_data: str) -> "TaskWrapper":
-        return cls.model_validate_json(serialized_data)
-
-
-class TenantIsolatedTaskQueue:
-    """
-    Simple queue for tenant isolated tasks, used for rag related tenant tasks isolation.
-    It uses Redis list to store tasks, and Redis key to store task waiting flag.
-    Support tasks that can be serialized by json.
-    """
-
-    def __init__(self, tenant_id: str, unique_key: str):
-        self._tenant_id = tenant_id
-        self._unique_key = unique_key
-        self._queue = f"tenant_self_{unique_key}_task_queue:{tenant_id}"
-        self._task_key = f"tenant_{unique_key}_task:{tenant_id}"
-
-    def get_task_key(self):
-        return redis_client.get(self._task_key)
-
-    def set_task_waiting_time(self, ttl: int = _DEFAULT_TASK_TTL):
-        redis_client.setex(self._task_key, ttl, 1)
-
-    def delete_task_key(self):
-        redis_client.delete(self._task_key)
-
-    def push_tasks(self, tasks: Sequence[Any]):
-        serialized_tasks = []
-        for task in tasks:
-            # Store str list directly, maintaining full compatibility for pipeline scenarios
-            if isinstance(task, str):
-                serialized_tasks.append(task)
-            else:
-                # Use TaskWrapper to do JSON serialization for non-string tasks
-                wrapper = TaskWrapper(data=task)
-                serialized_data = wrapper.serialize()
-                serialized_tasks.append(serialized_data)
-
-        redis_client.lpush(self._queue, *serialized_tasks)
-
-    def pull_tasks(self, count: int = 1) -> Sequence[Any]:
-        if count <= 0:
-            return []
-
-        tasks = []
-        for _ in range(count):
-            serialized_task = redis_client.rpop(self._queue)
-            if not serialized_task:
-                break
-
-            if isinstance(serialized_task, bytes):
-                serialized_task = serialized_task.decode("utf-8")
-
-            try:
-                wrapper = TaskWrapper.deserialize(serialized_task)
-                tasks.append(wrapper.data)
-            except (json.JSONDecodeError, ValidationError, TypeError, ValueError):
-                # Fall back to raw string for legacy format or invalid JSON
-                tasks.append(serialized_task)
-
-        return tasks

api/core/tools/__base/tool.py

@@ -210,13 +210,12 @@ class Tool(ABC):
             meta=meta,
         )
 
-    def create_json_message(self, object: dict, suppress_output: bool = False) -> ToolInvokeMessage:
+    def create_json_message(self, object: dict) -> ToolInvokeMessage:
         """
         create a json message
         """
         return ToolInvokeMessage(
-            type=ToolInvokeMessage.MessageType.JSON,
-            message=ToolInvokeMessage.JsonMessage(json_object=object, suppress_output=suppress_output),
+            type=ToolInvokeMessage.MessageType.JSON, message=ToolInvokeMessage.JsonMessage(json_object=object)
         )
 
     def create_variable_message(

api/core/tools/__base/tool_runtime.py

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