bump docker compose daemon version

This reverts commit 34d1f86f76.

.devcontainer/post_create_command.sh

@@ -1,6 +1,7 @@
 #!/bin/bash
 WORKSPACE_ROOT=$(pwd)
 
+npm add -g pnpm@10.15.0
 corepack enable
 cd web && pnpm install
 pipx install uv

.github/workflows/autofix.yml

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

.gitignore

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

api/.env.example

@@ -27,12 +27,12 @@ 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
 
-# Collaboration mode toggle
-ENABLE_COLLABORATION_MODE=false
-
 # Access token expiration time in minutes
 ACCESS_TOKEN_EXPIRE_MINUTES=60
 
@@ -374,12 +374,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,6 +463,9 @@ HTTP_REQUEST_NODE_MAX_BINARY_SIZE=10485760
 HTTP_REQUEST_NODE_MAX_TEXT_SIZE=1048576
 HTTP_REQUEST_NODE_SSL_VERIFY=True
 
+# Webhook request configuration
+WEBHOOK_REQUEST_BODY_MAX_SIZE=10485760
+
 # Respect X-* headers to redirect clients
 RESPECT_XFORWARD_HEADERS_ENABLED=false
 
@@ -546,6 +543,12 @@ ENABLE_CLEAN_MESSAGES=false
 ENABLE_MAIL_CLEAN_DOCUMENT_NOTIFY_TASK=false
 ENABLE_DATASETS_QUEUE_MONITOR=false
 ENABLE_CHECK_UPGRADABLE_PLUGIN_TASK=true
+ENABLE_WORKFLOW_SCHEDULE_POLLER_TASK=true
+# Interval time in minutes for polling scheduled workflows(default: 1 min)
+WORKFLOW_SCHEDULE_POLLER_INTERVAL=1
+WORKFLOW_SCHEDULE_POLLER_BATCH_SIZE=100
+# Maximum number of scheduled workflows to dispatch per tick (0 for unlimited)
+WORKFLOW_SCHEDULE_MAX_DISPATCH_PER_TICK=0
 
 # Position configuration
 POSITION_TOOL_PINS=
@@ -617,6 +620,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
-
-# 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,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,priority_pipeline,pipeline"
+                "dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor"
             ]
         }
     ]

api/AGENTS.md

@@ -0,0 +1,62 @@
+# 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/agent_skills/coding_style.md

@@ -0,0 +1,115 @@
+## 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

@@ -0,0 +1,96 @@
+## 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

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

api/agent_skills/plugin_oauth.md

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

api/agent_skills/trigger.md

@@ -0,0 +1,53 @@
+## 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,4 +1,3 @@
-import os
 import sys
 
 
@@ -9,16 +8,10 @@ def is_db_command():
 
 
 # create app
-celery = None
-flask_app = None
-socketio_app = None
-
 if is_db_command():
     from app_factory import create_migrations_app
 
     app = create_migrations_app()
-    socketio_app = app
-    flask_app = app
 else:
     # Gunicorn and Celery handle monkey patching automatically in production by
     # specifying the `gevent` worker class. Manual monkey patching is not required here.
@@ -29,15 +22,8 @@ else:
 
     from app_factory import create_app
 
-    socketio_app, flask_app = create_app()
-    app = flask_app
-    celery = flask_app.extensions["celery"]
+    app = create_app()
+    celery = app.extensions["celery"]
 
 if __name__ == "__main__":
-    from gevent import pywsgi
-    from geventwebsocket.handler import WebSocketHandler
-
-    host = os.environ.get("HOST", "0.0.0.0")
-    port = int(os.environ.get("PORT", 5001))
-    server = pywsgi.WSGIServer((host, port), socketio_app, handler_class=WebSocketHandler)
-    server.serve_forever()
+    app.run(host="0.0.0.0", port=5001)

api/app_factory.py

@@ -31,22 +31,14 @@ def create_flask_app_with_configs() -> DifyApp:
     return dify_app
 
 
-def create_app() -> tuple[any, DifyApp]:
+def create_app() -> DifyApp:
     start_time = time.perf_counter()
     app = create_flask_app_with_configs()
     initialize_extensions(app)
-
-    import socketio
-
-    from extensions.ext_socketio import sio
-
-    sio.app = app
-    socketio_app = socketio.WSGIApp(sio, app)
-
     end_time = time.perf_counter()
     if dify_config.DEBUG:
         logger.info("Finished create_app (%s ms)", round((end_time - start_time) * 1000, 2))
-    return socketio_app, app
+    return app
 
 
 def initialize_extensions(app: DifyApp):

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,6 +1229,55 @@ def setup_system_tool_oauth_client(provider, client_params):
     click.echo(click.style(f"OAuth client params setup successfully. id: {oauth_client.id}", fg="green"))
 
 
+@click.command("setup-system-trigger-oauth-client", help="Setup system trigger oauth client.")
+@click.option("--provider", prompt=True, help="Provider name")
+@click.option("--client-params", prompt=True, help="Client Params")
+def setup_system_trigger_oauth_client(provider, client_params):
+    """
+    Setup system trigger oauth client
+    """
+    from 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.
@@ -1601,7 +1650,7 @@ def transform_datasource_credentials():
                             "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,6 +174,33 @@ 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
@@ -263,6 +290,8 @@ 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):
     """
@@ -331,31 +360,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):
     """
@@ -871,16 +875,6 @@ class MailConfig(BaseSettings):
         default=None,
     )
 
-    ENABLE_TRIAL_APP: bool = Field(
-        description="Enable trial app",
-        default=False,
-    )
-
-    ENABLE_EXPLORE_BANNER: bool = Field(
-        description="Enable explore banner",
-        default=False,
-    )
-
 
 class RagEtlConfig(BaseSettings):
     """
@@ -955,11 +949,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):
     """
@@ -1035,6 +1024,44 @@ 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):
@@ -1093,13 +1120,6 @@ class PositionConfig(BaseSettings):
         return {item.strip() for item in self.POSITION_TOOL_EXCLUDES.split(",") if item.strip() != ""}
 
 
-class CollaborationConfig(BaseSettings):
-    ENABLE_COLLABORATION_MODE: bool = Field(
-        description="Whether to enable collaboration mode features across the workspace",
-        default=False,
-    )
-
-
 class LoginConfig(BaseSettings):
     ENABLE_EMAIL_CODE_LOGIN: bool = Field(
         description="whether to enable email code login",
@@ -1159,19 +1179,14 @@ class SwaggerUIConfig(BaseSettings):
     )
 
 
