feat: log aggregation and conditional alert rules

CHANGELOG.md

@@ -6,6 +6,21 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
 ## [Unreleased]
 
+## [1.2.0] - 2026-05-21
+
+### Added
+
+- **Conditional alert rules** — per-job rules can create alerts from HTTP status thresholds, execution duration thresholds, response body matches and consecutive failure counts.
+- **Alert rules API and UI** — CRUD endpoints under `/api/jobs/{job_id}/alert-rules` plus management controls on the job detail page.
+- **Alembic migration `c3d4e5f6a7b8`** — creates the `alert_rules` table with job and workspace foreign keys.
+- **Workspace-scoped alert dispatch** — alert notifications and escalation policies now dispatch only through channels/policies in the alert's workspace.
+
+### Changed
+
+- **Workspace isolation hardening** — job detail/update/delete/run, execution detail and alert rule endpoints now honor `X-Workspace-ID` resource scope.
+- **Worker parity** — queued Celery executions now evaluate conditional alert rules after final retry handling.
+- **Version alignment** — backend package metadata, `/api/version`, frontend package metadata and lockfile report `1.2.0`.
+
 ## [1.1.0] - 2026-05-20
 
 ### Added

README.md

@@ -30,6 +30,7 @@ AutoFlowOps centralises these routines in a single self-hosted platform:
 - **Executions** — persistent history with status, timings, response previews and masked secrets
 - **Webhooks** — receive external events, store payloads with token validation, reprocess events
 - **Alerts** — automatic alerts on job and webhook failures, with acknowledge and resolve workflows
+- **Conditional alert rules** — trigger alerts from HTTP status, runtime, response text or consecutive failure thresholds
 - **Notifications** — send critical alerts to Discord, Telegram, SMTP email and custom webhooks
 - **Escalation policies** — multi-step escalation with configurable delays per step
 - **Roles** — admin, operator and viewer roles enforced server-side; audit trail of all sensitive actions
@@ -48,6 +49,7 @@ AutoFlowOps centralises these routines in a single self-hosted platform:
 - Webhook receiver with secret token validation (SHA-256)
 - Execution history with masked secrets and response previews
 - Internal alerting system for failed executions
+- Per-job conditional alert rules for custom operational thresholds
 - External notification channels for critical operational alerts (Discord, Telegram, SMTP, custom webhooks)
 - Notification templates and multi-step escalation policies
 - Role-based access control (admin / operator / viewer) enforced server-side
@@ -130,9 +132,9 @@ The script asks for an image tag (default `latest`), copies `.env.example` to `.
 To run a specific release:
 
 ```bash
-IMAGE_TAG=v1.1.0 bash scripts/setup.sh
+IMAGE_TAG=v1.2.0 bash scripts/setup.sh
 # or
-IMAGE_TAG=v1.1.0 make registry-up
+IMAGE_TAG=v1.2.0 make registry-up
 ```
 
 ### Option B — Build from source
@@ -232,8 +234,8 @@ make test
 
 | Suite | Tests | Status |
 | --- | --- | --- |
-| Backend | 216 | Passing |
-| Frontend | 75 | Passing |
+| Backend | 260 | Passing |
+| Frontend | 76 | Passing |
 
 ---
 

README.pt-br.md

@@ -30,6 +30,7 @@ O AutoFlowOps centraliza essas rotinas em uma única plataforma self-hosted:
 - **Execuções** — histórico persistente com status, tempos, prévia de respostas e segredos mascarados
 - **Webhooks** — receba eventos externos, armazene payloads com validação por token, reprocesse eventos
 - **Alertas** — alertas automáticos em falhas de jobs e webhooks, com fluxos de reconhecimento e resolução
+- **Regras condicionais de alerta** — dispare alertas por status HTTP, duração, texto da resposta ou limites de falhas consecutivas
 - **Notificações** — envie alertas críticos para Discord, Telegram, SMTP e webhooks customizados
 - **Políticas de escalonamento** — escalonamento em múltiplos passos com delays configuráveis por etapa
 - **Permissões** — roles admin, operator e viewer aplicados server-side; trilha de auditoria de todas as ações sensíveis
@@ -48,6 +49,7 @@ O AutoFlowOps centraliza essas rotinas em uma única plataforma self-hosted:
 - Receptor de webhooks com validação por token secreto (SHA-256)
 - Histórico de execuções com segredos mascarados e prévia de respostas
 - Sistema interno de alertas para execuções com falha
+- Regras condicionais de alerta por job para limites operacionais customizados
 - Canais de notificação externos para alertas críticos (Discord, Telegram, SMTP, webhooks customizados)
 - Templates de notificação e políticas de escalonamento em múltiplos passos
 - Controle de acesso por roles (admin / operator / viewer) aplicado server-side
@@ -115,9 +117,9 @@ O script solicita a tag da imagem (padrão `latest`), copia `.env.example` para
 Para rodar uma versão específica:
 
 ```bash
