diff --git a/README.md b/README.md index 34a7906..ac360ae 100644 --- a/README.md +++ b/README.md @@ -114,13 +114,71 @@ If authentication is enabled (see below), send the taskowl API key as | **Task actions** | `revoke_task`, `retry_task`, `execute_task` | | **Workers** | `get_worker_status`, `list_workers`, `get_worker_stats`, `shutdown_worker`, `scale_worker_pool`, `restart_worker_pool`, `get_active_tasks`, `get_scheduled_tasks`, `get_reserved_tasks` | | **Queues** | `list_queues` | +| **Automations** | `list_automations`, `create_automation`, `get_automation`, `update_automation`, `delete_automation`, `toggle_automation`, `get_automation_runs`, `get_automation_status` | -**Total: 20 tools** +**Total: 28 tools** `list_tasks` supports exact filters (`state`, `name`, `worker`, `since`), a partial case-insensitive `search` on the task name, `offset` for pagination, and `sort_by` (`timestamp` [default, newest-first], `name`, `state`, `worker`). +### Automations + +Automations are declarative **trigger → conditions → actions** definitions that drive +workflow automation (a superset of the env-var alerts). They are managed via the +`/api/automations` endpoints and the `*_automation` MCP tools. + +Event triggers are evaluated by the consumer process: when an enabled automation's +`event_type` matches an incoming Celery event and all its `conditions` pass, its actions +fire. Every evaluation is recorded in the append-only `automation_runs` log (metadata only +— args, kwargs, results, and tracebacks are never stored), queryable via +`GET /api/automations/{id}/runs` and the `get_automation_runs` MCP tool. + +```bash +curl -X POST http://localhost:8000/api/automations \ + -H "Authorization: Bearer $API_KEY" \ + -H "Content-Type: application/json" \ + -d '{ + "name": "alert-on-failure", + "trigger_type": "event", + "event_type": "task-failed", + "conditions": [{"field": "name", "op": "eq", "value": "payments.charge"}], + "actions": [{"type": "log"}] + }' +``` + +Trigger types: `event` (with `event_type`) or `periodic` (with `schedule_seconds`). +Conditions use `{field, op, value}` against event fields (including dotted paths) with ops +`eq/neq/gt/gte/lt/lte/contains/matches/in/exists`. + +Action types: +- `log` — write to the application log (`level`, `message`) +- `slack_webhook` — Slack-formatted webhook (`webhook_url` or `ALERT_WEBHOOK_URL`, `text`, `fields`) +- `webhook` — generic JSON webhook (`url`, `payload`) +- `retry_task` — retry the event's task (`task_id` defaults to the event `uuid`) +- `execute_task` — send a task by name (`name`, `args`, `kwargs`, `queue`, `countdown`, `eta`, `expires`, `priority`) +- `revoke_task` — revoke the event's task (`task_id` defaults to the event `uuid`, `terminate`) +- `check_workers_offline` — scan for stale/offline workers and alert (used by the + seeded `alert-worker-offline-sweep` periodic automation) + +Action params support `{event.field}` interpolation (e.g. `"task_id": "{event.uuid}"`). + +Safety knobs prevent alert/action storms: +- `cooldown_seconds` — after firing, wait at least this long before firing again +- `max_runs_per_window` + `window_seconds` — fire at most `max_runs_per_window` times per `window_seconds` +- `circuit_breaker` — `{"failure_threshold": N, "window_seconds": W}`; skips actions + (`"circuit_open"`) once the automation has fired N times within W seconds, and auto-closes + once the window rolls past + +Skipped evaluations (cooldown, rate limit, or circuit open) are still recorded in +`automation_runs` with a `details.skipped` reason (`"cooldown"` / `"rate_limited"` / +`"circuit_open"`), keeping storm suppression auditable. + +**Periodic triggers**: automations with `trigger_type: "periodic"` and `schedule_seconds` +run on a schedule (evaluated by the consumer's periodic loop, checked every +`AUTOMATION_CHECK_SECONDS`). Their conditions are evaluated against an empty event context, +so they are typically used for schedule-driven actions (e.g. a heartbeat webhook). + ## Examples Questions you can ask your AI assistant when the MCP server is connected: @@ -140,6 +198,7 @@ Questions you can ask your AI assistant when the MCP server is connected: | "What's scheduled to run next?" | `get_scheduled_tasks`, `get_reserved_tasks` | | "Retry task abc" | `retry_task` | | "Run myapp.tasks.process now" | `execute_task` | +| "Create an automation that alerts on payment failures" | `create_automation` | ## Architecture @@ -178,6 +237,7 @@ All configuration is via environment variables: | `ALERT_ON_WORKER_OFFLINE` | Enable worker-offline alerts | `true` | No | | `ALERT_SLOW_TASK_SECONDS` | Alert when a succeeded task exceeds this runtime | None | No | | `ALERT_WORKER_CHECK_SECONDS` | Interval for the periodic stale-worker check | `30` | No | +| `AUTOMATION_CHECK_SECONDS` | Interval for the periodic automation evaluation loop | `5` | No | ### Brokers @@ -190,6 +250,12 @@ export CELERY_BROKER_URL="redis://localhost:6379/0" # Redis ### Alerts / Webhooks +> **Deprecated in favor of Automations.** The `ALERT_*` env vars below are +> legacy: on consumer startup they seed the equivalent built-in automations +> (`alert-task-failed`, `alert-slow-task`, `alert-worker-offline`, +> `alert-worker-offline-sweep`), which are then managed like any other +> automation via the API/MCP. Prefer defining automations directly. + Set `ALERT_WEBHOOK_URL` to a Slack incoming webhook to receive notifications on task failures, offline workers, and slow tasks. Alerting is **off by default**. @@ -201,7 +267,7 @@ Conditions: - `ALERT_ON_TASK_FAILED=true` (default) — notify when a task fails - `ALERT_ON_WORKER_OFFLINE=true` (default) — notify when a worker goes offline - (via an `worker-offline` event or a stale heartbeat detected every + (via a `worker-offline` event or a stale heartbeat detected every `ALERT_WORKER_CHECK_SECONDS`) - `ALERT_SLOW_TASK_SECONDS=30` — notify when a succeeded task exceeds 30s @@ -232,6 +298,8 @@ scrape_configs: | `taskowl_worker_status` | Gauge (1 = online, 0 = offline) | `worker` | | `taskowl_worker_active_tasks` | Gauge | `worker` | | `taskowl_worker_processed_total` | Counter | `worker` | +| `taskowl_automation_fired_total` | Counter (automation runs that fired actions) | `automation_id`, `trigger` | +| `taskowl_automation_skipped_total` | Counter (runs skipped by a safety mechanism) | `automation_id`, `reason` | > **Security**: `/metrics` is intentionally unauthenticated so Prometheus can > scrape it without the taskowl API key. Only expose it to trusted networks or @@ -267,6 +335,7 @@ retries, and metrics. Interactive docs are available at: | **Workers** | `GET /api/workers`, `GET /api/workers/list`, `GET /api/workers/{name}/stats`, `GET /api/workers/active-tasks`, `GET /api/workers/scheduled`, `GET /api/workers/reserved` | | **Worker actions** | `POST /api/workers/{name}/shutdown`, `POST /api/workers/{name}/scale`, `POST /api/workers/{name}/restart` | | **Queues** | `GET /api/queues` | +| **Automations** | `GET /api/automations`, `POST /api/automations`, `GET /api/automations/{id}`, `PUT /api/automations/{id}`, `DELETE /api/automations/{id}`, `POST /api/automations/{id}/toggle`, `GET /api/automations/{id}/runs`, `GET /api/automations/{id}/status` | | **Ops** | `GET /health`, `GET /metrics` | The `/openapi.json` schema is the authoritative reference — this README lists diff --git a/alembic/env.py b/alembic/env.py index a309006..6643720 100644 --- a/alembic/env.py +++ b/alembic/env.py @@ -8,7 +8,7 @@ from alembic import context from taskowl.config import settings from taskowl.database import Base -from taskowl.models import TaskEvent, WorkerEvent # noqa: F401 - import to register models +from taskowl.models import Automation, AutomationRun, TaskEvent, WorkerEvent # noqa: F401 # this is the Alembic Config object, which provides # access to the values within the .ini file in use. diff --git a/alembic/versions/003_workflow_automations.py b/alembic/versions/003_workflow_automations.py new file mode 100644 index 0000000..6d419df --- /dev/null +++ b/alembic/versions/003_workflow_automations.py @@ -0,0 +1,86 @@ +"""workflow automations schema + +Revision ID: 003_workflow_automations +Revises: 002_event_sourcing +Create Date: 2026-09-08 00:00:00.000000 + +""" + +from collections.abc import Sequence + +import sqlalchemy as sa + +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = "003_workflow_automations" +down_revision: str | None = "002_event_sourcing" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + """Create automation tables.""" + op.create_table( + "automations", + sa.Column("id", sa.Integer, primary_key=True, autoincrement=True), + sa.Column("name", sa.String(255), nullable=False, unique=True), + sa.Column("enabled", sa.Boolean, nullable=False, server_default=sa.text("true")), + sa.Column("trigger_type", sa.String(20), nullable=False), + sa.Column("event_type", sa.String(50), nullable=True), + sa.Column("schedule_seconds", sa.Integer, nullable=True), + sa.Column("conditions", sa.JSON, nullable=True), + sa.Column("actions", sa.JSON, nullable=True), + sa.Column("cooldown_seconds", sa.Integer, nullable=True), + sa.Column("max_runs_per_window", sa.Integer, nullable=True), + sa.Column("window_seconds", sa.Integer, nullable=True), + sa.Column("circuit_breaker", sa.JSON, nullable=True), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + ) + op.create_index("idx_automations_enabled", "automations", ["enabled"]) + op.create_index("idx_automations_trigger", "automations", ["trigger_type"]) + + op.create_table( + "automation_runs", + sa.Column("id", sa.Integer, primary_key=True, autoincrement=True), + sa.Column("automation_id", sa.Integer, nullable=False), + sa.Column("trigger", sa.String(50), nullable=False), + sa.Column("matched", sa.Boolean, nullable=False, server_default=sa.text("false")), + sa.Column( + "conditions_passed", + sa.Boolean, + nullable=False, + server_default=sa.text("false"), + ), + sa.Column("actions_fired", sa.JSON, nullable=True), + sa.Column("details", sa.JSON, nullable=True), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + ) + op.create_index("idx_automation_runs_automation_id", "automation_runs", ["automation_id"]) + op.create_index("idx_automation_runs_created_at", "automation_runs", ["created_at"]) + + +def downgrade() -> None: + """Drop automation tables.""" + op.drop_index("idx_automation_runs_created_at", table_name="automation_runs") + op.drop_index("idx_automation_runs_automation_id", table_name="automation_runs") + op.drop_table("automation_runs") + op.drop_index("idx_automations_trigger", table_name="automations") + op.drop_index("idx_automations_enabled", table_name="automations") + op.drop_table("automations") diff --git a/src/taskowl/automations.py b/src/taskowl/automations.py new file mode 100644 index 0000000..0d3b080 --- /dev/null +++ b/src/taskowl/automations.py @@ -0,0 +1,523 @@ +"""Workflow automation CRUD functions for taskowl. + +This module contains the core logic for managing workflow automations +(declarative trigger -> conditions -> actions definitions) that can be +used by both REST API endpoints and MCP tools. +""" + +import logging + +from sqlalchemy import delete, select +from sqlalchemy.ext.asyncio import AsyncSession + +from taskowl.database import async_session_maker +from taskowl.models import Automation, AutomationRun + +logger = logging.getLogger(__name__) + +_VALID_TRIGGERS = ("event", "periodic") +_VALID_EVENT_TYPES = ( + "task-sent", + "task-received", + "task-started", + "task-succeeded", + "task-failed", + "task-revoked", + "task-retried", + "task-rejected", + "worker-online", + "worker-heartbeat", + "worker-offline", +) + + +def _validate_automation(data: dict) -> str | None: + """Validate an automation definition. Returns an error message or None.""" + trigger_type = data.get("trigger_type") + if trigger_type not in _VALID_TRIGGERS: + return f"trigger_type must be one of {list(_VALID_TRIGGERS)}" + + if trigger_type == "event": + event_type = data.get("event_type") + if not event_type: + return "event_type is required for event triggers" + if event_type not in _VALID_EVENT_TYPES: + return f"event_type must be one of {list(_VALID_EVENT_TYPES)}" + elif trigger_type == "periodic": + schedule_seconds = data.get("schedule_seconds") + if not isinstance(schedule_seconds, int) or schedule_seconds <= 0: + return "schedule_seconds must be a positive integer for periodic triggers" + + cooldown = data.get("cooldown_seconds") + if cooldown is not None and (not isinstance(cooldown, int) or cooldown < 0): + return "cooldown_seconds must be a non-negative integer" + + max_runs = data.get("max_runs_per_window") + window = data.get("window_seconds") + if max_runs is not None and (not isinstance(max_runs, int) or max_runs <= 0): + return "max_runs_per_window must be a positive integer" + if max_runs is not None and (not isinstance(window, int) or window <= 0): + return "window_seconds is required when max_runs_per_window is set" + + return None + + +def _serialize(automation: Automation) -> dict: + """Convert an Automation model to a JSON-serializable dict.""" + return { + "id": automation.id, + "name": automation.name, + "enabled": automation.enabled, + "trigger_type": automation.trigger_type, + "event_type": automation.event_type, + "schedule_seconds": automation.schedule_seconds, + "conditions": automation.conditions, + "actions": automation.actions, + "cooldown_seconds": automation.cooldown_seconds, + "max_runs_per_window": automation.max_runs_per_window, + "window_seconds": automation.window_seconds, + "circuit_breaker": automation.circuit_breaker, + "created_at": automation.created_at.isoformat() if automation.created_at else None, + "updated_at": automation.updated_at.isoformat() if automation.updated_at else None, + } + + +def _serialize_run(run: AutomationRun) -> dict: + """Convert an AutomationRun model to a JSON-serializable dict.""" + return { + "id": run.id, + "automation_id": run.automation_id, + "trigger": run.trigger, + "matched": run.matched, + "conditions_passed": run.conditions_passed, + "actions_fired": run.actions_fired, + "details": run.details, + "created_at": run.created_at.isoformat() if run.created_at else None, + } + + +async def create_automation( + data: dict, + session: AsyncSession | None = None, +) -> dict: + """Create a new automation definition. + + Args: + data: Automation definition dict + session: Optional database session (for testing) + + Returns: + Dict with the created automation or an error + """ + error = _validate_automation(data) + if error: + return {"error": error} + if not data.get("name"): + return {"error": "name is required"} + + async def _create(session: AsyncSession) -> dict: + existing = await session.execute(select(Automation).where(Automation.name == data["name"])) + if existing.scalar_one_or_none() is not None: + return {"error": f"Automation with name '{data['name']}' already exists"} + + automation = Automation( + name=data["name"], + enabled=data.get("enabled", True), + trigger_type=data["trigger_type"], + event_type=data.get("event_type"), + schedule_seconds=data.get("schedule_seconds"), + conditions=data.get("conditions"), + actions=data.get("actions"), + cooldown_seconds=data.get("cooldown_seconds"), + max_runs_per_window=data.get("max_runs_per_window"), + window_seconds=data.get("window_seconds"), + circuit_breaker=data.get("circuit_breaker"), + ) + session.add(automation) + await session.commit() + await session.refresh(automation) + return _serialize(automation) + + if session is None: + async with async_session_maker() as db_session: + return await _create(db_session) + return await _create(session) + + +async def get_automation( + automation_id: int, + session: AsyncSession | None = None, +) -> dict: + """Get an automation by id. + + Args: + automation_id: ID of the automation + session: Optional database session (for testing) + + Returns: + Dict with the automation or an error + """ + + async def _get(session: AsyncSession) -> dict: + result = await session.execute(select(Automation).where(Automation.id == automation_id)) + automation = result.scalar_one_or_none() + if automation is None: + return {"error": f"Automation not found: {automation_id}"} + return _serialize(automation) + + if session is None: + async with async_session_maker() as db_session: + return await _get(db_session) + return await _get(session) + + +async def list_automations( + enabled: bool | None = None, + session: AsyncSession | None = None, +) -> list[dict]: + """List automations, optionally filtered by enabled state. + + Args: + enabled: If set, filter by enabled/disabled + session: Optional database session (for testing) + + Returns: + List of automation dicts + """ + + async def _list(session: AsyncSession) -> list[dict]: + query = select(Automation).order_by(Automation.id) + if enabled is not None: + query = query.where(Automation.enabled == enabled) + result = await session.execute(query) + return [_serialize(a) for a in result.scalars().all()] + + if session is None: + async with async_session_maker() as db_session: + return await _list(db_session) + return await _list(session) + + +async def update_automation( + automation_id: int, + data: dict, + session: AsyncSession | None = None, +) -> dict: + """Update an automation definition. + + Args: + automation_id: ID of the automation + data: Fields to update + session: Optional database session (for testing) + + Returns: + Dict with the updated automation or an error + """ + merged = {k: v for k, v in data.items() if v is not None} + if not merged: + return {"error": "No fields to update"} + + # Validate the merged definition (respect existing values for unset fields) + async def _update(session: AsyncSession) -> dict: + result = await session.execute(select(Automation).where(Automation.id == automation_id)) + automation = result.scalar_one_or_none() + if automation is None: + return {"error": f"Automation not found: {automation_id}"} + + current = _serialize(automation) + merged_def = {**current, **merged} + error = _validate_automation(merged_def) + if error: + return {"error": error} + + for field, value in merged.items(): + if field in ("created_at", "updated_at", "id"): + continue + setattr(automation, field, value) + + await session.commit() + await session.refresh(automation) + return _serialize(automation) + + if session is None: + async with async_session_maker() as db_session: + return await _update(db_session) + return await _update(session) + + +async def delete_automation( + automation_id: int, + session: AsyncSession | None = None, +) -> dict: + """Delete an automation and its run history. + + Args: + automation_id: ID of the automation + session: Optional database session (for testing) + + Returns: + Dict with status or an error + """ + + async def _delete(session: AsyncSession) -> dict: + result = await session.execute(select(Automation).where(Automation.id == automation_id)) + if result.scalar_one_or_none() is None: + return {"error": f"Automation not found: {automation_id}"} + await session.execute(delete(Automation).where(Automation.id == automation_id)) + await session.execute( + delete(AutomationRun).where(AutomationRun.automation_id == automation_id) + ) + await session.commit() + return {"status": "success", "message": f"Automation {automation_id} deleted"} + + if session is None: + async with async_session_maker() as db_session: + return await _delete(db_session) + return await _delete(session) + + +async def toggle_automation( + automation_id: int, + session: AsyncSession | None = None, +) -> dict: + """Toggle an automation's enabled state. + + Args: + automation_id: ID of the automation + session: Optional database session (for testing) + + Returns: + Dict with the updated automation or an error + """ + + async def _toggle(session: AsyncSession) -> dict: + result = await session.execute(select(Automation).where(Automation.id == automation_id)) + automation = result.scalar_one_or_none() + if automation is None: + return {"error": f"Automation not found: {automation_id}"} + automation.enabled = not automation.enabled + await session.commit() + await session.refresh(automation) + return _serialize(automation) + + if session is None: + async with async_session_maker() as db_session: + return await _toggle(db_session) + return await _toggle(session) + + +async def get_automation_status( + automation_id: int, + session: AsyncSession | None = None, +) -> dict: + """Get an automation's status, including current circuit-breaker state. + + Args: + automation_id: ID of the automation + session: Optional database session (for testing) + + Returns: + Dict with the automation plus its latest run and circuit state + """ + + async def _get(session: AsyncSession) -> dict: + result = await session.execute(select(Automation).where(Automation.id == automation_id)) + automation = result.scalar_one_or_none() + if automation is None: + return {"error": f"Automation not found: {automation_id}"} + + runs = await list_automation_runs(automation_id, limit=1, session=session) + latest_run = runs[0] if runs else None + + circuit_state = "closed" + circuit_config = automation.circuit_breaker + if ( + circuit_config + and latest_run + and (latest_run.get("details") or {}).get("skipped") == "circuit_open" + ): + circuit_state = "open" + + return { + **_serialize(automation), + "circuit_state": circuit_state, + "last_run": latest_run, + } + + if session is None: + async with async_session_maker() as db_session: + return await _get(db_session) + return await _get(session) + + +async def list_automation_runs( + automation_id: int | None = None, + limit: int = 50, + session: AsyncSession | None = None, +) -> list[dict]: + """List automation runs, optionally filtered by automation id. + + Args: + automation_id: If set, only runs for this automation + limit: Max number of runs to return (default: 50) + session: Optional database session (for testing) + + Returns: + List of run dicts (newest first) + """ + + async def _list(session: AsyncSession) -> list[dict]: + query = ( + select(AutomationRun) + .order_by(AutomationRun.created_at.desc(), AutomationRun.id.desc()) + .limit(limit) + ) + if automation_id is not None: + query = query.where(AutomationRun.automation_id == automation_id) + result = await session.execute(query) + return [_serialize_run(r) for r in result.scalars().all()] + + if session is None: + async with async_session_maker() as db_session: + return await _list(db_session) + return await _list(session) + + +async def seed_builtin_automations(session: AsyncSession | None = None) -> list[dict]: + """Create the built-in alert automations from the legacy env-var settings. + + Idempotent: only creates an automation if one with that name does not + already exist. Returns the list of newly created automations. + + These automations replicate the legacy ``ALERT_*`` behavior (task-failed, + slow-task, worker-offline event, and the periodic stale-worker sweep) so + the engine is the single alerting mechanism. + + Args: + session: Optional database session (for testing) + + Returns: + List of created automation dicts + """ + from taskowl.config import settings + + webhook_url = settings.alert_webhook_url + created: list[dict] = [] + + async def _ensure( + name: str, + definition: dict, + session: AsyncSession, + ) -> None: + existing = await session.execute(select(Automation).where(Automation.name == name)) + if existing.scalar_one_or_none() is not None: + return + automation = Automation( + name=name, + enabled=definition.get("enabled", True), + trigger_type=definition["trigger_type"], + event_type=definition.get("event_type"), + schedule_seconds=definition.get("schedule_seconds"), + conditions=definition.get("conditions"), + actions=definition.get("actions"), + ) + session.add(automation) + await session.flush() + created.append(_serialize(automation)) + + async def _seed(session: AsyncSession) -> None: + if not webhook_url: + return + + if settings.alert_on_task_failed: + await _ensure( + "alert-task-failed", + { + "trigger_type": "event", + "event_type": "task-failed", + "conditions": [], + "actions": [ + { + "type": "slack_webhook", + "webhook_url": webhook_url, + "text": "Task failed", + "fields": { + "Task": "{event.name}", + "Task ID": "{event.uuid}", + "Worker": "{event.hostname}", + "Error": "{event.exception}", + }, + } + ], + }, + session, + ) + + if settings.alert_slow_task_seconds is not None: + await _ensure( + "alert-slow-task", + { + "trigger_type": "event", + "event_type": "task-succeeded", + "conditions": [ + { + "field": "runtime", + "op": "gt", + "value": settings.alert_slow_task_seconds, + } + ], + "actions": [ + { + "type": "slack_webhook", + "webhook_url": webhook_url, + "text": "Slow task detected", + "fields": { + "Task": "{event.name}", + "Task ID": "{event.uuid}", + "Runtime (s)": "{event.runtime}", + "Worker": "{event.hostname}", + }, + } + ], + }, + session, + ) + + if settings.alert_on_worker_offline: + await _ensure( + "alert-worker-offline", + { + "trigger_type": "event", + "event_type": "worker-offline", + "conditions": [], + "actions": [ + { + "type": "slack_webhook", + "webhook_url": webhook_url, + "text": "Worker offline", + "fields": {"Worker": "{event.hostname}"}, + } + ], + }, + session, + ) + await _ensure( + "alert-worker-offline-sweep", + { + "trigger_type": "periodic", + "schedule_seconds": settings.alert_worker_check_seconds, + "conditions": [], + "actions": [{"type": "check_workers_offline", "webhook_url": webhook_url}], + }, + session, + ) + + await session.commit() + + if session is None: + async with async_session_maker() as db_session: + await _seed(db_session) + else: + await _seed(session) + + return created diff --git a/src/taskowl/config.py b/src/taskowl/config.py index 2160883..a5aff51 100644 --- a/src/taskowl/config.py +++ b/src/taskowl/config.py @@ -68,6 +68,10 @@ class Settings(BaseSettings): default=30, description="Interval for the periodic stale-worker check", ) + automation_check_seconds: int = Field( + default=5, + description="Interval for the periodic automation evaluation loop", + ) model_config = {"env_prefix": "", "case_sensitive": False} diff --git a/src/taskowl/consumer/receiver.py b/src/taskowl/consumer/receiver.py index a6c17f3..fa87638 100644 --- a/src/taskowl/consumer/receiver.py +++ b/src/taskowl/consumer/receiver.py @@ -14,13 +14,14 @@ from celery.events import EventReceiver from kombu import Connection -from taskowl.alerting import AlertNotifier +from taskowl.automations import seed_builtin_automations from taskowl.config import settings from taskowl.consumer.handlers import ( TASK_EVENT_HANDLERS, WORKER_EVENT_HANDLERS, ) from taskowl.database import async_session_maker +from taskowl.workflow import WorkflowEngine logger = logging.getLogger(__name__) @@ -41,8 +42,8 @@ def __init__(self, broker_url: str) -> None: self._main_loop: asyncio.AbstractEventLoop | None = None self.connection: Connection | None = None self.recv: EventReceiver | None = None - self.alert_notifier = AlertNotifier() - self._alert_task: asyncio.Task | None = None + self._automation_task: asyncio.Task | None = None + self.workflow_engine = WorkflowEngine() def _create_handlers(self) -> dict[str, Any]: """Create event handler mapping for Celery receiver.""" @@ -80,13 +81,15 @@ async def _handle_event(self, event: dict, handler_func: Any) -> None: async with async_session_maker() as session: await handler_func(event, session) - # Fire alerts after successful persistence event_type = event.get("type", "") - await self.alert_notifier.notify_event(event_type, event) + # Mark workers online so the offline sweep can re-alert on recovery if event_type in ("worker-online", "worker-heartbeat"): hostname = event.get("hostname") if hostname: - self.alert_notifier.mark_online(hostname) + self.workflow_engine.mark_online(hostname) + + # Evaluate workflow automations (alerts are now automations) + await self.workflow_engine.evaluate_event(event_type, event) except Exception: logger.exception(f"Error handling event: {event}") @@ -160,28 +163,34 @@ def handle_signal() -> None: for sig in (signal.SIGTERM, signal.SIGINT): loop.add_signal_handler(sig, handle_signal) - # Start periodic stale-worker alert check (no-op if alerting disabled) - self._alert_task = asyncio.create_task(self._periodic_worker_check()) + # Seed built-in alert automations (idempotent; no-op if alerting disabled) + try: + await seed_builtin_automations() + except Exception: + logger.exception("Failed to seed built-in automations") + + # Start periodic automation evaluation loop (no-op if no periodic automations) + self._automation_task = asyncio.create_task(self._periodic_automation_check()) # Run event capture in thread pool to avoid blocking await asyncio.to_thread(self._capture_events) - # Cancel periodic alert check on shutdown - if self._alert_task: - self._alert_task.cancel() + # Cancel periodic tasks on shutdown + if self._automation_task: + self._automation_task.cancel() with contextlib.suppress(asyncio.CancelledError): - await self._alert_task + await self._automation_task logger.info("Celery event consumer stopped") - async def _periodic_worker_check(self) -> None: - """Periodically check for stale/offline workers and fire alerts.""" - interval = settings.alert_worker_check_seconds + async def _periodic_automation_check(self) -> None: + """Periodically evaluate due periodic automations.""" + interval = settings.automation_check_seconds while True: try: - await self.alert_notifier.check_workers() + await self.workflow_engine.evaluate_periodic() except Exception: - logger.exception("Error in periodic worker alert check") + logger.exception("Error in periodic automation check") await asyncio.sleep(interval) async def stop(self) -> None: diff --git a/src/taskowl/main.py b/src/taskowl/main.py index 9f68170..721e023 100644 --- a/src/taskowl/main.py +++ b/src/taskowl/main.py @@ -11,6 +11,16 @@ from taskowl.actions import execute_task, retry_task, revoke_task from taskowl.auth import verify_api_key +from taskowl.automations import ( + create_automation, + delete_automation, + get_automation, + get_automation_status, + list_automation_runs, + list_automations, + toggle_automation, + update_automation, +) from taskowl.config import settings from taskowl.database import close_db, get_db, init_db from taskowl.metrics import generate_metrics @@ -411,6 +421,138 @@ async def api_list_queues( return result +class AutomationCreate(BaseModel): + """Request model for creating an automation.""" + + name: str + enabled: bool = True + trigger_type: str + event_type: str | None = None + schedule_seconds: int | None = None + conditions: list | None = None + actions: list | None = None + cooldown_seconds: int | None = None + max_runs_per_window: int | None = None + window_seconds: int | None = None + circuit_breaker: dict | None = None + + +class AutomationUpdate(BaseModel): + """Request model for updating an automation.""" + + name: str | None = None + enabled: bool | None = None + trigger_type: str | None = None + event_type: str | None = None + schedule_seconds: int | None = None + conditions: list | None = None + actions: list | None = None + cooldown_seconds: int | None = None + max_runs_per_window: int | None = None + window_seconds: int | None = None + circuit_breaker: dict | None = None + + +@app.get("/api/automations") +async def api_list_automations( + enabled: bool | None = None, + session: AsyncSession = Depends(get_db), + _: None = Depends(verify_api_key), +) -> list[dict]: + """List workflow automations, optionally filtered by enabled state.""" + return await list_automations(enabled, session) + + +@app.post("/api/automations") +async def api_create_automation( + request: AutomationCreate, + session: AsyncSession = Depends(get_db), + _: None = Depends(verify_api_key), +) -> dict: + """Create a new workflow automation.""" + result = await create_automation(request.model_dump(), session) + if "error" in result: + raise HTTPException(status_code=400, detail=result["error"]) + return result + + +@app.get("/api/automations/{automation_id}") +async def api_get_automation( + automation_id: int, + session: AsyncSession = Depends(get_db), + _: None = Depends(verify_api_key), +) -> dict: + """Get a workflow automation by id.""" + result = await get_automation(automation_id, session) + if "error" in result: + raise HTTPException(status_code=404, detail=result["error"]) + return result + + +@app.put("/api/automations/{automation_id}") +async def api_update_automation( + automation_id: int, + request: AutomationUpdate, + session: AsyncSession = Depends(get_db), + _: None = Depends(verify_api_key), +) -> dict: + """Update a workflow automation.""" + result = await update_automation(automation_id, request.model_dump(), session) + if "error" in result: + raise HTTPException(status_code=400, detail=result["error"]) + return result + + +@app.delete("/api/automations/{automation_id}") +async def api_delete_automation( + automation_id: int, + session: AsyncSession = Depends(get_db), + _: None = Depends(verify_api_key), +) -> dict: + """Delete a workflow automation.""" + result = await delete_automation(automation_id, session) + if "error" in result: + raise HTTPException(status_code=404, detail=result["error"]) + return result + + +@app.post("/api/automations/{automation_id}/toggle") +async def api_toggle_automation( + automation_id: int, + session: AsyncSession = Depends(get_db), + _: None = Depends(verify_api_key), +) -> dict: + """Toggle a workflow automation's enabled state.""" + result = await toggle_automation(automation_id, session) + if "error" in result: + raise HTTPException(status_code=404, detail=result["error"]) + return result + + +@app.get("/api/automations/{automation_id}/runs") +async def api_list_automation_runs( + automation_id: int, + limit: int = 50, + session: AsyncSession = Depends(get_db), + _: None = Depends(verify_api_key), +) -> list[dict]: + """List run history for a workflow automation.""" + return await list_automation_runs(automation_id, limit, session) + + +@app.get("/api/automations/{automation_id}/status") +async def api_get_automation_status( + automation_id: int, + session: AsyncSession = Depends(get_db), + _: None = Depends(verify_api_key), +) -> dict: + """Get a workflow automation's status, including circuit-breaker state.""" + result = await get_automation_status(automation_id, session) + if "error" in result: + raise HTTPException(status_code=404, detail=result["error"]) + return result + + # MCP server is now run separately via taskowl-mcp command # See taskowl.mcp.server:run_mcp_server() diff --git a/src/taskowl/mcp/tools.py b/src/taskowl/mcp/tools.py index 2934d3c..8949820 100644 --- a/src/taskowl/mcp/tools.py +++ b/src/taskowl/mcp/tools.py @@ -464,3 +464,196 @@ async def list_queues() -> dict: ) response.raise_for_status() return response.json() + + @server.tool( + name="list_automations", + description="List workflow automations, optionally filtered by enabled state", + ) + async def list_automations(enabled: bool | None = None) -> list[dict]: + """List workflow automations. + + Args: + enabled: If set, filter by enabled/disabled state + """ + async with httpx.AsyncClient() as client: + params = {} + if enabled is not None: + params["enabled"] = enabled + response = await client.get( + f"http://{settings.taskowl_host}:{settings.taskowl_port}/api/automations", + params=params, + headers=_get_headers(), + ) + response.raise_for_status() + return response.json() + + @server.tool( + name="create_automation", + description=( + "Create a workflow automation (trigger -> conditions -> actions). " + "trigger_type is 'event' (with event_type) or 'periodic' (with schedule_seconds)." + ), + ) + async def create_automation( + name: str, + trigger_type: str, + event_type: str | None = None, + schedule_seconds: int | None = None, + enabled: bool = True, + conditions: list | None = None, + actions: list | None = None, + cooldown_seconds: int | None = None, + max_runs_per_window: int | None = None, + window_seconds: int | None = None, + circuit_breaker: dict | None = None, + ) -> dict: + """Create a workflow automation. + + Args: + name: Unique automation name + trigger_type: 'event' or 'periodic' + event_type: Celery event type for event triggers + schedule_seconds: Interval for periodic triggers + enabled: Whether the automation starts enabled + conditions: List of condition dicts {field, op, value} + actions: List of action dicts {type, ...params} + cooldown_seconds: Minimum seconds between firings + max_runs_per_window: Max firings per window (with window_seconds) + window_seconds: Window size for max_runs_per_window + circuit_breaker: Optional per-automation circuit breaker config + """ + async with httpx.AsyncClient() as client: + payload = { + "name": name, + "trigger_type": trigger_type, + "enabled": enabled, + "event_type": event_type, + "schedule_seconds": schedule_seconds, + "conditions": conditions, + "actions": actions, + "cooldown_seconds": cooldown_seconds, + "max_runs_per_window": max_runs_per_window, + "window_seconds": window_seconds, + "circuit_breaker": circuit_breaker, + } + response = await client.post( + f"http://{settings.taskowl_host}:{settings.taskowl_port}/api/automations", + json=payload, + headers=_get_headers(), + ) + response.raise_for_status() + return response.json() + + @server.tool( + name="get_automation", + description="Get a workflow automation by id", + ) + async def get_automation(automation_id: int) -> dict: + """Get a workflow automation by id. + + Args: + automation_id: ID of the automation + """ + async with httpx.AsyncClient() as client: + response = await client.get( + f"http://{settings.taskowl_host}:{settings.taskowl_port}/api/automations/{automation_id}", + headers=_get_headers(), + ) + response.raise_for_status() + return response.json() + + @server.tool( + name="update_automation", + description="Update a workflow automation by id", + ) + async def update_automation(automation_id: int, **fields) -> dict: + """Update a workflow automation. + + Args: + automation_id: ID of the automation + **fields: Fields to update (name, enabled, event_type, conditions, actions, ...) + """ + async with httpx.AsyncClient() as client: + response = await client.put( + f"http://{settings.taskowl_host}:{settings.taskowl_port}/api/automations/{automation_id}", + json=fields, + headers=_get_headers(), + ) + response.raise_for_status() + return response.json() + + @server.tool( + name="delete_automation", + description="Delete a workflow automation by id", + ) + async def delete_automation(automation_id: int) -> dict: + """Delete a workflow automation. + + Args: + automation_id: ID of the automation + """ + async with httpx.AsyncClient() as client: + response = await client.delete( + f"http://{settings.taskowl_host}:{settings.taskowl_port}/api/automations/{automation_id}", + headers=_get_headers(), + ) + response.raise_for_status() + return response.json() + + @server.tool( + name="toggle_automation", + description="Toggle a workflow automation's enabled state", + ) + async def toggle_automation(automation_id: int) -> dict: + """Toggle a workflow automation's enabled state. + + Args: + automation_id: ID of the automation + """ + async with httpx.AsyncClient() as client: + response = await client.post( + f"http://{settings.taskowl_host}:{settings.taskowl_port}/api/automations/{automation_id}/toggle", + headers=_get_headers(), + ) + response.raise_for_status() + return response.json() + + @server.tool( + name="get_automation_runs", + description="List run history for a workflow automation", + ) + async def get_automation_runs(automation_id: int, limit: int = 50) -> list[dict]: + """List run history for a workflow automation. + + Args: + automation_id: ID of the automation + limit: Max number of runs to return (default: 50) + """ + async with httpx.AsyncClient() as client: + response = await client.get( + f"http://{settings.taskowl_host}:{settings.taskowl_port}/api/automations/{automation_id}/runs", + params={"limit": limit}, + headers=_get_headers(), + ) + response.raise_for_status() + return response.json() + + @server.tool( + name="get_automation_status", + description=( + "Get a workflow automation's status, including circuit-breaker state and its latest run" + ), + ) + async def get_automation_status(automation_id: int) -> dict: + """Get a workflow automation's status. + + Args: + automation_id: ID of the automation + """ + async with httpx.AsyncClient() as client: + response = await client.get( + f"http://{settings.taskowl_host}:{settings.taskowl_port}/api/automations/{automation_id}/status", + headers=_get_headers(), + ) + response.raise_for_status() + return response.json() diff --git a/src/taskowl/metrics.py b/src/taskowl/metrics.py index b246e0d..260f778 100644 --- a/src/taskowl/metrics.py +++ b/src/taskowl/metrics.py @@ -18,7 +18,7 @@ from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession -from taskowl.models import TaskEvent, WorkerEvent +from taskowl.models import AutomationRun, TaskEvent, WorkerEvent # NOTE: The /metrics endpoint is intentionally UNAUTHENTICATED so that # Prometheus can scrape it without sending the taskowl API key. Ensure this @@ -61,11 +61,24 @@ async def generate_metrics(session: AsyncSession) -> bytes: ["worker"], registry=registry, ) + automation_fired_total = Counter( + "taskowl_automation_fired_total", + "Number of automation runs that fired actions", + ["automation_id", "trigger"], + registry=registry, + ) + automation_skipped_total = Counter( + "taskowl_automation_skipped_total", + "Number of automation runs skipped by a safety mechanism", + ["automation_id", "reason"], + registry=registry, + ) await _populate_task_metrics(session, task_events_total, task_duration) await _populate_worker_metrics( session, worker_status, worker_active_tasks, worker_processed_total ) + await _populate_automation_metrics(session, automation_fired_total, automation_skipped_total) return generate_latest(registry) @@ -156,3 +169,30 @@ def _is_online(event: WorkerEvent, now: datetime, offline_timeout: timedelta) -> if ts.tzinfo is None: ts = ts.replace(tzinfo=UTC) return now - ts <= offline_timeout + + +async def _populate_automation_metrics( + session: AsyncSession, + automation_fired_total: Counter, + automation_skipped_total: Counter, +) -> None: + """Populate automation run metrics. + + Counts runs that fired actions (per automation + trigger) and runs skipped + by a safety mechanism (per automation + reason). The skip reason is read + from the run's ``details.skipped`` JSON field; databases without native + JSON operators fall back to scanning recent runs in Python. + """ + result = await session.execute( + select( + AutomationRun.automation_id, + AutomationRun.trigger, + AutomationRun.actions_fired, + AutomationRun.details, + ).order_by(AutomationRun.id) + ) + for automation_id, trigger, actions_fired, details in result.all(): + if actions_fired is not None: + automation_fired_total.labels(str(automation_id), trigger).inc() + elif details and details.get("skipped"): + automation_skipped_total.labels(str(automation_id), details["skipped"]).inc() diff --git a/src/taskowl/models.py b/src/taskowl/models.py index e649402..d5acd7d 100644 --- a/src/taskowl/models.py +++ b/src/taskowl/models.py @@ -83,3 +83,77 @@ class WorkerEvent(Base): Index("idx_worker_events_event_type_timestamp", "event_type", "timestamp"), Index("idx_worker_events_timestamp", "timestamp"), ) + + +class Automation(Base): + """Declarative workflow automation definition. + + An automation evaluates Celery events (or runs on a schedule) and, when + its conditions match, fires a list of actions. It is the general + superset of the original env-var alerts. + """ + + __tablename__ = "automations" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + name: Mapped[str] = mapped_column(String(255), nullable=False, unique=True) + enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True) + + # Trigger: "event" (fire on a Celery event type) or "periodic" (on a schedule) + trigger_type: Mapped[str] = mapped_column(String(20), nullable=False) + event_type: Mapped[str | None] = mapped_column(String(50), nullable=True) + schedule_seconds: Mapped[int | None] = mapped_column(Integer, nullable=True) + + # Conditions and actions as declarative JSON lists + conditions: Mapped[list | None] = mapped_column(JSON, nullable=True) + actions: Mapped[list | None] = mapped_column(JSON, nullable=True) + + # Safety: prevent storming + cooldown_seconds: Mapped[int | None] = mapped_column(Integer, nullable=True) + max_runs_per_window: Mapped[int | None] = mapped_column(Integer, nullable=True) + window_seconds: Mapped[int | None] = mapped_column(Integer, nullable=True) + + # Optional per-automation circuit breaker + circuit_breaker: Mapped[dict | None] = mapped_column(JSON, nullable=True) + + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + server_default=func.now(), + nullable=False, + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + server_default=func.now(), + onupdate=func.now(), + nullable=False, + ) + + __table_args__ = ( + Index("idx_automations_enabled", "enabled"), + Index("idx_automations_trigger", "trigger_type"), + ) + + +class AutomationRun(Base): + """Append-only audit log of automation evaluations and fired actions.""" + + __tablename__ = "automation_runs" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + automation_id: Mapped[int] = mapped_column(Integer, nullable=False) + trigger: Mapped[str] = mapped_column(String(50), nullable=False) + matched: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + conditions_passed: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + actions_fired: Mapped[list | None] = mapped_column(JSON, nullable=True) + details: Mapped[dict | None] = mapped_column(JSON, nullable=True) + + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + server_default=func.now(), + nullable=False, + ) + + __table_args__ = ( + Index("idx_automation_runs_automation_id", "automation_id"), + Index("idx_automation_runs_created_at", "created_at"), + ) diff --git a/src/taskowl/workflow.py b/src/taskowl/workflow.py new file mode 100644 index 0000000..0c0e116 --- /dev/null +++ b/src/taskowl/workflow.py @@ -0,0 +1,565 @@ +"""Workflow automation evaluation engine. + +Loads enabled automations and evaluates them against incoming Celery +events: trigger matching, condition evaluation, and action dispatch. +Every evaluation is recorded in the append-only ``automation_runs`` log. +""" + +import logging +import re +from datetime import UTC, datetime, timedelta +from typing import Any + +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from taskowl.actions import execute_task, retry_task, revoke_task +from taskowl.alerting import WebhookClient, build_worker_offline_payload +from taskowl.automations import _serialize +from taskowl.config import settings +from taskowl.database import async_session_maker +from taskowl.models import Automation, AutomationRun, WorkerEvent + +logger = logging.getLogger(__name__) + +# Fields that are safe to store in the run audit log. Args/kwargs/results +# and tracebacks are deliberately excluded (metadata-only, like alerts). +_SAFE_FIELDS = ( + "type", + "uuid", + "hostname", + "name", + "runtime", + "exception", + "retries", + "queue", + "pid", + "root_id", + "parent_id", +) + +_SUPPORTED_OPS = ( + "eq", + "neq", + "gt", + "gte", + "lt", + "lte", + "contains", + "matches", + "in", + "exists", +) + + +def _resolve_field(event: dict, field: str) -> Any: + """Resolve a (possibly dotted) field path against an event dict.""" + value: Any = event + for part in field.split("."): + if isinstance(value, dict): + value = value.get(part) + else: + return None + return value + + +def evaluate_condition(event: dict, condition: dict) -> bool: + """Evaluate a single condition ``{field, op, value}`` against an event.""" + field = condition.get("field") + op = condition.get("op") + value = condition.get("value") + + if not field or op not in _SUPPORTED_OPS: + logger.warning("Invalid condition: %s", condition) + return False + + actual = _resolve_field(event, field) + + if op == "exists": + return actual is not None + if op == "eq": + return actual == value + if op == "neq": + return actual != value + if op == "gt": + return _safe_compare(actual, value) and actual > value + if op == "gte": + return _safe_compare(actual, value) and actual >= value + if op == "lt": + return _safe_compare(actual, value) and actual < value + if op == "lte": + return _safe_compare(actual, value) and actual <= value + if op == "contains": + if isinstance(actual, str) and isinstance(value, str): + return value in actual + if isinstance(actual, (list, dict)): + return value in actual + return False + if op == "matches": + return isinstance(value, str) and bool(re.search(value, str(actual or ""))) + if op == "in": + return isinstance(value, (list, tuple)) and actual in value + return False + + +def _safe_compare(actual: Any, value: Any) -> bool: + """Return True if the two values are comparable (same numeric-ish type).""" + if actual is None or value is None: + return False + if isinstance(actual, (int, float)) and isinstance(value, (int, float)): + return True + return type(actual) is type(value) + + +def _safe_snapshot(event: dict) -> dict: + """Build a metadata-only snapshot of an event for the audit log.""" + return {field: event[field] for field in _SAFE_FIELDS if field in event} + + +class WorkflowEngine: + """Evaluates Celery events against automations and records runs.""" + + def __init__(self) -> None: + self._offline_notified: set[str] = set() + + def mark_online(self, hostname: str) -> None: + """Record that a worker is seen online, allowing future offline alerts.""" + self._offline_notified.discard(hostname) + + async def evaluate_event( + self, + event_type: str, + event: dict, + session: AsyncSession | None = None, + ) -> None: + """Evaluate an incoming event against all enabled event automations. + + Args: + event_type: Celery event type (e.g. 'task-failed') + event: The event payload dict + session: Optional database session (for testing) + """ + if session is None: + async with async_session_maker() as db_session: + await self._evaluate_event(db_session, event_type, event) + else: + await self._evaluate_event(session, event_type, event) + + async def _evaluate_event( + self, + session: AsyncSession, + event_type: str, + event: dict, + ) -> None: + """Internal implementation of evaluate_event.""" + result = await session.execute( + select(Automation).where( + Automation.enabled.is_(True), + Automation.trigger_type == "event", + Automation.event_type == event_type, + ) + ) + automations = result.scalars().all() + now = datetime.now(UTC) + + for automation in automations: + conditions = automation.conditions or [] + conditions_passed = all( + evaluate_condition(event, condition) for condition in conditions + ) + + actions_fired = None + skipped = None + if conditions_passed: + skipped = await self._check_circuit_breaker( + automation, now, session + ) or await self._check_rate_limits(automation, now, session) + if skipped is None: + actions_fired = await self._dispatch_actions(automation, event, session) + + details: dict = { + "event_type": event_type, + "event": _safe_snapshot(event), + "automation": _serialize(automation), + } + if skipped is not None: + details["skipped"] = skipped + + session.add( + AutomationRun( + automation_id=automation.id, + trigger=event_type, + matched=True, + conditions_passed=conditions_passed, + actions_fired=actions_fired, + details=details, + ) + ) + + await session.commit() + + async def evaluate_periodic( + self, + session: AsyncSession | None = None, + ) -> None: + """Evaluate enabled periodic automations whose schedule is due. + + Args: + session: Optional database session (for testing) + """ + if session is None: + async with async_session_maker() as db_session: + await self._evaluate_periodic(db_session) + else: + await self._evaluate_periodic(session) + + async def _evaluate_periodic(self, session: AsyncSession) -> None: + """Internal implementation of evaluate_periodic.""" + result = await session.execute( + select(Automation).where( + Automation.enabled.is_(True), + Automation.trigger_type == "periodic", + ) + ) + automations = result.scalars().all() + now = datetime.now(UTC) + + for automation in automations: + schedule = automation.schedule_seconds + if not schedule: + continue + last_run = await self._last_run_time(automation.id, session) + if last_run is not None and now - last_run < timedelta(seconds=schedule): + continue + + conditions = automation.conditions or [] + conditions_passed = all(evaluate_condition({}, condition) for condition in conditions) + + actions_fired = None + skipped = None + if conditions_passed: + skipped = await self._check_circuit_breaker( + automation, now, session + ) or await self._check_rate_limits(automation, now, session) + if skipped is None: + actions_fired = await self._dispatch_actions(automation, {}, session) + + details: dict = { + "event_type": "periodic", + "event": {}, + "automation": _serialize(automation), + } + if skipped is not None: + details["skipped"] = skipped + + session.add( + AutomationRun( + automation_id=automation.id, + trigger="periodic", + matched=True, + conditions_passed=conditions_passed, + actions_fired=actions_fired, + details=details, + ) + ) + + await session.commit() + + async def _check_circuit_breaker( + self, + automation: Automation, + now: datetime, + session: AsyncSession, + ) -> str | None: + """Check the per-automation circuit breaker. Returns skip reason or None. + + The breaker trips when the automation has fired ``failure_threshold`` + times within ``window_seconds``. While tripped, actions are skipped + (``circuit_open``) until the window rolls past (auto-close). + """ + config = automation.circuit_breaker + if not config: + return None + threshold = config.get("failure_threshold") + window = config.get("window_seconds") + if not threshold or not window: + return None + + fired = await self._count_fired_in_window(automation.id, window, now, session) + if fired >= threshold: + return "circuit_open" + return None + + async def _check_rate_limits( + self, + automation: Automation, + now: datetime, + session: AsyncSession, + ) -> str | None: + """Enforce cooldown and max-runs-per-window. Returns a skip reason or None.""" + cooldown = automation.cooldown_seconds + if cooldown is not None: + last_fire = await self._last_fire_time(automation.id, session) + if last_fire is not None and now - last_fire < timedelta(seconds=cooldown): + return "cooldown" + + max_runs = automation.max_runs_per_window + window = automation.window_seconds + if max_runs is not None and window is not None: + fired = await self._count_fired_in_window(automation.id, window, now, session) + if fired >= max_runs: + return "rate_limited" + + return None + + @staticmethod + async def _last_fire_time(automation_id: int, session: AsyncSession) -> datetime | None: + """Return the timestamp of the last run that fired actions (or None).""" + result = await session.execute( + select(AutomationRun.created_at) + .where( + AutomationRun.automation_id == automation_id, + AutomationRun.actions_fired.isnot(None), + ) + .order_by(AutomationRun.created_at.desc()) + .limit(1) + ) + ts = result.scalar_one_or_none() + if ts is None: + return None + if ts.tzinfo is None: + ts = ts.replace(tzinfo=UTC) + return ts + + @staticmethod + async def _last_run_time(automation_id: int, session: AsyncSession) -> datetime | None: + """Return the timestamp of the last run for an automation (any run).""" + result = await session.execute( + select(AutomationRun.created_at) + .where(AutomationRun.automation_id == automation_id) + .order_by(AutomationRun.created_at.desc()) + .limit(1) + ) + ts = result.scalar_one_or_none() + if ts is None: + return None + if ts.tzinfo is None: + ts = ts.replace(tzinfo=UTC) + return ts + + @staticmethod + async def _count_fired_in_window( + automation_id: int, + window_seconds: int, + now: datetime, + session: AsyncSession, + ) -> int: + """Count runs that fired actions within the given window.""" + since = now - timedelta(seconds=window_seconds) + result = await session.execute( + select(func.count()) + .select_from(AutomationRun) + .where( + AutomationRun.automation_id == automation_id, + AutomationRun.actions_fired.isnot(None), + AutomationRun.created_at >= since, + ) + ) + return result.scalar_one() or 0 + + async def _dispatch_actions( + self, + automation: Automation, + event: dict, + session: AsyncSession, + ) -> list[dict]: + """Dispatch an automation's actions. Returns a list of fired actions.""" + fired = [] + for action in automation.actions or []: + action_type = action.get("type") + try: + if action_type == "log": + self._fire_log(automation, event, action) + fired.append({"type": "log"}) + elif action_type == "slack_webhook": + await self._fire_slack_webhook(automation, event, action) + fired.append({"type": "slack_webhook"}) + elif action_type == "webhook": + await self._fire_webhook(automation, event, action) + fired.append({"type": "webhook"}) + elif action_type == "retry_task": + result = await self._fire_retry_task(event, action, session) + fired.append({"type": "retry_task", **result}) + elif action_type == "execute_task": + result = await self._fire_execute_task(event, action) + fired.append({"type": "execute_task", **result}) + elif action_type == "revoke_task": + result = await self._fire_revoke_task(event, action, session) + fired.append({"type": "revoke_task", **result}) + elif action_type == "check_workers_offline": + result = await self._fire_check_workers_offline(action, session) + fired.append({"type": "check_workers_offline", **result}) + else: + logger.warning( + "Unsupported action type %r for automation %s", + action_type, + automation.name, + ) + except Exception: + logger.exception("Action %r failed for automation %s", action_type, automation.name) + fired.append({"type": action_type, "error": "action failed"}) + return fired + + @staticmethod + def _interpolate(value: Any, event: dict) -> Any: + """Substitute ``{event.field}`` references in action params with event values.""" + if isinstance(value, str): + + def _sub(match: re.Match) -> str: + field = match.group(1) + resolved = _resolve_field(event, field) + return str(resolved) if resolved is not None else match.group(0) + + return re.sub(r"\{event\.([\w.]+)\}", _sub, value) + if isinstance(value, list): + return [WorkflowEngine._interpolate(item, event) for item in value] + if isinstance(value, dict): + return {k: WorkflowEngine._interpolate(v, event) for k, v in value.items()} + return value + + @staticmethod + def _fire_log(automation: Automation, event: dict, action: dict) -> None: + """Fire a 'log' action: write a message to the application log.""" + level = (action.get("level") or "info").upper() + message = action.get("message") or ( + f"Automation '{automation.name}' fired on {event.get('type')}" + ) + logger.log(getattr(logging, level, logging.INFO), message) + + @staticmethod + async def _fire_slack_webhook(automation: Automation, event: dict, action: dict) -> None: + """Fire a 'slack_webhook' action.""" + url = action.get("webhook_url") or settings.alert_webhook_url + if not url: + raise ValueError("slack_webhook action requires webhook_url (or ALERT_WEBHOOK_URL)") + + text = action.get("text") or f"Automation '{automation.name}' fired" + fields = [ + {"title": key, "value": WorkflowEngine._interpolate(val, event)} + for key, val in (action.get("fields") or {}).items() + ] + payload = { + "text": text, + "attachments": [{"color": "warning", "fields": fields or None}], + } + await WebhookClient(url).send(payload) + + @staticmethod + async def _fire_webhook(automation: Automation, event: dict, action: dict) -> None: + """Fire a generic 'webhook' action.""" + url = action.get("url") + if not url: + raise ValueError("webhook action requires url") + payload = action.get("payload") or { + "automation": automation.name, + "event_type": event.get("type"), + "event": _safe_snapshot(event), + } + payload = WorkflowEngine._interpolate(payload, event) + await WebhookClient(url).send(payload) + + @staticmethod + async def _fire_retry_task(event: dict, action: dict, session: AsyncSession) -> dict: + """Fire a 'retry_task' action using the event's task id.""" + task_id = action.get("task_id") or event.get("uuid") + if not task_id: + return {"error": "no task_id available"} + result = await retry_task(task_id, session) + return {"error": result["error"]} if "error" in result else {"task_id": task_id} + + @staticmethod + async def _fire_execute_task(event: dict, action: dict) -> dict: + """Fire an 'execute_task' action by name.""" + name = action.get("name") or event.get("name") + if not name: + return {"error": "no task name available"} + result = await execute_task( + name, + args=action.get("args"), + kwargs=action.get("kwargs"), + queue=action.get("queue"), + countdown=action.get("countdown"), + eta=action.get("eta"), + expires=action.get("expires"), + priority=action.get("priority"), + ) + if "error" in result: + return {"error": result["error"]} + return {"task_id": result.get("task_id")} + + @staticmethod + async def _fire_revoke_task(event: dict, action: dict, session: AsyncSession) -> dict: + """Fire a 'revoke_task' action using the event's task id.""" + task_id = action.get("task_id") or event.get("uuid") + if not task_id: + return {"error": "no task_id available"} + terminate = action.get("terminate", False) + result = await revoke_task(task_id, terminate, session) + return {"error": result["error"]} if "error" in result else {"task_id": task_id} + + async def _fire_check_workers_offline( + self, + action: dict, + session: AsyncSession, + ) -> dict: + """Fire a 'check_workers_offline' action: alert for stale/offline workers. + + Scans worker events for workers whose heartbeat has gone stale and sends + a Slack webhook for each newly-offline worker (deduplicated in-memory + until the worker is seen online again). + """ + url = action.get("webhook_url") or settings.alert_webhook_url + if not url: + return {"error": "check_workers_offline requires webhook_url (or ALERT_WEBHOOK_URL)"} + + now = datetime.now(UTC) + offline_timeout = timedelta(seconds=settings.worker_offline_timeout_seconds) + + result = await session.execute(select(WorkerEvent.hostname).distinct()) + hostnames = [row[0] for row in result.all()] + + alerted = 0 + for hostname in hostnames: + if hostname in self._offline_notified: + continue + if await self._worker_stale(session, hostname, now, offline_timeout): + self._offline_notified.add(hostname) + await WebhookClient(url).send(build_worker_offline_payload(hostname)) + alerted += 1 + + return {"workers_alerted": alerted} + + @staticmethod + async def _worker_stale( + session: AsyncSession, + hostname: str, + now: datetime, + offline_timeout: timedelta, + ) -> bool: + """Check whether a worker's latest event indicates it is stale/offline.""" + query = ( + select(WorkerEvent) + .where(WorkerEvent.hostname == hostname) + .order_by(WorkerEvent.timestamp.desc()) + .limit(1) + ) + result = await session.execute(query) + event = result.scalar_one_or_none() + if event is None: + return True + if event.event_type == "offline": + return True + ts = event.timestamp + if ts.tzinfo is None: + ts = ts.replace(tzinfo=UTC) + return now - ts > offline_timeout diff --git a/tests/test_api.py b/tests/test_api.py index de14f7d..a809eb2 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -1173,3 +1173,157 @@ async def test_api_get_task_includes_chain_fields(client: AsyncClient, db_sessio assert response.status_code == 200 data = response.json() assert data["root_id"] == str(task_id) + + +# Automation endpoint tests + + +@pytest.mark.asyncio +async def test_api_list_automations_empty(client: AsyncClient): + """Test GET /api/automations with no data.""" + response = await client.get("/api/automations") + assert response.status_code == 200 + assert response.json() == [] + + +@pytest.mark.asyncio +async def test_api_create_automation(client: AsyncClient): + """Test POST /api/automations.""" + response = await client.post( + "/api/automations", + json={ + "name": "alert-on-failure", + "trigger_type": "event", + "event_type": "task-failed", + "conditions": [{"field": "name", "op": "eq", "value": "payments.charge"}], + "actions": [{"type": "log"}], + }, + ) + assert response.status_code == 200 + data = response.json() + assert data["name"] == "alert-on-failure" + assert data["event_type"] == "task-failed" + + +@pytest.mark.asyncio +async def test_api_create_automation_invalid(client: AsyncClient): + """Test POST /api/automations with an invalid trigger.""" + response = await client.post( + "/api/automations", + json={"name": "bad", "trigger_type": "cron", "event_type": "task-failed"}, + ) + assert response.status_code == 400 + assert "trigger_type" in response.json()["detail"] + + +@pytest.mark.asyncio +async def test_api_get_automation(client: AsyncClient, db_session: AsyncSession): + """Test GET /api/automations/{id}.""" + from taskowl.models import Automation + + db_session.add(Automation(name="getme", trigger_type="event", event_type="task-failed")) + await db_session.commit() + + response = await client.get("/api/automations/1") + assert response.status_code == 200 + assert response.json()["name"] == "getme" + + +@pytest.mark.asyncio +async def test_api_get_automation_not_found(client: AsyncClient): + """Test GET /api/automations/{id} for a non-existent automation.""" + response = await client.get("/api/automations/999") + assert response.status_code == 404 + + +@pytest.mark.asyncio +async def test_api_update_automation(client: AsyncClient, db_session: AsyncSession): + """Test PUT /api/automations/{id}.""" + from taskowl.models import Automation + + db_session.add(Automation(name="update-me", trigger_type="event", event_type="task-failed")) + await db_session.commit() + + response = await client.put( + "/api/automations/1", json={"enabled": False, "cooldown_seconds": 60} + ) + assert response.status_code == 200 + data = response.json() + assert data["enabled"] is False + assert data["cooldown_seconds"] == 60 + + +@pytest.mark.asyncio +async def test_api_delete_automation(client: AsyncClient, db_session: AsyncSession): + """Test DELETE /api/automations/{id}.""" + from taskowl.models import Automation + + db_session.add(Automation(name="delete-me", trigger_type="event", event_type="task-failed")) + await db_session.commit() + + response = await client.delete("/api/automations/1") + assert response.status_code == 200 + assert response.json()["status"] == "success" + + response = await client.get("/api/automations") + assert response.json() == [] + + +@pytest.mark.asyncio +async def test_api_toggle_automation(client: AsyncClient, db_session: AsyncSession): + """Test POST /api/automations/{id}/toggle.""" + from taskowl.models import Automation + + db_session.add(Automation(name="toggle-me", trigger_type="event", event_type="task-failed")) + await db_session.commit() + + response = await client.post("/api/automations/1/toggle") + assert response.status_code == 200 + assert response.json()["enabled"] is False + + +@pytest.mark.asyncio +async def test_api_list_automation_runs(client: AsyncClient, db_session: AsyncSession): + """Test GET /api/automations/{id}/runs.""" + from taskowl.models import Automation, AutomationRun + + db_session.add(Automation(name="with-runs", trigger_type="event", event_type="task-failed")) + await db_session.commit() + db_session.add(AutomationRun(automation_id=1, trigger="task-failed", matched=True)) + await db_session.commit() + + response = await client.get("/api/automations/1/runs") + assert response.status_code == 200 + data = response.json() + assert len(data) == 1 + assert data[0]["matched"] is True + + +@pytest.mark.asyncio +async def test_api_get_automation_status(client: AsyncClient, db_session: AsyncSession): + """Test GET /api/automations/{id}/status.""" + from taskowl.models import Automation + + db_session.add( + Automation( + name="status-api", + trigger_type="event", + event_type="task-failed", + circuit_breaker={"failure_threshold": 2, "window_seconds": 60}, + ) + ) + await db_session.commit() + + response = await client.get("/api/automations/1/status") + assert response.status_code == 200 + data = response.json() + assert data["name"] == "status-api" + assert data["circuit_state"] == "closed" + assert data["last_run"] is None + + +@pytest.mark.asyncio +async def test_api_get_automation_status_not_found(client: AsyncClient): + """Test GET /api/automations/{id}/status for a non-existent automation.""" + response = await client.get("/api/automations/999/status") + assert response.status_code == 404 diff --git a/tests/test_automations.py b/tests/test_automations.py new file mode 100644 index 0000000..1943448 --- /dev/null +++ b/tests/test_automations.py @@ -0,0 +1,364 @@ +"""Tests for workflow automation CRUD functions.""" + +from unittest.mock import patch + +import pytest +from sqlalchemy.ext.asyncio import AsyncSession + +from taskowl.automations import ( + create_automation, + delete_automation, + get_automation, + get_automation_status, + list_automation_runs, + list_automations, + seed_builtin_automations, + toggle_automation, + update_automation, +) +from taskowl.models import AutomationRun + + +@pytest.mark.asyncio +async def test_create_automation_event(db_session: AsyncSession): + """Test creating an event-triggered automation.""" + result = await create_automation( + { + "name": "alert-on-failure", + "trigger_type": "event", + "event_type": "task-failed", + "conditions": [{"field": "name", "op": "eq", "value": "payments.charge"}], + "actions": [{"type": "log"}], + }, + session=db_session, + ) + + assert "error" not in result + assert result["name"] == "alert-on-failure" + assert result["trigger_type"] == "event" + assert result["event_type"] == "task-failed" + assert result["enabled"] is True + + +@pytest.mark.asyncio +async def test_create_automation_periodic(db_session: AsyncSession): + """Test creating a periodic automation.""" + result = await create_automation( + { + "name": "stale-worker-check", + "trigger_type": "periodic", + "schedule_seconds": 30, + "actions": [{"type": "log"}], + }, + session=db_session, + ) + + assert "error" not in result + assert result["trigger_type"] == "periodic" + assert result["schedule_seconds"] == 30 + + +@pytest.mark.asyncio +async def test_create_automation_duplicate_name(db_session: AsyncSession): + """Test creating an automation with a duplicate name.""" + await create_automation( + {"name": "dup", "trigger_type": "event", "event_type": "task-failed"}, + session=db_session, + ) + result = await create_automation( + {"name": "dup", "trigger_type": "event", "event_type": "task-failed"}, + session=db_session, + ) + + assert "error" in result + assert "already exists" in result["error"] + + +@pytest.mark.asyncio +async def test_create_automation_missing_name(db_session: AsyncSession): + """Test creating an automation without a name.""" + result = await create_automation( + {"trigger_type": "event", "event_type": "task-failed"}, + session=db_session, + ) + assert "error" in result + assert "name is required" in result["error"] + + +@pytest.mark.asyncio +async def test_create_automation_invalid_trigger(db_session: AsyncSession): + """Test creating an automation with an invalid trigger type.""" + result = await create_automation( + {"name": "bad", "trigger_type": "cron", "event_type": "task-failed"}, + session=db_session, + ) + assert "error" in result + assert "trigger_type" in result["error"] + + +@pytest.mark.asyncio +async def test_create_automation_event_missing_type(db_session: AsyncSession): + """Test creating an event automation without an event type.""" + result = await create_automation( + {"name": "bad", "trigger_type": "event"}, + session=db_session, + ) + assert "error" in result + assert "event_type is required" in result["error"] + + +@pytest.mark.asyncio +async def test_create_automation_periodic_bad_schedule(db_session: AsyncSession): + """Test creating a periodic automation with a bad schedule.""" + result = await create_automation( + {"name": "bad", "trigger_type": "periodic", "schedule_seconds": 0}, + session=db_session, + ) + assert "error" in result + assert "schedule_seconds" in result["error"] + + +@pytest.mark.asyncio +async def test_get_automation(db_session: AsyncSession): + """Test getting an automation by id.""" + created = await create_automation( + {"name": "getme", "trigger_type": "event", "event_type": "task-failed"}, + session=db_session, + ) + result = await get_automation(created["id"], session=db_session) + assert result["id"] == created["id"] + assert result["name"] == "getme" + + +@pytest.mark.asyncio +async def test_get_automation_not_found(db_session: AsyncSession): + """Test getting a non-existent automation.""" + result = await get_automation(999, session=db_session) + assert "error" in result + assert "not found" in result["error"] + + +@pytest.mark.asyncio +async def test_list_automations(db_session: AsyncSession): + """Test listing automations.""" + await create_automation( + {"name": "a", "trigger_type": "event", "event_type": "task-failed"}, + session=db_session, + ) + await create_automation( + {"name": "b", "trigger_type": "periodic", "schedule_seconds": 30}, + session=db_session, + ) + result = await list_automations(session=db_session) + assert len(result) == 2 + + +@pytest.mark.asyncio +async def test_list_automations_filter_enabled(db_session: AsyncSession): + """Test listing automations filtered by enabled state.""" + a = await create_automation( + {"name": "a", "trigger_type": "event", "event_type": "task-failed"}, + session=db_session, + ) + await toggle_automation(a["id"], session=db_session) + + result = await list_automations(enabled=True, session=db_session) + assert all(item["enabled"] for item in result) + assert len(result) == 0 + + result = await list_automations(enabled=False, session=db_session) + assert len(result) == 1 + assert result[0]["name"] == "a" + + +@pytest.mark.asyncio +async def test_update_automation(db_session: AsyncSession): + """Test updating an automation.""" + created = await create_automation( + {"name": "update-me", "trigger_type": "event", "event_type": "task-failed"}, + session=db_session, + ) + result = await update_automation( + created["id"], + {"enabled": False, "cooldown_seconds": 60}, + session=db_session, + ) + assert result["enabled"] is False + assert result["cooldown_seconds"] == 60 + + +@pytest.mark.asyncio +async def test_update_automation_not_found(db_session: AsyncSession): + """Test updating a non-existent automation.""" + result = await update_automation(999, {"enabled": False}, session=db_session) + assert "error" in result + assert "not found" in result["error"] + + +@pytest.mark.asyncio +async def test_delete_automation(db_session: AsyncSession): + """Test deleting an automation.""" + created = await create_automation( + {"name": "delete-me", "trigger_type": "event", "event_type": "task-failed"}, + session=db_session, + ) + db_session.add(AutomationRun(automation_id=created["id"], trigger="task-failed", matched=True)) + await db_session.commit() + + result = await delete_automation(created["id"], session=db_session) + assert result["status"] == "success" + + remaining = await list_automations(session=db_session) + assert remaining == [] + runs = await list_automation_runs(created["id"], session=db_session) + assert runs == [] + + +@pytest.mark.asyncio +async def test_toggle_automation(db_session: AsyncSession): + """Test toggling an automation's enabled state.""" + created = await create_automation( + {"name": "toggle-me", "trigger_type": "event", "event_type": "task-failed"}, + session=db_session, + ) + result = await toggle_automation(created["id"], session=db_session) + assert result["enabled"] is False + result = await toggle_automation(created["id"], session=db_session) + assert result["enabled"] is True + + +@pytest.mark.asyncio +async def test_list_automation_runs(db_session: AsyncSession): + """Test listing automation runs.""" + created = await create_automation( + {"name": "runs", "trigger_type": "event", "event_type": "task-failed"}, + session=db_session, + ) + db_session.add_all( + [ + AutomationRun(automation_id=created["id"], trigger="task-failed", matched=True), + AutomationRun(automation_id=created["id"], trigger="task-failed", matched=False), + ] + ) + await db_session.commit() + + runs = await list_automation_runs(created["id"], session=db_session) + assert len(runs) == 2 + assert runs[0]["matched"] is False # newest first + + +@pytest.mark.asyncio +async def test_list_automation_runs_no_runs(db_session: AsyncSession): + """Test listing automation runs when there are none.""" + created = await create_automation( + {"name": "no-runs", "trigger_type": "event", "event_type": "task-failed"}, + session=db_session, + ) + runs = await list_automation_runs(created["id"], session=db_session) + assert runs == [] + + +@pytest.mark.asyncio +async def test_seed_builtin_automations(db_session: AsyncSession): + """Test seeding built-in alert automations from env settings.""" + with patch( + "taskowl.config.settings", + alert_webhook_url="http://hooks.test/x", + alert_on_task_failed=True, + alert_on_worker_offline=True, + alert_slow_task_seconds=30.0, + alert_worker_check_seconds=60, + ): + created = await seed_builtin_automations(session=db_session) + + names = {a["name"] for a in created} + assert { + "alert-task-failed", + "alert-slow-task", + "alert-worker-offline", + "alert-worker-offline-sweep", + } <= names + + # Idempotent: seeding again creates nothing new + with patch( + "taskowl.config.settings", + alert_webhook_url="http://hooks.test/x", + alert_on_task_failed=True, + alert_on_worker_offline=True, + alert_slow_task_seconds=30.0, + alert_worker_check_seconds=60, + ): + second = await seed_builtin_automations(session=db_session) + assert second == [] + + +@pytest.mark.asyncio +async def test_seed_builtin_automations_no_webhook(db_session: AsyncSession): + """Test seeding does nothing without a webhook URL.""" + with patch("taskowl.config.settings", alert_webhook_url=None): + created = await seed_builtin_automations(session=db_session) + assert created == [] + + +@pytest.mark.asyncio +async def test_get_automation_status_circuit_open(db_session: AsyncSession): + """Test automation status reports circuit state from the latest run.""" + created = await create_automation( + { + "name": "status-demo", + "trigger_type": "event", + "event_type": "task-failed", + "actions": [{"type": "log"}], + "circuit_breaker": {"failure_threshold": 2, "window_seconds": 60}, + }, + session=db_session, + ) + db_session.add( + AutomationRun( + automation_id=created["id"], + trigger="task-failed", + matched=True, + actions_fired=None, + details={"skipped": "circuit_open"}, + ) + ) + await db_session.commit() + + result = await get_automation_status(created["id"], session=db_session) + assert result["circuit_state"] == "open" + assert result["last_run"]["details"]["skipped"] == "circuit_open" + + +@pytest.mark.asyncio +async def test_get_automation_status_circuit_closed(db_session: AsyncSession): + """Test automation status reports closed circuit without a skip.""" + created = await create_automation( + { + "name": "status-closed", + "trigger_type": "event", + "event_type": "task-failed", + "actions": [{"type": "log"}], + "circuit_breaker": {"failure_threshold": 2, "window_seconds": 60}, + }, + session=db_session, + ) + db_session.add( + AutomationRun( + automation_id=created["id"], + trigger="task-failed", + matched=True, + actions_fired=[{"type": "log"}], + ) + ) + await db_session.commit() + + result = await get_automation_status(created["id"], session=db_session) + assert result["circuit_state"] == "closed" + + +@pytest.mark.asyncio +async def test_get_automation_status_not_found(db_session: AsyncSession): + """Test automation status for a non-existent automation.""" + result = await get_automation_status(999, session=db_session) + assert "error" in result + assert "not found" in result["error"] diff --git a/tests/test_metrics.py b/tests/test_metrics.py index 8cb9f70..ac7d7b9 100644 --- a/tests/test_metrics.py +++ b/tests/test_metrics.py @@ -8,7 +8,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from taskowl.metrics import generate_metrics -from taskowl.models import TaskEvent, WorkerEvent +from taskowl.models import AutomationRun, TaskEvent, WorkerEvent @pytest.mark.asyncio @@ -102,6 +102,34 @@ async def test_metrics_worker_status_offline(db_session: AsyncSession): assert 'taskowl_worker_status{worker="celery@w2"} 0.0' in text +@pytest.mark.asyncio +async def test_metrics_automation_runs(db_session: AsyncSession): + """Automation fired and skipped runs should be counted.""" + db_session.add( + AutomationRun( + automation_id=1, + trigger="task-failed", + matched=True, + actions_fired=[{"type": "log"}], + ) + ) + db_session.add( + AutomationRun( + automation_id=1, + trigger="task-failed", + matched=True, + actions_fired=None, + details={"skipped": "circuit_open"}, + ) + ) + await db_session.commit() + + text = (await generate_metrics(db_session)).decode() + + assert 'taskowl_automation_fired_total{automation_id="1",trigger="task-failed"} 1.0' in text + assert 'taskowl_automation_skipped_total{automation_id="1",reason="circuit_open"} 1.0' in text + + @pytest.mark.asyncio async def test_api_metrics_endpoint(client: AsyncClient): """GET /metrics should return 200 with Prometheus content type.""" diff --git a/tests/test_workflow.py b/tests/test_workflow.py new file mode 100644 index 0000000..613cbf8 --- /dev/null +++ b/tests/test_workflow.py @@ -0,0 +1,879 @@ +"""Tests for the workflow automation evaluation engine.""" + +from datetime import UTC, datetime +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from taskowl.automations import create_automation +from taskowl.models import AutomationRun +from taskowl.workflow import WorkflowEngine, _resolve_field, evaluate_condition + + +@pytest.mark.asyncio +async def test_resolve_field_top_level(): + """Test resolving a top-level event field.""" + event = {"type": "task-failed", "name": "payments.charge", "runtime": 1.5} + assert _resolve_field(event, "name") == "payments.charge" + assert _resolve_field(event, "runtime") == 1.5 + assert _resolve_field(event, "missing") is None + + +@pytest.mark.asyncio +async def test_resolve_field_dotted(): + """Test resolving a dotted field path.""" + event = {"request": {"id": "abc", "name": "nested.task"}} + assert _resolve_field(event, "request.id") == "abc" + assert _resolve_field(event, "request.name") == "nested.task" + assert _resolve_field(event, "request.missing") is None + + +def test_condition_eq(): + """Test eq condition.""" + event = {"name": "payments.charge"} + assert evaluate_condition(event, {"field": "name", "op": "eq", "value": "payments.charge"}) + assert not evaluate_condition(event, {"field": "name", "op": "eq", "value": "other"}) + + +def test_condition_neq(): + """Test neq condition.""" + event = {"name": "payments.charge"} + assert evaluate_condition(event, {"field": "name", "op": "neq", "value": "other"}) + assert not evaluate_condition(event, {"field": "name", "op": "neq", "value": "payments.charge"}) + + +def test_condition_gt_gte_lt_lte(): + """Test numeric comparison conditions.""" + event = {"retries": 3, "runtime": 10.5} + assert evaluate_condition(event, {"field": "retries", "op": "gt", "value": 2}) + assert not evaluate_condition(event, {"field": "retries", "op": "gt", "value": 3}) + assert evaluate_condition(event, {"field": "retries", "op": "gte", "value": 3}) + assert evaluate_condition(event, {"field": "runtime", "op": "lt", "value": 20}) + assert evaluate_condition(event, {"field": "runtime", "op": "lte", "value": 10.5}) + assert not evaluate_condition(event, {"field": "runtime", "op": "lt", "value": 10.5}) + + +def test_condition_contains(): + """Test contains condition.""" + event = {"name": "payments.charge", "hosts": ["a", "b"]} + assert evaluate_condition(event, {"field": "name", "op": "contains", "value": "charge"}) + assert not evaluate_condition(event, {"field": "name", "op": "contains", "value": "refund"}) + assert evaluate_condition(event, {"field": "hosts", "op": "contains", "value": "a"}) + + +def test_condition_matches(): + """Test matches (regex) condition.""" + event = {"exception": "ValueError: bad input"} + assert evaluate_condition( + event, {"field": "exception", "op": "matches", "value": r"ValueError"} + ) + assert not evaluate_condition( + event, {"field": "exception", "op": "matches", "value": r"TimeoutError"} + ) + + +def test_condition_in(): + """Test in condition.""" + event = {"queue": "high"} + assert evaluate_condition(event, {"field": "queue", "op": "in", "value": ["high", "default"]}) + assert not evaluate_condition(event, {"field": "queue", "op": "in", "value": ["low"]}) + + +def test_condition_exists(): + """Test exists condition.""" + event = {"exception": "boom"} + assert evaluate_condition(event, {"field": "exception", "op": "exists"}) + assert not evaluate_condition(event, {"field": "traceback", "op": "exists"}) + + +def test_condition_invalid(): + """Test an invalid condition returns False.""" + event = {"name": "x"} + assert not evaluate_condition(event, {"field": "name", "op": "bogus", "value": 1}) + assert not evaluate_condition(event, {"field": None, "op": "eq", "value": 1}) + + +@pytest.mark.asyncio +async def test_evaluate_event_matching_trigger_writes_run(db_session: AsyncSession): + """Test a matching trigger + passing conditions writes a run.""" + await create_automation( + { + "name": "alert-failures", + "trigger_type": "event", + "event_type": "task-failed", + "conditions": [{"field": "name", "op": "eq", "value": "payments.charge"}], + "actions": [{"type": "log"}], + }, + session=db_session, + ) + + engine = WorkflowEngine() + await engine.evaluate_event( + "task-failed", + {"type": "task-failed", "name": "payments.charge", "hostname": "w1"}, + session=db_session, + ) + + result = await db_session.execute(select(AutomationRun)) + runs = result.scalars().all() + assert len(runs) == 1 + assert runs[0].trigger == "task-failed" + assert runs[0].matched is True + assert runs[0].conditions_passed is True + assert runs[0].actions_fired == [{"type": "log"}] + # metadata-only snapshot (no args/kwargs/results) + assert "hostname" in runs[0].details["event"] + assert runs[0].details["event"].get("name") == "payments.charge" + + +@pytest.mark.asyncio +async def test_evaluate_event_conditions_fail(db_session: AsyncSession): + """Test a matching trigger with failing conditions records a run with no actions.""" + await create_automation( + { + "name": "alert-failures", + "trigger_type": "event", + "event_type": "task-failed", + "conditions": [{"field": "name", "op": "eq", "value": "payments.charge"}], + "actions": [{"type": "log"}], + }, + session=db_session, + ) + + engine = WorkflowEngine() + await engine.evaluate_event( + "task-failed", + {"type": "task-failed", "name": "other.task"}, + session=db_session, + ) + + result = await db_session.execute(select(AutomationRun)) + run = result.scalars().one() + assert run.matched is True + assert run.conditions_passed is False + assert run.actions_fired is None + + +@pytest.mark.asyncio +async def test_evaluate_event_non_matching_trigger(db_session: AsyncSession): + """Test an event that doesn't match any automation writes nothing.""" + await create_automation( + { + "name": "alert-failures", + "trigger_type": "event", + "event_type": "task-failed", + "actions": [{"type": "log"}], + }, + session=db_session, + ) + + engine = WorkflowEngine() + await engine.evaluate_event( + "task-succeeded", + {"type": "task-succeeded", "name": "payments.charge"}, + session=db_session, + ) + + result = await db_session.execute(select(AutomationRun)) + assert result.scalars().all() == [] + + +@pytest.mark.asyncio +async def test_evaluate_event_disabled_automation_skipped(db_session: AsyncSession): + """Test disabled automations are not evaluated.""" + from taskowl.automations import toggle_automation + + created = await create_automation( + { + "name": "disabled", + "trigger_type": "event", + "event_type": "task-failed", + "actions": [{"type": "log"}], + }, + session=db_session, + ) + await toggle_automation(created["id"], session=db_session) + + engine = WorkflowEngine() + await engine.evaluate_event( + "task-failed", {"type": "task-failed", "name": "x"}, session=db_session + ) + + result = await db_session.execute(select(AutomationRun)) + assert result.scalars().all() == [] + + +@pytest.mark.asyncio +async def test_evaluate_event_multiple_automations(db_session: AsyncSession): + """Test multiple automations matching the same event each write a run.""" + await create_automation( + { + "name": "a", + "trigger_type": "event", + "event_type": "task-failed", + "actions": [{"type": "log"}], + }, + session=db_session, + ) + await create_automation( + { + "name": "b", + "trigger_type": "event", + "event_type": "task-failed", + "actions": [{"type": "log"}], + }, + session=db_session, + ) + + engine = WorkflowEngine() + await engine.evaluate_event( + "task-failed", {"type": "task-failed", "name": "x"}, session=db_session + ) + + result = await db_session.execute(select(AutomationRun)) + assert len(result.scalars().all()) == 2 + + +@pytest.mark.asyncio +async def test_evaluate_event_unsupported_action_logged_not_fired(db_session: AsyncSession): + """Test unsupported action types are skipped without crashing.""" + await create_automation( + { + "name": "future-action", + "trigger_type": "event", + "event_type": "task-failed", + "actions": [{"type": "does_not_exist"}], + }, + session=db_session, + ) + + engine = WorkflowEngine() + await engine.evaluate_event( + "task-failed", {"type": "task-failed", "name": "x"}, session=db_session + ) + + result = await db_session.execute(select(AutomationRun)) + run = result.scalars().one() + assert run.conditions_passed is True + assert run.actions_fired == [] + + +@pytest.mark.asyncio +async def test_evaluate_event_snapshot_omits_sensitive_fields(db_session: AsyncSession): + """Test the run snapshot never includes args/kwargs/result/traceback.""" + await create_automation( + { + "name": "safe", + "trigger_type": "event", + "event_type": "task-failed", + "actions": [{"type": "log"}], + }, + session=db_session, + ) + + engine = WorkflowEngine() + await engine.evaluate_event( + "task-failed", + { + "type": "task-failed", + "name": "x", + "args": ["secret"], + "kwargs": {"key": "secret"}, + "result": {"data": "secret"}, + "traceback": "Traceback...", + "exception": "ValueError", + }, + session=db_session, + ) + + result = await db_session.execute(select(AutomationRun)) + snapshot = result.scalars().one().details["event"] + assert "exception" in snapshot + for sensitive in ("args", "kwargs", "result", "traceback"): + assert sensitive not in snapshot + + +@pytest.mark.asyncio +async def test_interpolate_event_fields(): + """Test {event.field} interpolation in action params.""" + event = {"uuid": "abc-123", "name": "payments.charge", "retries": 2} + result = WorkflowEngine._interpolate( + {"task_id": "{event.uuid}", "label": "processing {event.name}"}, event + ) + assert result["task_id"] == "abc-123" + assert result["label"] == "processing payments.charge" + + +@pytest.mark.asyncio +async def test_fire_slack_webhook(db_session: AsyncSession): + """Test slack_webhook action sends a payload via WebhookClient.""" + await create_automation( + { + "name": "slack-failures", + "trigger_type": "event", + "event_type": "task-failed", + "actions": [ + { + "type": "slack_webhook", + "webhook_url": "http://hooks.test/x", + "text": "A task failed", + "fields": {"Task": "{event.name}", "Task ID": "{event.uuid}"}, + } + ], + }, + session=db_session, + ) + + with patch("taskowl.workflow.WebhookClient") as mock_client_cls: + mock_client = MagicMock() + mock_client.send = AsyncMock() + mock_client_cls.return_value = mock_client + + engine = WorkflowEngine() + await engine.evaluate_event( + "task-failed", + {"type": "task-failed", "name": "payments.charge", "uuid": "abc"}, + session=db_session, + ) + + mock_client_cls.assert_called_once_with("http://hooks.test/x") + mock_client.send.assert_awaited_once() + payload = mock_client.send.call_args.args[0] + assert payload["text"] == "A task failed" + assert payload["attachments"][0]["fields"][0]["value"] == "payments.charge" + + +@pytest.mark.asyncio +async def test_fire_slack_webhook_no_url(db_session: AsyncSession): + """Test slack_webhook without a URL records an error.""" + await create_automation( + { + "name": "slack-no-url", + "trigger_type": "event", + "event_type": "task-failed", + "actions": [{"type": "slack_webhook"}], + }, + session=db_session, + ) + + engine = WorkflowEngine() + await engine.evaluate_event( + "task-failed", {"type": "task-failed", "name": "x"}, session=db_session + ) + + result = await db_session.execute(select(AutomationRun)) + run = result.scalars().one() + assert run.actions_fired[0]["type"] == "slack_webhook" + assert "error" in run.actions_fired[0] + + +@pytest.mark.asyncio +async def test_fire_generic_webhook(db_session: AsyncSession): + """Test generic webhook action sends the payload.""" + await create_automation( + { + "name": "generic-webhook", + "trigger_type": "event", + "event_type": "task-failed", + "actions": [ + { + "type": "webhook", + "url": "http://hooks.test/y", + "payload": {"task": "{event.name}"}, + } + ], + }, + session=db_session, + ) + + with patch("taskowl.workflow.WebhookClient") as mock_client_cls: + mock_client = MagicMock() + mock_client.send = AsyncMock() + mock_client_cls.return_value = mock_client + + engine = WorkflowEngine() + await engine.evaluate_event( + "task-failed", + {"type": "task-failed", "name": "payments.charge"}, + session=db_session, + ) + + mock_client_cls.assert_called_once_with("http://hooks.test/y") + mock_client.send.assert_awaited_once() + payload = mock_client.send.call_args.args[0] + assert payload["task"] == "payments.charge" + + +@pytest.mark.asyncio +async def test_fire_retry_task(db_session: AsyncSession): + """Test retry_task action dispatches to the actions module.""" + await create_automation( + { + "name": "retry-failures", + "trigger_type": "event", + "event_type": "task-failed", + "actions": [{"type": "retry_task"}], + }, + session=db_session, + ) + + with patch("taskowl.workflow.retry_task") as mock_retry: + mock_retry.return_value = {"status": "success", "new_task_id": "new-1"} + + engine = WorkflowEngine() + await engine.evaluate_event( + "task-failed", + {"type": "task-failed", "name": "x", "uuid": "abc"}, + session=db_session, + ) + + mock_retry.assert_awaited_once_with("abc", db_session) + result = await db_session.execute(select(AutomationRun)) + run = result.scalars().one() + assert run.actions_fired == [{"type": "retry_task", "task_id": "abc"}] + + +@pytest.mark.asyncio +async def test_fire_execute_task(db_session: AsyncSession): + """Test execute_task action dispatches to the actions module.""" + await create_automation( + { + "name": "execute-cleanup", + "trigger_type": "event", + "event_type": "task-failed", + "actions": [{"type": "execute_task", "name": "cleanup.run", "kwargs": {"x": 1}}], + }, + session=db_session, + ) + + with patch("taskowl.workflow.execute_task") as mock_execute: + mock_execute.return_value = {"status": "success", "task_id": "new-1"} + + engine = WorkflowEngine() + await engine.evaluate_event( + "task-failed", {"type": "task-failed", "name": "x", "uuid": "abc"}, session=db_session + ) + + mock_execute.assert_awaited_once_with( + "cleanup.run", + args=None, + kwargs={"x": 1}, + queue=None, + countdown=None, + eta=None, + expires=None, + priority=None, + ) + result = await db_session.execute(select(AutomationRun)) + run = result.scalars().one() + assert run.actions_fired == [{"type": "execute_task", "task_id": "new-1"}] + + +@pytest.mark.asyncio +async def test_fire_revoke_task(db_session: AsyncSession): + """Test revoke_task action dispatches to the actions module.""" + await create_automation( + { + "name": "revoke-failures", + "trigger_type": "event", + "event_type": "task-failed", + "actions": [{"type": "revoke_task"}], + }, + session=db_session, + ) + + with patch("taskowl.workflow.revoke_task") as mock_revoke: + mock_revoke.return_value = {"status": "success"} + + engine = WorkflowEngine() + await engine.evaluate_event( + "task-failed", + {"type": "task-failed", "name": "x", "uuid": "abc"}, + session=db_session, + ) + + mock_revoke.assert_awaited_once_with("abc", False, db_session) + result = await db_session.execute(select(AutomationRun)) + run = result.scalars().one() + assert run.actions_fired == [{"type": "revoke_task", "task_id": "abc"}] + + +@pytest.mark.asyncio +async def test_cooldown_blocks_second_fire(db_session: AsyncSession): + """Test cooldown_seconds prevents re-firing within the window.""" + await create_automation( + { + "name": "cooldown-demo", + "trigger_type": "event", + "event_type": "task-failed", + "actions": [{"type": "log"}], + "cooldown_seconds": 60, + }, + session=db_session, + ) + + engine = WorkflowEngine() + # First event fires + await engine.evaluate_event( + "task-failed", {"type": "task-failed", "name": "x"}, session=db_session + ) + # Second event within cooldown is skipped + await engine.evaluate_event( + "task-failed", {"type": "task-failed", "name": "x"}, session=db_session + ) + + result = await db_session.execute(select(AutomationRun)) + runs = result.scalars().all() + assert len(runs) == 2 + assert runs[0].actions_fired == [{"type": "log"}] + assert runs[1].actions_fired is None + assert runs[1].details["skipped"] == "cooldown" + + +@pytest.mark.asyncio +async def test_cooldown_expires(db_session: AsyncSession): + """Test cooldown no longer blocks once the window has passed.""" + from datetime import timedelta + + from taskowl.models import AutomationRun as Run + + await create_automation( + { + "name": "cooldown-expiry", + "trigger_type": "event", + "event_type": "task-failed", + "actions": [{"type": "log"}], + "cooldown_seconds": 10, + }, + session=db_session, + ) + + engine = WorkflowEngine() + await engine.evaluate_event( + "task-failed", {"type": "task-failed", "name": "x"}, session=db_session + ) + + # Backdate the run so the cooldown has expired + run = (await db_session.execute(select(Run))).scalars().one() + run.created_at = datetime.now(UTC) - timedelta(seconds=30) + await db_session.commit() + + await engine.evaluate_event( + "task-failed", {"type": "task-failed", "name": "x"}, session=db_session + ) + + runs = (await db_session.execute(select(Run))).scalars().all() + assert runs[1].actions_fired == [{"type": "log"}] + + +@pytest.mark.asyncio +async def test_rate_limit_blocks_after_max(db_session: AsyncSession): + """Test max_runs_per_window blocks once the cap is reached.""" + await create_automation( + { + "name": "rate-limit-demo", + "trigger_type": "event", + "event_type": "task-failed", + "actions": [{"type": "log"}], + "max_runs_per_window": 2, + "window_seconds": 60, + }, + session=db_session, + ) + + engine = WorkflowEngine() + for _ in range(3): + await engine.evaluate_event( + "task-failed", {"type": "task-failed", "name": "x"}, session=db_session + ) + + result = await db_session.execute(select(AutomationRun)) + runs = result.scalars().all() + assert len(runs) == 3 + assert runs[0].actions_fired == [{"type": "log"}] + assert runs[1].actions_fired == [{"type": "log"}] + assert runs[2].actions_fired is None + assert runs[2].details["skipped"] == "rate_limited" + + +@pytest.mark.asyncio +async def test_rate_limit_no_skip_within_budget(db_session: AsyncSession): + """Test firings within the budget are not rate-limited.""" + await create_automation( + { + "name": "rate-budget", + "trigger_type": "event", + "event_type": "task-failed", + "actions": [{"type": "log"}], + "max_runs_per_window": 5, + "window_seconds": 60, + }, + session=db_session, + ) + + engine = WorkflowEngine() + for _ in range(3): + await engine.evaluate_event( + "task-failed", {"type": "task-failed", "name": "x"}, session=db_session + ) + + runs = (await db_session.execute(select(AutomationRun))).scalars().all() + assert all(run.actions_fired == [{"type": "log"}] for run in runs) + + +@pytest.mark.asyncio +async def test_circuit_breaker_trips_after_threshold(db_session: AsyncSession): + """Test the circuit breaker skips actions once the failure threshold is hit.""" + await create_automation( + { + "name": "breaker-demo", + "trigger_type": "event", + "event_type": "task-failed", + "actions": [{"type": "log"}], + "circuit_breaker": {"failure_threshold": 2, "window_seconds": 60}, + }, + session=db_session, + ) + + engine = WorkflowEngine() + # Fire twice (reaches threshold) + for _ in range(2): + await engine.evaluate_event( + "task-failed", {"type": "task-failed", "name": "x"}, session=db_session + ) + # Third should be circuit_open + await engine.evaluate_event( + "task-failed", {"type": "task-failed", "name": "x"}, session=db_session + ) + + runs = (await db_session.execute(select(AutomationRun))).scalars().all() + assert runs[0].actions_fired == [{"type": "log"}] + assert runs[1].actions_fired == [{"type": "log"}] + assert runs[2].actions_fired is None + assert runs[2].details["skipped"] == "circuit_open" + + +@pytest.mark.asyncio +async def test_circuit_breaker_auto_closes_after_window(db_session: AsyncSession): + """Test the circuit breaker reopens once the window has passed.""" + from datetime import timedelta + + from taskowl.models import AutomationRun as Run + + await create_automation( + { + "name": "breaker-closes", + "trigger_type": "event", + "event_type": "task-failed", + "actions": [{"type": "log"}], + "circuit_breaker": {"failure_threshold": 2, "window_seconds": 10}, + }, + session=db_session, + ) + + engine = WorkflowEngine() + for _ in range(2): + await engine.evaluate_event( + "task-failed", {"type": "task-failed", "name": "x"}, session=db_session + ) + + # Backdate the fired runs so the window has passed + runs = (await db_session.execute(select(Run))).scalars().all() + old = datetime.now(UTC) - timedelta(seconds=30) + for run in runs: + run.created_at = old + await db_session.commit() + + await engine.evaluate_event( + "task-failed", {"type": "task-failed", "name": "x"}, session=db_session + ) + + runs = (await db_session.execute(select(Run))).scalars().all() + assert runs[2].actions_fired == [{"type": "log"}] + + +@pytest.mark.asyncio +async def test_periodic_automation_fires_on_schedule(db_session: AsyncSession): + """Test a periodic automation fires when its schedule is due.""" + await create_automation( + { + "name": "heartbeat", + "trigger_type": "periodic", + "schedule_seconds": 5, + "actions": [{"type": "log"}], + }, + session=db_session, + ) + + engine = WorkflowEngine() + # First evaluation: no prior run -> fires + await engine.evaluate_periodic(session=db_session) + # Second evaluation: within schedule -> not due, no run recorded + await engine.evaluate_periodic(session=db_session) + + runs = (await db_session.execute(select(AutomationRun))).scalars().all() + assert len(runs) == 1 + assert runs[0].trigger == "periodic" + assert runs[0].conditions_passed is True + assert runs[0].actions_fired == [{"type": "log"}] + + +@pytest.mark.asyncio +async def test_periodic_automation_fires_again_after_schedule(db_session: AsyncSession): + """Test a periodic automation fires again once the schedule has elapsed.""" + from datetime import timedelta + + from taskowl.models import AutomationRun as Run + + await create_automation( + { + "name": "heartbeat-again", + "trigger_type": "periodic", + "schedule_seconds": 5, + "actions": [{"type": "log"}], + }, + session=db_session, + ) + + engine = WorkflowEngine() + await engine.evaluate_periodic(session=db_session) + + # Backdate the run so the schedule has elapsed + run = (await db_session.execute(select(Run))).scalars().one() + run.created_at = datetime.now(UTC) - timedelta(seconds=10) + await db_session.commit() + + await engine.evaluate_periodic(session=db_session) + + runs = (await db_session.execute(select(Run))).scalars().all() + assert len(runs) == 2 + assert runs[1].actions_fired == [{"type": "log"}] + + +@pytest.mark.asyncio +async def test_periodic_automation_conditions_evaluated(db_session: AsyncSession): + """Test a periodic automation with failing conditions records no actions.""" + await create_automation( + { + "name": "periodic-condition", + "trigger_type": "periodic", + "schedule_seconds": 5, + "conditions": [{"field": "runtime", "op": "gt", "value": 10}], + "actions": [{"type": "log"}], + }, + session=db_session, + ) + + engine = WorkflowEngine() + await engine.evaluate_periodic(session=db_session) + + runs = (await db_session.execute(select(AutomationRun))).scalars().all() + assert len(runs) == 1 + assert runs[0].conditions_passed is False + assert runs[0].actions_fired is None + + +@pytest.mark.asyncio +async def test_check_workers_offline_alerts_stale_worker(db_session: AsyncSession): + """Test the check_workers_offline action alerts for stale workers.""" + from datetime import timedelta + + from taskowl.models import WorkerEvent + + now = datetime.now(UTC) + db_session.add( + WorkerEvent( + event_type="heartbeat", + hostname="celery@w1", + timestamp=now - timedelta(seconds=300), + ) + ) + db_session.add( + WorkerEvent( + event_type="heartbeat", + hostname="celery@w2", + timestamp=now, + ) + ) + await db_session.commit() + + await create_automation( + { + "name": "offline-sweep", + "trigger_type": "periodic", + "schedule_seconds": 5, + "actions": [{"type": "check_workers_offline", "webhook_url": "http://hooks.test/z"}], + }, + session=db_session, + ) + + with patch("taskowl.workflow.WebhookClient") as mock_client_cls: + mock_client = MagicMock() + mock_client.send = AsyncMock() + mock_client_cls.return_value = mock_client + + engine = WorkflowEngine() + await engine.evaluate_periodic(session=db_session) + + # Only the stale worker is alerted + assert mock_client.send.await_count == 1 + payload = mock_client.send.call_args.args[0] + assert payload["text"] == "⚠️ Worker offline" + assert "celery@w1" in payload["attachments"][0]["fields"][0]["value"] + + +@pytest.mark.asyncio +async def test_check_workers_offline_dedupes(db_session: AsyncSession): + """Test the check_workers_offline action dedupes until the worker recovers.""" + from datetime import timedelta + + from taskowl.models import WorkerEvent + + db_session.add( + WorkerEvent( + event_type="heartbeat", + hostname="celery@w1", + timestamp=datetime.now(UTC) - timedelta(seconds=300), + ) + ) + await db_session.commit() + + await create_automation( + { + "name": "offline-sweep", + "trigger_type": "periodic", + "schedule_seconds": 5, + "actions": [{"type": "check_workers_offline", "webhook_url": "http://hooks.test/z"}], + }, + session=db_session, + ) + + engine = WorkflowEngine() + with patch("taskowl.workflow.WebhookClient") as mock_client_cls: + mock_client = MagicMock() + mock_client.send = AsyncMock() + mock_client_cls.return_value = mock_client + + await engine.evaluate_periodic(session=db_session) + await engine.evaluate_periodic(session=db_session) + + assert mock_client.send.await_count == 1 + + # Worker recovers -> can be alerted again (backdate the last run first) + engine.mark_online("celery@w1") + from datetime import timedelta + + run = (await db_session.execute(select(AutomationRun))).scalars().all()[-1] + run.created_at = datetime.now(UTC) - timedelta(seconds=30) + await db_session.commit() + + with patch("taskowl.workflow.WebhookClient") as mock_client_cls: + mock_client = MagicMock() + mock_client.send = AsyncMock() + mock_client_cls.return_value = mock_client + + await engine.evaluate_periodic(session=db_session) + + assert mock_client.send.await_count == 1