-class TenantSelfTaskQueueConfig(BaseSettings):
-    TENANT_SELF_TASK_QUEUE_PULL_SIZE: int = Field(
-        description="Default batch size for tenant self task queue pull operations",
-        default=1,
-    )
-
-
 class FeatureConfig(
     # place the configs in alphabet order
     AppExecutionConfig,
     AuthConfig,  # Changed from OAuthConfig to AuthConfig
     BillingConfig,
     CodeExecutionSandboxConfig,
+    TriggerConfig,
+    AsyncWorkflowConfig,
     PluginConfig,
     MarketplaceConfig,
     DataSetConfig,
@@ -1190,13 +1205,11 @@ class FeatureConfig(
     RagEtlConfig,
     RepositoryConfig,
     SecurityConfig,
-    TenantSelfTaskQueueConfig,
     ToolConfig,
     UpdateConfig,
     WorkflowConfig,
     WorkflowNodeExecutionConfig,
     WorkspaceConfig,
-    CollaborationConfig,
     LoginConfig,
     AccountConfig,
     SwaggerUIConfig,

api/configs/feature/hosted_service/__init__.py

@@ -8,11 +8,6 @@ class HostedCreditConfig(BaseSettings):
         default="",
     )
 
-    HOSTED_POOL_CREDITS: int = Field(
-        description="Pool credits for hosted service",
-        default=200,
-    )
-
     def get_model_credits(self, model_name: str) -> int:
         """
         Get credit value for a specific model name.
@@ -65,46 +60,19 @@ class HostedOpenAiConfig(BaseSettings):
 
     HOSTED_OPENAI_TRIAL_MODELS: str = Field(
         description="Comma-separated list of available models for trial access",
-        default="gpt-4,"
-        "gpt-4-turbo-preview,"
-        "gpt-4-turbo-2024-04-09,"
-        "gpt-4-1106-preview,"
-        "gpt-4-0125-preview,"
-        "gpt-4-turbo,"
-        "gpt-4.1,"
-        "gpt-4.1-2025-04-14,"
-        "gpt-4.1-mini,"
-        "gpt-4.1-mini-2025-04-14,"
-        "gpt-4.1-nano,"
-        "gpt-4.1-nano-2025-04-14,"
-        "gpt-3.5-turbo,"
+        default="gpt-3.5-turbo,"
+        "gpt-3.5-turbo-1106,"
+        "gpt-3.5-turbo-instruct,"
         "gpt-3.5-turbo-16k,"
         "gpt-3.5-turbo-16k-0613,"
-        "gpt-3.5-turbo-1106,"
         "gpt-3.5-turbo-0613,"
         "gpt-3.5-turbo-0125,"
-        "gpt-3.5-turbo-instruct,"
-        "text-davinci-003,"
-        "chatgpt-4o-latest,"
-        "gpt-4o,"
-        "gpt-4o-2024-05-13,"
-        "gpt-4o-2024-08-06,"
-        "gpt-4o-2024-11-20,"
-        "gpt-4o-audio-preview,"
-        "gpt-4o-audio-preview-2025-06-03,"
-        "gpt-4o-mini,"
-        "gpt-4o-mini-2024-07-18,"
-        "o3-mini,"
-        "o3-mini-2025-01-31,"
-        "gpt-5-mini-2025-08-07,"
-        "gpt-5-mini,"
-        "o4-mini,"
-        "o4-mini-2025-04-16,"
-        "gpt-5-chat-latest,"
-        "gpt-5,"
-        "gpt-5-2025-08-07,"
-        "gpt-5-nano,"
-        "gpt-5-nano-2025-08-07",
+        "text-davinci-003",
+    )
+
+    HOSTED_OPENAI_QUOTA_LIMIT: NonNegativeInt = Field(
+        description="Quota limit for hosted OpenAI service usage",
+        default=200,
     )
 
     HOSTED_OPENAI_PAID_ENABLED: bool = Field(
@@ -119,13 +87,6 @@ class HostedOpenAiConfig(BaseSettings):
         "gpt-4-turbo-2024-04-09,"
         "gpt-4-1106-preview,"
         "gpt-4-0125-preview,"
-        "gpt-4-turbo,"
-        "gpt-4.1,"
-        "gpt-4.1-2025-04-14,"
-        "gpt-4.1-mini,"
-        "gpt-4.1-mini-2025-04-14,"
-        "gpt-4.1-nano,"
-        "gpt-4.1-nano-2025-04-14,"
         "gpt-3.5-turbo,"
         "gpt-3.5-turbo-16k,"
         "gpt-3.5-turbo-16k-0613,"
@@ -133,150 +94,7 @@ class HostedOpenAiConfig(BaseSettings):
         "gpt-3.5-turbo-0613,"
         "gpt-3.5-turbo-0125,"
         "gpt-3.5-turbo-instruct,"
-        "text-davinci-003,"
-        "chatgpt-4o-latest,"
-        "gpt-4o,"
-        "gpt-4o-2024-05-13,"
-        "gpt-4o-2024-08-06,"
-        "gpt-4o-2024-11-20,"
-        "gpt-4o-audio-preview,"
-        "gpt-4o-audio-preview-2025-06-03,"
-        "gpt-4o-mini,"
-        "gpt-4o-mini-2024-07-18,"
-        "o3-mini,"
-        "o3-mini-2025-01-31,"
-        "gpt-5-mini-2025-08-07,"
-        "gpt-5-mini,"
-        "o4-mini,"
-        "o4-mini-2025-04-16,"
-        "gpt-5-chat-latest,"
-        "gpt-5,"
-        "gpt-5-2025-08-07,"
-        "gpt-5-nano,"
-        "gpt-5-nano-2025-08-07",
-    )
-
-
-class HostedGeminiConfig(BaseSettings):
-    """
-    Configuration for fetching Gemini service
-    """
-
-    HOSTED_GEMINI_API_KEY: str | None = Field(
-        description="API key for hosted Gemini service",
-        default=None,
-    )
-
-    HOSTED_GEMINI_API_BASE: str | None = Field(
-        description="Base URL for hosted Gemini API",
-        default=None,
-    )
-
-    HOSTED_GEMINI_API_ORGANIZATION: str | None = Field(
-        description="Organization ID for hosted Gemini service",
-        default=None,
-    )
-
-    HOSTED_GEMINI_TRIAL_ENABLED: bool = Field(
-        description="Enable trial access to hosted Gemini service",
-        default=False,
-    )
-
-    HOSTED_GEMINI_TRIAL_MODELS: str = Field(
-        description="Comma-separated list of available models for trial access",
-        default="gemini-2.5-flash,gemini-2.0-flash,gemini-2.0-flash-lite,",
-    )
-
-    HOSTED_GEMINI_PAID_ENABLED: bool = Field(
-        description="Enable paid access to hosted gemini service",
-        default=False,
-    )
-
-    HOSTED_GEMINI_PAID_MODELS: str = Field(
-        description="Comma-separated list of available models for paid access",
-        default="gemini-2.5-flash,gemini-2.0-flash,gemini-2.0-flash-lite,",
-    )
-
-
-class HostedXAIConfig(BaseSettings):
-    """
-    Configuration for fetching XAI service
-    """
-
-    HOSTED_XAI_API_KEY: str | None = Field(
-        description="API key for hosted XAI service",
-        default=None,
-    )
-
-    HOSTED_XAI_API_BASE: str | None = Field(
-        description="Base URL for hosted XAI API",
-        default=None,
-    )
-
-    HOSTED_XAI_API_ORGANIZATION: str | None = Field(
-        description="Organization ID for hosted XAI service",
-        default=None,
-    )
-
-    HOSTED_XAI_TRIAL_ENABLED: bool = Field(
-        description="Enable trial access to hosted XAI service",
-        default=False,
-    )
-
-    HOSTED_XAI_TRIAL_MODELS: str = Field(
-        description="Comma-separated list of available models for trial access",
-        default="grok-3,grok-3-mini,grok-3-mini-fast",
-    )
-
-    HOSTED_XAI_PAID_ENABLED: bool = Field(
-        description="Enable paid access to hosted XAI service",
-        default=False,
-    )
-
-    HOSTED_XAI_PAID_MODELS: str = Field(
-        description="Comma-separated list of available models for paid access",
-        default="grok-3,grok-3-mini,grok-3-mini-fast",
-    )
-
-
-class HostedDeepseekConfig(BaseSettings):
-    """
-    Configuration for fetching Deepseek service
-    """
-
-    HOSTED_DEEPSEEK_API_KEY: str | None = Field(
-        description="API key for hosted Deepseek service",
-        default=None,
-    )
-
-    HOSTED_DEEPSEEK_API_BASE: str | None = Field(
-        description="Base URL for hosted Deepseek API",
-        default=None,
-    )
-
-    HOSTED_DEEPSEEK_API_ORGANIZATION: str | None = Field(
-        description="Organization ID for hosted Deepseek service",
-        default=None,
-    )
-
-    HOSTED_DEEPSEEK_TRIAL_ENABLED: bool = Field(
-        description="Enable trial access to hosted Deepseek service",
-        default=False,
-    )
-
-    HOSTED_DEEPSEEK_TRIAL_MODELS: str = Field(
-        description="Comma-separated list of available models for trial access",
-        default="deepseek-chat,deepseek-reasoner",
-    )
-
-    HOSTED_DEEPSEEK_PAID_ENABLED: bool = Field(
-        description="Enable paid access to hosted XAI service",
-        default=False,
-    )
-
-    HOSTED_DEEPSEEK_PAID_MODELS: str = Field(
-        description="Comma-separated list of available models for paid access",
-        default="deepseek-chat,deepseek-reasoner",
+        "text-davinci-003",
     )
 
 
@@ -326,30 +144,16 @@ class HostedAnthropicConfig(BaseSettings):
         default=False,
     )
 
+    HOSTED_ANTHROPIC_QUOTA_LIMIT: NonNegativeInt = Field(
+        description="Quota limit for hosted Anthropic service usage",
+        default=600000,
+    )
+
     HOSTED_ANTHROPIC_PAID_ENABLED: bool = Field(
         description="Enable paid access to hosted Anthropic service",
         default=False,
     )
 
-    HOSTED_ANTHROPIC_TRIAL_MODELS: str = Field(
-        description="Comma-separated list of available models for paid access",
-        default="claude-opus-4-20250514,"
-        "claude-sonnet-4-20250514,"
-        "claude-3-5-haiku-20241022,"
-        "claude-3-opus-20240229,"
-        "claude-3-7-sonnet-20250219,"
-        "claude-3-haiku-20240307",
-    )
-    HOSTED_ANTHROPIC_PAID_MODELS: str = Field(
-        description="Comma-separated list of available models for paid access",
-        default="claude-opus-4-20250514,"
-        "claude-sonnet-4-20250514,"
-        "claude-3-5-haiku-20241022,"
-        "claude-3-opus-20240229,"
-        "claude-3-7-sonnet-20250219,"
-        "claude-3-haiku-20240307",
-    )
-
 
 class HostedMinmaxConfig(BaseSettings):
     """
@@ -446,8 +250,5 @@ class HostedServiceConfig(
     HostedModerationConfig,
     # credit config
     HostedCreditConfig,
-    HostedGeminiConfig,
-    HostedXAIConfig,
-    HostedDeepseekConfig,
 ):
     pass

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,6 +9,7 @@ if TYPE_CHECKING:
     from core.model_runtime.entities.model_entities import AIModelEntity
     from core.plugin.entities.plugin_daemon import PluginModelProviderEntity
     from core.tools.plugin_tool.provider import PluginToolProviderController
+    from core.trigger.provider import PluginTriggerProviderController
 
 
 """
@@ -41,3 +42,11 @@ 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

@@ -58,16 +58,15 @@ from .app import (
     mcp_server,
     message,
     model_config,
-    online_user,
     ops_trace,
     site,
     statistic,
     workflow,
     workflow_app_log,
-    workflow_comment,
     workflow_draft_variable,
     workflow_run,
     workflow_statistic,
+    workflow_trigger,
 )
 
 # Import auth controllers
@@ -108,12 +107,10 @@ from .datasets.rag_pipeline import (
 
 # Import explore controllers
 from .explore import (
-    banner,
     installed_app,
     parameter,
     recommended_app,
     saved_message,
-    trial,
 )
 
 # Import tag controllers
@@ -130,6 +127,7 @@ from .workspace import (
     models,
     plugin,
     tool_providers,
+    trigger_providers,
     workspace,
 )
 
@@ -147,7 +145,6 @@ __all__ = [
     "apikey",
     "app",
     "audio",
-    "banner",
     "billing",
     "bp",
     "completion",
@@ -201,7 +198,7 @@ __all__ = [
     "statistic",
     "tags",
     "tool_providers",
-    "trial",
+    "trigger_providers",
     "version",
     "website",
     "workflow",
@@ -209,5 +206,6 @@ __all__ = [
     "workflow_draft_variable",
     "workflow_run",
     "workflow_statistic",
+    "workflow_trigger",
     "workspace",
 ]

api/controllers/console/admin.py

@@ -16,7 +16,7 @@ from controllers.console import api, console_ns
 from controllers.console.wraps import only_edition_cloud
 from extensions.ext_database import db
 from libs.token import extract_access_token
-from models.model import App, ExporleBanner, InstalledApp, RecommendedApp, TrialApp
+from models.model import App, InstalledApp, RecommendedApp
 
 
 def admin_required(view: Callable[P, R]):
@@ -52,8 +52,6 @@ class InsertExploreAppListApi(Resource):
                 "language": fields.String(required=True, description="Language code"),
                 "category": fields.String(required=True, description="App category"),
                 "position": fields.Integer(required=True, description="Display position"),
-                "can_trial": fields.Boolean(required=True, description="Can trial"),
-                "trial_limit": fields.Integer(required=True, description="Trial limit"),
             },
         )
     )
@@ -73,8 +71,6 @@ class InsertExploreAppListApi(Resource):
             .add_argument("language", type=supported_language, required=True, nullable=False, location="json")
             .add_argument("category", type=str, required=True, nullable=False, location="json")
             .add_argument("position", type=int, required=True, nullable=False, location="json")
-            .add_argument("can_trial", type=bool, required=True, nullable=False, location="json")
-            .add_argument("trial_limit", type=int, required=True, nullable=False, location="json")
         )
         args = parser.parse_args()
 
@@ -112,20 +108,6 @@ class InsertExploreAppListApi(Resource):
                 )
 
                 db.session.add(recommended_app)
-                if args["can_trial"]:
-                    trial_app = db.session.execute(
-                        select(TrialApp).where(TrialApp.app_id == args["app_id"])
-                    ).scalar_one_or_none()
-                    if not trial_app:
-                        db.session.add(
-                            TrialApp(
-                                app_id=args["app_id"],
-                                tenant_id=app.tenant_id,
-                                trial_limit=args["trial_limit"],
-                            )
-                        )
-                    else:
-                        trial_app.trial_limit = args["trial_limit"]
 
                 app.is_public = True
                 db.session.commit()
@@ -140,20 +122,6 @@ class InsertExploreAppListApi(Resource):
                 recommended_app.category = args["category"]
                 recommended_app.position = args["position"]
 
-                if args["can_trial"]:
-                    trial_app = db.session.execute(
-                        select(TrialApp).where(TrialApp.app_id == args["app_id"])
-                    ).scalar_one_or_none()
-                    if not trial_app:
-                        db.session.add(
-                            TrialApp(
-                                app_id=args["app_id"],
-                                tenant_id=app.tenant_id,
-                                trial_limit=args["trial_limit"],
-                            )
-                        )
-                    else:
-                        trial_app.trial_limit = args["trial_limit"]
                 app.is_public = True
 
                 db.session.commit()
@@ -199,83 +167,7 @@ class InsertExploreAppApi(Resource):
             for installed_app in installed_apps:
                 session.delete(installed_app)
 
-            trial_app = session.execute(
-                select(TrialApp).where(TrialApp.app_id == recommended_app.app_id)
-            ).scalar_one_or_none()
-            if trial_app:
-                session.delete(trial_app)
-
         db.session.delete(recommended_app)
         db.session.commit()
 
         return {"result": "success"}, 204
-
-
-@console_ns.route("/admin/insert-explore-banner")
-class InsertExploreBanner(Resource):
-    @api.doc("insert_explore_banner")
-    @api.doc(description="Insert an explore banner")
-    @api.expect(
-        api.model(
-            "InsertExploreBannerRequest",
-            {
-                "content": fields.String(required=True, description="Banner content"),
-                "link": fields.String(required=True, description="Banner link"),
-                "sort": fields.Integer(required=True, description="Banner sort"),
-            },
-        )
-    )
-    @api.response(200, "Banner inserted successfully")
-    @admin_required
-    @only_edition_cloud
-    def post(self):
-        parser = reqparse.RequestParser()
-        parser.add_argument("category", type=str, required=True, nullable=False, location="json")
-        parser.add_argument("title", type=str, required=True, nullable=False, location="json")
-        parser.add_argument("description", type=str, required=True, nullable=False, location="json")
-        parser.add_argument("img-src", type=str, required=True, nullable=False, location="json")
-
-        parser.add_argument("language", type=str, required=True, nullable=False, location="json")
-        parser.add_argument("link", type=str, required=True, nullable=False, location="json")
-        parser.add_argument("sort", type=int, required=True, nullable=False, location="json")
-
-        args = parser.parse_args()
-
-        content = {
-            "category": args["category"],
-            "title": args["title"],
-            "description": args["description"],
-            "img-src": args["img-src"],
-        }
-
-        if not args["language"]:
-            args["language"] = "en-US"
-
-        banner = ExporleBanner(
-            content=content,
-            link=args["link"],
-            sort=args["sort"],
-            language=args["language"],
-        )
-        db.session.add(banner)
-        db.session.commit()
-
-        return {"result": "success"}, 200
-
-
-@console_ns.route("/admin/delete-explore-banner/<uuid:banner_id>")
-class DeleteExploreBanner(Resource):
-    @api.doc("delete_explore_banner")
-    @api.doc(description="Delete an explore banner")
-    @api.response(204, "Banner deleted successfully")
-    @admin_required
-    @only_edition_cloud
-    def delete(self, banner_id):
-        banner = db.session.execute(select(ExporleBanner).where(ExporleBanner.id == banner_id)).scalar_one_or_none()
-        if not banner:
-            raise NotFound(f"Banner '{banner_id}' is not found")
-
-        db.session.delete(banner)
-        db.session.commit()
-
-        return {"result": "success"}, 204

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/error.py

@@ -115,9 +115,3 @@ class InvokeRateLimitError(BaseHTTPException):
     error_code = "rate_limit_error"
     description = "Rate Limit Error"
     code = 429
-
-
-class NeedAddIdsError(BaseHTTPException):
-    error_code = "need_add_ids"
-    description = "Need to add ids."
-    code = 400

api/controllers/console/app/generator.py

@@ -11,6 +11,7 @@ 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
@@ -206,13 +207,11 @@ class InstructionGenerateApi(Resource):
         )
         args = parser.parse_args()
         _, current_tenant_id = current_account_with_tenant()
-        code_template = (
-            Python3CodeProvider.get_default_code()
-            if args["language"] == "python"
-            else (JavascriptCodeProvider.get_default_code())
-            if args["language"] == "javascript"
-            else ""
+        providers: list[type[CodeNodeProvider]] = [Python3CodeProvider, JavascriptCodeProvider]
+        code_provider: type[CodeNodeProvider] | None = next(
+            (p for p in providers if p.is_accept_language(args["language"])), None
         )
+        code_template = code_provider.get_default_code() if code_provider else ""
         try:
             # Generate from nothing for a workflow node
             if (args["current"] == code_template or args["current"] == "") and args["node_id"] != "":

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/online_user.py

@@ -1,339 +0,0 @@
-import json
-import time
-
-from werkzeug.wrappers import Request as WerkzeugRequest
-
-from extensions.ext_redis import redis_client
-from extensions.ext_socketio import sio
-from libs.passport import PassportService
-from libs.token import extract_access_token
-from services.account_service import AccountService
-
-SESSION_STATE_TTL_SECONDS = 3600
-WORKFLOW_ONLINE_USERS_PREFIX = "workflow_online_users:"
-WORKFLOW_LEADER_PREFIX = "workflow_leader:"
-WS_SID_MAP_PREFIX = "ws_sid_map:"
-
-
-def _workflow_key(workflow_id: str) -> str:
-    return f"{WORKFLOW_ONLINE_USERS_PREFIX}{workflow_id}"
-
-
-def _leader_key(workflow_id: str) -> str:
-    return f"{WORKFLOW_LEADER_PREFIX}{workflow_id}"
-
-
-def _sid_key(sid: str) -> str:
-    return f"{WS_SID_MAP_PREFIX}{sid}"
-
-
-def _refresh_session_state(workflow_id: str, sid: str) -> None:
-    """
-    Refresh TTLs for workflow + session keys so healthy sessions do not linger forever after crashes.
-    """
-    workflow_key = _workflow_key(workflow_id)
-    sid_key = _sid_key(sid)
-    if redis_client.exists(workflow_key):
-        redis_client.expire(workflow_key, SESSION_STATE_TTL_SECONDS)
-    if redis_client.exists(sid_key):
-        redis_client.expire(sid_key, SESSION_STATE_TTL_SECONDS)
-
-
-@sio.on("connect")
-def socket_connect(sid, environ, auth):
-    """
-    WebSocket connect event, do authentication here.
-    """
-    token = None
-    if auth and isinstance(auth, dict):
-        token = auth.get("token")
-
-    if not token:
-        try:
-            request_environ = WerkzeugRequest(environ)
-            token = extract_access_token(request_environ)
-        except Exception:
-            token = None
-
-    if not token:
-        return False
-
-    try:
-        decoded = PassportService().verify(token)
-        user_id = decoded.get("user_id")
-        if not user_id:
-            return False
-
-        with sio.app.app_context():
-            user = AccountService.load_logged_in_account(account_id=user_id)
-            if not user:
-                return False
-
-            sio.save_session(sid, {"user_id": user.id, "username": user.name, "avatar": user.avatar})
-
-            return True
-
-    except Exception:
-        return False
-
-
-@sio.on("user_connect")
-def handle_user_connect(sid, data):
-    """
-    Handle user connect event. Each session (tab) is treated as an independent collaborator.
-    """
-
-    workflow_id = data.get("workflow_id")
-    if not workflow_id:
-        return {"msg": "workflow_id is required"}, 400
-
-    session = sio.get_session(sid)
-    user_id = session.get("user_id")
-
-    if not user_id:
-        return {"msg": "unauthorized"}, 401
-
-    # Each session is stored independently with sid as key
-    session_info = {
-        "user_id": user_id,
-        "username": session.get("username", "Unknown"),
-        "avatar": session.get("avatar", None),
-        "sid": sid,
-        "connected_at": int(time.time()),  # Add timestamp to differentiate tabs
-    }
-
-    workflow_key = _workflow_key(workflow_id)
-    # Store session info with sid as key
-    redis_client.hset(workflow_key, sid, json.dumps(session_info))
-    redis_client.set(
-        _sid_key(sid),
-        json.dumps({"workflow_id": workflow_id, "user_id": user_id}),
-        ex=SESSION_STATE_TTL_SECONDS,
-    )
-    _refresh_session_state(workflow_id, sid)
-
-    # Leader election: first session becomes the leader
-    leader_sid = get_or_set_leader(workflow_id, sid)
-    is_leader = leader_sid == sid
-
-    sio.enter_room(sid, workflow_id)
-    broadcast_online_users(workflow_id)
-
-    # Notify this session of their leader status
-    sio.emit("status", {"isLeader": is_leader}, room=sid)
-
-    return {"msg": "connected", "user_id": user_id, "sid": sid, "isLeader": is_leader}
-
-
-@sio.on("disconnect")
-def handle_disconnect(sid):
-    """
-    Handle session disconnect event. Remove the specific session from online users.
-    """
-    mapping = redis_client.get(_sid_key(sid))
-    if mapping:
-        data = json.loads(mapping)
-        workflow_id = data["workflow_id"]
-
-        # Remove this specific session
-        redis_client.hdel(_workflow_key(workflow_id), sid)
-        redis_client.delete(_sid_key(sid))
-
-        # Handle leader re-election if the leader session disconnected
-        handle_leader_disconnect(workflow_id, sid)
-
-        broadcast_online_users(workflow_id)
-
-
-def _clear_session_state(workflow_id: str, sid: str) -> None:
-    redis_client.hdel(_workflow_key(workflow_id), sid)
-    redis_client.delete(_sid_key(sid))
-
-
-def _is_session_active(workflow_id: str, sid: str) -> bool:
-    if not sid:
-        return False
-
-    try:
-        if not sio.manager.is_connected(sid, "/"):
-            return False
-    except AttributeError:
-        return False
-
-    if not redis_client.hexists(_workflow_key(workflow_id), sid):
-        return False
-
-    if not redis_client.exists(_sid_key(sid)):
-        return False
-
-    return True
-
-
-def get_or_set_leader(workflow_id: str, sid: str) -> str:
-    """
-    Get current leader session or set this session as leader if no valid leader exists.
-    Returns the leader session id (sid).
-    """
-    raw_leader = redis_client.get(_leader_key(workflow_id))
-    current_leader = raw_leader.decode("utf-8") if isinstance(raw_leader, bytes) else raw_leader
-    leader_replaced = False
-
-    if current_leader and not _is_session_active(workflow_id, current_leader):
-        _clear_session_state(workflow_id, current_leader)
-        redis_client.delete(_leader_key(workflow_id))
-        current_leader = None
-        leader_replaced = True
-
-    if not current_leader:
-        redis_client.set(_leader_key(workflow_id), sid, ex=SESSION_STATE_TTL_SECONDS)  # Expire in 1 hour
-        if leader_replaced:
-            broadcast_leader_change(workflow_id, sid)
-        return sid
-
-    return current_leader
-
-
-def handle_leader_disconnect(workflow_id, disconnected_sid):
-    """
-    Handle leader re-election when a session disconnects.
-    If the disconnected session was the leader, elect a new leader from remaining sessions.
-    """
-    current_leader = redis_client.get(_leader_key(workflow_id))
-
-    if current_leader:
-        current_leader = current_leader.decode("utf-8") if isinstance(current_leader, bytes) else current_leader
-
-        if current_leader == disconnected_sid:
-            # Leader session disconnected, elect a new leader
-            sessions_json = redis_client.hgetall(_workflow_key(workflow_id))
-
-            if sessions_json:
-                # Get the first remaining session as new leader
-                new_leader_sid = list(sessions_json.keys())[0]
-                if isinstance(new_leader_sid, bytes):
-                    new_leader_sid = new_leader_sid.decode("utf-8")
-
-                redis_client.set(_leader_key(workflow_id), new_leader_sid, ex=SESSION_STATE_TTL_SECONDS)
-
-                # Notify all sessions about the new leader
-                broadcast_leader_change(workflow_id, new_leader_sid)
-            else:
-                # No sessions left, remove leader
-                redis_client.delete(_leader_key(workflow_id))
-
-
-def broadcast_leader_change(workflow_id, new_leader_sid):
-    """
-    Broadcast leader change to all sessions in the workflow.
-    """
-    sessions_json = redis_client.hgetall(_workflow_key(workflow_id))
-
-    for sid, session_info_json in sessions_json.items():
-        try:
-            sid_str = sid.decode("utf-8") if isinstance(sid, bytes) else sid
-            is_leader = sid_str == new_leader_sid
-            # Emit to each session whether they are the new leader
-            sio.emit("status", {"isLeader": is_leader}, room=sid_str)
-        except Exception:
-            continue
-
-
-def get_current_leader(workflow_id):
-    """
-    Get the current leader for a workflow.
-    """
-    leader = redis_client.get(_leader_key(workflow_id))
-    return leader.decode("utf-8") if leader and isinstance(leader, bytes) else leader
-
-
-def broadcast_online_users(workflow_id):
-    """
-    Broadcast online users to the workflow room.
-    Each session is shown as a separate user (even if same person has multiple tabs).
-    """
-    sessions_json = redis_client.hgetall(_workflow_key(workflow_id))
-    users = []
-
-    for sid, session_info_json in sessions_json.items():
-        try:
-            session_info = json.loads(session_info_json)
-            # Each session appears as a separate "user" in the UI
-            users.append(
-                {
-                    "user_id": session_info["user_id"],
-                    "username": session_info["username"],
-                    "avatar": session_info.get("avatar"),
-                    "sid": session_info["sid"],
-                    "connected_at": session_info.get("connected_at"),
-                }
-            )
-        except Exception:
-            continue
-
-    # Sort by connection time to maintain consistent order
-    users.sort(key=lambda x: x.get("connected_at") or 0)
-
-    # Get current leader session
-    leader_sid = get_current_leader(workflow_id)
-
-    sio.emit("online_users", {"workflow_id": workflow_id, "users": users, "leader": leader_sid}, room=workflow_id)
-
-
-@sio.on("collaboration_event")
-def handle_collaboration_event(sid, data):
-    """
-    Handle general collaboration events, include:
-    1. mouse_move
-    2. vars_and_features_update
-    3. sync_request (ask leader to update graph)
-    4. app_state_update
-    5. mcp_server_update
-    6. workflow_update
-    7. comments_update
-    8. node_panel_presence
-
-    """
-    mapping = redis_client.get(_sid_key(sid))
-
-    if not mapping:
-        return {"msg": "unauthorized"}, 401
-
-    mapping_data = json.loads(mapping)
-    workflow_id = mapping_data["workflow_id"]
-    user_id = mapping_data["user_id"]
-    _refresh_session_state(workflow_id, sid)
-
-    event_type = data.get("type")
-    event_data = data.get("data")
-    timestamp = data.get("timestamp", int(time.time()))
-
-    if not event_type:
-        return {"msg": "invalid event type"}, 400
-
-    sio.emit(
-        "collaboration_update",
-        {"type": event_type, "userId": user_id, "data": event_data, "timestamp": timestamp},
-        room=workflow_id,
-        skip_sid=sid,
-    )
-
-    return {"msg": "event_broadcasted"}
-
-
-@sio.on("graph_event")
-def handle_graph_event(sid, data):
-    """
-    Handle graph events - simple broadcast relay.
-    """
-    mapping = redis_client.get(_sid_key(sid))
-
-    if not mapping:
-        return {"msg": "unauthorized"}, 401
-
-    mapping_data = json.loads(mapping)
-    workflow_id = mapping_data["workflow_id"]
-    _refresh_session_state(workflow_id, sid)
-
-    sio.emit("graph_update", data, room=workflow_id, skip_sid=sid)
-
-    return {"msg": "graph_update_broadcasted"}

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

@@ -5,12 +5,10 @@ from typing import cast
 
 from flask import abort, request
 from flask_restx import Resource, fields, inputs, marshal_with, reqparse
-from pydantic_core import ValidationError
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
 
 import services
-from configs import dify_config
 from controllers.console import api, console_ns
 from controllers.console.app.error import ConversationCompletedError, DraftWorkflowNotExist, DraftWorkflowNotSync
 from controllers.console.app.wraps import get_app_model
@@ -21,11 +19,18 @@ from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.file.models import File
 from core.helper.trace_id_helper import get_external_trace_id
+from core.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 extensions.ext_redis import redis_client
 from factories import file_factory, variable_factory
-from fields.online_user_fields import online_user_list_fields
 from fields.workflow_fields import workflow_fields, workflow_pagination_fields
 from fields.workflow_run_fields import workflow_run_node_execution_fields
 from libs import helper
@@ -41,6 +46,7 @@ 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
@@ -103,22 +109,10 @@ class DraftWorkflowApi(Resource):
                 "hash": fields.String(description="Workflow hash for validation"),
                 "environment_variables": fields.List(fields.Raw, required=True, description="Environment variables"),
                 "conversation_variables": fields.List(fields.Raw, description="Conversation variables"),
-                "memory_blocks": fields.List(fields.Raw, description="Memory blocks"),
             },
         )
     )
-    @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
@@ -138,8 +132,6 @@ class DraftWorkflowApi(Resource):
                 .add_argument("hash", type=str, required=False, location="json")
                 .add_argument("environment_variables", type=list, required=True, location="json")
                 .add_argument("conversation_variables", type=list, required=False, location="json")
-                .add_argument("force_upload", type=bool, required=False, default=False, location="json")
-                .add_argument("memory_blocks", type=list, required=False, location="json")
             )
             args = parser.parse_args()
         elif "text/plain" in content_type:
@@ -157,8 +149,6 @@ class DraftWorkflowApi(Resource):
                     "hash": data.get("hash"),
                     "environment_variables": data.get("environment_variables"),
                     "conversation_variables": data.get("conversation_variables"),
-                    "memory_blocks": data.get("memory_blocks"),
-                    "force_upload": data.get("force_upload", False),
                 }
             except json.JSONDecodeError:
                 return {"message": "Invalid JSON data"}, 400
@@ -175,10 +165,6 @@ class DraftWorkflowApi(Resource):
             conversation_variables = [
                 variable_factory.build_conversation_variable_from_mapping(obj) for obj in conversation_variables_list
             ]
-            memory_blocks_list = args.get("memory_blocks") or []
-            from core.memory.entities import MemoryBlockSpec
-
-            memory_blocks = [MemoryBlockSpec.model_validate(obj) for obj in memory_blocks_list]
             workflow = workflow_service.sync_draft_workflow(
                 app_model=app_model,
                 graph=args["graph"],
@@ -187,13 +173,9 @@ class DraftWorkflowApi(Resource):
                 account=current_user,
                 environment_variables=environment_variables,
                 conversation_variables=conversation_variables,
-                force_upload=args.get("force_upload", False),
-                memory_blocks=memory_blocks,
             )
         except WorkflowHashNotEqualError:
             raise DraftWorkflowNotSync()
-        except ValidationError as e:
-            return {"message": str(e)}, 400
 
         return {
             "result": "success",
@@ -762,45 +744,6 @@ class ConvertToWorkflowApi(Resource):
         }
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/config")
-class WorkflowConfigApi(Resource):
-    """Resource for workflow configuration."""
-
-    @api.doc("get_workflow_config")
-    @api.doc(description="Get workflow configuration")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.response(200, "Workflow configuration retrieved successfully")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def get(self, app_model: App):
-        return {
-            "parallel_depth_limit": dify_config.WORKFLOW_PARALLEL_DEPTH_LIMIT,
-        }
-
-
-class WorkflowFeaturesApi(Resource):
-    """Update draft workflow features."""
-
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def post(self, app_model: App):
-        parser = reqparse.RequestParser()
-        parser.add_argument("features", type=dict, required=True, location="json")
-        args = parser.parse_args()
-
-        features = args.get("features")
-
-        # Update draft workflow features
-        workflow_service = WorkflowService()
-        workflow_service.update_draft_workflow_features(app_model=app_model, features=features, account=current_user)
-
-        return {"result": "success"}
-
-
 @console_ns.route("/apps/<uuid:app_id>/workflows")
 class PublishedAllWorkflowApi(Resource):
     @api.doc("get_all_published_workflows")
@@ -984,103 +927,228 @@ class DraftWorkflowNodeLastRunApi(Resource):
         return node_exec
 
 
-class WorkflowOnlineUsersApi(Resource):
+@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
-    @marshal_with(online_user_list_fields)
-    def get(self):
+    @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("workflow_ids", type=str, required=True, location="args")
+        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")
 
-        workflow_ids = [id.strip() for id in args["workflow_ids"].split(",")]
-
-        results = []
-        for workflow_id in workflow_ids:
-            users_json = redis_client.hgetall(f"workflow_online_users:{workflow_id}")
-
-            users = []
-            for _, user_info_json in users_json.items():
-                try:
-                    users.append(json.loads(user_info_json))
-                except Exception:
-                    continue
-            results.append({"workflow_id": workflow_id, "users": users})
-
-        return {"data": results}
+        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})
+            return helper.compact_generate_response(
+                AppGenerateService.generate(
+                    app_model=app_model,
+                    user=current_user,
+                    args=event.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:
+            raise ValueError(e.to_user_friendly_error())
+        except Exception as e:
+            logger.exception("Error polling trigger debug event")
+            raise e
 
 
-api.add_resource(
-    DraftWorkflowApi,
-    "/apps/<uuid:app_id>/workflows/draft",
-)
-api.add_resource(
-    WorkflowConfigApi,
-    "/apps/<uuid:app_id>/workflows/draft/config",
-)
-api.add_resource(
-    WorkflowFeaturesApi,
-    "/apps/<uuid:app_id>/workflows/draft/features",
-)
-api.add_resource(
-    AdvancedChatDraftWorkflowRunApi,
-    "/apps/<uuid:app_id>/advanced-chat/workflows/draft/run",
-)
-api.add_resource(
-    DraftWorkflowRunApi,
-    "/apps/<uuid:app_id>/workflows/draft/run",
-)
-api.add_resource(
-    WorkflowTaskStopApi,
-    "/apps/<uuid:app_id>/workflow-runs/tasks/<string:task_id>/stop",
-)
-api.add_resource(
-    DraftWorkflowNodeRunApi,
-    "/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/run",
-)
-api.add_resource(
-    AdvancedChatDraftRunIterationNodeApi,
-    "/apps/<uuid:app_id>/advanced-chat/workflows/draft/iteration/nodes/<string:node_id>/run",
-)
-api.add_resource(
-    WorkflowDraftRunIterationNodeApi,
-    "/apps/<uuid:app_id>/workflows/draft/iteration/nodes/<string:node_id>/run",
-)
-api.add_resource(
-    AdvancedChatDraftRunLoopNodeApi,
-    "/apps/<uuid:app_id>/advanced-chat/workflows/draft/loop/nodes/<string:node_id>/run",
-)
-api.add_resource(
-    WorkflowDraftRunLoopNodeApi,
-    "/apps/<uuid:app_id>/workflows/draft/loop/nodes/<string:node_id>/run",
-)
-api.add_resource(
-    PublishedWorkflowApi,
-    "/apps/<uuid:app_id>/workflows/publish",
-)
-api.add_resource(
-    PublishedAllWorkflowApi,
-    "/apps/<uuid:app_id>/workflows",
-)
-api.add_resource(
-    DefaultBlockConfigsApi,
-    "/apps/<uuid:app_id>/workflows/default-workflow-block-configs",
-)
-api.add_resource(
-    DefaultBlockConfigApi,
-    "/apps/<uuid:app_id>/workflows/default-workflow-block-configs/<string:block_type>",
-)
-api.add_resource(
-    ConvertToWorkflowApi,
-    "/apps/<uuid:app_id>/convert-to-workflow",
-)
-api.add_resource(
-    WorkflowByIdApi,
-    "/apps/<uuid:app_id>/workflows/<string:workflow_id>",
-)
-api.add_resource(
-    DraftWorkflowNodeLastRunApi,
-    "/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/last-run",
-)
-api.add_resource(WorkflowOnlineUsersApi, "/apps/workflows/online-users")
+@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()}), 500
+            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."}
+            ), 500
+
+
+@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:
+            raise ValueError(e.to_user_friendly_error())
+        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:
+            response = AppGenerateService.generate(
+                app_model=app_model,
+                user=current_user,
+                args=trigger_debug_event.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",
+                }
+            ), 500

api/controllers/console/app/workflow_app_log.py

@@ -28,6 +28,7 @@ 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)",
         }
@@ -68,6 +69,7 @@ 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")
         )
@@ -92,6 +94,7 @@ 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_comment.py

@@ -1,240 +0,0 @@
-import logging
-
-from flask_restx import Resource, fields, marshal_with, reqparse
-
-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 fields.member_fields import account_with_role_fields
-from fields.workflow_comment_fields import (
-    workflow_comment_basic_fields,
-    workflow_comment_create_fields,
-    workflow_comment_detail_fields,
-    workflow_comment_reply_create_fields,
-    workflow_comment_reply_update_fields,
-    workflow_comment_resolve_fields,
-    workflow_comment_update_fields,
-)
-from libs.login import current_user, login_required
-from models import App
-from services.account_service import TenantService
-from services.workflow_comment_service import WorkflowCommentService
-
-logger = logging.getLogger(__name__)
-
-
-class WorkflowCommentListApi(Resource):
-    """API for listing and creating workflow comments."""
-
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model
-    @marshal_with(workflow_comment_basic_fields, envelope="data")
-    def get(self, app_model: App):
-        """Get all comments for a workflow."""
-        comments = WorkflowCommentService.get_comments(tenant_id=current_user.current_tenant_id, app_id=app_model.id)
-
-        return comments
-
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model
-    @marshal_with(workflow_comment_create_fields)
-    def post(self, app_model: App):
-        """Create a new workflow comment."""
-        parser = reqparse.RequestParser()
-        parser.add_argument("position_x", type=float, required=True, location="json")
-        parser.add_argument("position_y", type=float, required=True, location="json")
-        parser.add_argument("content", type=str, required=True, location="json")
-        parser.add_argument("mentioned_user_ids", type=list, location="json", default=[])
-        args = parser.parse_args()
-
-        result = WorkflowCommentService.create_comment(
-            tenant_id=current_user.current_tenant_id,
-            app_id=app_model.id,
-            created_by=current_user.id,
-            content=args.content,
-            position_x=args.position_x,
-            position_y=args.position_y,
-            mentioned_user_ids=args.mentioned_user_ids,
-        )
-
-        return result, 201
-
-
-class WorkflowCommentDetailApi(Resource):
-    """API for managing individual workflow comments."""
-
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model
-    @marshal_with(workflow_comment_detail_fields)
-    def get(self, app_model: App, comment_id: str):
-        """Get a specific workflow comment."""
-        comment = WorkflowCommentService.get_comment(
-            tenant_id=current_user.current_tenant_id, app_id=app_model.id, comment_id=comment_id
-        )
-
-        return comment
-
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model
-    @marshal_with(workflow_comment_update_fields)
-    def put(self, app_model: App, comment_id: str):
-        """Update a workflow comment."""
-        parser = reqparse.RequestParser()
-        parser.add_argument("content", type=str, required=True, location="json")
-        parser.add_argument("position_x", type=float, required=False, location="json")
-        parser.add_argument("position_y", type=float, required=False, location="json")
-        parser.add_argument("mentioned_user_ids", type=list, location="json", default=[])
-        args = parser.parse_args()
-
-        result = WorkflowCommentService.update_comment(
-            tenant_id=current_user.current_tenant_id,
-            app_id=app_model.id,
-            comment_id=comment_id,
-            user_id=current_user.id,
-            content=args.content,
-            position_x=args.position_x,
-            position_y=args.position_y,
-            mentioned_user_ids=args.mentioned_user_ids,
-        )
-
-        return result
-
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model
-    def delete(self, app_model: App, comment_id: str):
-        """Delete a workflow comment."""
-        WorkflowCommentService.delete_comment(
-            tenant_id=current_user.current_tenant_id,
-            app_id=app_model.id,
-            comment_id=comment_id,
-            user_id=current_user.id,
-        )
-
-        return {"result": "success"}, 204
-
-
-class WorkflowCommentResolveApi(Resource):
-    """API for resolving and reopening workflow comments."""
-
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model
-    @marshal_with(workflow_comment_resolve_fields)
-    def post(self, app_model: App, comment_id: str):
-        """Resolve a workflow comment."""
-        comment = WorkflowCommentService.resolve_comment(
-            tenant_id=current_user.current_tenant_id,
-            app_id=app_model.id,
-            comment_id=comment_id,
-            user_id=current_user.id,
-        )
-
-        return comment
-
-
-class WorkflowCommentReplyApi(Resource):
-    """API for managing comment replies."""
-
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model
-    @marshal_with(workflow_comment_reply_create_fields)
-    def post(self, app_model: App, comment_id: str):
-        """Add a reply to a workflow comment."""
-        # Validate comment access first
-        WorkflowCommentService.validate_comment_access(
-            comment_id=comment_id, tenant_id=current_user.current_tenant_id, app_id=app_model.id
-        )
-
-        parser = reqparse.RequestParser()
-        parser.add_argument("content", type=str, required=True, location="json")
-        parser.add_argument("mentioned_user_ids", type=list, location="json", default=[])
-        args = parser.parse_args()
-
-        result = WorkflowCommentService.create_reply(
-            comment_id=comment_id,
-            content=args.content,
-            created_by=current_user.id,
-            mentioned_user_ids=args.mentioned_user_ids,
-        )
-
-        return result, 201
-
-
-class WorkflowCommentReplyDetailApi(Resource):
-    """API for managing individual comment replies."""
-
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model
-    @marshal_with(workflow_comment_reply_update_fields)
-    def put(self, app_model: App, comment_id: str, reply_id: str):
-        """Update a comment reply."""
-        # Validate comment access first
-        WorkflowCommentService.validate_comment_access(
-            comment_id=comment_id, tenant_id=current_user.current_tenant_id, app_id=app_model.id
-        )
-
-        parser = reqparse.RequestParser()
-        parser.add_argument("content", type=str, required=True, location="json")
-        parser.add_argument("mentioned_user_ids", type=list, location="json", default=[])
-        args = parser.parse_args()
-
-        reply = WorkflowCommentService.update_reply(
-            reply_id=reply_id, user_id=current_user.id, content=args.content, mentioned_user_ids=args.mentioned_user_ids
-        )
-
-        return reply
-
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model
-    def delete(self, app_model: App, comment_id: str, reply_id: str):
-        """Delete a comment reply."""
-        # Validate comment access first
-        WorkflowCommentService.validate_comment_access(
-            comment_id=comment_id, tenant_id=current_user.current_tenant_id, app_id=app_model.id
-        )
-
-        WorkflowCommentService.delete_reply(reply_id=reply_id, user_id=current_user.id)
-
-        return {"result": "success"}, 204
-
-
-class WorkflowCommentMentionUsersApi(Resource):
-    """API for getting mentionable users for workflow comments."""
-
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model
-    @marshal_with({"users": fields.List(fields.Nested(account_with_role_fields))})
-    def get(self, app_model: App):
-        """Get all users in current tenant for mentions."""
-        members = TenantService.get_tenant_members(current_user.current_tenant)
-        return {"users": members}
-
-
-# Register API routes
-api.add_resource(WorkflowCommentListApi, "/apps/<uuid:app_id>/workflow/comments")
-api.add_resource(WorkflowCommentDetailApi, "/apps/<uuid:app_id>/workflow/comments/<string:comment_id>")
-api.add_resource(WorkflowCommentResolveApi, "/apps/<uuid:app_id>/workflow/comments/<string:comment_id>/resolve")
-api.add_resource(WorkflowCommentReplyApi, "/apps/<uuid:app_id>/workflow/comments/<string:comment_id>/replies")
-api.add_resource(
-    WorkflowCommentReplyDetailApi, "/apps/<uuid:app_id>/workflow/comments/<string:comment_id>/replies/<string:reply_id>"
-)
-api.add_resource(WorkflowCommentMentionUsersApi, "/apps/<uuid:app_id>/workflow/comments/mention-users")

api/controllers/console/app/workflow_draft_variable.py

@@ -19,8 +19,8 @@ from core.variables.segments import ArrayFileSegment, FileSegment, Segment
 from core.variables.types import SegmentType
 from core.workflow.constants import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
 from extensions.ext_database import db
-from factories import variable_factory
 from factories.file_factory import build_from_mapping, build_from_mappings
+from factories.variable_factory import build_segment_with_type
 from libs.login import current_user, login_required
 from models import Account, App, AppMode
 from models.workflow import WorkflowDraftVariable
@@ -355,7 +355,7 @@ class VariableApi(Resource):
                 if len(raw_value) > 0 and not isinstance(raw_value[0], dict):
                     raise InvalidArgumentError(description=f"expected dict for files[0], got {type(raw_value)}")
                 raw_value = build_from_mappings(mappings=raw_value, tenant_id=app_model.tenant_id)
-            new_value = variable_factory.build_segment_with_type(variable.value_type, raw_value)
+            new_value = build_segment_with_type(variable.value_type, raw_value)
         draft_var_srv.update_variable(variable, name=new_name, value=new_value)
         db.session.commit()
         return variable
@@ -448,35 +448,8 @@ class ConversationVariableCollectionApi(Resource):
         db.session.commit()
         return _get_variable_list(app_model, CONVERSATION_VARIABLE_NODE_ID)
 
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=AppMode.ADVANCED_CHAT)
-    def post(self, app_model: App):
-        # The role of the current user in the ta table must be admin, owner, or editor
-        if not current_user.is_editor:
-            raise Forbidden()
-
-        parser = reqparse.RequestParser()
-        parser.add_argument("conversation_variables", type=list, required=True, location="json")
-        args = parser.parse_args()
-
-        workflow_service = WorkflowService()
-
-        conversation_variables_list = args.get("conversation_variables") or []
-        conversation_variables = [
-            variable_factory.build_conversation_variable_from_mapping(obj) for obj in conversation_variables_list
-        ]
-
-        workflow_service.update_draft_workflow_conversation_variables(
-            app_model=app_model,
-            account=current_user,
-            conversation_variables=conversation_variables,
-        )
-
-        return {"result": "success"}
-
 
+@console_ns.route("/apps/<uuid:app_id>/workflows/draft/system-variables")
 class SystemVariableCollectionApi(Resource):
     @api.doc("get_system_variables")
     @api.doc(description="Get system variables for workflow")
@@ -526,44 +499,3 @@ class EnvironmentVariableCollectionApi(Resource):
             )
 
         return {"items": env_vars_list}
-
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def post(self, app_model: App):
-        # The role of the current user in the ta table must be admin, owner, or editor
-        if not current_user.is_editor:
-            raise Forbidden()
-
-        parser = reqparse.RequestParser()
-        parser.add_argument("environment_variables", type=list, required=True, location="json")
-        args = parser.parse_args()
-
-        workflow_service = WorkflowService()
-
-        environment_variables_list = args.get("environment_variables") or []
-        environment_variables = [
-            variable_factory.build_environment_variable_from_mapping(obj) for obj in environment_variables_list
-        ]
-
-        workflow_service.update_draft_workflow_environment_variables(
-            app_model=app_model,
-            account=current_user,
-            environment_variables=environment_variables,
-        )
-
-        return {"result": "success"}
-
-
-api.add_resource(
-    WorkflowVariableCollectionApi,
-    "/apps/<uuid:app_id>/workflows/draft/variables",
-)
-api.add_resource(NodeVariableCollectionApi, "/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/variables")
-api.add_resource(VariableApi, "/apps/<uuid:app_id>/workflows/draft/variables/<uuid:variable_id>")
-api.add_resource(VariableResetApi, "/apps/<uuid:app_id>/workflows/draft/variables/<uuid:variable_id>/reset")
-
-api.add_resource(ConversationVariableCollectionApi, "/apps/<uuid:app_id>/workflows/draft/conversation-variables")
-api.add_resource(SystemVariableCollectionApi, "/apps/<uuid:app_id>/workflows/draft/system-variables")
-api.add_resource(EnvironmentVariableCollectionApi, "/apps/<uuid:app_id>/workflows/draft/environment-variables")

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

@@ -0,0 +1,145 @@
+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/app/wraps.py

@@ -23,15 +23,6 @@ def _load_app_model(app_id: str) -> App | None:
     return app_model
 
 
-def _load_app_model_with_trial(app_id: str) -> App | None:
-    app_model = (
-        db.session.query(App)
-        .where(App.id == app_id, App.status == "normal")
-        .first()
-    )
-    return app_model
-
-
 def get_app_model(view: Callable[P, R] | None = None, *, mode: Union[AppMode, list[AppMode], None] = None):
     def decorator(view_func: Callable[P1, R1]):
         @wraps(view_func)
@@ -71,44 +62,3 @@ def get_app_model(view: Callable[P, R] | None = None, *, mode: Union[AppMode, li
         return decorator
     else:
         return decorator(view)
-
-
-def get_app_model_with_trial(view: Callable[P, R] | None = None, *, mode: Union[AppMode, list[AppMode], None] = None):
-    def decorator(view_func: Callable[P, R]):
-        @wraps(view_func)
-        def decorated_view(*args: P.args, **kwargs: P.kwargs):
-            if not kwargs.get("app_id"):
-                raise ValueError("missing app_id in path parameters")
-
-            app_id = kwargs.get("app_id")
-            app_id = str(app_id)
-
-            del kwargs["app_id"]
-
-            app_model = _load_app_model_with_trial(app_id)
-
-            if not app_model:
-                raise AppNotFoundError()
-
-            app_mode = AppMode.value_of(app_model.mode)
-
-            if mode is not None:
-                if isinstance(mode, list):
-                    modes = mode
-                else:
-                    modes = [mode]
-
-                if app_mode not in modes:
-                    mode_values = {m.value for m in modes}
-                    raise AppNotFoundError(f"App mode is not in the supported list: {mode_values}")
-
-            kwargs["app_model"] = app_model
-
-            return view_func(*args, **kwargs)
-
-        return decorated_view
-
-    if view is None:
-        return decorator
-    else:
-        return decorator(view)

api/controllers/console/auth/login.py

@@ -25,13 +25,10 @@ from controllers.console.wraps import email_password_login_enabled, setup_requir
 from events.tenant_event import tenant_was_created
 from libs.helper import email, extract_remote_ip
 from libs.login import current_account_with_tenant
-from libs.passport import PassportService
 from libs.token import (
-    check_csrf_token,
     clear_access_token_from_cookie,
     clear_csrf_token_from_cookie,
     clear_refresh_token_from_cookie,
-    extract_access_token,
     extract_refresh_token,
     set_access_token_to_cookie,
     set_csrf_token_to_cookie,
@@ -292,18 +289,3 @@ class RefreshTokenApi(Resource):
             return response
         except Exception as e:
             return {"result": "fail", "message": str(e)}, 401
-
-
-# this api helps frontend to check whether user is authenticated
-# TODO: remove in the future. frontend should redirect to login page by catching 401 status
-@console_ns.route("/login/status")
-class LoginStatus(Resource):
-    def get(self):
-        token = extract_access_token(request)
-        res = True
-        try:
-            validated = PassportService().verify(token=token)
-            check_csrf_token(request=request, user_id=validated.get("user_id"))
-        except:
-            res = False
-        return {"logged_in": res}

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/explore/banner.py

@@ -1,43 +0,0 @@
-from flask import request
-from flask_restx import Resource
-
-from controllers.console import api
-from controllers.console.explore.wraps import explore_banner_enabled
-from extensions.ext_database import db
-from models.model import ExporleBanner
-
-
-class BannerApi(Resource):
-    """Resource for banner list."""
-
-    @explore_banner_enabled
-    def get(self):
-        """Get banner list."""
-        language = request.args.get("language", "en-US")
-
-        # Build base query for enabled banners
-        base_query = db.session.query(ExporleBanner).where(ExporleBanner.status == "enabled")
-        
-        # Try to get banners in the requested language
-        banners = base_query.where(ExporleBanner.language == language).order_by(ExporleBanner.sort).all()
-        
-        # Fallback to en-US if no banners found and language is not en-US
-        if not banners and language != "en-US":
-            banners = base_query.where(ExporleBanner.language == "en-US").order_by(ExporleBanner.sort).all()
-        # Convert banners to serializable format
-        result = []
-        for banner in banners:
-            banner_data = {
-                "id": banner.id,
-                "content": banner.content,  # Already parsed as JSON by SQLAlchemy
-                "link": banner.link,
-                "sort": banner.sort,
-                "status": banner.status,
-                "created_at": banner.created_at.isoformat() if banner.created_at else None,
-            }
-            result.append(banner_data)
-
-        return result
-
-
-api.add_resource(BannerApi, "/explore/banners")

api/controllers/console/explore/error.py

@@ -29,25 +29,3 @@ class AppAccessDeniedError(BaseHTTPException):
     error_code = "access_denied"
     description = "App access denied."
     code = 403
-
-
-class TrialAppNotAllowed(BaseHTTPException):
-    """*403* `Trial App Not Allowed`
-
-    Raise if the user has reached the trial app limit.
-    """
-
-    error_code = "trial_app_not_allowed"
-    code = 403
-    description = "the app is not allowed to be trial."
-
-
-class TrialAppLimitExceeded(BaseHTTPException):
-    """*403* `Trial App Limit Exceeded`
-
-    Raise if the user has exceeded the trial app limit.
-    """
-
-    error_code = "trial_app_limit_exceeded"
-    code = 403
-    description = "The user has exceeded the trial app limit."

api/controllers/console/explore/recommended_app.py

@@ -27,7 +27,6 @@ recommended_app_fields = {
     "category": fields.String,
     "position": fields.Integer,
     "is_listed": fields.Boolean,
-    "can_trial": fields.Boolean,
 }
 
 recommended_app_list_fields = {

api/controllers/console/explore/trial.py

@@ -1,514 +0,0 @@
-import logging
-from typing import Any, cast
-
-from flask import request
-from flask_restx import Resource, marshal, marshal_with, reqparse
-from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
-
-import services
-from controllers.common import fields
-from controllers.common.fields import build_site_model
-from controllers.console import api
-from controllers.console.app.error import (
-    AppUnavailableError,
-    AudioTooLargeError,
-    CompletionRequestError,
-    ConversationCompletedError,
-    NeedAddIdsError,
-    NoAudioUploadedError,
-    ProviderModelCurrentlyNotSupportError,
-    ProviderNotInitializeError,
-    ProviderNotSupportSpeechToTextError,
-    ProviderQuotaExceededError,
-    UnsupportedAudioTypeError,
-)
-from controllers.console.app.wraps import get_app_model_with_trial
-from controllers.console.explore.error import (
-    AppSuggestedQuestionsAfterAnswerDisabledError,
-    NotChatAppError,
-    NotCompletionAppError,
-    NotWorkflowAppError,
-)
-from controllers.console.explore.wraps import TrialAppResource, trial_feature_enable
-from controllers.service_api import service_api_ns
-from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
-from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
-from core.app.apps.base_app_queue_manager import AppQueueManager
-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 core.workflow.graph_engine.manager import GraphEngineManager
-from extensions.ext_database import db
-from fields.app_fields import app_detail_fields_with_site
-from fields.dataset_fields import dataset_fields
-from fields.workflow_fields import workflow_fields
-from libs import helper
-from libs.helper import uuid_value
-from libs.login import current_user
-from models import Account
-from models.account import TenantStatus
-from models.model import AppMode, Site
-from models.workflow import Workflow
-from services.app_generate_service import AppGenerateService
-from services.app_service import AppService
-from services.audio_service import AudioService
-from services.dataset_service import DatasetService
-from services.errors.audio import (
-    AudioTooLargeServiceError,
-    NoAudioUploadedServiceError,
-    ProviderNotSupportSpeechToTextServiceError,
-    UnsupportedAudioTypeServiceError,
-)
-from services.errors.conversation import ConversationNotExistsError
-from services.errors.llm import InvokeRateLimitError
-from services.errors.message import (
-    MessageNotExistsError,
-    SuggestedQuestionsAfterAnswerDisabledError,
-)
-from services.message_service import MessageService
-from services.recommended_app_service import RecommendedAppService
-
-logger = logging.getLogger(__name__)
-
-
-class TrialAppWorkflowRunApi(TrialAppResource):
-    def post(self, trial_app):
-        """
-        Run workflow
-        """
-        app_model = trial_app
-        if not app_model:
-            raise NotWorkflowAppError()
-        app_mode = AppMode.value_of(app_model.mode)
-        if app_mode != AppMode.WORKFLOW:
-            raise NotWorkflowAppError()
-
-        parser = reqparse.RequestParser()
-        parser.add_argument("inputs", type=dict, required=True, nullable=False, location="json")
-        parser.add_argument("files", type=list, required=False, location="json")
-        args = parser.parse_args()
-        assert current_user is not None
-        try:
-            app_id = app_model.id
-            user_id = current_user.id
-            response = AppGenerateService.generate(
-                app_model=app_model, user=current_user, args=args, invoke_from=InvokeFrom.EXPLORE, streaming=True
-            )
-            RecommendedAppService.add_trial_app_record(app_id, user_id)
-            return helper.compact_generate_response(response)
-        except ProviderTokenNotInitError as ex:
-            raise ProviderNotInitializeError(ex.description)
-        except QuotaExceededError:
-            raise ProviderQuotaExceededError()
-        except ModelCurrentlyNotSupportError:
-            raise ProviderModelCurrentlyNotSupportError()
-        except InvokeError as e:
-            raise CompletionRequestError(e.description)
-        except InvokeRateLimitError as ex:
-            raise InvokeRateLimitHttpError(ex.description)
-        except ValueError as e:
-            raise e
-        except Exception:
-            logger.exception("internal server error.")
-            raise InternalServerError()
-
-
-class TrialAppWorkflowTaskStopApi(TrialAppResource):
-    def post(self, trial_app, task_id: str):
-        """
-        Stop workflow task
-        """
-        app_model = trial_app
-        if not app_model:
-            raise NotWorkflowAppError()
-        app_mode = AppMode.value_of(app_model.mode)
-        if app_mode != AppMode.WORKFLOW:
-            raise NotWorkflowAppError()
-        assert current_user is not None
-
-        # Stop using both mechanisms for backward compatibility
-        # Legacy stop flag mechanism (without user check)
-        AppQueueManager.set_stop_flag_no_user_check(task_id)
-
-        # New graph engine command channel mechanism
-        GraphEngineManager.send_stop_command(task_id)
-
-        return {"result": "success"}
-
-
-class TrialChatApi(TrialAppResource):
-    @trial_feature_enable
-    def post(self, trial_app):
-        app_model = trial_app
-        app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
-            raise NotChatAppError()
-
-        parser = reqparse.RequestParser()
-        parser.add_argument("inputs", type=dict, required=True, location="json")
-        parser.add_argument("query", type=str, required=True, location="json")
-        parser.add_argument("files", type=list, required=False, location="json")
-        parser.add_argument("conversation_id", type=uuid_value, location="json")
-        parser.add_argument("parent_message_id", type=uuid_value, required=False, location="json")
-        parser.add_argument("retriever_from", type=str, required=False, default="explore_app", location="json")
-        args = parser.parse_args()
-
-        args["auto_generate_name"] = False
-
-        try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account instance")
-            
-            # Get IDs before they might be detached from session
-            app_id = app_model.id
-            user_id = current_user.id
-            
-            response = AppGenerateService.generate(
-                app_model=app_model, user=current_user, args=args, invoke_from=InvokeFrom.EXPLORE, streaming=True
-            )
-            RecommendedAppService.add_trial_app_record(app_id, user_id)
-            return helper.compact_generate_response(response)
-        except services.errors.conversation.ConversationNotExistsError:
-            raise NotFound("Conversation Not Exists.")
-        except services.errors.conversation.ConversationCompletedError:
-            raise ConversationCompletedError()
-        except services.errors.app_model_config.AppModelConfigBrokenError:
-            logger.exception("App model config broken.")
-            raise AppUnavailableError()
-        except ProviderTokenNotInitError as ex:
-            raise ProviderNotInitializeError(ex.description)
-        except QuotaExceededError:
-            raise ProviderQuotaExceededError()
-        except ModelCurrentlyNotSupportError:
-            raise ProviderModelCurrentlyNotSupportError()
-        except InvokeError as e:
-            raise CompletionRequestError(e.description)
-        except InvokeRateLimitError as ex:
-            raise InvokeRateLimitHttpError(ex.description)
-        except ValueError as e:
-            raise e
-        except Exception:
-            logger.exception("internal server error.")
-            raise InternalServerError()
-
-
-class TrialMessageSuggestedQuestionApi(TrialAppResource):
-    @trial_feature_enable
-    def get(self, trial_app, message_id):
-        app_model = trial_app
-        app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
-            raise NotChatAppError()
-
-        message_id = str(message_id)
-
-        try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account instance")
-            questions = MessageService.get_suggested_questions_after_answer(
-                app_model=app_model, user=current_user, message_id=message_id, invoke_from=InvokeFrom.EXPLORE
-            )
-        except MessageNotExistsError:
-            raise NotFound("Message not found")
-        except ConversationNotExistsError:
-            raise NotFound("Conversation not found")
-        except SuggestedQuestionsAfterAnswerDisabledError:
-            raise AppSuggestedQuestionsAfterAnswerDisabledError()
-        except ProviderTokenNotInitError as ex:
-            raise ProviderNotInitializeError(ex.description)
-        except QuotaExceededError:
-            raise ProviderQuotaExceededError()
-        except ModelCurrentlyNotSupportError:
-            raise ProviderModelCurrentlyNotSupportError()
-        except InvokeError as e:
-            raise CompletionRequestError(e.description)
-        except Exception:
-            logger.exception("internal server error.")
-            raise InternalServerError()
-
-        return {"data": questions}
-
-
-class TrialChatAudioApi(TrialAppResource):
-    @trial_feature_enable
-    def post(self, trial_app):
-        app_model = trial_app
-
-        file = request.files["file"]
-
-        try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account instance")
-            
-            # Get IDs before they might be detached from session
-            app_id = app_model.id
-            user_id = current_user.id
-            
-            response = AudioService.transcript_asr(app_model=app_model, file=file, end_user=None)
-            RecommendedAppService.add_trial_app_record(app_id, user_id)
-            return response
-        except services.errors.app_model_config.AppModelConfigBrokenError:
-            logger.exception("App model config broken.")
-            raise AppUnavailableError()
-        except NoAudioUploadedServiceError:
-            raise NoAudioUploadedError()
-        except AudioTooLargeServiceError as e:
-            raise AudioTooLargeError(str(e))
-        except UnsupportedAudioTypeServiceError:
-            raise UnsupportedAudioTypeError()
-        except ProviderNotSupportSpeechToTextServiceError:
-            raise ProviderNotSupportSpeechToTextError()
-        except ProviderTokenNotInitError as ex:
-            raise ProviderNotInitializeError(ex.description)
-        except QuotaExceededError:
-            raise ProviderQuotaExceededError()
-        except ModelCurrentlyNotSupportError:
-            raise ProviderModelCurrentlyNotSupportError()
-        except InvokeError as e:
-            raise CompletionRequestError(e.description)
-        except ValueError as e:
-            raise e
-        except Exception as e:
-            logger.exception("internal server error.")
-            raise InternalServerError()
-
-
-class TrialChatTextApi(TrialAppResource):
-    @trial_feature_enable
-    def post(self, trial_app):
-        app_model = trial_app
-        try:
-            parser = reqparse.RequestParser()
-            parser.add_argument("message_id", type=str, required=False, location="json")
-            parser.add_argument("voice", type=str, location="json")
-            parser.add_argument("text", type=str, location="json")
-            parser.add_argument("streaming", type=bool, location="json")
-            args = parser.parse_args()
-
-            message_id = args.get("message_id", None)
-            text = args.get("text", None)
-            voice = args.get("voice", None)
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account instance")
-            
-            # Get IDs before they might be detached from session
-            app_id = app_model.id
-            user_id = current_user.id
-            
-            response = AudioService.transcript_tts(app_model=app_model, text=text, voice=voice, message_id=message_id)
-            RecommendedAppService.add_trial_app_record(app_id, user_id)
-            return response
-        except services.errors.app_model_config.AppModelConfigBrokenError:
-            logger.exception("App model config broken.")
-            raise AppUnavailableError()
-        except NoAudioUploadedServiceError:
-            raise NoAudioUploadedError()
-        except AudioTooLargeServiceError as e:
-            raise AudioTooLargeError(str(e))
-        except UnsupportedAudioTypeServiceError:
-            raise UnsupportedAudioTypeError()
-        except ProviderNotSupportSpeechToTextServiceError:
-            raise ProviderNotSupportSpeechToTextError()
-        except ProviderTokenNotInitError as ex:
-            raise ProviderNotInitializeError(ex.description)
-        except QuotaExceededError:
-            raise ProviderQuotaExceededError()
-        except ModelCurrentlyNotSupportError:
-            raise ProviderModelCurrentlyNotSupportError()
-        except InvokeError as e:
-            raise CompletionRequestError(e.description)
-        except ValueError as e:
-            raise e
-        except Exception as e:
-            logger.exception("internal server error.")
-            raise InternalServerError()
-
-
-class TrialCompletionApi(TrialAppResource):
-    @trial_feature_enable
-    def post(self, trial_app):
-        app_model = trial_app
-        if app_model.mode != "completion":
-            raise NotCompletionAppError()
-
-        parser = reqparse.RequestParser()
-        parser.add_argument("inputs", type=dict, required=True, location="json")
-        parser.add_argument("query", type=str, location="json", default="")
-        parser.add_argument("files", type=list, required=False, location="json")
-        parser.add_argument("response_mode", type=str, choices=["blocking", "streaming"], location="json")
-        parser.add_argument("retriever_from", type=str, required=False, default="explore_app", location="json")
-        args = parser.parse_args()
-
-        streaming = args["response_mode"] == "streaming"
-        args["auto_generate_name"] = False
-
-        try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account instance")
-            
-            # Get IDs before they might be detached from session
-            app_id = app_model.id
-            user_id = current_user.id
-            
-            response = AppGenerateService.generate(
-                app_model=app_model, user=current_user, args=args, invoke_from=InvokeFrom.EXPLORE, streaming=streaming
-            )
-
-            RecommendedAppService.add_trial_app_record(app_id, user_id)
-            return helper.compact_generate_response(response)
-        except services.errors.conversation.ConversationNotExistsError:
-            raise NotFound("Conversation Not Exists.")
-        except services.errors.conversation.ConversationCompletedError:
-            raise ConversationCompletedError()
-        except services.errors.app_model_config.AppModelConfigBrokenError:
-            logger.exception("App model config broken.")
-            raise AppUnavailableError()
-        except ProviderTokenNotInitError as ex:
-            raise ProviderNotInitializeError(ex.description)
-        except QuotaExceededError:
-            raise ProviderQuotaExceededError()
-        except ModelCurrentlyNotSupportError:
-            raise ProviderModelCurrentlyNotSupportError()
-        except InvokeError as e:
-            raise CompletionRequestError(e.description)
-        except ValueError as e:
-            raise e
-        except Exception:
-            logger.exception("internal server error.")
-            raise InternalServerError()
-
-
-class TrialSitApi(Resource):
-    """Resource for trial app sites."""
-
-    @trial_feature_enable
-    @get_app_model_with_trial
-    @service_api_ns.marshal_with(build_site_model(service_api_ns))
-    def get(self, app_model):
-        """Retrieve app site info.
-
-        Returns the site configuration for the application including theme, icons, and text.
-        """
-        site = db.session.query(Site).where(Site.app_id == app_model.id).first()
-
-        if not site:
-            raise Forbidden()
-
-        assert app_model.tenant
-        if app_model.tenant.status == TenantStatus.ARCHIVE:
-            raise Forbidden()
-
-        return site
-
-
-class TrialAppParameterApi(Resource):
-    """Resource for app variables."""
-
-    @trial_feature_enable
-    @get_app_model_with_trial
-    @marshal_with(fields.parameters_fields)
-    def get(self, app_model):
-        """Retrieve app parameters."""
-
-        if app_model is None:
-            raise AppUnavailableError()
-
-        if app_model.mode in {AppMode.ADVANCED_CHAT, AppMode.WORKFLOW}:
-            workflow = app_model.workflow
-            if workflow is None:
-                raise AppUnavailableError()
-
-            features_dict = workflow.features_dict
-            user_input_form = workflow.user_input_form(to_old_structure=True)
-        else:
-            app_model_config = app_model.app_model_config
-            if app_model_config is None:
-                raise AppUnavailableError()
-
-            features_dict = app_model_config.to_dict()
-
-            user_input_form = features_dict.get("user_input_form", [])
-
-        return get_parameters_from_feature_dict(features_dict=features_dict, user_input_form=user_input_form)
-
-
-class AppApi(Resource):
-    @trial_feature_enable
-    @get_app_model_with_trial
-    @marshal_with(app_detail_fields_with_site)
-    def get(self, app_model):
-        """Get app detail"""
-
-        app_service = AppService()
-        app_model = app_service.get_app(app_model)
-
-        return app_model
-
-
-class AppWorkflowApi(Resource):
-    @trial_feature_enable
-    @get_app_model_with_trial
-    @marshal_with(workflow_fields)
-    def get(self, app_model):
-        """Get workflow detail"""
-        if not app_model.workflow_id:
-            raise AppUnavailableError()
-
-        workflow = (
-            db.session.query(Workflow)
-            .where(
-                Workflow.id == app_model.workflow_id,
-            )
-            .first()
-        )
-        return workflow
-
-
-class DatasetListApi(Resource):
-    @trial_feature_enable
-    @get_app_model_with_trial
-    def get(self, app_model):
-        page = request.args.get("page", default=1, type=int)
-        limit = request.args.get("limit", default=20, type=int)
-        ids = request.args.getlist("ids")
-      
-        tenant_id = app_model.tenant_id
-        if ids:
-            datasets, total = DatasetService.get_datasets_by_ids(ids, tenant_id)
-        else:
-            raise NeedAddIdsError()
-
-        data = cast(list[dict[str, Any]], marshal(datasets, dataset_fields))
-
-        response = {"data": data, "has_more": len(datasets) == limit, "limit": limit, "total": total, "page": page}
-        return response
-
-
-api.add_resource(TrialChatApi, "/trial-apps/<uuid:app_id>/chat-messages", endpoint="trial_app_chat_completion")
-
-api.add_resource(
-    TrialMessageSuggestedQuestionApi,
-    "/trial-apps/<uuid:app_id>/messages/<uuid:message_id>/suggested-questions",
-    endpoint="trial_app_suggested_question",
-)
-
-api.add_resource(TrialChatAudioApi, "/trial-apps/<uuid:app_id>/audio-to-text", endpoint="trial_app_audio")
-api.add_resource(TrialChatTextApi, "/trial-apps/<uuid:app_id>/text-to-audio", endpoint="trial_app_text")
-
-api.add_resource(TrialCompletionApi, "/trial-apps/<uuid:app_id>/completion-messages", endpoint="trial_app_completion")
-
-api.add_resource(TrialSitApi, "/trial-apps/<uuid:app_id>/site")
-
-api.add_resource(TrialAppParameterApi, "/trial-apps/<uuid:app_id>/parameters", endpoint="trial_app_parameters")
-
-api.add_resource(AppApi, "/trial-apps/<uuid:app_id>", endpoint="trial_app")
-
-api.add_resource(TrialAppWorkflowRunApi, "/trial-apps/<uuid:app_id>/workflows/run", endpoint="trial_app_workflow_run")
-api.add_resource(TrialAppWorkflowTaskStopApi, "/trial-apps/<uuid:app_id>/workflows/tasks/<string:task_id>/stop")
-
-api.add_resource(AppWorkflowApi, "/trial-apps/<uuid:app_id>/workflows", endpoint="trial_app_workflow")
-api.add_resource(DatasetListApi, "/trial-apps/<uuid:app_id>/datasets", endpoint="trial_app_datasets")

api/controllers/console/explore/wraps.py

@@ -2,16 +2,14 @@ from collections.abc import Callable
 from functools import wraps
 from typing import Concatenate, ParamSpec, TypeVar
 
-from flask import abort
-from flask_login import current_user
 from flask_restx import Resource
 from werkzeug.exceptions import NotFound
 
-from controllers.console.explore.error import AppAccessDeniedError, TrialAppLimitExceeded, TrialAppNotAllowed
+from controllers.console.explore.error import AppAccessDeniedError
 from controllers.console.wraps import account_initialization_required
 from extensions.ext_database import db
 from libs.login import current_account_with_tenant, login_required
-from models import AccountTrialAppRecord, App, InstalledApp, TrialApp
+from models import InstalledApp
 from services.enterprise.enterprise_service import EnterpriseService
 from services.feature_service import FeatureService
 
@@ -73,59 +71,6 @@ def user_allowed_to_access_app(view: Callable[Concatenate[InstalledApp, P], R] |
     return decorator
 
 
-def trial_app_required(view: Callable[Concatenate[App, P], R] | None = None):
-    def decorator(view: Callable[Concatenate[App, P], R]):
-        @wraps(view)
-        def decorated(app_id: str, *args: P.args, **kwargs: P.kwargs):
-            trial_app = db.session.query(TrialApp).where(TrialApp.app_id == str(app_id)).first()
-
-            if trial_app is None:
-                raise TrialAppNotAllowed()
-            app = trial_app.app
-
-            if app is None:
-                raise TrialAppNotAllowed()
-
-            account_trial_app_record = (
-                db.session.query(AccountTrialAppRecord)
-                .where(AccountTrialAppRecord.account_id == current_user.id, AccountTrialAppRecord.app_id == app_id)
-                .first()
-            )
-            if account_trial_app_record:
-                if account_trial_app_record.count >= trial_app.trial_limit:
-                    raise TrialAppLimitExceeded()
-
-            return view(app, *args, **kwargs)
-
-        return decorated
-
-    if view:
-        return decorator(view)
-    return decorator
-
-
-def trial_feature_enable(view: Callable[..., R]) -> Callable[..., R]:
-    @wraps(view)
-    def decorated(*args, **kwargs):
-        features = FeatureService.get_system_features()
-        if not features.enable_trial_app:
-            abort(403, "Trial app feature is not enabled.")
-        return view(*args, **kwargs)
-
-    return decorated
-
-
-def explore_banner_enabled(view: Callable[..., R]) -> Callable[..., R]:
-    @wraps(view)
-    def decorated(*args, **kwargs):
-        features = FeatureService.get_system_features()
-        if not features.enable_explore_banner:
-            abort(403, "Explore banner feature is not enabled.")
-        return view(*args, **kwargs)
-
-    return decorated
-
-
 class InstalledAppResource(Resource):
     # must be reversed if there are multiple decorators
 
@@ -135,13 +80,3 @@ class InstalledAppResource(Resource):
         account_initialization_required,
         login_required,
     ]
-
-
-class TrialAppResource(Resource):
-    # must be reversed if there are multiple decorators
-
-    method_decorators = [
-        trial_app_required,
-        account_initialization_required,
-        login_required,
-    ]

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/account.py

@@ -32,7 +32,6 @@ from controllers.console.wraps import (
     only_edition_cloud,
     setup_required,
 )
-from core.file import helpers as file_helpers
 from extensions.ext_database import db
 from fields.member_fields import account_fields
 from libs.datetime_utils import naive_utc_now
@@ -129,17 +128,6 @@ class AccountNameApi(Resource):
 
 @console_ns.route("/account/avatar")
 class AccountAvatarApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def get(self):
-        parser = reqparse.RequestParser()
-        parser.add_argument("avatar", type=str, required=True, location="args")
-        args = parser.parse_args()
-
-        avatar_url = file_helpers.get_signed_file_url(args["avatar"])
-        return {"avatar_url": avatar_url}
-
     @setup_required
     @login_required
     @account_initialization_required

api/controllers/console/workspace/plugin.py

@@ -114,6 +114,25 @@ 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
@@ -558,19 +577,21 @@ 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,
-                user_id,
-                args["plugin_id"],
-                args["provider"],
-                args["action"],
-                args["parameter"],
-                args["provider_type"],
+                tenant_id=tenant_id,
+                user_id=user_id,
+                plugin_id=args["plugin_id"],
+                provider=args["provider"],
+                action=args["action"],
+                parameter=args["parameter"],
+                credential_id=args["credential_id"],
+                provider_type=args["provider_type"],
             )
         except PluginDaemonClientSideError as e:
             raise ValueError(e)
@@ -686,3 +707,23 @@ 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,12 +21,14 @@ 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

@@ -0,0 +1,592 @@
+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
@@ -51,9 +50,6 @@ tenant_fields = {
     "in_trial": fields.Boolean,
     "trial_end_reason": fields.String,
     "custom_config": fields.Raw(attribute="custom_config"),
-    "trial_credits": fields.Integer,
-    "trial_credits_used": fields.Integer,
-    "next_credit_reset_date": fields.Integer,
 }
 
 tenants_fields = {
@@ -87,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/__init__.py

@@ -19,7 +19,6 @@ from .app import (
     annotation,
     app,
     audio,
-    chatflow_memory,
     completion,
     conversation,
     file,
@@ -41,7 +40,6 @@ __all__ = [
     "annotation",
     "app",
     "audio",
-    "chatflow_memory",
     "completion",
     "conversation",
     "dataset",

api/controllers/service_api/app/chatflow_memory.py

@@ -1,109 +0,0 @@
-from flask_restx import Resource, reqparse
-
-from controllers.service_api import api
-from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate_app_token
-from core.memory.entities import MemoryBlock, MemoryCreatedBy
-from core.workflow.runtime.variable_pool import VariablePool
-from models import App, EndUser
-from services.chatflow_memory_service import ChatflowMemoryService
-from services.workflow_service import WorkflowService
-
-
-class MemoryListApi(Resource):
-    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))
-    def get(self, app_model: App, end_user: EndUser):
-        parser = reqparse.RequestParser()
-        parser.add_argument("conversation_id", required=False, type=str | None, default=None)
-        parser.add_argument("memory_id", required=False, type=str | None, default=None)
-        parser.add_argument("version", required=False, type=int | None, default=None)
-        args = parser.parse_args()
-        conversation_id: str | None = args.get("conversation_id")
-        memory_id = args.get("memory_id")
-        version = args.get("version")
-
-        if conversation_id:
-            result = ChatflowMemoryService.get_persistent_memories_with_conversation(
-                app_model, MemoryCreatedBy(end_user_id=end_user.id), conversation_id, version
-            )
-            session_memories = ChatflowMemoryService.get_session_memories_with_conversation(
-                app_model, MemoryCreatedBy(end_user_id=end_user.id), conversation_id, version
-            )
-            result = [*result, *session_memories]
-        else:
-            result = ChatflowMemoryService.get_persistent_memories(
-                app_model, MemoryCreatedBy(end_user_id=end_user.id), version
-            )
-
-        if memory_id:
-            result = [it for it in result if it.spec.id == memory_id]
-        return [it for it in result if it.spec.end_user_visible]
-
-
-class MemoryEditApi(Resource):
-    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))
-    def put(self, app_model: App, end_user: EndUser):
-        parser = reqparse.RequestParser()
-        parser.add_argument("id", type=str, required=True)
-        parser.add_argument("conversation_id", type=str | None, required=False, default=None)
-        parser.add_argument("node_id", type=str | None, required=False, default=None)
-        parser.add_argument("update", type=str, required=True)
-        args = parser.parse_args()
-        workflow = WorkflowService().get_published_workflow(app_model)
-        update = args.get("update")
-        conversation_id = args.get("conversation_id")
-        node_id = args.get("node_id")
-        if not isinstance(update, str):
-            return {"error": "Invalid update"}, 400
-        if not workflow:
-            return {"error": "Workflow not found"}, 404
-        memory_spec = next((it for it in workflow.memory_blocks if it.id == args["id"]), None)
-        if not memory_spec:
-            return {"error": "Memory not found"}, 404
-
-        # First get existing memory
-        existing_memory = ChatflowMemoryService.get_memory_by_spec(
-            spec=memory_spec,
-            tenant_id=app_model.tenant_id,
-            app_id=app_model.id,
-            created_by=MemoryCreatedBy(end_user_id=end_user.id),
-            conversation_id=conversation_id,
-            node_id=node_id,
-            is_draft=False,
-        )
-
-        # Create updated memory instance with incremented version
-        updated_memory = MemoryBlock(
-            spec=existing_memory.spec,
-            tenant_id=existing_memory.tenant_id,
-            app_id=existing_memory.app_id,
-            conversation_id=existing_memory.conversation_id,
-            node_id=existing_memory.node_id,
-            value=update,  # New value
-            version=existing_memory.version + 1,  # Increment version for update
-            edited_by_user=True,
-            created_by=existing_memory.created_by,
-        )
-
-        ChatflowMemoryService.save_memory(updated_memory, VariablePool(), False)
-        return "", 204
-
-
-class MemoryDeleteApi(Resource):
-    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))
-    def delete(self, app_model: App, end_user: EndUser):
-        parser = reqparse.RequestParser()
-        parser.add_argument("id", type=str, required=False, default=None)
-        args = parser.parse_args()
-        memory_id = args.get("id")
-
-        if memory_id:
-            ChatflowMemoryService.delete_memory(app_model, memory_id, MemoryCreatedBy(end_user_id=end_user.id))
-            return "", 204
-        else:
-            ChatflowMemoryService.delete_all_user_memories(app_model, MemoryCreatedBy(end_user_id=end_user.id))
-            return "", 200
-
-
-api.add_resource(MemoryListApi, "/memories")
-api.add_resource(MemoryEditApi, "/memory-edit")
-api.add_resource(MemoryDeleteApi, "/memories")

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,14 +13,14 @@ 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, DefaultEndUserSessionID, EndUser
+from models.model import ApiToken, App
+from services.end_user_service import EndUserService
 from services.feature_service import FeatureService
 
 P = ParamSpec("P")
@@ -67,7 +67,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")
@@ -76,6 +75,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:
@@ -84,34 +84,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 = create_or_update_end_user_for_user_id(app_model, user_id)
+                end_user = EndUserService.get_or_create_end_user(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)
 
@@ -161,7 +139,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."
                         )
@@ -331,39 +309,6 @@ 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

@@ -0,0 +1,12 @@
+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

@@ -0,0 +1,44 @@
+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:
+        logger.exception("Endpoint processing failed for {endpoint_id}: {e}")
+        return jsonify({"error": "Endpoint processing failed", "message": str(e)}), 500
+    except Exception as e:
+        logger.exception("Webhook processing failed for {endpoint_id}")
+        return jsonify({"error": "Internal server error"}), 500

api/controllers/trigger/webhook.py

@@ -0,0 +1,105 @@
+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/controllers/web/__init__.py

@@ -18,7 +18,6 @@ web_ns = Namespace("web", description="Web application API operations", path="/"
 from . import (
     app,
     audio,
-    chatflow_memory,
     completion,
     conversation,
     feature,
@@ -40,7 +39,6 @@ __all__ = [
     "app",
     "audio",
     "bp",
-    "chatflow_memory",
     "completion",
     "conversation",
     "feature",

api/controllers/web/chatflow_memory.py

@@ -1,108 +0,0 @@
-from flask_restx import reqparse
-
-from controllers.web import api
-from controllers.web.wraps import WebApiResource
-from core.memory.entities import MemoryBlock, MemoryCreatedBy
-from core.workflow.runtime.variable_pool import VariablePool
-from models import App, EndUser
-from services.chatflow_memory_service import ChatflowMemoryService
-from services.workflow_service import WorkflowService
-
-
-class MemoryListApi(WebApiResource):
-    def get(self, app_model: App, end_user: EndUser):
-        parser = reqparse.RequestParser()
-        parser.add_argument("conversation_id", required=False, type=str | None, default=None)
-        parser.add_argument("memory_id", required=False, type=str | None, default=None)
-        parser.add_argument("version", required=False, type=int | None, default=None)
-        args = parser.parse_args()
-        conversation_id: str | None = args.get("conversation_id")
-        memory_id = args.get("memory_id")
-        version = args.get("version")
-
-        if conversation_id:
-            result = ChatflowMemoryService.get_persistent_memories_with_conversation(
-                app_model, MemoryCreatedBy(end_user_id=end_user.id), conversation_id, version
-            )
-            session_memories = ChatflowMemoryService.get_session_memories_with_conversation(
-                app_model, MemoryCreatedBy(end_user_id=end_user.id), conversation_id, version
-            )
-            result = [*result, *session_memories]
-        else:
-            result = ChatflowMemoryService.get_persistent_memories(
-                app_model, MemoryCreatedBy(end_user_id=end_user.id), version
-            )
-
-        if memory_id:
-            result = [it for it in result if it.spec.id == memory_id]
-        return [it for it in result if it.spec.end_user_visible]
-
-
-class MemoryEditApi(WebApiResource):
-    def put(self, app_model: App, end_user: EndUser):
-        parser = reqparse.RequestParser()
-        parser.add_argument("id", type=str, required=True)
-        parser.add_argument("conversation_id", type=str | None, required=False, default=None)
-        parser.add_argument("node_id", type=str | None, required=False, default=None)
-        parser.add_argument("update", type=str, required=True)
-        args = parser.parse_args()
-        workflow = WorkflowService().get_published_workflow(app_model)
-        update = args.get("update")
-        conversation_id = args.get("conversation_id")
-        node_id = args.get("node_id")
-        if not isinstance(update, str):
-            return {"error": "Update must be a string"}, 400
-        if not workflow:
-            return {"error": "Workflow not found"}, 404
-        memory_spec = next((it for it in workflow.memory_blocks if it.id == args["id"]), None)
-        if not memory_spec:
-            return {"error": "Memory not found"}, 404
-        if not memory_spec.end_user_editable:
-            return {"error": "Memory not editable"}, 403
-
-        # First get existing memory
-        existing_memory = ChatflowMemoryService.get_memory_by_spec(
-            spec=memory_spec,
-            tenant_id=app_model.tenant_id,
-            app_id=app_model.id,
-            created_by=MemoryCreatedBy(end_user_id=end_user.id),
-            conversation_id=conversation_id,
-            node_id=node_id,
-            is_draft=False,
-        )
-
-        # Create updated memory instance with incremented version
-        updated_memory = MemoryBlock(
-            spec=existing_memory.spec,
-            tenant_id=existing_memory.tenant_id,
-            app_id=existing_memory.app_id,
-            conversation_id=existing_memory.conversation_id,
-            node_id=existing_memory.node_id,
-            value=update,  # New value
-            version=existing_memory.version + 1,  # Increment version for update
-            edited_by_user=True,
-            created_by=existing_memory.created_by,
-        )
-
-        ChatflowMemoryService.save_memory(updated_memory, VariablePool(), False)
-        return "", 204
-
-
-class MemoryDeleteApi(WebApiResource):
-    def delete(self, app_model: App, end_user: EndUser):
-        parser = reqparse.RequestParser()
-        parser.add_argument("id", type=str, required=False, default=None)
-        args = parser.parse_args()
-        memory_id = args.get("id")
-
-        if memory_id:
-            ChatflowMemoryService.delete_memory(app_model, memory_id, MemoryCreatedBy(end_user_id=end_user.id))
-            return "", 204
-        else:
-            ChatflowMemoryService.delete_all_user_memories(app_model, MemoryCreatedBy(end_user_id=end_user.id))
-            return "", 200
-
-
-api.add_resource(MemoryListApi, "/memories")
-api.add_resource(MemoryEditApi, "/memory-edit")
-api.add_resource(MemoryDeleteApi, "/memories")

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

@@ -1,11 +1,10 @@
 import logging
 import time
-from collections.abc import Mapping, MutableMapping, Sequence
+from collections.abc import Mapping, Sequence
 from typing import Any, cast
 
 from sqlalchemy import select
 from sqlalchemy.orm import Session
-from typing_extensions import override
 
 from core.app.apps.advanced_chat.app_config_manager import AdvancedChatAppConfig
 from core.app.apps.base_app_queue_manager import AppQueueManager
@@ -21,8 +20,6 @@ from core.app.entities.queue_entities import (
     QueueTextChunkEvent,
 )
 from core.app.features.annotation_reply.annotation_reply import AnnotationReplyFeature
-from core.memory.entities import MemoryCreatedBy, MemoryScope
-from core.model_runtime.entities import AssistantPromptMessage, UserPromptMessage
 from core.moderation.base import ModerationError
 from core.moderation.input_moderation import InputModeration
 from core.variables.variables import VariableUnion
@@ -30,7 +27,6 @@ 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.graph_events import GraphRunSucceededEvent
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 from core.workflow.runtime import GraphRuntimeState, VariablePool
@@ -43,8 +39,6 @@ from models import Workflow
 from models.enums import UserFrom
 from models.model import App, Conversation, Message, MessageAnnotation
 from models.workflow import ConversationVariable
-from services.chatflow_history_service import ChatflowHistoryService
-from services.chatflow_memory_service import ChatflowMemoryService
 
 logger = logging.getLogger(__name__)
 
@@ -87,10 +81,6 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
         self._workflow_node_execution_repository = workflow_node_execution_repository
 
     def run(self):
-        ChatflowMemoryService.wait_for_sync_memory_completion(
-            workflow=self._workflow, conversation_id=self.conversation.id
-        )
-
         app_config = self.application_generate_entity.app_config
         app_config = cast(AdvancedChatAppConfig, app_config)
 
@@ -153,7 +143,6 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
                 # Based on the definition of `VariableUnion`,
                 # `list[Variable]` can be safely used as `list[VariableUnion]` since they are compatible.
                 conversation_variables=conversation_variables,
-                memory_blocks=self._fetch_memory_blocks(),
             )
 
             # init graph
@@ -217,31 +206,6 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
         for event in generator:
             self._handle_event(workflow_entry, event)
 
-        try:
-            self._check_app_memory_updates(variable_pool)
-        except Exception as e:
-            logger.exception("Failed to check app memory updates", exc_info=e)
-
-    @override
-    def _handle_event(self, workflow_entry: WorkflowEntry, event: Any) -> None:
-        super()._handle_event(workflow_entry, event)
-        if isinstance(event, GraphRunSucceededEvent):
-            workflow_outputs = event.outputs
-            if not workflow_outputs:
-                logger.warning("Chatflow output is empty.")
-                return
-            assistant_message = workflow_outputs.get("answer")
-            if not assistant_message:
-                logger.warning("Chatflow output does not contain 'answer'.")
-                return
-            if not isinstance(assistant_message, str):
-                logger.warning("Chatflow output 'answer' is not a string.")
-                return
-            try:
-                self._sync_conversation_to_chatflow_tables(assistant_message)
-            except Exception as e:
-                logger.exception("Failed to sync conversation to memory tables", exc_info=e)
-
     def handle_input_moderation(
         self,
         app_record: App,
@@ -439,67 +403,3 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
 
         # Return combined list
         return existing_variables + new_variables
-
-    def _fetch_memory_blocks(self) -> Mapping[str, str]:
-        """fetch all memory blocks for current app"""
-
-        memory_blocks_dict: MutableMapping[str, str] = {}
-        is_draft = self.application_generate_entity.invoke_from == InvokeFrom.DEBUGGER
-        conversation_id = self.conversation.id
-        memory_block_specs = self._workflow.memory_blocks
-        # Get runtime memory values
-        memories = ChatflowMemoryService.get_memories_by_specs(
-            memory_block_specs=memory_block_specs,
-            tenant_id=self._workflow.tenant_id,
-            app_id=self._workflow.app_id,
-            node_id=None,
-            conversation_id=conversation_id,
-            is_draft=is_draft,
-            created_by=self._get_created_by(),
-        )
-
-        # Build memory_id -> value mapping
-        for memory in memories:
-            if memory.spec.scope == MemoryScope.APP:
-                # App level: use memory_id directly
-                memory_blocks_dict[memory.spec.id] = memory.value
-            else:  # NODE scope
-                node_id = memory.node_id
-                if not node_id:
-                    logger.warning("Memory block %s has no node_id, skip.", memory.spec.id)
-                    continue
-                key = f"{node_id}.{memory.spec.id}"
-                memory_blocks_dict[key] = memory.value
-
-        return memory_blocks_dict
-
-    def _sync_conversation_to_chatflow_tables(self, assistant_message: str):
-        ChatflowHistoryService.save_app_message(
-            prompt_message=UserPromptMessage(content=(self.application_generate_entity.query)),
-            conversation_id=self.conversation.id,
-            app_id=self._workflow.app_id,
-            tenant_id=self._workflow.tenant_id,
-        )
-        ChatflowHistoryService.save_app_message(
-            prompt_message=AssistantPromptMessage(content=assistant_message),
-            conversation_id=self.conversation.id,
-            app_id=self._workflow.app_id,
-            tenant_id=self._workflow.tenant_id,
-        )
-
-    def _check_app_memory_updates(self, variable_pool: VariablePool):
-        is_draft = self.application_generate_entity.invoke_from == InvokeFrom.DEBUGGER
-
-        ChatflowMemoryService.update_app_memory_if_needed(
-            workflow=self._workflow,
-            conversation_id=self.conversation.id,
-            variable_pool=variable_pool,
-            is_draft=is_draft,
-            created_by=self._get_created_by(),
-        )
-
-    def _get_created_by(self) -> MemoryCreatedBy:
-        if self.application_generate_entity.invoke_from in {InvokeFrom.DEBUGGER, InvokeFrom.EXPLORE}:
-            return MemoryCreatedBy(account_id=self.application_generate_entity.user_id)
-        else:
-            return MemoryCreatedBy(end_user_id=self.application_generate_entity.user_id)

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,6 +37,7 @@ 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,
@@ -303,6 +304,11 @@ 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

@@ -40,15 +40,19 @@ from core.workflow.repositories.draft_variable_repository import DraftVariableSa
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 from core.workflow.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader
-from enums.cloud_plan import CloudPlan
 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.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__)
 
@@ -251,7 +255,7 @@ class PipelineGenerator(BaseAppGenerator):
             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 == CloudPlan.SANDBOX:
+            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}"
 

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

@@ -27,6 +27,7 @@ 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
@@ -53,7 +54,10 @@ class WorkflowAppGenerator(BaseAppGenerator):
         invoke_from: InvokeFrom,
         streaming: Literal[True],
         call_depth: int,
-    ) -> Generator[Mapping | str, None, None]: ...
+        triggered_from: WorkflowRunTriggeredFrom | None = None,
+        root_node_id: str | None = None,
+        graph_engine_layers: Sequence[GraphEngineLayer] = (),
+    ) -> Generator[Mapping[str, Any] | str, None, None]: ...
 
     @overload
     def generate(
@@ -66,6 +70,9 @@ 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
@@ -79,7 +86,10 @@ class WorkflowAppGenerator(BaseAppGenerator):
         invoke_from: InvokeFrom,
         streaming: bool,
         call_depth: int,
-    ) -> Union[Mapping[str, Any], Generator[Mapping | str, None, None]]: ...
+        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]]: ...
 
     def generate(
         self,
@@ -91,7 +101,10 @@ class WorkflowAppGenerator(BaseAppGenerator):
         invoke_from: InvokeFrom,
         streaming: bool = True,
         call_depth: int = 0,
-    ) -> Union[Mapping[str, Any], Generator[Mapping | str, None, None]]:
+        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]]:
         files: Sequence[Mapping[str, Any]] = args.get("files") or []
 
         # parse files
@@ -126,17 +139,20 @@ class WorkflowAppGenerator(BaseAppGenerator):
             **extract_external_trace_id_from_args(args),
         }
         workflow_run_id = str(uuid.uuid4())
+        if triggered_from in (WorkflowRunTriggeredFrom.DEBUGGING, WorkflowRunTriggeredFrom.APP_RUN):
+            # start node get inputs
+            inputs = self._prepare_user_inputs(
+                user_inputs=inputs,
+                variables=app_config.variables,
+                tenant_id=app_model.tenant_id,
+                strict_type_validation=True if invoke_from == InvokeFrom.SERVICE_API else False,
+            )
         # init application generate entity
         application_generate_entity = WorkflowAppGenerateEntity(
             task_id=str(uuid.uuid4()),
             app_config=app_config,
             file_upload_config=file_extra_config,
-            inputs=self._prepare_user_inputs(
-                user_inputs=inputs,
-                variables=app_config.variables,
-                tenant_id=app_model.tenant_id,
-                strict_type_validation=True if invoke_from == InvokeFrom.SERVICE_API else False,
-            ),
+            inputs=inputs,
             files=list(system_files),
             user_id=user.id,
             stream=streaming,
@@ -155,7 +171,10 @@ class WorkflowAppGenerator(BaseAppGenerator):
         # Create session factory
         session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
         # Create workflow execution(aka workflow run) repository
-        if invoke_from == InvokeFrom.DEBUGGER:
+        if triggered_from is not None:
+            # Use explicitly provided triggered_from (for async triggers)
+            workflow_triggered_from = triggered_from
+        elif invoke_from == InvokeFrom.DEBUGGER:
             workflow_triggered_from = WorkflowRunTriggeredFrom.DEBUGGING
         else:
             workflow_triggered_from = WorkflowRunTriggeredFrom.APP_RUN
@@ -182,8 +201,16 @@ 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,
         *,
@@ -196,6 +223,8 @@ 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.
@@ -231,8 +260,10 @@ 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,
             },
         )
 
@@ -426,6 +457,8 @@ 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.
@@ -469,6 +502,8 @@ 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,5 +1,6 @@
 import logging
 import time
+from collections.abc import Sequence
 from typing import cast
 
 from core.app.apps.base_app_queue_manager import AppQueueManager
@@ -8,6 +9,7 @@ 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
@@ -16,6 +18,7 @@ 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
 
@@ -35,17 +38,21 @@ 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
 
@@ -60,6 +67,7 @@ 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,
         )
@@ -92,6 +100,7 @@ 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

api/core/app/apps/workflow_app_runner.py

@@ -84,6 +84,7 @@ class WorkflowBasedAppRunner:
         workflow_id: str = "",
         tenant_id: str = "",
         user_id: str = "",
+        root_node_id: str | None = None,
     ) -> Graph:
         """
         Init graph
@@ -117,7 +118,7 @@ class WorkflowBasedAppRunner:
         )
 
         # init graph