-IMAGE_TAG=v1.1.0 bash scripts/setup.sh
+IMAGE_TAG=v1.2.0 bash scripts/setup.sh
 # ou
-IMAGE_TAG=v1.1.0 make registry-up
+IMAGE_TAG=v1.2.0 make registry-up
 ```
 
 ### Opção B — Build a partir do código-fonte
@@ -217,8 +219,8 @@ make test
 
 | Suite | Testes | Status |
 | --- | --- | --- |
-| Backend | 216 | Passando |
-| Frontend | 75 | Passando |
+| Backend | 260 | Passando |
+| Frontend | 76 | Passando |
 
 ---
 

backend/alembic/versions/b2c3d4e5f6a7_add_workspaces.py

@@ -84,20 +84,31 @@ def upgrade() -> None:
     )
 
     for table in _DOMAIN_TABLES:
-        op.add_column(
-            table,
-            sa.Column(
-                "workspace_id",
-                sa.Uuid(),
-                sa.ForeignKey("workspaces.id", ondelete="SET NULL"),
-                nullable=True,
-            ),
-        )
+        with op.batch_alter_table(table) as batch_op:
+            batch_op.add_column(
+                sa.Column(
+                    "workspace_id",
+                    sa.Uuid(),
+                    nullable=True,
+                )
+            )
+            batch_op.create_foreign_key(
+                f"fk_{table}_workspace_id_workspaces",
+                "workspaces",
+                ["workspace_id"],
+                ["id"],
+                ondelete="SET NULL",
+            )
 
 
 def downgrade() -> None:
     for table in reversed(_DOMAIN_TABLES):
-        op.drop_column(table, "workspace_id")
+        with op.batch_alter_table(table) as batch_op:
+            batch_op.drop_constraint(
+                f"fk_{table}_workspace_id_workspaces",
+                type_="foreignkey",
+            )
+            batch_op.drop_column("workspace_id")
 
     op.drop_index(
         op.f("ix_workspace_memberships_user_id"), table_name="workspace_memberships"

backend/alembic/versions/c3d4e5f6a7b8_add_alert_rules.py

@@ -0,0 +1,54 @@
+"""add_alert_rules
+
+Revision ID: c3d4e5f6a7b8
+Revises: b2c3d4e5f6a7
+Create Date: 2026-05-21 00:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "c3d4e5f6a7b8"
+down_revision: Union[str, None] = "b2c3d4e5f6a7"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "alert_rules",
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("job_id", sa.Uuid(), nullable=False),
+        sa.Column("condition_type", sa.String(length=50), nullable=False),
+        sa.Column("condition_value", sa.String(length=500), nullable=False),
+        sa.Column("severity", sa.String(length=50), nullable=False),
+        sa.Column("message", sa.Text(), nullable=True),
+        sa.Column("is_enabled", sa.Boolean(), nullable=False),
+        sa.Column("workspace_id", sa.Uuid(), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("(CURRENT_TIMESTAMP)"),
+            nullable=False,
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("(CURRENT_TIMESTAMP)"),
+            nullable=False,
+        ),
+        sa.ForeignKeyConstraint(["job_id"], ["jobs.id"], ondelete="CASCADE"),
+        sa.ForeignKeyConstraint(
+            ["workspace_id"], ["workspaces.id"], ondelete="SET NULL"
+        ),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(op.f("ix_alert_rules_job_id"), "alert_rules", ["job_id"])
+
+
+def downgrade() -> None:
+    op.drop_index(op.f("ix_alert_rules_job_id"), table_name="alert_rules")
+    op.drop_table("alert_rules")

backend/app/__init__.py

@@ -1 +1 @@
-__version__ = "1.1.0"
+__version__ = "1.2.0"

backend/app/api/alert_rules.py

@@ -0,0 +1,176 @@
+import uuid
+
+from fastapi import APIRouter, Depends, HTTPException, Request
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.database import get_db
+from app.dependencies import get_active_workspace, get_current_user, require_operator
+from app.models.alert_rule import AlertRule
+from app.models.job import Job
+from app.models.user import User
+from app.models.workspace import Workspace
+from app.schemas.alert_rule import (
+    AlertRuleConditionType,
+    AlertRuleCreate,
+    AlertRuleRead,
+    AlertRuleUpdate,
+    validate_alert_rule_condition_value,
+)
+from app.services.audit import client_ip, log_action
+
+router = APIRouter(prefix="/jobs/{job_id}/alert-rules", tags=["alert-rules"])
+
+
+async def _get_job_or_404(
+    session: AsyncSession,
+    job_id: uuid.UUID,
+    workspace: Workspace | None = None,
+) -> Job:
+    result = await session.execute(select(Job).where(Job.id == job_id))
+    job = result.scalar_one_or_none()
+    if job is None or (workspace is not None and job.workspace_id != workspace.id):
+        raise HTTPException(status_code=404, detail="Job not found")
+    return job
+
+
+async def _get_rule_or_404(
+    session: AsyncSession,
+    job_id: uuid.UUID,
+    rule_id: uuid.UUID,
+    workspace: Workspace | None = None,
+) -> AlertRule:
+    stmt = select(AlertRule).where(
+        AlertRule.id == rule_id, AlertRule.job_id == job_id
+    )
+    if workspace is not None:
+        stmt = stmt.where(AlertRule.workspace_id == workspace.id)
+    result = await session.execute(stmt)
+    rule = result.scalar_one_or_none()
+    if rule is None:
+        raise HTTPException(status_code=404, detail="Alert rule not found")
+    return rule
+
+
+def _ensure_valid_condition_value(
+    condition_type: AlertRuleConditionType | str,
+    condition_value: str,
+) -> None:
+    try:
+        validate_alert_rule_condition_value(
+            condition_type,  # type: ignore[arg-type]
+            condition_value,
+        )
+    except ValueError as exc:
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
+
+
+@router.get("", response_model=list[AlertRuleRead])
+async def list_alert_rules(
+    job_id: uuid.UUID,
+    session: AsyncSession = Depends(get_db),
+    _user: User = Depends(get_current_user),
+    workspace: Workspace | None = Depends(get_active_workspace),
+) -> list[AlertRuleRead]:
+    await _get_job_or_404(session, job_id, workspace)
+    result = await session.execute(
+        select(AlertRule)
+        .where(AlertRule.job_id == job_id)
+        .order_by(AlertRule.created_at.asc())
+    )
+    rules = result.scalars().all()
+    return [AlertRuleRead.model_validate(r) for r in rules]
+
+
+@router.post("", response_model=AlertRuleRead, status_code=201)
+async def create_alert_rule(
+    job_id: uuid.UUID,
+    payload: AlertRuleCreate,
+    request: Request,
+    session: AsyncSession = Depends(get_db),
+    current_user: User = Depends(require_operator),
+    workspace: Workspace | None = Depends(get_active_workspace),
+) -> AlertRuleRead:
+    job = await _get_job_or_404(session, job_id, workspace)
+    rule = AlertRule(
+        job_id=job.id,
+        condition_type=payload.condition_type,
+        condition_value=payload.condition_value,
+        severity=payload.severity,
+        message=payload.message,
+        is_enabled=payload.is_enabled,
+        workspace_id=job.workspace_id,
+    )
+    session.add(rule)
+    await session.flush()
+    await log_action(
+        session,
+        action="alert_rules.create",
+        resource_type="alert_rule",
+        resource_id=str(rule.id),
+        user_id=current_user.id,
+        ip_address=client_ip(request),
+        user_agent=request.headers.get("User-Agent"),
+        metadata={"job_id": str(job_id), "condition_type": rule.condition_type},
+    )
+    await session.commit()
+    await session.refresh(rule)
+    return AlertRuleRead.model_validate(rule)
+
+
+@router.patch("/{rule_id}", response_model=AlertRuleRead)
+async def update_alert_rule(
+    job_id: uuid.UUID,
+    rule_id: uuid.UUID,
+    payload: AlertRuleUpdate,
+    request: Request,
+    session: AsyncSession = Depends(get_db),
+    current_user: User = Depends(require_operator),
+    workspace: Workspace | None = Depends(get_active_workspace),
+) -> AlertRuleRead:
+    await _get_job_or_404(session, job_id, workspace)
+    rule = await _get_rule_or_404(session, job_id, rule_id, workspace)
+    updates = payload.model_dump(exclude_unset=True)
+    next_condition_type = updates.get("condition_type", rule.condition_type)
+    next_condition_value = updates.get("condition_value", rule.condition_value)
+    _ensure_valid_condition_value(next_condition_type, next_condition_value)
+    for field, value in updates.items():
+        setattr(rule, field, value)
+    await log_action(
+        session,
+        action="alert_rules.update",
+        resource_type="alert_rule",
+        resource_id=str(rule_id),
+        user_id=current_user.id,
+        ip_address=client_ip(request),
+        user_agent=request.headers.get("User-Agent"),
+        metadata={"updated_fields": list(updates.keys())},
+    )
+    await session.commit()
+    await session.refresh(rule)
+    return AlertRuleRead.model_validate(rule)
+
+
+@router.delete("/{rule_id}", status_code=204)
+async def delete_alert_rule(
+    job_id: uuid.UUID,
+    rule_id: uuid.UUID,
+    request: Request,
+    session: AsyncSession = Depends(get_db),
+    current_user: User = Depends(require_operator),
+    workspace: Workspace | None = Depends(get_active_workspace),
+) -> None:
+    await _get_job_or_404(session, job_id, workspace)
+    rule = await _get_rule_or_404(session, job_id, rule_id, workspace)
+    await log_action(
+        session,
+        action="alert_rules.delete",
+        resource_type="alert_rule",
+        resource_id=str(rule_id),
+        user_id=current_user.id,
+        ip_address=client_ip(request),
+        user_agent=request.headers.get("User-Agent"),
+        metadata={"job_id": str(job_id)},
+    )
+    await session.delete(rule)
+    await session.commit()

backend/app/api/executions.py

@@ -41,10 +41,12 @@ async def get_execution(
     execution_id: uuid.UUID,
     current_user: User = Depends(get_current_user),
     session: AsyncSession = Depends(get_db),
+    workspace: Workspace | None = Depends(get_active_workspace),
 ) -> ExecutionRead:
-    result = await session.execute(
-        select(Execution).where(Execution.id == execution_id)
-    )
+    stmt = select(Execution).where(Execution.id == execution_id)
+    if workspace is not None:
+        stmt = stmt.where(Execution.workspace_id == workspace.id)
+    result = await session.execute(stmt)
     execution = result.scalar_one_or_none()
     if execution is None:
         raise HTTPException(status_code=404, detail="Execution not found")