Agent Zero
307 lines
11 KiB
Python
307 lines
11 KiB
Python
"""Trigger dispatcher — routes events to matching automation definitions.
|
|
|
|
This module provides the **generic** bridge between the EventBus and the
|
|
automation execution engine. It replaces the previous hard-coded event
|
|
handler stubs (``on_contact_created`` etc.) with a single wildcard
|
|
subscriber that queries the database for matching
|
|
``AutomationDefinition`` rows and dispatches them through the common
|
|
``run_automation`` execution core.
|
|
|
|
Two event categories are supported:
|
|
|
|
1. **Domain events** (durable, via Outbox → Worker → EventBus)
|
|
- trigger_type = ``"event"``
|
|
- trigger_config = ``{"event_name": "contact.created"}``
|
|
- Any event published to the EventBus (either directly or via the
|
|
outbox processor) can trigger an automation.
|
|
|
|
2. **UI events** (ephemeral, via WebSocket → EventBus only)
|
|
- trigger_type = ``"ui"``
|
|
- trigger_config = ``{"event_name": "ui.contact_selected"}``
|
|
- UI events are **never** written to the outbox. They flow directly
|
|
from the frontend WebSocket handler through the EventBus to this
|
|
dispatcher.
|
|
|
|
All four trigger types (event, ui, schedule, manual) converge on the
|
|
same ``run_automation`` execution engine, ensuring consistent condition
|
|
evaluation, action execution, and run logging.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from sqlalchemy import select
|
|
|
|
if TYPE_CHECKING:
|
|
from app.core.event_bus import EventBus
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Prefix that distinguishes ephemeral UI events from domain events.
|
|
# UI events must NEVER be enqueued into the transactional outbox.
|
|
_UI_EVENT_PREFIX = "ui."
|
|
|
|
|
|
class TriggerDispatcher:
|
|
"""Generic event-to-automation dispatcher.
|
|
|
|
Subscribes to the ``*`` wildcard on the EventBus so that **every**
|
|
event — domain or UI — is evaluated for matching automations.
|
|
|
|
The dispatcher is stateless after construction; it queries the
|
|
database on each event to find matching ``AutomationDefinition``
|
|
rows. This is intentionally generic: no hard-coded event list is
|
|
maintained, and any registered outbox event can trigger an
|
|
automation.
|
|
"""
|
|
|
|
def __init__(self, event_bus: EventBus) -> None:
|
|
self._event_bus = event_bus
|
|
self._handler: Any = None
|
|
|
|
def register(self) -> None:
|
|
"""Subscribe the wildcard handler on the event bus."""
|
|
self._handler = self._on_event
|
|
self._event_bus.subscribe("*", self._handler)
|
|
logger.info("TriggerDispatcher registered — listening to all events")
|
|
|
|
def unregister(self) -> None:
|
|
"""Unsubscribe from the event bus (idempotent)."""
|
|
if self._handler is not None:
|
|
self._event_bus.unsubscribe("*", self._handler)
|
|
self._handler = None
|
|
|
|
async def _on_event(self, payload: dict[str, Any]) -> None:
|
|
"""Wildcard handler invoked for every EventBus event.
|
|
|
|
Determines the event name from the payload envelope, classifies
|
|
it as domain or UI, queries matching automation definitions, and
|
|
dispatches each through ``run_automation``.
|
|
"""
|
|
event_name: str = payload.get("event_name", "")
|
|
if not event_name:
|
|
logger.debug("TriggerDispatcher: event without event_name, skipping")
|
|
return
|
|
|
|
is_ui_event = event_name.startswith(_UI_EVENT_PREFIX)
|
|
trigger_type = "ui" if is_ui_event else "event"
|
|
|
|
logger.debug(
|
|
"TriggerDispatcher: evaluating event '%s' (trigger_type=%s)",
|
|
event_name,
|
|
trigger_type,
|
|
)
|
|
|
|
try:
|
|
await self._dispatch_matching_automations(
|
|
event_name=event_name,
|
|
trigger_type=trigger_type,
|
|
payload=payload,
|
|
)
|
|
# F-PROACTIVE: Also check for matching agent definitions on context/UI events
|
|
if is_ui_event or event_name.startswith("context."):
|
|
await self._dispatch_matching_agents(
|
|
event_name=event_name,
|
|
trigger_type=trigger_type,
|
|
payload=payload,
|
|
)
|
|
except Exception:
|
|
logger.exception(
|
|
"TriggerDispatcher: error dispatching event '%s'", event_name
|
|
)
|
|
|
|
async def _dispatch_matching_automations(
|
|
self,
|
|
event_name: str,
|
|
trigger_type: str,
|
|
payload: dict[str, Any],
|
|
) -> None:
|
|
"""Query DB for active automations matching *event_name* and dispatch."""
|
|
from app.core.db import get_session_factory
|
|
from app.plugins.builtins.contracts import get_contract
|
|
automation_contract = get_contract("automation")
|
|
AutomationDefinition = automation_contract.Automation # noqa: N806
|
|
if automation_contract is None:
|
|
logger.debug("Automation plugin not available — trigger skipped")
|
|
return
|
|
|
|
factory = get_session_factory()
|
|
tenant_id = payload.get("tenant_id")
|
|
|
|
async with factory() as db:
|
|
query = (
|
|
select(AutomationDefinition)
|
|
.where(AutomationDefinition.is_active.is_(True))
|
|
.where(AutomationDefinition.trigger_type == trigger_type)
|
|
)
|
|
if tenant_id is not None:
|
|
query = query.where(
|
|
AutomationDefinition.tenant_id == tenant_id
|
|
)
|
|
|
|
result = await db.execute(query)
|
|
automations = list(result.scalars().all())
|
|
|
|
if not automations:
|
|
logger.debug(
|
|
"TriggerDispatcher: no active automations for event '%s' (type=%s)",
|
|
event_name,
|
|
trigger_type,
|
|
)
|
|
return
|
|
|
|
for automation in automations:
|
|
config = automation.trigger_config or {}
|
|
configured_event = config.get("event_name", "")
|
|
if configured_event != event_name:
|
|
continue
|
|
|
|
logger.info(
|
|
"TriggerDispatcher: dispatching automation '%s' (%s) for event '%s'",
|
|
automation.name,
|
|
automation.id,
|
|
event_name,
|
|
)
|
|
|
|
await self._enqueue_automation(
|
|
automation_id=str(automation.id),
|
|
trigger_type=trigger_type,
|
|
trigger_data=payload,
|
|
)
|
|
|
|
async def _dispatch_matching_agents(
|
|
self,
|
|
event_name: str,
|
|
trigger_type: str,
|
|
payload: dict[str, Any],
|
|
) -> None:
|
|
"""F-PROACTIVE: Query DB for active agents matching *event_name* and dispatch.
|
|
|
|
Checks AgentDefinition.trigger_config for matching context/UI events.
|
|
If match found, creates an AgentRun and dispatches via run_agent.
|
|
"""
|
|
from app.core.db import get_session_factory
|
|
from app.plugins.builtins.contracts import get_contract
|
|
|
|
automation_contract = get_contract("automation")
|
|
if automation_contract is None:
|
|
return
|
|
|
|
AgentDefinition = automation_contract.AgentDefinition # noqa: N806
|
|
if AgentDefinition is None:
|
|
return
|
|
|
|
factory = get_session_factory()
|
|
tenant_id = payload.get("tenant_id")
|
|
|
|
async with factory() as db:
|
|
query = (
|
|
select(AgentDefinition)
|
|
.where(AgentDefinition.is_active.is_(True))
|
|
.where(AgentDefinition.mode == "proactive")
|
|
)
|
|
if tenant_id is not None:
|
|
query = query.where(AgentDefinition.tenant_id == tenant_id)
|
|
|
|
result = await db.execute(query)
|
|
agents = list(result.scalars().all())
|
|
|
|
if not agents:
|
|
return
|
|
|
|
for agent in agents:
|
|
config = agent.trigger_config or {}
|
|
configured_event = config.get("event_name", "")
|
|
if configured_event != event_name:
|
|
continue
|
|
|
|
logger.info(
|
|
"TriggerDispatcher: dispatching agent '%s' (%s) for event '%s'",
|
|
agent.name,
|
|
agent.id,
|
|
event_name,
|
|
)
|
|
|
|
try:
|
|
await automation_contract.run_agent(
|
|
ctx={},
|
|
agent_id=str(agent.id),
|
|
trigger_type=trigger_type,
|
|
trigger_data=payload,
|
|
)
|
|
except Exception:
|
|
logger.exception(
|
|
"TriggerDispatcher: run_agent failed for agent_id=%s",
|
|
agent.id,
|
|
)
|
|
|
|
async def _enqueue_automation(
|
|
self,
|
|
automation_id: str,
|
|
trigger_type: str,
|
|
trigger_data: dict[str, Any],
|
|
) -> None:
|
|
"""Dispatch automation through the common execution core.
|
|
|
|
Uses ``run_automation`` directly (in-process) for low latency.
|
|
For production workloads with back-pressure, the caller may
|
|
alternatively enqueue via ``enqueue_job``.
|
|
"""
|
|
try:
|
|
from app.plugins.builtins.contracts import get_contract
|
|
automation_contract = get_contract("automation")
|
|
if automation_contract is not None:
|
|
await automation_contract.run_automation(
|
|
ctx={},
|
|
automation_id=automation_id,
|
|
trigger_type=trigger_type,
|
|
trigger_data=trigger_data,
|
|
)
|
|
except Exception:
|
|
logger.exception(
|
|
"TriggerDispatcher: run_automation failed for automation_id=%s",
|
|
automation_id,
|
|
)
|
|
|
|
|
|
# ── Module-level helpers ─────────────────────────────────────────────────────
|
|
|
|
_dispatcher: TriggerDispatcher | None = None
|
|
|
|
|
|
def get_trigger_dispatcher() -> TriggerDispatcher | None:
|
|
"""Return the singleton dispatcher, or ``None`` if not registered."""
|
|
return _dispatcher
|
|
|
|
|
|
def register_trigger_dispatcher(event_bus: EventBus) -> TriggerDispatcher:
|
|
"""Create and register the trigger dispatcher on *event_bus*.
|
|
|
|
Safe to call multiple times — subsequent calls are no-ops.
|
|
"""
|
|
global _dispatcher
|
|
if _dispatcher is not None:
|
|
logger.debug("TriggerDispatcher already registered")
|
|
return _dispatcher
|
|
_dispatcher = TriggerDispatcher(event_bus)
|
|
_dispatcher.register()
|
|
return _dispatcher
|
|
|
|
|
|
def unregister_trigger_dispatcher() -> None:
|
|
"""Unregister and discard the singleton dispatcher."""
|
|
global _dispatcher
|
|
if _dispatcher is not None:
|
|
_dispatcher.unregister()
|
|
_dispatcher = None
|
|
|
|
|
|
def is_ui_event(event_name: str) -> bool:
|
|
"""Return ``True`` if *event_name* is an ephemeral UI event.
|
|
|
|
UI events must never be written to the transactional outbox.
|
|
"""
|
|
return event_name.startswith(_UI_EVENT_PREFIX)
|