-        graph = Graph.init(graph_config=graph_config, node_factory=node_factory)
+        graph = Graph.init(graph_config=graph_config, node_factory=node_factory, root_node_id=root_node_id)
 
         if not graph:
             raise ValueError("graph not found in workflow")

api/core/app/entities/app_invoke_entities.py

@@ -32,6 +32,10 @@ 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"
@@ -65,6 +69,8 @@ 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"
 
@@ -129,7 +135,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/layers/pause_state_persist_layer.py

@@ -1,5 +1,5 @@
 from sqlalchemy import Engine
-from sqlalchemy.orm import sessionmaker
+from sqlalchemy.orm import Session, sessionmaker
 
 from core.workflow.graph_engine.layers.base import GraphEngineLayer
 from core.workflow.graph_events.base import GraphEngineEvent
@@ -9,7 +9,7 @@ from repositories.factory import DifyAPIRepositoryFactory
 
 
 class PauseStatePersistenceLayer(GraphEngineLayer):
-    def __init__(self, session_factory: Engine | sessionmaker, state_owner_user_id: str):
+    def __init__(self, session_factory: Engine | sessionmaker[Session], state_owner_user_id: str):
         """Create a PauseStatePersistenceLayer.
 
         The `state_owner_user_id` is used when creating state file for pause.

api/core/app/layers/suspend_layer.py

@@ -0,0 +1,21 @@
+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

@@ -0,0 +1,88 @@
+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

@@ -0,0 +1,88 @@
+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,13 +0,0 @@
-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: list[str]

api/core/entities/mcp_provider.py

@@ -14,7 +14,6 @@ from core.helper.provider_cache import NoOpProviderCredentialCache
 from core.mcp.types import OAuthClientInformation, OAuthClientMetadata, OAuthTokens
 from core.tools.entities.common_entities import I18nObject
 from core.tools.entities.tool_entities import ToolProviderType
-from core.tools.utils.encryption import create_provider_encrypter
 
 if TYPE_CHECKING:
     from models.tools import MCPToolProvider
@@ -272,6 +271,7 @@ class MCPProviderEntity(BaseModel):
 
     def _decrypt_dict(self, data: dict[str, Any]) -> dict[str, Any]:
         """Generic method to decrypt dictionary fields"""
+        from core.tools.utils.encryption import create_provider_encrypter
 
         if not data:
             return {}

api/core/entities/parameter_entities.py

@@ -14,6 +14,7 @@ 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_entities.py

@@ -107,7 +107,7 @@ class CustomModelConfiguration(BaseModel):
 
     model: str
     model_type: ModelType
-    credentials: dict | None = None
+    credentials: dict | None
     current_credential_id: str | None = None
     current_credential_name: str | None = None
     available_model_credentials: list[CredentialConfiguration] = []
@@ -207,6 +207,7 @@ 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/name_generator.py

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

api/core/helper/provider_encryption.py

@@ -0,0 +1,129 @@
+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/hosting_configuration.py

@@ -56,9 +56,6 @@ class HostingConfiguration:
         self.provider_map[f"{DEFAULT_PLUGIN_ID}/minimax/minimax"] = self.init_minimax()
         self.provider_map[f"{DEFAULT_PLUGIN_ID}/spark/spark"] = self.init_spark()
         self.provider_map[f"{DEFAULT_PLUGIN_ID}/zhipuai/zhipuai"] = self.init_zhipuai()
-        self.provider_map[f"{DEFAULT_PLUGIN_ID}/gemini/google"] = self.init_gemini()
-        self.provider_map[f"{DEFAULT_PLUGIN_ID}/x/x"] = self.init_xai()
-        self.provider_map[f"{DEFAULT_PLUGIN_ID}/deepseek/deepseek"] = self.init_deepseek()
 
         self.moderation_config = self.init_moderation_config()
 
@@ -131,7 +128,7 @@ class HostingConfiguration:
         quotas: list[HostingQuota] = []
 
         if dify_config.HOSTED_OPENAI_TRIAL_ENABLED:
-            hosted_quota_limit = 0
+            hosted_quota_limit = dify_config.HOSTED_OPENAI_QUOTA_LIMIT
             trial_models = self.parse_restrict_models_from_env("HOSTED_OPENAI_TRIAL_MODELS")
             trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit, restrict_models=trial_models)
             quotas.append(trial_quota)
@@ -159,49 +156,18 @@ class HostingConfiguration:
             quota_unit=quota_unit,
         )
 
-    def init_gemini(self) -> HostingProvider:
-        quota_unit = QuotaUnit.CREDITS
-        quotas: list[HostingQuota] = []
-
-        if dify_config.HOSTED_GEMINI_TRIAL_ENABLED:
-            hosted_quota_limit = 0
-            trial_models = self.parse_restrict_models_from_env("HOSTED_GEMINI_TRIAL_MODELS")
-            trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit, restrict_models=trial_models)
-            quotas.append(trial_quota)
-
-        if dify_config.HOSTED_GEMINI_PAID_ENABLED:
-            paid_models = self.parse_restrict_models_from_env("HOSTED_GEMINI_PAID_MODELS")
-            paid_quota = PaidHostingQuota(restrict_models=paid_models)
-            quotas.append(paid_quota)
-
-        if len(quotas) > 0:
-            credentials = {
-                "google_api_key": dify_config.HOSTED_GEMINI_API_KEY,
-            }
-
-            if dify_config.HOSTED_GEMINI_API_BASE:
-                credentials["google_base_url"] = dify_config.HOSTED_GEMINI_API_BASE
-
-            return HostingProvider(enabled=True, credentials=credentials, quota_unit=quota_unit, quotas=quotas)
-
-        return HostingProvider(
-            enabled=False,
-            quota_unit=quota_unit,
-        )
-
-    def init_anthropic(self) -> HostingProvider:
-        quota_unit = QuotaUnit.CREDITS
+    @staticmethod
+    def init_anthropic() -> HostingProvider:
+        quota_unit = QuotaUnit.TOKENS
         quotas: list[HostingQuota] = []
 
         if dify_config.HOSTED_ANTHROPIC_TRIAL_ENABLED:
-            hosted_quota_limit = 0
-            trail_models = self.parse_restrict_models_from_env("HOSTED_ANTHROPIC_TRIAL_MODELS")
-            trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit, restrict_models=trail_models)
+            hosted_quota_limit = dify_config.HOSTED_ANTHROPIC_QUOTA_LIMIT
+            trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit)
             quotas.append(trial_quota)
 
         if dify_config.HOSTED_ANTHROPIC_PAID_ENABLED:
-            paid_models = self.parse_restrict_models_from_env("HOSTED_ANTHROPIC_PAID_MODELS")
-            paid_quota = PaidHostingQuota(restrict_models=paid_models)
+            paid_quota = PaidHostingQuota()
             quotas.append(paid_quota)
 
         if len(quotas) > 0:
@@ -219,66 +185,6 @@ class HostingConfiguration:
             quota_unit=quota_unit,
         )
 
-    def init_xai(self) -> HostingProvider:
-        quota_unit = QuotaUnit.CREDITS
-        quotas: list[HostingQuota] = []
-
-        if dify_config.HOSTED_XAI_TRIAL_ENABLED:
-            hosted_quota_limit = 0
-            trail_models = self.parse_restrict_models_from_env("HOSTED_XAI_TRIAL_MODELS")
-            trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit, restrict_models=trail_models)
-            quotas.append(trial_quota)
-
-        if dify_config.HOSTED_XAI_PAID_ENABLED:
-            paid_models = self.parse_restrict_models_from_env("HOSTED_XAI_PAID_MODELS")
-            paid_quota = PaidHostingQuota(restrict_models=paid_models)
-            quotas.append(paid_quota)
-
-        if len(quotas) > 0:
-            credentials = {
-                "api_key": dify_config.HOSTED_XAI_API_KEY,
-            }
-
-            if dify_config.HOSTED_XAI_API_BASE:
-                credentials["endpoint_url"] = dify_config.HOSTED_XAI_API_BASE
-
-            return HostingProvider(enabled=True, credentials=credentials, quota_unit=quota_unit, quotas=quotas)
-
-        return HostingProvider(
-            enabled=False,
-            quota_unit=quota_unit,
-        )
-
-    def init_deepseek(self) -> HostingProvider:
-        quota_unit = QuotaUnit.CREDITS
-        quotas: list[HostingQuota] = []
-
-        if dify_config.HOSTED_DEEPSEEK_TRIAL_ENABLED:
-            hosted_quota_limit = 0
-            trail_models = self.parse_restrict_models_from_env("HOSTED_DEEPSEEK_TRIAL_MODELS")
-            trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit, restrict_models=trail_models)
-            quotas.append(trial_quota)
-
-        if dify_config.HOSTED_DEEPSEEK_PAID_ENABLED:
-            paid_models = self.parse_restrict_models_from_env("HOSTED_DEEPSEEK_PAID_MODELS")
-            paid_quota = PaidHostingQuota(restrict_models=paid_models)
-            quotas.append(paid_quota)
-
-        if len(quotas) > 0:
-            credentials = {
-                "api_key": dify_config.HOSTED_DEEPSEEK_API_KEY,
-            }
-
-            if dify_config.HOSTED_DEEPSEEK_API_BASE:
-                credentials["endpoint_url"] = dify_config.HOSTED_DEEPSEEK_API_BASE
-
-            return HostingProvider(enabled=True, credentials=credentials, quota_unit=quota_unit, quotas=quotas)
-
-        return HostingProvider(
-            enabled=False,
-            quota_unit=quota_unit,
-        )
-
     @staticmethod
     def init_minimax() -> HostingProvider:
         quota_unit = QuotaUnit.TOKENS

api/core/llm_generator/llm_generator.py

@@ -14,12 +14,10 @@ from core.llm_generator.prompts import (
     JAVASCRIPT_CODE_GENERATOR_PROMPT_TEMPLATE,
     LLM_MODIFY_CODE_SYSTEM,
     LLM_MODIFY_PROMPT_SYSTEM,
-    MEMORY_UPDATE_PROMPT,
     PYTHON_CODE_GENERATOR_PROMPT_TEMPLATE,
     SYSTEM_STRUCTURED_OUTPUT_GENERATE,
     WORKFLOW_RULE_CONFIG_PROMPT_GENERATE_TEMPLATE,
 )
-from core.memory.entities import MemoryBlock, MemoryBlockSpec
 from core.model_manager import ModelManager
 from core.model_runtime.entities.llm_entities import LLMResult
 from core.model_runtime.entities.message_entities import PromptMessage, SystemPromptMessage, UserPromptMessage
@@ -30,7 +28,6 @@ from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
 from core.ops.utils import measure_time
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
 from core.workflow.entities.workflow_node_execution import WorkflowNodeExecutionMetadataKey
-from core.workflow.runtime.variable_pool import VariablePool
 from extensions.ext_database import db
 from extensions.ext_storage import storage
 from models import App, Message, WorkflowNodeExecutionModel
@@ -565,35 +562,3 @@ class LLMGenerator:
                 "Failed to invoke LLM model, model: %s", json.dumps(model_config.get("name")), exc_info=True
             )
             return {"error": f"An unexpected error occurred: {str(e)}"}
-
-    @staticmethod
-    def update_memory_block(
-        tenant_id: str,
-        visible_history: Sequence[tuple[str, str]],
-        variable_pool: VariablePool,
-        memory_block: MemoryBlock,
-        memory_spec: MemoryBlockSpec,
-    ) -> str:
-        model_instance = ModelManager().get_model_instance(
-            tenant_id=tenant_id,
-            provider=memory_spec.model.provider,
-            model=memory_spec.model.name,
-            model_type=ModelType.LLM,
-        )
-        formatted_history = ""
-        for sender, message in visible_history:
-            formatted_history += f"{sender}: {message}\n"
-        filled_instruction = variable_pool.convert_template(memory_spec.instruction).text
-        formatted_prompt = PromptTemplateParser(MEMORY_UPDATE_PROMPT).format(
-            inputs={
-                "formatted_history": formatted_history,
-                "current_value": memory_block.value,
-                "instruction": filled_instruction,
-            }
-        )
-        llm_result = model_instance.invoke_llm(
-            prompt_messages=[UserPromptMessage(content=formatted_prompt)],
-            model_parameters=memory_spec.model.completion_params,
-            stream=False,
-        )
-        return llm_result.message.get_text_content()

api/core/llm_generator/prompts.py

@@ -422,18 +422,3 @@ INSTRUCTION_GENERATE_TEMPLATE_PROMPT = """The output of this prompt is not as ex
 You should edit the prompt according to the IDEAL OUTPUT."""
 
 INSTRUCTION_GENERATE_TEMPLATE_CODE = """Please fix the errors in the {{#error_message#}}."""
-
-MEMORY_UPDATE_PROMPT = """
-Based on the following conversation history, update the memory content:
-
-Conversation history:
-{{formatted_history}}
-
-Current memory:
-{{current_value}}
-
-Update instruction:
-{{instruction}}
-
-Please output only the updated memory content, no other text like greeting:
-"""

api/core/memory/entities.py

@@ -1,119 +0,0 @@
-from __future__ import annotations
-
-from enum import StrEnum
-from typing import TYPE_CHECKING, Optional
-from uuid import uuid4
-
-from pydantic import BaseModel, Field
-
-if TYPE_CHECKING:
-    from core.app.app_config.entities import ModelConfig
-
-
-class MemoryScope(StrEnum):
-    """Memory scope determined by node_id field"""
-    APP = "app"  # node_id is None
-    NODE = "node"  # node_id is not None
-
-
-class MemoryTerm(StrEnum):
-    """Memory term determined by conversation_id field"""
-    SESSION = "session"  # conversation_id is not None
-    PERSISTENT = "persistent"  # conversation_id is None
-
-
-class MemoryStrategy(StrEnum):
-    ON_TURNS = "on_turns"
-
-
-class MemoryScheduleMode(StrEnum):
-    SYNC = "sync"
-    ASYNC = "async"
-
-
-class MemoryBlockSpec(BaseModel):
-    """Memory block specification for workflow configuration"""
-    id: str = Field(
-        default_factory=lambda: str(uuid4()),
-        description="Unique identifier for the memory block",
-    )
-    name: str = Field(description="Display name of the memory block")
-    description: str = Field(default="", description="Description of the memory block")
-    template: str = Field(description="Initial template content for the memory")
-    instruction: str = Field(description="Instructions for updating the memory")
-    scope: MemoryScope = Field(description="Scope of the memory (app or node level)")
-    term: MemoryTerm = Field(description="Term of the memory (session or persistent)")
-    strategy: MemoryStrategy = Field(description="Update strategy for the memory")
-    update_turns: int = Field(gt=0, description="Number of turns between updates")
-    preserved_turns: int = Field(gt=0, description="Number of conversation turns to preserve")
-    schedule_mode: MemoryScheduleMode = Field(description="Synchronous or asynchronous update mode")
-    model: ModelConfig = Field(description="Model configuration for memory updates")
-    end_user_visible: bool = Field(default=False, description="Whether memory is visible to end users")
-    end_user_editable: bool = Field(default=False, description="Whether memory is editable by end users")
-
-
-class MemoryCreatedBy(BaseModel):
-    end_user_id: str | None = None
-    account_id: str | None = None
-
-
-class MemoryBlock(BaseModel):
-    """Runtime memory block instance
-
-    Design Rules:
-    - app_id = None: Global memory (future feature, not implemented yet)
-    - app_id = str: App-specific memory
-    - conversation_id = None: Persistent memory (cross-conversation)
-    - conversation_id = str: Session memory (conversation-specific)
-    - node_id = None: App-level scope
-    - node_id = str: Node-level scope
-
-    These rules implicitly determine scope and term without redundant storage.
-    """
-    spec: MemoryBlockSpec
-    tenant_id: str
-    value: str
-    app_id: str
-    conversation_id: Optional[str] = None
-    node_id: Optional[str] = None
-    edited_by_user: bool = False
-    created_by: MemoryCreatedBy
-    version: int = Field(description="Memory block version number")
-
-
-class MemoryValueData(BaseModel):
-    value: str
-    edited_by_user: bool = False
-
-
-class ChatflowConversationMetadata(BaseModel):
-    """Metadata for chatflow conversation with visible message count"""
-    type: str = "mutable_visible_window"
-    visible_count: int = Field(gt=0, description="Number of visible messages to keep")
-
-
-class MemoryBlockWithConversation(MemoryBlock):
-    """MemoryBlock with optional conversation metadata for session memories"""
-    conversation_metadata: ChatflowConversationMetadata = Field(
-        description="Conversation metadata, only present for session memories"
-    )
-
-    @classmethod
-    def from_memory_block(
-        cls,
-        memory_block: MemoryBlock,
-        conversation_metadata: ChatflowConversationMetadata
-    ) -> MemoryBlockWithConversation:
-        """Create MemoryBlockWithConversation from MemoryBlock"""
-        return cls(
-            spec=memory_block.spec,
-            tenant_id=memory_block.tenant_id,
-            value=memory_block.value,
-            app_id=memory_block.app_id,
-            conversation_id=memory_block.conversation_id,
-            node_id=memory_block.node_id,
-            edited_by_user=memory_block.edited_by_user,
-            created_by=memory_block.created_by,
-            version=memory_block.version,
-            conversation_metadata=conversation_metadata
-        )

api/core/memory/errors.py

@@ -1,6 +0,0 @@
-class MemorySyncTimeoutError(Exception):
-    def __init__(self, app_id: str, conversation_id: str):
-        self.app_id = app_id
-        self.conversation_id = conversation_id
-        self.message = "Memory synchronization timeout after 50 seconds"
-        super().__init__(self.message)

api/core/ops/tencent_trace/client.py

@@ -5,7 +5,6 @@ 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
@@ -111,7 +110,6 @@ class TencentTraceClient:
         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
@@ -121,6 +119,7 @@ class TencentTraceClient:
 
         # Metrics exporter and instruments
         try:
+            from opentelemetry import metrics
             from opentelemetry.sdk.metrics import Histogram, MeterProvider
             from opentelemetry.sdk.metrics.export import AggregationTemporality, PeriodicExportingMetricReader
 
@@ -203,11 +202,9 @@ 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)
+                metrics.set_meter_provider(provider)
+                self.meter = metrics.get_meter("dify-sdk", dify_config.project.version)
 
                 # LLM operation duration histogram
                 self.hist_llm_duration = self.meter.create_histogram(
@@ -247,7 +244,6 @@ class TencentTraceClient:
                 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
@@ -257,7 +253,6 @@ class TencentTraceClient:
         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
@@ -284,14 +279,6 @@ 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)
@@ -330,13 +317,6 @@ class TencentTraceClient:
                 "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)
@@ -364,13 +344,6 @@ class TencentTraceClient:
                 "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)
@@ -398,13 +371,6 @@ class TencentTraceClient:
                 "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)
@@ -424,14 +390,6 @@ 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",
-                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)
@@ -516,19 +474,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/plugin/backwards_invocation/app.py

