refactor(api): inject plugin trigger endpoint dispatch (#44547)

api/.importlinter

@@ -1447,10 +1447,11 @@ forbidden_modules =
 allow_indirect_imports = True
 
 [importlinter:contract:trigger-subscription-builder-persistence-boundary]
-name = Trigger subscription builder orchestration uses injected persistence
+name = Trigger builder and endpoint orchestration use injected persistence
 type = forbidden
 source_modules =
     services.trigger.trigger_subscription_builder_service
+    services.trigger.endpoint_service
 forbidden_modules =
     controllers
     extensions

api/controllers/trigger/trigger.py

@@ -1,12 +1,17 @@
 import logging
 import re
+from http import HTTPStatus
 
 from flask import Response, jsonify, request
 
 from controllers.common.access_response import plugin_endpoint_not_found_response
 from controllers.trigger import bp
 from extensions.ext_application_services import application_services
-from services.trigger.trigger_service import TriggerService
+from services.trigger.endpoint_service import (
+    TriggerEndpointEventNotFoundError,
+    TriggerEndpointLoadError,
+    TriggerEndpointProviderNotFoundError,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -23,17 +28,31 @@ def trigger_endpoint(endpoint_id: str) -> Response | tuple[Response, int]:
     if not UUID_MATCHER.match(endpoint_id):
         return plugin_endpoint_not_found_response()
     try:
-        response = TriggerService.process_endpoint(endpoint_id, request)
-        if not response:
+        response = application_services().workspaces.trigger_endpoints.process(endpoint_id, request)
+        if response is None:
             response = application_services().workspaces.trigger_builders.process_builder_validation_endpoint(
                 endpoint_id, request
             )
-        if not response:
+        if response is None:
             logger.info("Endpoint not found for %s", endpoint_id)
             return plugin_endpoint_not_found_response()
         return response
-    except ValueError as e:
-        return jsonify({"error": "Endpoint processing failed", "message": str(e)}), 400
+    except TriggerEndpointProviderNotFoundError:
+        return Response(status=HTTPStatus.NOT_FOUND, response="Trigger provider not found")
+    except TriggerEndpointLoadError:
+        logger.exception("Failed to load trigger subscription for endpoint %s", endpoint_id)
+        return Response(status=HTTPStatus.INTERNAL_SERVER_ERROR, response="Failed to get subscription by endpoint")
+    except TriggerEndpointEventNotFoundError:
+        return (
+            jsonify({"error": "Endpoint processing failed", "message": "Plugin returned an undeclared trigger event."}),
+            HTTPStatus.BAD_REQUEST,
+        )
+    except ValueError:
+        logger.exception("Invalid webhook request or plugin response for endpoint %s", endpoint_id)
+        return (
+            jsonify({"error": "Endpoint processing failed", "message": "Invalid webhook request or plugin response."}),
+            HTTPStatus.BAD_REQUEST,
+        )
     except Exception:
         logger.exception("Webhook processing failed for %s", endpoint_id)
-        return jsonify({"error": "Internal server error"}), 500
+        return jsonify({"error": "Internal server error"}), HTTPStatus.INTERNAL_SERVER_ERROR

api/extensions/application_services/workspace.py

@@ -19,6 +19,8 @@ from repositories.workspace.workspace_repository import WorkspaceRepository
 from services.account.adapters import RedisInvitationTokenStore
 from services.credentials.query import CredentialQuery
 from services.file_service import FileService
+from services.trigger.endpoint_gateway import PluginTriggerEndpointGateway, enqueue_triggered_workflows
+from services.trigger.endpoint_service import TriggerEndpointService
 from services.trigger.oauth_client_service import TriggerOAuthClientService
 from services.trigger.oauth_gateway import (
     OAuthProxyContextGateway,
@@ -32,6 +34,7 @@ from services.trigger.subscription_codec import ProviderTriggerSubscriptionCodec
 from services.trigger.subscription_gateway import PluginTriggerSubscriptionGateway
 from services.trigger.subscription_query_service import TriggerSubscriptionQueryService
 from services.trigger.subscription_service import TriggerSubscriptionService
+from services.trigger.trigger_request_service import TriggerHttpRequestCachingService
 from services.trigger.trigger_subscription_builder_service import TriggerSubscriptionBuilderService
 from services.workspace.gateways import (
     DeploymentWorkspaceCreationPolicy,
@@ -72,6 +75,7 @@ class WorkspaceServices:
     trigger_subscriptions: TriggerSubscriptionQueryService
     trigger_subscription_management: TriggerSubscriptionService
     trigger_builders: TriggerSubscriptionBuilderService
+    trigger_endpoints: TriggerEndpointService
     trigger_oauth_clients: TriggerOAuthClientService
     trigger_oauth: TriggerOAuthService
 
@@ -112,8 +116,9 @@ def build_workspace_services(
     credentials: CredentialQuery,
 ) -> WorkspaceServices:
     trigger_providers = PluginTriggerProviderQuery()
+    subscriptions = TriggerSubscriptionRepository(session_factory=database_client)
     trigger_subscription_management = TriggerSubscriptionService(
-        subscriptions=TriggerSubscriptionRepository(session_factory=database_client),
+        subscriptions=subscriptions,
         providers=PluginTriggerSubscriptionGateway(),
         locks=RedisTriggerSubscriptionLocks(redis=redis),
         codec=ProviderTriggerSubscriptionCodec(),
@@ -166,6 +171,12 @@ def build_workspace_services(
         trigger_subscriptions=TriggerSubscriptionQueryService(credentials=credentials, schemas=trigger_providers),
         trigger_subscription_management=trigger_subscription_management,
         trigger_builders=trigger_builders,
+        trigger_endpoints=TriggerEndpointService(
+            subscriptions=subscriptions,
+            provider=PluginTriggerEndpointGateway(),
+            requests=TriggerHttpRequestCachingService,
+            publish=enqueue_triggered_workflows,
+        ),
         trigger_oauth_clients=trigger_oauth_clients,
         trigger_oauth=TriggerOAuthService(
             clients=trigger_oauth_clients,

api/repositories/trigger/subscription_repository.py

@@ -1,4 +1,4 @@
-"""Short SQL transactions and Redis serialization for trigger subscription commands."""
+"""Short SQL transactions and Redis serialization for trigger subscriptions."""
 
 import logging
 import sqlite3
@@ -70,6 +70,12 @@ class TriggerSubscriptionRepository(TriggerSubscriptionStore):
             )
             return self._snapshot(row) if row is not None else None
 
+    def get_by_endpoint(self, *, endpoint_id: str) -> TriggerSubscriptionSnapshot | None:
+        """Resolve a public endpoint capability, deriving its owner from the stored row."""
+        with self._session_factory() as session:
+            row = session.scalar(select(TriggerSubscription).where(TriggerSubscription.endpoint_id == endpoint_id))
+            return self._snapshot(row) if row is not None else None
+
     @override
     def create(self, *, subscription: TriggerSubscriptionSnapshot, max_count: int) -> None:
         try:

api/services/trigger/endpoint_gateway.py

@@ -0,0 +1,87 @@
+"""Adapt persisted endpoint snapshots to the plugin runtime and existing task queue."""
+
+from flask import Request
+
+from core.helper.provider_encryption import create_provider_encrypter
+from core.plugin.impl.exc import PluginNotFoundError
+from core.trigger.entities.entities import Subscription
+from core.trigger.trigger_manager import TriggerManager
+from core.trigger.utils.encryption import TriggerProviderCredentialsCache, TriggerProviderPropertiesCache
+from core.trigger.utils.endpoint import generate_plugin_trigger_endpoint_url
+from models.provider_ids import TriggerProviderID
+from services.trigger.endpoint_service import (
+    TriggerEndpointDispatch,
+    TriggerEndpointLoadError,
+    TriggerEndpointProviderNotFoundError,
+)
+from services.trigger.subscription_service import TriggerSubscriptionSnapshot
+from services.workflow.entities import PluginTriggerDispatchData
+
+
+class PluginTriggerEndpointGateway:
+    """Use existing credential caches after the subscription query has closed.
+
+    TODO: The shared RSA key provider still owns a scoped DB session; remove that
+    dependency when tenant key reads return detached values.
+    """
+
+    def dispatch(self, *, subscription: TriggerSubscriptionSnapshot, request: Request) -> TriggerEndpointDispatch:
+        try:
+            provider = TriggerManager.get_trigger_provider(
+                tenant_id=subscription.tenant_id, provider_id=TriggerProviderID(subscription.provider_id)
+            )
+            credentials_encrypter, _ = create_provider_encrypter(
+                tenant_id=subscription.tenant_id,
+                config=provider.get_credential_schema_config(subscription.credential_type),
+                cache=TriggerProviderCredentialsCache(
+                    tenant_id=subscription.tenant_id,
+                    provider_id=str(provider.get_provider_id()),
+                    credential_id=subscription.id,
+                ),
+            )
+            credentials = credentials_encrypter.decrypt(subscription.credentials)
+            properties_encrypter, _ = create_provider_encrypter(
+                tenant_id=subscription.tenant_id,
+                config=provider.get_properties_schema(),
+                cache=TriggerProviderPropertiesCache(
+                    tenant_id=subscription.tenant_id,
+                    provider_id=str(provider.get_provider_id()),
+                    subscription_id=subscription.id,
+                ),
+            )
+            properties = properties_encrypter.decrypt(subscription.properties)
+        except Exception as error:
+            # TriggerManager wraps a missing plugin in ValueError, preserving its cause.
+            if isinstance(error, PluginNotFoundError) or (
+                isinstance(error, ValueError) and isinstance(error.__cause__, PluginNotFoundError)
+            ):
+                raise TriggerEndpointProviderNotFoundError(
+                    f"Trigger provider {subscription.provider_id} not found."
+                ) from error
+            raise TriggerEndpointLoadError(f"Failed to prepare trigger subscription {subscription.id}.") from error
+
+        try:
+            result = provider.dispatch(
+                request=request,
+                subscription=Subscription(
+                    expires_at=subscription.expires_at,
+                    endpoint=generate_plugin_trigger_endpoint_url(subscription.endpoint_id),
+                    parameters=subscription.parameters,
+                    properties=properties,
+                ),
+                credentials=credentials,
+                credential_type=subscription.credential_type,
+            )
+        except PluginNotFoundError as error:
+            raise TriggerEndpointProviderNotFoundError(
+                f"Trigger provider {subscription.provider_id} not found."
+            ) from error
+        return TriggerEndpointDispatch(
+            result=result, event_names=frozenset(event.identity.name for event in provider.get_events())
+        )
+
+
+def enqueue_triggered_workflows(data: PluginTriggerDispatchData) -> None:
+    from tasks.trigger_processing_tasks import dispatch_triggered_workflows_async
+
+    dispatch_triggered_workflows_async.delay(data.model_dump(mode="json"))

api/services/trigger/endpoint_service.py

@@ -0,0 +1,112 @@
+"""Dispatch persisted trigger endpoints without holding a subscription transaction."""
+
+import logging
+import secrets
+import time
+from collections.abc import Callable, Mapping
+from dataclasses import dataclass
+from typing import Any, Protocol
+
+from flask import Request, Response
+
+from core.plugin.entities.request import TriggerDispatchResponse
+from services.trigger.subscription_service import TriggerSubscriptionSnapshot
+from services.workflow.entities import PluginTriggerDispatchData
+
+logger = logging.getLogger(__name__)
+
+
+class TriggerEndpointLoadError(Exception):
+    """A persisted subscription or its provider configuration could not be loaded."""
+
+
+class TriggerEndpointProviderNotFoundError(Exception):
+    """The subscription's plugin provider is no longer available."""
+
+
+class TriggerEndpointEventNotFoundError(Exception):
+    """The plugin dispatched an event absent from its provider declaration."""
+
+
+class TriggerEndpointQuery(Protocol):
+    def get_by_endpoint(self, *, endpoint_id: str) -> TriggerSubscriptionSnapshot | None:
+        """Return the capability's detached owner/configuration, or None when absent."""
+        ...
+
+
+@dataclass(frozen=True, slots=True)
+class TriggerEndpointDispatch:
+    result: TriggerDispatchResponse
+    event_names: frozenset[str]
+
+
+class TriggerEndpointGateway(Protocol):
+    def dispatch(self, *, subscription: TriggerSubscriptionSnapshot, request: Request) -> TriggerEndpointDispatch: ...
+
+
+class TriggerEndpointRequestStore(Protocol):
+    def persist_request(self, request_id: str, request: Request) -> None: ...
+
+    def persist_payload(self, request_id: str, payload: Mapping[str, Any]) -> None: ...
+
+
+class TriggerEndpointService:
+    def __init__(
+        self,
+        *,
+        subscriptions: TriggerEndpointQuery,
+        provider: TriggerEndpointGateway,
+        requests: TriggerEndpointRequestStore,
+        publish: Callable[[PluginTriggerDispatchData], None],
+    ) -> None:
+        self._subscriptions = subscriptions
+        self._provider = provider
+        self._requests = requests
+        self._publish = publish
+
+    def process(self, endpoint_id: str, request: Request) -> Response | None:
+        """Return the plugin response, or None so the caller can try a temporary builder."""
+        timestamp = int(time.time())
+        try:
+            subscription = self._subscriptions.get_by_endpoint(endpoint_id=endpoint_id)
+        except Exception as error:
+            raise TriggerEndpointLoadError(f"Failed to load trigger endpoint {endpoint_id}.") from error
+        if subscription is None:
+            return None
+
+        dispatched = self._provider.dispatch(subscription=subscription, request=request)
+        result = dispatched.result
+        if result.events:
+            request_id = f"trigger_request_{timestamp}_{secrets.token_hex(6)}"
+            self._requests.persist_request(request_id, request)
+            self._requests.persist_payload(request_id, result.payload)
+            for event_name in result.events:
+                if event_name not in dispatched.event_names:
+                    logger.error(
+                        "Event name %s not found in provider %s for endpoint %s",
+                        event_name,
+                        subscription.provider_id,
+                        endpoint_id,
+                    )
+                    raise TriggerEndpointEventNotFoundError(
+                        f"Event name {event_name} not found in provider {subscription.provider_id}"
+                    )
+            self._publish(
+                PluginTriggerDispatchData(
+                    user_id=result.user_id,
+                    tenant_id=subscription.tenant_id,
+                    endpoint_id=subscription.endpoint_id,
+                    provider_id=subscription.provider_id,
+                    subscription_id=subscription.id,
+                    timestamp=timestamp,
+                    events=list(result.events),
+                    request_id=request_id,
+                )
+            )
+            logger.info(
+                "Queued async dispatching for %d triggers on endpoint %s with request_id %s",
+                len(result.events),
+                endpoint_id,
+                request_id,
+            )
+        return result.response

api/services/trigger/trigger_provider_service.py

@@ -38,7 +38,7 @@ class TriggerProviderService:
     """Legacy subscription reads and renewal tasks.
 
     Console management and builder writes use the injected TriggerSubscriptionService.
-    TODO: Retire these ORM-returning reads when webhook dispatch, dynamic options and
+    TODO: Retire these ORM-returning reads when debug invocation, dynamic options and
     renewal tasks migrate to detached subscription queries and explicit dependencies.
     """
 
@@ -317,32 +317,3 @@ class TriggerProviderService:
                 if oauth_params is not None
                 else None
             )
-
-    @classmethod
-    def get_subscription_by_endpoint(cls, endpoint_id: str) -> TriggerSubscription | None:
-        """
-        Get a trigger subscription by the endpoint ID.
-        """
-        with Session(db.engine, expire_on_commit=False) as session:
-            subscription = session.scalar(
-                select(TriggerSubscription).where(TriggerSubscription.endpoint_id == endpoint_id).limit(1)
-            )
-            if not subscription:
-                return None
-            provider_controller: PluginTriggerProviderController = TriggerManager.get_trigger_provider(
-                tenant_id=subscription.tenant_id, provider_id=TriggerProviderID(subscription.provider_id)
-            )
-            credential_encrypter, _ = create_trigger_provider_encrypter_for_subscription(
-                tenant_id=subscription.tenant_id,
-                controller=provider_controller,
-                subscription=subscription,
-            )
-            subscription.credentials = dict(credential_encrypter.decrypt(subscription.credentials))
-
-            properties_encrypter, _ = create_trigger_provider_encrypter_for_properties(
-                tenant_id=subscription.tenant_id,
-                controller=provider_controller,
-                subscription=subscription,
-            )
-            subscription.properties = dict(properties_encrypter.decrypt(subscription.properties))
-            return subscription

api/services/trigger/trigger_service.py

@@ -1,22 +1,17 @@
 import logging
-import secrets
-import time
 from collections.abc import Mapping
 from typing import Any
 
-from flask import Request, Response
 from pydantic import BaseModel
 from sqlalchemy import select
 from sqlalchemy.orm import sessionmaker
 
 from core.plugin.entities.plugin_daemon import CredentialType
-from core.plugin.entities.request import TriggerDispatchResponse, TriggerInvokeEventResponse
-from core.plugin.impl.exc import PluginNotFoundError
+from core.plugin.entities.request import TriggerInvokeEventResponse
 from core.trigger.constants import TRIGGER_PLUGIN_NODE_TYPE
 from core.trigger.debug.events import PluginTriggerDebugEvent
 from core.trigger.provider import PluginTriggerProviderController
 from core.trigger.trigger_manager import TriggerManager
-from core.trigger.utils.encryption import create_trigger_provider_encrypter_for_subscription
 from core.workflow.nodes.trigger_plugin.entities import TriggerEventNodeData
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
@@ -27,8 +22,6 @@ from models.trigger import TriggerSubscription, WorkflowPluginTrigger
 from models.workflow import Workflow
 from services.trigger.trigger_provider_service import TriggerProviderService
 from services.trigger.trigger_request_service import TriggerHttpRequestCachingService
-from services.workflow.entities import PluginTriggerDispatchData
-from tasks.trigger_processing_tasks import dispatch_triggered_workflows_async
 
 logger = logging.getLogger(__name__)
 
@@ -73,82 +66,6 @@ class TriggerService:
             payload=payload,
         )
 
-    @classmethod
-    def process_endpoint(cls, endpoint_id: str, request: Request) -> Response | None:
-        """
-        Extract and process data from incoming endpoint request.
-
-        Args:
-            endpoint_id: Endpoint ID
-            request: Request
-        """
-        timestamp = int(time.time())
-        subscription: TriggerSubscription | None = None
-        try:
-            subscription = TriggerProviderService.get_subscription_by_endpoint(endpoint_id)
-        except PluginNotFoundError:
-            return Response(status=404, response="Trigger provider not found")
-        except Exception:
-            return Response(status=500, response="Failed to get subscription by endpoint")
-
-        if not subscription:
-            return None
-
-        provider_id = TriggerProviderID(subscription.provider_id)
-        controller: PluginTriggerProviderController = TriggerManager.get_trigger_provider(
-            tenant_id=subscription.tenant_id, provider_id=provider_id
-        )
-        encrypter, _ = create_trigger_provider_encrypter_for_subscription(
-            tenant_id=subscription.tenant_id,
-            controller=controller,
-            subscription=subscription,
-        )
-        dispatch_response: TriggerDispatchResponse = controller.dispatch(
-            request=request,
-            subscription=subscription.to_entity(),
-            credentials=encrypter.decrypt(subscription.credentials),
-            credential_type=CredentialType.of(subscription.credential_type),
-        )
-
-        if dispatch_response.events:
-            request_id = f"trigger_request_{timestamp}_{secrets.token_hex(6)}"
-
-            # save the request and payload to storage as persistent data
-            TriggerHttpRequestCachingService.persist_request(request_id, request)
-            TriggerHttpRequestCachingService.persist_payload(request_id, dispatch_response.payload)
-
-            # Validate event names
-            for event_name in dispatch_response.events:
-                if controller.get_event(event_name) is None:
-                    logger.error(
-                        "Event name %s not found in provider %s for endpoint %s",
-                        event_name,
-                        subscription.provider_id,
-                        endpoint_id,
-                    )
-                    raise ValueError(f"Event name {event_name} not found in provider {subscription.provider_id}")
-
-            plugin_trigger_dispatch_data = PluginTriggerDispatchData(
-                user_id=dispatch_response.user_id,
-                tenant_id=subscription.tenant_id,
-                endpoint_id=endpoint_id,
-                provider_id=subscription.provider_id,
-                subscription_id=subscription.id,
-                timestamp=timestamp,
-                events=list(dispatch_response.events),
-                request_id=request_id,
-            )
-            dispatch_data = plugin_trigger_dispatch_data.model_dump(mode="json")
-            dispatch_triggered_workflows_async.delay(dispatch_data)
-
-            logger.info(
-                "Queued async dispatching for %d triggers on endpoint %s with request_id %s",
-                len(dispatch_response.events),
-                endpoint_id,
-                request_id,
-            )
-        return dispatch_response.response
-
     @classmethod
     def sync_plugin_trigger_relationships(cls, app: App, workflow: Workflow):
         """

api/tests/e2e_tests/containers/services/test_trigger_builder_callers.py

@@ -316,7 +316,7 @@ def test_public_callback_invalid_builder_data_is_a_bad_request(
 
     assert response.status_code == HTTPStatus.BAD_REQUEST
     assert response.get_json()["error"] == "Endpoint processing failed"
-    assert "Invalid JSON" in response.get_json()["message"]
+    assert response.get_json()["message"] == "Invalid webhook request or plugin response."
     assert daemon.requests == []
 
 

api/tests/e2e_tests/containers/services/test_trigger_endpoint.py

@@ -0,0 +1,382 @@
+"""Persisted webhook endpoints use real SQL, Redis, RSA, storage, plugin HTTP and Celery."""
+
+import json
+from collections.abc import Iterator
+from contextvars import Context
+from dataclasses import dataclass, field
+from http import HTTPStatus
+from queue import Empty
+from time import time
+from uuid import uuid4
+
+import pytest
+from celery import Celery
+from celery.signals import before_task_publish
+from kombu.message import Message
+from kombu.simple import SimpleQueue
+from sqlalchemy import text
+
+from controllers.trigger.trigger import trigger_endpoint
+from core.helper.encrypter import encrypt_token
+from core.plugin.entities.plugin_daemon import CredentialType
+from core.plugin.utils.http_parser import deserialize_request
+from core.trigger.utils.encryption import TriggerProviderCredentialsCache, TriggerProviderPropertiesCache
+from core.trigger.utils.endpoint import generate_plugin_trigger_endpoint_url
+from extensions.ext_redis import redis_client
+from extensions.ext_storage import storage
+from models.trigger import TriggerSubscription
+from services.trigger.trigger_request_service import TriggerHttpRequestCachingService
+from services.workflow.entities import PluginTriggerDispatchData
+from tasks.trigger_processing_tasks import TRIGGER_QUEUE, dispatch_triggered_workflows_async
+from tests.e2e_tests.containers.services import test_trigger_subscription_builder
+from tests.e2e_tests.containers.services.test_trigger_subscription_query import _provider_with_secrets
+from tests.unit_tests.controllers.console.workspace import test_trigger_oauth_client
+from tests.unit_tests.controllers.console.workspace.test_trigger_oauth_client import _PROVIDER, _Harness
+from tests.unit_tests.controllers.console.workspace.test_trigger_subscription_builder import _Daemon
+from tests.unit_tests.controllers.console.workspace.test_trigger_subscription_query import _add_subscription
+
+application_redis = test_trigger_subscription_builder.application_redis
+sqlite_engine = test_trigger_subscription_builder.sqlite_engine
+sqlite_session_factory = test_trigger_subscription_builder.sqlite_session_factory
+harness = test_trigger_oauth_client.harness
+daemon = test_trigger_subscription_builder.daemon
+
+
+@dataclass(frozen=True)
+class _Publication:
+    data: PluginTriggerDispatchData
+    persisted_request: bytes
+    persisted_payload: bytes
+    open_transactions: bool
+
+
+@dataclass
+class _Runtime:
+    queue: SimpleQueue
+    publications: list[_Publication] = field(default_factory=list)
+
+    def take_messages(self) -> list[Message]:
+        messages: list[Message] = []
+        while True:
+            try:
+                message = self.queue.get(block=False)
+            except Empty:
+                return messages
+            messages.append(message)
+            message.ack()
+
+
+@pytest.fixture
+def runtime(harness: _Harness) -> Iterator[_Runtime]:
+    harness.app.add_url_rule(
+        "/triggers/plugin/<string:endpoint_id>",
+        view_func=trigger_endpoint,
+        methods=["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"],
+    )
+    # Exercise the production shared task and enqueue callback with a real
+    # in-memory broker. No worker executes the queued workflow in this test.
+    previous_app = dispatch_triggered_workflows_async.app
+    celery_app = Celery("trigger-endpoint-tests", broker="memory://", set_as_current=False)
+    celery_app.conf.update(task_always_eager=False, task_serializer="json")
+    celery_app.set_current()
+    try:
+        with celery_app.connection_for_write() as connection:
+            with SimpleQueue(connection.channel(), TRIGGER_QUEUE) as queue:
+                queue.clear()
+                runtime = _Runtime(queue)
+
+                def observe_publication(sender: str, body: object, **_kwargs: object) -> None:
+                    if sender != dispatch_triggered_workflows_async.name:
+                        return
+                    assert isinstance(body, tuple)
+                    args = body[0]
+                    assert isinstance(args, tuple)
+                    data = PluginTriggerDispatchData.model_validate(args[0])
+                    runtime.publications.append(
+                        _Publication(
+                            data=data,
+                            persisted_request=storage.load_once(f"triggers/{data.request_id}.raw"),
+                            persisted_payload=storage.load_once(f"triggers/{data.request_id}.payload"),
+                            open_transactions=any(session.in_transaction() for session in harness.metadata.sessions),
+                        )
+                    )
+
+                before_task_publish.connect(observe_publication, weak=False)
+                try:
+                    yield runtime
+                finally:
+                    # Installed Celery stubs omit the public Signal.disconnect API.
+                    before_task_publish.disconnect(observe_publication)  # pyrefly: ignore[missing-attribute]
+                    queue.clear()
+    finally:
+        previous_app.set_current()
+        celery_app.close()
+
+
+def _seed(harness: _Harness, daemon: _Daemon, credential_type: CredentialType) -> TriggerSubscription:
+    daemon.provider = _provider_with_secrets()
+    with harness.app.app_context():
+        token = encrypt_token(harness.tenant_id, "endpoint-token")
+        secret = encrypt_token(harness.tenant_id, "endpoint-property")
+    return _add_subscription(
+        harness,
+        user_id=str(uuid4()),
+        credential_type=credential_type,
+        credentials={} if credential_type == CredentialType.UNAUTHORIZED else {"token": token, "username": "octocat"},
+        properties={"secret": secret, "repository": "dify"},
+    )
+
+
+def _dispatch(daemon: _Daemon, *, events: list[str]) -> bytes:
+    body = b"accepted\x00\xff"
+    raw_response = (
+        b"HTTP/1.1 202 Accepted\r\nContent-Type: application/octet-stream\r\n"
+        b"X-Webhook-Result: accepted\r\nSet-Cookie: first=1\r\nSet-Cookie: second=2\r\n"
+        + f"Content-Length: {len(body)}\r\n\r\n".encode()
+        + body
+    )
+    daemon.response_data = {
+        "user_id": "plugin-user",
+        "events": events,
+        "response": raw_response.hex(),
+        "payload": {"repository": "dify", "nested": {"count": 0, "empty": []}},
+    }
+    return body
+
+
+@pytest.mark.parametrize("credential_type", list(CredentialType))
+def test_callback_preserves_plugin_response_and_uses_subscription_owner_and_cached_secrets(
+    harness: _Harness, daemon: _Daemon, runtime: _Runtime, credential_type: CredentialType
+) -> None:
+    subscription = _seed(harness, daemon, credential_type)
+    expected_body = _dispatch(daemon, events=[])
+    url = f"/triggers/plugin/{subscription.endpoint_id}?tenant_id=untrusted-workspace"
+
+    response = Context().run(
+        harness.client.post, url, data=b"webhook-body", headers={"X-Webhook-Signature": "signature"}
+    )
+
+    assert response.status_code == HTTPStatus.ACCEPTED
+    assert response.data == expected_body
+    assert response.content_type == "application/octet-stream"
+    assert response.headers["X-Webhook-Result"] == "accepted"
+    assert response.headers.getlist("Set-Cookie") == ["first=1", "second=2"]
+    assert response.headers["Content-Length"] == str(len(expected_body))
+    sent = next(request for request in daemon.requests if request.method == "POST")
+    assert sent.path == f"/plugin/{harness.tenant_id}/dispatch/trigger/dispatch_event"
+    assert sent.headers["X-Plugin-ID"] == "langgenius/github"
+    assert sent.headers["X-Plugin-Unique-Identifier"] == daemon.provider["plugin_unique_identifier"]
+    data = sent.payload["data"]
+    assert isinstance(data, dict)
+    expected_credentials: dict[str, str] = (
+        {} if credential_type == CredentialType.UNAUTHORIZED else {"token": "endpoint-token", "username": "octocat"}
+    )
+    assert data["credentials"] == expected_credentials
+    assert data["credential_type"] == credential_type.value
+    assert data["provider"] == "github"
+    assert data["subscription"] == {
+        "expires_at": subscription.expires_at,
+        "endpoint": generate_plugin_trigger_endpoint_url(subscription.endpoint_id),
+        "parameters": {"event": "push"},
+        "properties": {"secret": "endpoint-property", "repository": "dify"},
+    }
+    raw_request = data["raw_http_request"]
+    assert isinstance(raw_request, str)
+    restored = deserialize_request(bytes.fromhex(raw_request))
+    assert restored.method == "POST"
+    assert restored.full_path == url
+    assert restored.headers["X-Webhook-Signature"] == "signature"
+    assert restored.get_data() == b"webhook-body"
+    assert all(not request.open_transactions for request in daemon.requests)
+
+    credential_key = TriggerProviderCredentialsCache(harness.tenant_id, _PROVIDER, subscription.id).cache_key
+    properties_key = TriggerProviderPropertiesCache(harness.tenant_id, _PROVIDER, subscription.id).cache_key
+    cached_credentials = redis_client.get(credential_key)
+    cached_properties = redis_client.get(properties_key)
+    assert cached_credentials is not None
+    assert cached_properties is not None
+    assert json.loads(cached_credentials) == expected_credentials
+    assert json.loads(cached_properties) == {"secret": "endpoint-property", "repository": "dify"}
+    repeated = Context().run(harness.client.post, url, data=b"webhook-body")
+    assert repeated.status_code == HTTPStatus.ACCEPTED
+    assert repeated.data == expected_body
+    repeated_data = daemon.requests[-1].payload["data"]
+    assert isinstance(repeated_data, dict)
+    assert repeated_data["credentials"] == expected_credentials
+    assert repeated_data["subscription"] == data["subscription"]
+    assert runtime.take_messages() == []
+    assert runtime.publications == []
+    assert not storage.exists("triggers/")
+    with harness.session_factory() as session:
+        saved = session.get(TriggerSubscription, subscription.id)
+        assert saved is not None
+        assert saved.credentials == subscription.credentials
+        assert saved.properties == subscription.properties
+
+
+def test_events_are_persisted_before_real_task_publication_with_trusted_subscription_identity(
+    harness: _Harness, daemon: _Daemon, runtime: _Runtime
+) -> None:
+    subscription = _seed(harness, daemon, CredentialType.API_KEY)
+    expected_body = _dispatch(daemon, events=["push", "push"])
+    before = int(time())
+
+    response = Context().run(
+        harness.client.post,
+        f"/triggers/plugin/{subscription.endpoint_id}?tenant_id=untrusted",
+        data=b"persisted-webhook-body",
+        headers={"X-Webhook-Signature": "signed"},
+    )
+
+    assert response.status_code == HTTPStatus.ACCEPTED
+    assert response.data == expected_body
+    assert len(runtime.publications) == 1
+    publication = runtime.publications[0]
+    data = publication.data
+    assert before <= data.timestamp <= int(time())
+    assert data.request_id.startswith(f"trigger_request_{data.timestamp}_")
+    assert len(data.request_id.rsplit("_", 1)[1]) == 12
+    assert data.model_dump(exclude={"request_id", "timestamp"}) == {
+        "user_id": "plugin-user",
+        "tenant_id": harness.tenant_id,
+        "endpoint_id": subscription.endpoint_id,
+        "provider_id": _PROVIDER,
+        "subscription_id": subscription.id,
+        "events": ["push", "push"],
+    }
+    assert not publication.open_transactions
+    assert json.loads(publication.persisted_payload) == {"repository": "dify", "nested": {"count": 0, "empty": []}}
+    assert deserialize_request(publication.persisted_request).get_data() == b"persisted-webhook-body"
+    restored = TriggerHttpRequestCachingService.get_request(data.request_id)
+    assert restored.get_data() == b"persisted-webhook-body"
+    assert restored.headers["X-Webhook-Signature"] == "signed"
+    assert TriggerHttpRequestCachingService.get_payload(data.request_id) == json.loads(publication.persisted_payload)
+    messages = runtime.take_messages()
+    assert len(messages) == 1
+    assert messages[0].headers["task"] == dispatch_triggered_workflows_async.name
+    assert messages[0].delivery_info["routing_key"] == TRIGGER_QUEUE
+    assert messages[0].payload[0] == [data.model_dump(mode="json")]
+    assert all(not request.open_transactions for request in daemon.requests)
+
+
+def test_unknown_event_returns_bad_request_after_persistence_without_publishing(
+    harness: _Harness, daemon: _Daemon, runtime: _Runtime, caplog: pytest.LogCaptureFixture
+) -> None:
+    subscription = _seed(harness, daemon, CredentialType.API_KEY)
+    _dispatch(daemon, events=["push", "undeclared"])
+
+    response = Context().run(harness.client.post, f"/triggers/plugin/{subscription.endpoint_id}", data=b"invalid-event")
+
+    assert response.status_code == HTTPStatus.BAD_REQUEST
+    assert response.get_json() == {
+        "error": "Endpoint processing failed",
+        "message": "Plugin returned an undeclared trigger event.",
+    }
+    expected_log = f"Event name undeclared not found in provider {_PROVIDER} for endpoint {subscription.endpoint_id}"
+    assert expected_log in caplog.text
+    assert runtime.take_messages() == []
+    assert runtime.publications == []
+    files = storage.scan("triggers/")
+    assert len(files) == 2
+    assert {filename.rsplit(".", 1)[1] for filename in files} == {"raw", "payload"}
+    raw_file = next(filename for filename in files if filename.endswith(".raw"))
+    assert deserialize_request(storage.load_once(raw_file)).get_data() == b"invalid-event"
+
+
+@pytest.mark.parametrize("failure", ["provider-unavailable", "invalid-provider", "invalid-credentials"])
+def test_subscription_preparation_failures_keep_raw_error_response_and_do_not_fall_back(
+    harness: _Harness, daemon: _Daemon, runtime: _Runtime, failure: str
+) -> None:
+    subscription = _seed(harness, daemon, CredentialType.API_KEY)
+    _dispatch(daemon, events=[])
+    if failure == "provider-unavailable":
+        daemon.status = HTTPStatus.SERVICE_UNAVAILABLE
+    else:
+        with harness.session_factory.begin() as session:
+            if failure == "invalid-provider":
+                saved = session.get(TriggerSubscription, subscription.id)
+                assert saved is not None
+                saved.provider_id = "not/a/valid/provider"
+            else:
+                # Historical corrupt JSON bypasses the application-owned type.
+                session.execute(
+                    text("UPDATE trigger_subscriptions SET credentials = '42' WHERE id = :id"),
+                    {"id": subscription.id},
+                )
+
+    response = Context().run(harness.client.post, f"/triggers/plugin/{subscription.endpoint_id}", data=b"body")
+
+    assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
+    assert response.data == b"Failed to get subscription by endpoint"
+    assert response.mimetype == "text/html"
+    assert not any(request.method == "POST" for request in daemon.requests)
+    assert all(not request.open_transactions for request in daemon.requests)
+    assert runtime.take_messages() == []
+    assert not storage.exists("triggers/")
+
+
+@pytest.mark.parametrize("failure", ["rejected", "unavailable", "empty", "malformed"])
+def test_plugin_dispatch_errors_keep_json_response_without_persisting_or_publishing(
+    harness: _Harness, daemon: _Daemon, runtime: _Runtime, caplog: pytest.LogCaptureFixture, failure: str
+) -> None:
+    subscription = _seed(harness, daemon, CredentialType.API_KEY)
+    diagnostic = "private-plugin-diagnostic-for-test"
+    daemon.response_data = None
+    expected_status = HTTPStatus.BAD_REQUEST
+    expected_body = {
+        "error": "Endpoint processing failed",
+        "message": "Invalid webhook request or plugin response.",
+    }
+    if failure == "rejected":
+        daemon.response_code = 1
+        daemon.response_message = f"signature rejected: {diagnostic}"
+    elif failure == "unavailable":
+        daemon.response_code = -500
+        daemon.response_message = json.dumps(
+            {"error_type": "PluginDaemonInternalServerError", "message": f"unavailable: {diagnostic}"}
+        )
+        expected_status = HTTPStatus.INTERNAL_SERVER_ERROR
+        expected_body = {"error": "Internal server error"}
+    elif failure == "malformed":
+        _dispatch(daemon, events=[])
+        assert daemon.response_data is not None
+        daemon.response_data["response"] = "not-hex"
+        daemon.response_data["payload"] = {"secret": diagnostic}
+
+    response = Context().run(harness.client.post, f"/triggers/plugin/{subscription.endpoint_id}", data=b"body")
+
+    assert response.status_code == expected_status
+    assert response.get_json() == expected_body
+    assert diagnostic not in response.get_data(as_text=True)
+    expected_detail = "got empty data from plugin daemon" if failure == "empty" else diagnostic
+    assert any(
+        record.name == "controllers.trigger.trigger"
+        and subscription.endpoint_id in record.getMessage()
+        and record.exc_info
+        and expected_detail in str(record.exc_info[1])
+        for record in caplog.records
+    )
+    assert runtime.take_messages() == []
+    assert runtime.publications == []
+    assert not storage.exists("triggers/")
+
+
+def test_plugin_removed_before_dispatch_returns_not_found_without_persisting_or_publishing(
+    harness: _Harness, daemon: _Daemon, runtime: _Runtime
+) -> None:
+    subscription = _seed(harness, daemon, CredentialType.API_KEY)
+    daemon.response_data = None
+    daemon.response_code = -500
+    daemon.response_message = json.dumps({"error_type": "PluginNotFoundError", "message": "plugin uninstalled"})
+
+    response = Context().run(harness.client.post, f"/triggers/plugin/{subscription.endpoint_id}", data=b"body")
+
+    assert response.status_code == HTTPStatus.NOT_FOUND
+    assert response.data == b"Trigger provider not found"
+    assert response.content_type == "text/html; charset=utf-8"
+    assert [request.method for request in daemon.requests] == ["GET", "POST"]
+    assert daemon.requests[-1].path == f"/plugin/{harness.tenant_id}/dispatch/trigger/dispatch_event"
+    assert all(not request.open_transactions for request in daemon.requests)
+    assert runtime.take_messages() == []
+    assert runtime.publications == []
+    assert not storage.exists("triggers/")

api/tests/e2e_tests/containers/trigger/test_trigger_e2e.py

@@ -8,13 +8,13 @@ from types import SimpleNamespace
 from typing import Any
 
 import pytest
-from flask import Flask, Response
+from flask import Flask
 from flask.testing import FlaskClient
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from configs import dify_config
-from core.plugin.entities.request import TriggerInvokeEventResponse
+from core.plugin.entities.plugin_daemon import CredentialType
 from core.trigger.constants import (
     TRIGGER_PLUGIN_NODE_TYPE,
     TRIGGER_SCHEDULE_NODE_TYPE,
@@ -33,7 +33,6 @@ from models.model import App
 from models.trigger import (
     AppTrigger,
     TriggerSubscription,
-    WorkflowPluginTrigger,
     WorkflowSchedulePlan,
     WorkflowTriggerLog,
     WorkflowWebhookTrigger,
@@ -47,7 +46,7 @@ from services.trigger.schedule_service import ScheduleService
 from services.workflow_service import WorkflowService
 from tasks import trigger_processing_tasks
 
-from .conftest import MockPluginSubscription, RecordingScheduleTask
+from .conftest import RecordingScheduleTask
 
 # Test constants
 WEBHOOK_ID_PRODUCTION = "wh1234567890123456789012"
@@ -311,65 +310,76 @@ def test_schedule_visual_debug_poll_generates_event(
     assert event.workflow_args["inputs"] == {}
 
 
-def test_plugin_trigger_dispatches_and_debug_events(
-    test_client_with_containers: FlaskClient,
-    mock_plugin_subscription: MockPluginSubscription,
-    monkeypatch: pytest.MonkeyPatch,
+def test_plugin_dispatch_delivers_each_debug_event_to_its_waiting_pool(
+    db_session_with_containers: Session,
+    tenant_and_account: tuple[Tenant, Account],
+    app_model: App,
 ) -> None:
-    """Plugin trigger endpoint should dispatch events and generate debug events."""
-    endpoint_id = "1cc7fa12-3f7b-4f6a-9c8d-1234567890ab"
-
-    debug_events: list[dict[str, Any]] = []
-    dispatched_payloads: list[dict[str, Any]] = []
-
-    def _fake_process_endpoint(_endpoint_id: str, _request: Any) -> Response:
-        dispatch_data = {
-            "user_id": "end-user",
-            "tenant_id": mock_plugin_subscription.tenant_id,
-            "endpoint_id": _endpoint_id,
-            "provider_id": mock_plugin_subscription.provider_id,
-            "subscription_id": mock_plugin_subscription.id,
-            "timestamp": int(time.time()),
-            "events": ["created", "updated"],
-            "request_id": f"req-{_endpoint_id}",
-        }
-        trigger_processing_tasks.dispatch_triggered_workflows_async.delay(dispatch_data)
-        return Response("ok", status=202)
-
-    monkeypatch.setattr(
-        "services.trigger.trigger_service.TriggerService.process_endpoint",
-        staticmethod(_fake_process_endpoint),
+    """The real task helper preserves dispatch identity through the Redis event bus."""
+    tenant, account = tenant_and_account
+    subscription = TriggerSubscription(
+        name="debug-dispatch",
+        tenant_id=tenant.id,
+        user_id=account.id,
+        provider_id="langgenius/test-provider/test-provider",
+        endpoint_id="1cc7fa12-3f7b-4f6a-9c8d-1234567890ab",
+        parameters={},
+        properties={},
+        credentials={},
+        credential_type=CredentialType.UNAUTHORIZED,
     )
-
-    monkeypatch.setattr(
-        trigger_processing_tasks.TriggerDebugEventBus,
-        "dispatch",
-        staticmethod(lambda **kwargs: debug_events.append(kwargs) or 1),
-    )
-
-    def _fake_delay(dispatch_data: dict[str, Any]) -> None:
-        dispatched_payloads.append(dispatch_data)
-        trigger_processing_tasks.dispatch_trigger_debug_event(
-            events=dispatch_data["events"],
-            user_id=dispatch_data["user_id"],
-            timestamp=dispatch_data["timestamp"],
-            request_id=dispatch_data["request_id"],
-            subscription=mock_plugin_subscription,
+    db_session_with_containers.add(subscription)
+    db_session_with_containers.commit()
+    event_names = ["created", "updated"]
+    pools = {
+        name: build_plugin_pool_key(
+            tenant_id=tenant.id,
+            provider_id=subscription.provider_id,
+            subscription_id=subscription.id,
+            name=name,
+        )
+        for name in event_names
+    }
+    for name, pool_key in pools.items():
+        assert (
+            TriggerDebugEventBus.poll(
+                event_type=PluginTriggerDebugEvent,
+                pool_key=pool_key,
+                tenant_id=tenant.id,
+                user_id=account.id,
+                app_id=app_model.id,
+                node_id=f"plugin-{name}",
+            )
+            is None
         )
 
-    monkeypatch.setattr(
-        trigger_processing_tasks.dispatch_triggered_workflows_async,
-        "delay",
-        staticmethod(_fake_delay),
+    timestamp = int(time.time())
+    dispatched = trigger_processing_tasks.dispatch_trigger_debug_event(
+        events=event_names,
+        user_id="external-user",
+        timestamp=timestamp,
+        request_id="request-with-two-events",
+        subscription=subscription,
     )
 
-    response = test_client_with_containers.post(f"/triggers/plugin/{endpoint_id}", json={"hello": "world"})
-
-    assert response.status_code == 202
-    assert dispatched_payloads, "Plugin trigger should enqueue workflow dispatch payload"
-    assert debug_events, "Plugin trigger should dispatch debug events"
-    dispatched_event_names = {event["event"].name for event in debug_events}
-    assert dispatched_event_names == {"created", "updated"}
+    assert dispatched == 2
+    for name, pool_key in pools.items():
+        event = TriggerDebugEventBus.poll(
+            event_type=PluginTriggerDebugEvent,
+            pool_key=pool_key,
+            tenant_id=tenant.id,
+            user_id=account.id,
+            app_id=app_model.id,
+            node_id=f"plugin-{name}",
+        )
+        assert event == PluginTriggerDebugEvent(
+            timestamp=timestamp,
+            user_id="external-user",
+            name=name,
+            request_id="request-with-two-events",
+            subscription_id=subscription.id,
+            provider_id=subscription.provider_id,
+        )
 
 
 def test_webhook_debug_dispatches_event(
@@ -668,257 +678,3 @@ def test_schedule_visual_cron_conversion(
     assert config.cron_expression == expected_cron, f"Expected {expected_cron}, got {config.cron_expression}"
     assert config.timezone == "UTC"
     assert config.node_id == "schedule-node"
-
-
-def test_plugin_trigger_full_chain_with_db_verification(
-    test_client_with_containers: FlaskClient,
-    db_session_with_containers: Session,
-    tenant_and_account: tuple[Tenant, Account],
-    app_model: App,
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    """Plugin trigger should create WorkflowTriggerLog and WorkflowPluginTrigger records."""
-
-    tenant, account = tenant_and_account
-
-    # Create published workflow with plugin trigger node
-    plugin_node_id = "plugin-trigger-node"
-    provider_id = "langgenius/test-provider/test-provider"
-    subscription_id = "sub-plugin-test"
-    endpoint_id = "2cc7fa12-3f7b-4f6a-9c8d-1234567890ab"
-
-    graph = {
-        "nodes": [
-            {
-                "id": plugin_node_id,
-                "data": {
-                    "type": TRIGGER_PLUGIN_NODE_TYPE,
-                    "title": "plugin",
-                    "plugin_id": "test-plugin",
-                    "plugin_unique_identifier": "test-plugin",
-                    "provider_id": provider_id,
-                    "event_name": "test_event",
-                    "subscription_id": subscription_id,
-                    "parameters": {},
-                },
-            },
-            {"id": "answer-1", "data": {"type": BuiltinNodeTypes.ANSWER, "title": "answer"}},
-        ],
-        "edges": [{"source": plugin_node_id, "target": "answer-1", "sourceHandle": "success"}],
-    }
-    published_workflow = Workflow.new(
-        tenant_id=tenant.id,
-        app_id=app_model.id,
-        type="workflow",
-        version=Workflow.version_from_datetime(naive_utc_now()),
-        graph=json.dumps(graph),
-        features=json.dumps({}),
-        created_by=account.id,
-        environment_variables=[],
-        conversation_variables=[],
-        rag_pipeline_variables=[],
-    )
-    db_session_with_containers.add(published_workflow)
-    app_model.workflow_id = published_workflow.id
-    db_session_with_containers.commit()
-
-    # Create trigger subscription
-    subscription = TriggerSubscription(
-        name="test-subscription",
-        tenant_id=tenant.id,
-        user_id=account.id,
-        provider_id=provider_id,
-        endpoint_id=endpoint_id,
-        parameters={},
-        properties={},
-        credentials={"token": "test-secret"},
-        credential_type="api-key",
-    )
-    db_session_with_containers.add(subscription)
-    db_session_with_containers.commit()
-
-    # Update subscription_id to match the created subscription
-    graph["nodes"][0]["data"]["subscription_id"] = subscription.id
-    published_workflow.graph = json.dumps(graph)
-    db_session_with_containers.commit()
-
-    # Create WorkflowPluginTrigger
-    plugin_trigger = WorkflowPluginTrigger(
-        app_id=app_model.id,
-        tenant_id=tenant.id,
-        node_id=plugin_node_id,
-        provider_id=provider_id,
-        event_name="test_event",
-        subscription_id=subscription.id,
-    )
-    app_trigger = AppTrigger(
-        tenant_id=tenant.id,
-        app_id=app_model.id,
-        node_id=plugin_node_id,
-        trigger_type=AppTriggerType.TRIGGER_PLUGIN,
-        status=AppTriggerStatus.ENABLED,
-        title="plugin",
-    )
-    db_session_with_containers.add_all([plugin_trigger, app_trigger])
-    db_session_with_containers.commit()
-
-    # Track dispatched data
-    dispatched_data: list[dict[str, Any]] = []
-
-    def _fake_process_endpoint(_endpoint_id: str, _request: Any) -> Response:
-        dispatch_data = {
-            "user_id": "end-user",
-            "tenant_id": tenant.id,
-            "endpoint_id": _endpoint_id,
-            "provider_id": provider_id,
-            "subscription_id": subscription.id,
-            "timestamp": int(time.time()),
-            "events": ["test_event"],
-            "request_id": f"req-{_endpoint_id}",
-        }
-        dispatched_data.append(dispatch_data)
-        return Response("ok", status=202)
-
-    monkeypatch.setattr(
-        "services.trigger.trigger_service.TriggerService.process_endpoint",
-        staticmethod(_fake_process_endpoint),
-    )
-
-    response = test_client_with_containers.post(f"/triggers/plugin/{endpoint_id}", json={"test": "data"})
-
-    assert response.status_code == 202
-    assert dispatched_data, "Plugin trigger should dispatch event data"
-    assert dispatched_data[0]["subscription_id"] == subscription.id
-    assert dispatched_data[0]["events"] == ["test_event"]
-
-    # Verify database records exist
-    db_session_with_containers.expire_all()
-    plugin_triggers = db_session_with_containers.scalars(
-        select(WorkflowPluginTrigger).where(
-            WorkflowPluginTrigger.app_id == app_model.id,
-            WorkflowPluginTrigger.node_id == plugin_node_id,
-        )
-    ).all()
-    assert plugin_triggers, "WorkflowPluginTrigger record should exist"
-    assert plugin_triggers[0].provider_id == provider_id
-    assert plugin_triggers[0].event_name == "test_event"
-
-
-def test_plugin_debug_via_http_endpoint(
-    test_client_with_containers: FlaskClient,
-    db_session_with_containers: Session,
-    tenant_and_account: tuple[Tenant, Account],
-    app_model: App,
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    """Plugin single-step debug via HTTP endpoint should dispatch debug event and be pollable."""
-
-    tenant, account = tenant_and_account
-
-    provider_id = "langgenius/debug-provider/debug-provider"
-    endpoint_id = "3cc7fa12-3f7b-4f6a-9c8d-1234567890ab"
-    event_name = "debug_event"
-
-    # Create subscription
-    subscription = TriggerSubscription(
-        name="debug-subscription",
-        tenant_id=tenant.id,
-        user_id=account.id,
-        provider_id=provider_id,
-        endpoint_id=endpoint_id,
-        parameters={},
-        properties={},
-        credentials={"token": "debug-secret"},
-        credential_type="api-key",
-    )
-    db_session_with_containers.add(subscription)
-    db_session_with_containers.commit()
-
-    # Create plugin trigger node config
-    node_id = "plugin-debug-node"
-    node_config = {
-        "id": node_id,
-        "data": {
-            "type": TRIGGER_PLUGIN_NODE_TYPE,
-            "title": "plugin-debug",
-            "plugin_id": "debug-plugin",
-            "plugin_unique_identifier": "debug-plugin",
-            "provider_id": provider_id,
-            "event_name": event_name,
-            "subscription_id": subscription.id,
-            "parameters": {},
-        },
-    }
-
-    # Start listening with poller
-
-    poller = PluginTriggerDebugEventPoller(
-        tenant_id=tenant.id,
-        user_id=account.id,
-        app_id=app_model.id,
-        node_config=node_config,
-        node_id=node_id,
-    )
-    assert poller.poll() is None, "First poll should return None (waiting)"
-
-    # Track debug events dispatched
-    debug_events: list[dict[str, Any]] = []
-    original_dispatch = TriggerDebugEventBus.dispatch
-
-    def _tracking_dispatch(**kwargs: Any) -> int:
-        debug_events.append(kwargs)
-        return original_dispatch(**kwargs)
-
-    monkeypatch.setattr(TriggerDebugEventBus, "dispatch", staticmethod(_tracking_dispatch))
-
-    # Mock process_endpoint to trigger debug event dispatch
-    def _fake_process_endpoint(_endpoint_id: str, _request: Any) -> Response:
-        # Simulate what happens inside process_endpoint + dispatch_triggered_workflows_async
-        pool_key = build_plugin_pool_key(
-            tenant_id=tenant.id,
-            provider_id=provider_id,
-            subscription_id=subscription.id,
-            name=event_name,
-        )
-        TriggerDebugEventBus.dispatch(
-            tenant_id=tenant.id,
-            event=PluginTriggerDebugEvent(
-                timestamp=int(time.time()),
-                user_id="end-user",
-                name=event_name,
-                request_id=f"req-{_endpoint_id}",
-                subscription_id=subscription.id,
-                provider_id=provider_id,
-            ),
-            pool_key=pool_key,
-        )
-        return Response("ok", status=202)
-
-    monkeypatch.setattr(
-        "services.trigger.trigger_service.TriggerService.process_endpoint",
-        staticmethod(_fake_process_endpoint),
-    )
-
-    # Call HTTP endpoint
-    response = test_client_with_containers.post(f"/triggers/plugin/{endpoint_id}", json={"debug": "payload"})
-
-    assert response.status_code == 202
-    assert debug_events, "Debug event should be dispatched via HTTP endpoint"
-    assert debug_events[0]["event"].name == event_name
-
-    # Mock invoke_trigger_event for poller
-
-    monkeypatch.setattr(
-        "services.trigger.trigger_service.TriggerService.invoke_trigger_event",
-        staticmethod(
-            lambda **_kwargs: TriggerInvokeEventResponse(
-                variables={"http_debug": "success"},
-                cancelled=False,
-            )
-        ),
-    )
-
-    # Second poll should receive the event
-    event = poller.poll()
-    assert event is not None, "Poller should receive debug event after HTTP trigger"
-    assert event.workflow_args["inputs"]["http_debug"] == "success"

api/tests/unit_tests/controllers/console/workspace/test_trigger_subscription_builder.py

@@ -54,6 +54,8 @@ class _Daemon:
     observed_lock_key: str | None = None
     response_code: int = 0
     response_message: str = ""
+    get_response_code: int = 0
+    get_response_message: str = ""
     status: HTTPStatus = HTTPStatus.OK
     post_status: HTTPStatus = HTTPStatus.OK
     requests: list[_DaemonRequest] = field(default_factory=list)
@@ -89,8 +91,10 @@ def daemon(harness: _Harness) -> Iterator[_Daemon]:
             )
             body = json.dumps(
                 {
-                    "code": daemon.response_codes_by_path.get(self.path, daemon.response_code) if streamed else 0,
-                    "message": daemon.response_message if streamed else "",
+                    "code": daemon.response_codes_by_path.get(self.path, daemon.response_code)
+                    if streamed
+                    else daemon.get_response_code,
+                    "message": daemon.response_message if streamed else daemon.get_response_message,
                     "data": data,
                 }
             )

api/tests/unit_tests/controllers/inner_api/workspace/test_workspace.py

@@ -33,6 +33,8 @@ from services.errors.workspace import (
     WorkspaceOwnerNotFoundError,
     WorkspacesLimitExceededError,
 )
+from services.trigger.endpoint_gateway import PluginTriggerEndpointGateway, enqueue_triggered_workflows
+from services.trigger.endpoint_service import TriggerEndpointService
 from services.trigger.oauth_client_service import TriggerOAuthClientService
 from services.trigger.oauth_gateway import (
     OAuthProxyContextGateway,
@@ -45,6 +47,7 @@ from services.trigger.subscription_codec import ProviderTriggerSubscriptionCodec
 from services.trigger.subscription_gateway import PluginTriggerSubscriptionGateway
 from services.trigger.subscription_query_service import TriggerSubscriptionQueryService
 from services.trigger.subscription_service import TriggerSubscriptionService
+from services.trigger.trigger_request_service import TriggerHttpRequestCachingService
 from services.trigger.trigger_subscription_builder_service import TriggerSubscriptionBuilderService
 from services.workspace.contracts import CreatedWorkspace, WorkspaceMembership
 from services.workspace.provisioning_service import WorkspaceProvisioningService
@@ -151,6 +154,12 @@ class TestWorkspaceCases:
                 schemas=PluginTriggerProviderQuery(),
             ),
             trigger_builders=builders,
+            trigger_endpoints=TriggerEndpointService(
+                subscriptions=TriggerSubscriptionRepository(session_factory=sqlite_session_factory),
+                provider=PluginTriggerEndpointGateway(),
+                requests=TriggerHttpRequestCachingService,
+                publish=enqueue_triggered_workflows,
+            ),
             trigger_subscription_management=subscription_management,
             trigger_oauth_clients=oauth_clients,
             trigger_oauth=TriggerOAuthService(

api/tests/unit_tests/controllers/trigger/test_trigger.py

@@ -1,11 +1,30 @@
-"""Invalid trigger capabilities fail through the real Flask route before dependencies are accessed."""
+"""Trigger capability failures exercise the real HTTP, SQL and daemon boundaries."""
 
+import json
+from contextvars import Context
 from http import HTTPStatus
+from uuid import uuid4
 
 import pytest
 from flask import Flask
+from sqlalchemy import Engine
 
 from controllers.trigger.trigger import trigger_endpoint
+from models.trigger import TriggerSubscription
+from tests.unit_tests.controllers.console.workspace import test_trigger_oauth_client, test_trigger_subscription_builder
+from tests.unit_tests.controllers.console.workspace.test_trigger_oauth_client import _Harness
+from tests.unit_tests.controllers.console.workspace.test_trigger_subscription_builder import _Daemon
+from tests.unit_tests.controllers.console.workspace.test_trigger_subscription_query import _add_subscription
+
+application_redis = test_trigger_oauth_client.application_redis
+harness = test_trigger_oauth_client.harness
+daemon = test_trigger_subscription_builder.daemon
+
+
+@pytest.fixture
+def endpoint_harness(harness: _Harness) -> _Harness:
+    harness.app.add_url_rule("/triggers/plugin/<string:endpoint_id>", view_func=trigger_endpoint, methods=["POST"])
+    return harness
 
 
 @pytest.mark.parametrize("endpoint_id", ["not-a-uuid", "123e4567-e89b-12d3-a456-426614174000"])
@@ -19,3 +38,78 @@ def test_invalid_endpoint_preserves_json_error_contract(endpoint_id: str) -> Non
     assert response.data == b'{"error":"Endpoint not found"}'
     assert response.mimetype == "application/json"
     assert response.headers["Cache-Control"] == "no-store"
+
+
+def test_subscription_query_failure_preserves_text_error_without_plugin_lookup(
+    endpoint_harness: _Harness, daemon: _Daemon, sqlite_engine: Engine
+) -> None:
+    TriggerSubscription.__table__.drop(sqlite_engine)
+
+    response = endpoint_harness.client.post(f"/triggers/plugin/{uuid4()}", json={})
+
+    assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
+    assert response.data == b"Failed to get subscription by endpoint"
+    assert response.content_type == "text/html; charset=utf-8"
+    assert daemon.requests == []
+
+
+def test_provider_load_failure_preserves_text_error_after_subscription_transaction_closes(
+    endpoint_harness: _Harness, daemon: _Daemon
+) -> None:
+    subscription = _add_subscription(endpoint_harness)
+    daemon.status = HTTPStatus.SERVICE_UNAVAILABLE
+
+    response = Context().run(endpoint_harness.client.post, f"/triggers/plugin/{subscription.endpoint_id}", json={})
+
+    assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
+    assert response.data == b"Failed to get subscription by endpoint"
+    assert response.content_type == "text/html; charset=utf-8"
+    assert len(daemon.requests) == 1
+    assert daemon.requests[0].method == "GET"
+    assert not daemon.requests[0].open_transactions
+
+
+@pytest.mark.parametrize(
+    ("error_type", "expected_status", "expected_body"),
+    [
+        pytest.param("PluginNotFoundError", HTTPStatus.NOT_FOUND, b"Trigger provider not found", id="plugin-not-found"),
+        pytest.param(
+            "PluginDaemonInternalServerError",
+            HTTPStatus.INTERNAL_SERVER_ERROR,
+            b"Failed to get subscription by endpoint",
+            id="daemon-failure",
+        ),
+    ],
+)
+def test_provider_metadata_daemon_errors_reach_http_mapping_after_subscription_transaction_closes(
+    endpoint_harness: _Harness,
+    daemon: _Daemon,
+    error_type: str,
+    expected_status: HTTPStatus,
+    expected_body: bytes,
+) -> None:
+    subscription = _add_subscription(endpoint_harness)
+    daemon.get_response_code = -500
+    daemon.get_response_message = json.dumps({"error_type": error_type, "message": "provider unavailable"})
+
+    response = Context().run(endpoint_harness.client.post, f"/triggers/plugin/{subscription.endpoint_id}", json={})
+
+    assert response.status_code == expected_status
+    assert response.data == expected_body
+    assert response.content_type == "text/html; charset=utf-8"
+    assert len(daemon.requests) == 1
+    assert daemon.requests[0].method == "GET"
+    assert not daemon.requests[0].open_transactions
+
+
+def test_invalid_persisted_provider_is_load_failure_not_missing_plugin(
+    endpoint_harness: _Harness, daemon: _Daemon
+) -> None:
+    subscription = _add_subscription(endpoint_harness, provider_id="not/a/valid/provider")
+
+    response = Context().run(endpoint_harness.client.post, f"/triggers/plugin/{subscription.endpoint_id}", json={})
+
+    assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
+    assert response.data == b"Failed to get subscription by endpoint"
+    assert response.content_type == "text/html; charset=utf-8"
+    assert daemon.requests == []

api/tests/unit_tests/extensions/test_ext_application_services.py

@@ -1128,6 +1128,10 @@ def test_build_application_services_wires_credential_query(
         services.workspaces.trigger_builders._create_subscription
         == services.workspaces.trigger_subscription_management.create
     )
+    assert (
+        services.workspaces.trigger_endpoints._subscriptions
+        is services.workspaces.trigger_subscription_management._subscriptions
+    )
 
 
 def test_build_application_services_uses_supplied_redis_for_both_workflow_stop_signals(

api/tests/unit_tests/repositories/trigger/test_subscription_repository.py

@@ -149,6 +149,49 @@ def test_get_requires_tenant_and_subscription_id_with_valid_decoys(
     assert repository.get(tenant_id=stored.tenant_id, subscription_id=stored.id) == stored
 
 
+def test_get_by_endpoint_returns_detached_encrypted_snapshot_and_releases_connection(
+    repository: TriggerSubscriptionRepository,
+    sqlite_session_factory: sessionmaker[Session],
+    sqlite_engine: Engine,
+) -> None:
+    expected = _snapshot()
+    decoy = _snapshot()
+    _persist(sqlite_session_factory, decoy, expected)
+    pool = sqlite_engine.pool
+    assert isinstance(pool, QueuePool)
+
+    snapshot = repository.get_by_endpoint(endpoint_id=expected.endpoint_id)
+
+    assert pool.checkedout() == 0
+    assert snapshot is not None
+    assert snapshot == expected
+    assert snapshot.tenant_id != decoy.tenant_id
+    assert snapshot.user_id != decoy.user_id
+    assert snapshot.credentials == {"token": "encrypted-token"}
+    snapshot.parameters["events"].append("issues")
+
+    # The detached result cannot mutate either persisted subscription.
+    assert repository.get(tenant_id=expected.tenant_id, subscription_id=expected.id) == expected
+    assert repository.get(tenant_id=decoy.tenant_id, subscription_id=decoy.id) == decoy
+    assert pool.checkedout() == 0
+
+
+def test_get_by_endpoint_does_not_fall_back_to_another_valid_subscription(
+    repository: TriggerSubscriptionRepository,
+    sqlite_session_factory: sessionmaker[Session],
+    sqlite_engine: Engine,
+) -> None:
+    decoy = _snapshot()
+    _persist(sqlite_session_factory, decoy)
+
+    assert repository.get_by_endpoint(endpoint_id=str(uuid4())) is None
+
+    pool = sqlite_engine.pool
+    assert isinstance(pool, QueuePool)
+    assert pool.checkedout() == 0
+    assert repository.get_by_endpoint(endpoint_id=decoy.endpoint_id) == decoy
+
+
 def test_create_enforces_provider_limit_without_counting_other_tenants_or_providers(
     repository: TriggerSubscriptionRepository,
     sqlite_session_factory: sessionmaker[Session],

api/tests/unit_tests/services/trigger/test_endpoint_gateway.py

@@ -0,0 +1,45 @@
+"""The endpoint publisher preserves the worker's real Celery contract."""
+
+from celery import Celery
+from kombu.simple import SimpleQueue
+
+from services.trigger.endpoint_gateway import enqueue_triggered_workflows
+from services.workflow.entities import PluginTriggerDispatchData
+from tasks.trigger_processing_tasks import TRIGGER_QUEUE, dispatch_triggered_workflows_async
+
+
+def test_enqueue_triggered_workflows_publishes_worker_payload_to_existing_queue() -> None:
+    data = PluginTriggerDispatchData(
+        user_id="plugin-user",
+        tenant_id="tenant-id",
+        endpoint_id="endpoint-id",
+        provider_id="langgenius/github/github",
+        subscription_id="subscription-id",
+        timestamp=1234567890,
+        events=["push", "push"],
+        request_id="trigger_request_1234567890_abcdef012345",
+    )
+    previous_app = dispatch_triggered_workflows_async.app
+    celery_app = Celery("trigger-endpoint-publisher-test", broker="memory://", set_as_current=False)
+    celery_app.conf.update(task_always_eager=False, task_serializer="json")
+    celery_app.set_current()
+    try:
+        with celery_app.connection_for_write() as connection:
+            with SimpleQueue(connection.channel(), TRIGGER_QUEUE) as queue:
+                queue.clear()
+                try:
+                    enqueue_triggered_workflows(data)
+
+                    message = queue.get(block=False)
+                    assert message.headers["task"] == dispatch_triggered_workflows_async.name
+                    assert message.delivery_info["routing_key"] == TRIGGER_QUEUE
+                    assert message.content_type == "application/json"
+                    assert message.payload[0] == [data.model_dump(mode="json")]
+                    assert message.payload[1] == {}
+                    message.ack()
+                    assert queue.qsize() == 0
+                finally:
+                    queue.clear()
+    finally:
+        previous_app.set_current()
+        celery_app.close()

api/tests/unit_tests/services/trigger/test_trigger_provider_service.py

@@ -300,16 +300,3 @@ def test_unverified_plugin_cannot_read_system_oauth(
     mocker.patch.object(service_module.PluginService, "is_plugin_verified", return_value=False)
 
     assert TriggerProviderService.get_oauth_client(trigger_db.tenant_id, trigger_db.provider_id) is None
-
-
-def test_endpoint_lookup_decrypts_persisted_subscription(
-    mocker, trigger_db: TriggerDatabase, provider_controller: Mock
-) -> None:
-    subscription = trigger_db.add_subscription(endpoint_id="lookup-endpoint")
-    _patch_provider(mocker, provider_controller)
-    _patch_identity_encryption(mocker)
-
-    assert TriggerProviderService.get_subscription_by_endpoint("missing") is None
-    found = TriggerProviderService.get_subscription_by_endpoint("lookup-endpoint")
-    assert found is not None
-    assert found.id == subscription.id