refactor(api): inject trigger workflow dispatch and execution logging

api/.importlinter

@@ -1224,6 +1224,8 @@ source_modules =
     services.trigger.subscription_renewal_service
     services.trigger.subscription_refresh_service
     services.trigger.subscription_binding_service
+    services.async_workflow_service
+    services.trigger.workflow_dispatch_service
 forbidden_modules =
     controllers
     extensions

api/controllers/trigger/webhook.py

@@ -78,6 +78,7 @@ def handle_webhook(webhook_id: str):
             workflow,
             end_users=application_services().app_scoped_end_users.commands,
             trigger_status=application_services().apps.trigger_status,
+            workflow_dispatch=application_services().workspaces.workflow_dispatch,
         )
 
         # Return configured response

api/extensions/application_services/workspace.py

@@ -14,6 +14,7 @@
 from repositories.app.console_repository import ConsoleAppRepository
 from repositories.credentials.query_repository import CredentialQueryRepository
 from repositories.knowledge.dataset_repository import SQLAlchemyDatasetRepository
+from repositories.sqlalchemy_workflow_trigger_log_repository import SQLAlchemyWorkflowTriggerDispatchRepository
 from repositories.trigger.oauth_client_repository import TriggerOAuthClientRepository
 from repositories.trigger.subscription_builder_repository import RedisTriggerSubscriptionBuilderRepository
 from repositories.trigger.subscription_repository import (
@@ -24,9 +25,11 @@
 from repositories.workspace.resource_cleanup_repository import WorkspaceResourceCleanupRepository
 from repositories.workspace.workspace_repository import WorkspaceRepository
 from services.account.adapters import RedisInvitationTokenStore
+from services.async_workflow_service import AsyncWorkflowService, TriggerWorkflowQuery
 from services.file_service import FileService
 from services.plugin.plugin_parameter_gateway import DaemonPluginParameterGateway
 from services.plugin.plugin_parameter_service import PluginParameterService
+from services.quota_service import QuotaService
 from services.trigger.endpoint_gateway import PluginTriggerEndpointGateway, enqueue_triggered_workflows
 from services.trigger.endpoint_service import TriggerEndpointService
 from services.trigger.event_gateway import PluginTriggerEventGateway
@@ -55,6 +58,8 @@
 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.trigger.workflow_dispatch_service import PluginTriggerWorkflowDispatchService, TriggerEndUserProvisioner
+from services.workflow.queue_dispatcher import CeleryWorkflowQueuePublisher
 from services.workspace.gateways import (
     DeploymentWorkspaceCreationPolicy,
     DeploymentWorkspaceFeatureGateway,
@@ -101,6 +106,8 @@ class WorkspaceServices:
     trigger_builders: TriggerSubscriptionBuilderService
     trigger_endpoints: TriggerEndpointService
     trigger_events: TriggerEventService
+    workflow_dispatch: AsyncWorkflowService
+    trigger_workflow_dispatch: PluginTriggerWorkflowDispatchService
     trigger_oauth_clients: TriggerOAuthClientService
     trigger_oauth: TriggerOAuthService
 
@@ -139,6 +146,9 @@ def build_workspace_services(
     registration: InvitedAccountRegistration,
     invitation_tokens: RedisInvitationTokenStore,
     credentials: CredentialQueryRepository,
+    workflow_queries: TriggerWorkflowQuery,
+    trigger_end_users: TriggerEndUserProvisioner,
+    mark_rate_limited: Callable[[str], None],
 ) -> WorkspaceServices:
     trigger_providers = PluginTriggerProviderQuery()
     subscriptions = TriggerSubscriptionRepository(session_factory=database_client)
@@ -171,6 +181,16 @@ def build_workspace_services(
         invalidate_cache=delete_cache_for_subscription,
         invalidate_properties=invalidate_subscription_properties,
     )
+    trigger_events = TriggerEventService(
+        subscriptions=trigger_subscriptions,
+        plugins=PluginTriggerEventGateway(),
+        requests=TriggerHttpRequestCachingService,
+    )
+    execution_logs = SQLAlchemyWorkflowTriggerDispatchRepository(session_factory=database_client)
+    queues = CeleryWorkflowQueuePublisher()
+    workflow_dispatch = AsyncWorkflowService(
+        workflows=workflow_queries, logs=execution_logs, queues=queues, reserve_quota=QuotaService.reserve
+    )
     return WorkspaceServices(
         queries=WorkspaceQueryService(workspaces=workspaces, plans=DeploymentWorkspacePlanGateway()),
         management=WorkspaceService(
@@ -216,10 +236,18 @@ def build_workspace_services(
         create_trigger_binding_service=lambda session: TriggerSubscriptionBindingService(
             sync_bindings=partial(subscriptions.sync_workflow_bindings, session=session)
         ),
-        trigger_events=TriggerEventService(
-            subscriptions=trigger_subscriptions,
-            plugins=PluginTriggerEventGateway(),
-            requests=TriggerHttpRequestCachingService,
+        trigger_events=trigger_events,
+        workflow_dispatch=workflow_dispatch,
+        trigger_workflow_dispatch=PluginTriggerWorkflowDispatchService(
+            workflows=workflow_queries,
+            logs=execution_logs,
+            queues=queues,
+            dispatch=workflow_dispatch,
+            events=trigger_events,
+            bindings=subscriptions,
+            end_users=trigger_end_users,
+            reserve_quota=QuotaService.reserve,
+            mark_rate_limited=mark_rate_limited,
         ),
         trigger_subscription_management=trigger_subscription_management,
         trigger_subscription_renewal=renewal,

api/extensions/ext_application_services.py

@@ -187,6 +187,7 @@
 from services.system_feature_service import SystemFeatureService
 from services.tag_application_service import TagApplicationService
 from services.tool_file_download_service import ToolFileDownloadService
+from services.trigger.workflow_dispatch_gateway import AppScopedTriggerEndUsers
 from services.upload_file_delivery_service import UploadFileDeliveryService
 from services.web_app_runtime_query_service import WebAppRuntimeQueryService
 from services.web_authentication_adapters import (
@@ -523,6 +524,11 @@ def build_application_services(
         registration=account_services.lifecycle,
         invitation_tokens=invitation_tokens,
         credentials=credential_queries,
+        workflow_queries=app_definition_repository,
+        mark_rate_limited=apps.trigger_status.mark_tenant_triggers_rate_limited,
+        trigger_end_users=AppScopedTriggerEndUsers(
+            end_users=AppScopedEndUserService(end_users=app_scoped_end_user_repository)
+        ),
     )
     return ApplicationServices(
         rbac=rbac,

api/repositories/app_definition_query_repository.py

@@ -1,5 +1,6 @@
 """Database repository for externally visible app definitions."""
 
+from collections.abc import Mapping, Sequence
 from typing import Any, cast, override
 
 from sqlalchemy import select
@@ -22,7 +23,9 @@
     AppSiteConfiguration,
     AppToolIconSource,
 )
+from services.errors.app import IsDraftWorkflowError, WorkflowNotFoundError
 from services.web_app_runtime_query_service import WebAppRuntimeRecord
+from services.workflow.entities import TriggerWorkflowSnapshot
 
 
 def map_site_configuration(site: Site) -> AppSiteConfiguration:
@@ -118,6 +121,62 @@ class AppDefinitionQueryRepository(AppDefinitionQuery):
     def __init__(self, *, session_factory: sessionmaker[Session]) -> None:
         self._session_factory = session_factory
 
+    def get_trigger_workflow(self, *, tenant_id: str, app_id: str, workflow_id: str | None) -> TriggerWorkflowSnapshot:
+        with self._session_factory() as session:
+            app = session.scalar(select(App).where(App.id == app_id, App.tenant_id == tenant_id))
+            if app is None:
+                raise WorkflowNotFoundError(f"App not found: {app_id}")
+            selected_id = workflow_id or app.workflow_id
+            workflow = (
+                session.scalar(
+                    select(Workflow).where(
+                        Workflow.id == selected_id, Workflow.app_id == app_id, Workflow.tenant_id == tenant_id
+                    )
+                )
+                if selected_id
+                else None
+            )
+            if workflow is None:
+                if workflow_id:
+                    raise WorkflowNotFoundError(f"Published workflow not found: {workflow_id}")
+                raise WorkflowNotFoundError(f"No published workflow found for app: {app_id}")
+            if workflow_id and workflow.version == Workflow.VERSION_DRAFT:
+                raise IsDraftWorkflowError(
+                    f"Cannot use draft workflow version. Workflow ID: {workflow_id}. "
+                    "Please use a published workflow version or leave workflow_id empty."
+                )
+            return self._trigger_workflow_snapshot(workflow)
+
+    def get_published_trigger_workflows(
+        self, *, tenant_id: str, app_ids: Sequence[str]
+    ) -> Mapping[str, TriggerWorkflowSnapshot]:
+        if not app_ids:
+            return {}
+        with self._session_factory() as session:
+            workflows = session.scalars(
+                select(Workflow)
+                .join(App, App.workflow_id == Workflow.id)
+                .where(
+                    App.id.in_(app_ids),
+                    App.tenant_id == tenant_id,
+                    Workflow.app_id == App.id,
+                    Workflow.tenant_id == App.tenant_id,
+                    Workflow.version != Workflow.VERSION_DRAFT,
+                )
+            ).all()
+            return {workflow.app_id: self._trigger_workflow_snapshot(workflow) for workflow in workflows}
+
+    @staticmethod
+    def _trigger_workflow_snapshot(workflow: Workflow) -> TriggerWorkflowSnapshot:
+        return TriggerWorkflowSnapshot(
+            id=workflow.id,
+            tenant_id=workflow.tenant_id,
+            app_id=workflow.app_id,
+            type=workflow.type,
+            version=workflow.version,
+            graph=workflow.graph,
+        )
+
     @override
     def get_mode(self, app_id: str) -> str | None:
         with self._session_factory() as session:

api/repositories/sqlalchemy_workflow_trigger_log_repository.py

@@ -2,17 +2,21 @@
 SQLAlchemy implementation of WorkflowTriggerLogRepository.
 """
 
-from collections.abc import Sequence
+import json
+from collections.abc import Mapping, Sequence
 from datetime import UTC, datetime, timedelta
 from typing import cast, override
 
-from sqlalchemy import and_, delete, func, select
+from sqlalchemy import and_, delete, func, select, update
 from sqlalchemy.engine import CursorResult
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import Session, sessionmaker
 
-from models.enums import WorkflowTriggerStatus
+from graphon.enums import WorkflowExecutionStatus
+from models.enums import WorkflowRunTriggeredFrom, WorkflowTriggerStatus
 from models.trigger import WorkflowTriggerLog
+from models.workflow import WorkflowAppLog, WorkflowAppLogCreatedFrom, WorkflowRun
 from repositories.workflow_trigger_log_repository import WorkflowTriggerLogRepository
+from services.workflow.entities import TriggerActor, TriggerData, TriggerWorkflowSnapshot
 
 
 class SQLAlchemyWorkflowTriggerLogRepository(WorkflowTriggerLogRepository):
@@ -142,3 +146,127 @@ def count_by_run_ids(self, run_ids: Sequence[str]) -> int:
             select(func.count()).select_from(WorkflowTriggerLog).where(WorkflowTriggerLog.workflow_run_id.in_(run_ids))
         )
         return int(count or 0)
+
+
+class SQLAlchemyWorkflowTriggerDispatchRepository:
+    """Transaction-owning trigger log writes for producers.
+
+    Runtime and retention code still borrow a caller session through the repository
+    above. These producer operations close their sessions before broker or billing I/O.
+    """
+
+    def __init__(self, *, session_factory: sessionmaker[Session]) -> None:
+        self._session_factory = session_factory
+
+    def create_pending(self, *, actor: TriggerActor, trigger_data: TriggerData, workflow_id: str, queue: str) -> str:
+        log = self._new_log(actor=actor, trigger_data=trigger_data, workflow_id=workflow_id, queue=queue)
+        with self._session_factory.begin() as session:
+            SQLAlchemyWorkflowTriggerLogRepository(session).create(log)
+            log_id = log.id
+        return log_id
+
+    def mark_failed(
+        self, *, tenant_id: str, app_id: str, log_id: str, status: WorkflowTriggerStatus, error: str
+    ) -> None:
+        with self._session_factory.begin() as session:
+            session.execute(
+                update(WorkflowTriggerLog)
+                .where(
+                    WorkflowTriggerLog.id == log_id,
+                    WorkflowTriggerLog.tenant_id == tenant_id,
+                    WorkflowTriggerLog.app_id == app_id,
+                )
+                .values(status=status, error=error)
+            )
+
+    def mark_queued(self, *, tenant_id: str, app_id: str, log_id: str, task_id: str) -> None:
+        with self._session_factory.begin() as session:
+            session.execute(
+                update(WorkflowTriggerLog)
+                .where(
+                    WorkflowTriggerLog.id == log_id,
+                    WorkflowTriggerLog.tenant_id == tenant_id,
+                    WorkflowTriggerLog.app_id == app_id,
+                )
+                .values(status=WorkflowTriggerStatus.QUEUED, celery_task_id=task_id, triggered_at=datetime.now(UTC))
+            )
+
+    def record_invocation_failure(
+        self,
+        *,
+        workflow: TriggerWorkflowSnapshot,
+        actor: TriggerActor,
+        trigger_data: TriggerData,
+        failure_inputs: Mapping[str, str],
+        error: str,
+        queue: str,
+    ) -> None:
+        now = datetime.now(UTC)
+        run = WorkflowRun(
+            tenant_id=workflow.tenant_id,
+            app_id=workflow.app_id,
+            workflow_id=workflow.id,
+            type=workflow.type,
+            triggered_from=WorkflowRunTriggeredFrom.PLUGIN.value,
+            version=workflow.version,
+            graph=workflow.graph,
+            inputs=json.dumps(dict(failure_inputs)),
+            status=WorkflowExecutionStatus.FAILED.value,
+            outputs="{}",
+            error=error,
+            elapsed_time=0.0,
+            total_tokens=0,
+            total_steps=0,
+            created_by_role=actor.role,
+            created_by=actor.id,
+            created_at=now,
+            finished_at=now,
+            exceptions_count=0,
+        )
+        log = self._new_log(actor=actor, trigger_data=trigger_data, workflow_id=workflow.id, queue=queue)
+        log.status = WorkflowTriggerStatus.FAILED
+        log.error = error
+        log.triggered_at = now
+        log.finished_at = now
+        log.elapsed_time = 0.0
+        log.total_tokens = 0
+        with self._session_factory.begin() as session:
+            session.add(run)
+            session.flush()
+            session.add(
+                WorkflowAppLog(
+                    tenant_id=workflow.tenant_id,
+                    app_id=workflow.app_id,
+                    workflow_id=workflow.id,
+                    workflow_run_id=run.id,
+                    created_from=WorkflowAppLogCreatedFrom.SERVICE_API,
+                    created_by_role=actor.role,
+                    created_by=actor.id,
+                )
+            )
+            log.workflow_run_id = run.id
+            SQLAlchemyWorkflowTriggerLogRepository(session).create(log)
+
+    @staticmethod
+    def _new_log(*, actor: TriggerActor, trigger_data: TriggerData, workflow_id: str, queue: str) -> WorkflowTriggerLog:
+        return WorkflowTriggerLog(
+            tenant_id=trigger_data.tenant_id,
+            app_id=trigger_data.app_id,
+            workflow_id=workflow_id,
+            root_node_id=trigger_data.root_node_id,
+            trigger_metadata=trigger_data.trigger_metadata.model_dump_json() if trigger_data.trigger_metadata else "{}",
+            trigger_type=trigger_data.trigger_type,
+            workflow_run_id=None,
+            outputs=None,
+            trigger_data=trigger_data.model_dump_json(),
+            inputs=json.dumps(dict(trigger_data.inputs)),
+            status=WorkflowTriggerStatus.PENDING,
+            queue_name=queue,
+            retry_count=0,
+            created_by_role=actor.role,
+            created_by=actor.id,
+            celery_task_id=None,
+            error=None,
+            elapsed_time=None,
+            total_tokens=None,
+        )