@@ -4,7 +4,6 @@ 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
@@ -16,6 +15,7 @@ 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 = create_or_update_end_user_for_user_id(app)
+            user = EndUserService.get_or_create_end_user(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,6 +94,7 @@ def as_normal_type(typ: StrEnum):
     if typ.value in {
         PluginParameterType.SECRET_INPUT,
         PluginParameterType.SELECT,
+        PluginParameterType.CHECKBOX,
     }:
         return "string"
     return typ.value
@@ -102,7 +103,13 @@ 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:
+            case (
+                PluginParameterType.STRING
+                | PluginParameterType.SECRET_INPUT
+                | PluginParameterType.SELECT
+                | PluginParameterType.CHECKBOX
+                | PluginParameterType.DYNAMIC_SELECT
+            ):
                 if value is None:
                     return ""
                 else:

api/core/plugin/entities/plugin.py

@@ -13,6 +13,7 @@ from core.plugin.entities.base import BasePluginEntity
 from core.plugin.entities.endpoint import EndpointProviderDeclaration
 from core.tools.entities.common_entities import I18nObject
 from core.tools.entities.tool_entities import ToolProviderEntity
+from core.trigger.entities.entities import TriggerProviderEntity
 
 
 class PluginInstallationSource(StrEnum):
@@ -63,6 +64,7 @@ class PluginCategory(StrEnum):
     Extension = auto()
     AgentStrategy = "agent-strategy"
     Datasource = "datasource"
+    Trigger = "trigger"
 
 
 class PluginDeclaration(BaseModel):
@@ -71,6 +73,7 @@ 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)
@@ -106,6 +109,7 @@ 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")
@@ -129,6 +133,8 @@ 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,3 +1,4 @@
+import enum
 from collections.abc import Mapping, Sequence
 from datetime import datetime
 from enum import StrEnum
@@ -14,6 +15,7 @@ from core.plugin.entities.parameters import PluginParameterOption
 from core.plugin.entities.plugin import PluginDeclaration, PluginEntity
 from core.tools.entities.common_entities import I18nObject
 from core.tools.entities.tool_entities import ToolProviderEntityWithPlugin
+from core.trigger.entities.entities import TriggerProviderEntity
 
 T = TypeVar("T", bound=(BaseModel | dict | list | bool | str))
 
@@ -205,3 +207,53 @@ 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,5 +1,9 @@
+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
@@ -13,6 +17,7 @@ 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,
 )
@@ -237,3 +242,43 @@ class RequestFetchAppInfo(BaseModel):
     """
 
     app_id: str
+
+
+class TriggerInvokeEventResponse(BaseModel):
+    variables: Mapping[str, Any] = Field(default_factory=dict)
+    cancelled: bool | None = 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,3 +10,13 @@ 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,6 +29,12 @@ 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(
@@ -43,7 +49,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 | list | bool | str))
+T = TypeVar("T", bound=(BaseModel | dict[str, Any] | list[Any] | bool | str))
 
 logger = logging.getLogger(__name__)
 
@@ -53,10 +59,10 @@ class BasePluginClient:
         self,
         method: str,
         path: str,
-        headers: dict | None = None,
-        data: bytes | dict | str | None = None,
-        params: dict | None = None,
-        files: dict | None = None,
+        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,
     ) -> httpx.Response:
         """
         Make a request to the plugin daemon inner API.
@@ -87,17 +93,17 @@ class BasePluginClient:
     def _prepare_request(
         self,
         path: str,
-        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]:
+        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]:
         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 | None = (
+        prepared_data: bytes | dict[str, Any] | str | None = (
             data if isinstance(data, (bytes, str, dict)) or data is None else None
         )
         if isinstance(data, dict):
@@ -112,10 +118,10 @@ class BasePluginClient:
         self,
         method: str,
         path: str,
-        params: dict | None = None,
-        headers: dict | None = None,
-        data: bytes | dict | None = None,
-        files: dict | None = None,
+        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,
     ) -> Generator[str, None, None]:
         """
         Make a stream request to the plugin daemon inner API
@@ -138,7 +144,7 @@ class BasePluginClient:
         try:
             with httpx.stream(**stream_kwargs) as response:
                 for raw_line in response.iter_lines():
-                    if raw_line is None:
+                    if not raw_line:
                         continue
                     line = raw_line.decode("utf-8") if isinstance(raw_line, bytes) else raw_line
                     line = line.strip()
@@ -155,10 +161,10 @@ class BasePluginClient:
         method: str,
         path: str,
         type_: type[T],
-        headers: dict | None = None,
-        data: bytes | dict | None = None,
-        params: dict | None = None,
-        files: dict | None = None,
+        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,
     ) -> Generator[T, None, None]:
         """
         Make a stream request to the plugin daemon inner API and yield the response as a model.
@@ -171,10 +177,10 @@ class BasePluginClient:
         method: str,
         path: str,
         type_: type[T],
-        headers: dict | None = None,
+        headers: dict[str, str] | None = None,
         data: bytes | None = None,
-        params: dict | None = None,
-        files: dict | None = None,
+        params: dict[str, Any] | None = None,
+        files: dict[str, Any] | None = None,
     ) -> T:
         """
         Make a request to the plugin daemon inner API and return the response as a model.
@@ -187,11 +193,11 @@ class BasePluginClient:
         method: str,
         path: str,
         type_: type[T],
-        headers: dict | None = None,
-        data: bytes | dict | None = None,
-        params: dict | None = None,
-        files: dict | None = None,
-        transformer: Callable[[dict], dict] | None = None,
+        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,
     ) -> T:
         """
         Make a request to the plugin daemon inner API and return the response as a model.
@@ -239,10 +245,10 @@ class BasePluginClient:
         method: str,
         path: str,
         type_: type[T],
-        headers: dict | None = None,
-        data: bytes | dict | None = None,
-        params: dict | None = None,
-        files: dict | None = None,
+        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,
     ) -> Generator[T, None, None]:
         """
         Make a stream request to the plugin daemon inner API and yield the response as a model.
@@ -302,6 +308,14 @@ 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,6 +15,7 @@ class DynamicSelectClient(BasePluginClient):
         provider: str,
         action: str,
         credentials: Mapping[str, Any],
+        credential_type: str,
         parameter: str,
     ) -> PluginDynamicSelectOptionsResponse:
         """
@@ -29,6 +30,7 @@ class DynamicSelectClient(BasePluginClient):
                 "data": {
                     "provider": GenericProviderID(provider).provider_name,
                     "credentials": credentials,
+                    "credential_type": credential_type,
                     "provider_action": action,
                     "parameter": parameter,
                 },