feat(alerts): dispatch after commit; maintenance mutes, not skips; actions via trigger_run (#5063)

Notifications and alert actions now start only once the evaluating
transaction commits, and never for one that rolls back. record_metrics
queues them on the session (_after_commit), and after_commit/after_rollback
listeners on the sync session spawn or drop them. Spawned tasks are held in a
module set so they cannot be garbage-collected before they run.

A maintenance window no longer skips evaluation. Alerts in its scope still
change state, but their notifications and actions are held, and each event
says so ("notification muted: maintenance window <name>"; also on no-data
and resume events). notify_after_maintenance, run every minute with the
silence check as the new core task alert_reconcile, sends one notification
for an alert still firing after its window ends and runs its action. The
event's note records the release.

Alert actions go through runner.trigger_run like every other run path, so
the playbook and the inventory must be ones the source lists (#5034 guards).
The firing event links to the run, or its note says why the action did not
start. The maintenance page and form wording now say what a window does.

Tests:
- integration: a rolled-back evaluation notifies nobody, and dispatch waits
  for the commit; maintenance mutes while the state moves and notifies once
  when the window ends; an action whose playbook is outside its source is
  refused, with the note set;
- the alert-action test uses a listed inventory name and waits, with a
  deadline, for the run the action starts;
- unit: _muted_by.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tests/core/test_alert_no_data.py

@@ -96,3 +96,14 @@ def test_a_sample_after_silence_clears_it_and_records_the_resume():
     assert value == 42.0 and note == "data resumed after 15m silent"
     [event] = session.added
     assert (event.id, event.from_state, event.to_state) == (event_id, "no_data", "firing")
+
+
+def test_muted_by_names_the_window_that_covers_the_rule():
+    from steward.core.alerts import _muted_by
+    rule = NS(source_module="host_agent", resource_name="nas")
+    windows = [NS(name="other", scope_source="docker", scope_resource=None),
+               NS(name="nas work", scope_source="host_agent", scope_resource="nas")]
+    assert _muted_by(windows, rule) == "notification muted: maintenance window nas work"
+    assert _muted_by(windows[:1], rule) is None
+    assert _muted_by([NS(name="all", scope_source=None, scope_resource=None)], rule) == (
+        "notification muted: maintenance window all")

tests/integration/test_alert_ansible_action.py

@@ -48,7 +48,8 @@ def test_firing_triggers_ansible_action(app, tmp_path):
             - debug:
                 msg: "FIRED-{{ steward_alert_metric }}"
     """))
-    (tmp_path / "inv.ini").write_text("localhost ansible_connection=local\n")
+    # inventory.ini: an inventory name the source lists, as trigger_run's guard requires.
+    (tmp_path / "inventory.ini").write_text("localhost ansible_connection=local\n")
 
     # An ansible source the action resolves against.
     app.config["ANSIBLE"] = {
@@ -57,7 +58,7 @@ def test_firing_triggers_ansible_action(app, tmp_path):
 
     uid = str(uuid.uuid4())
     rule_id = str(uuid.uuid4())
-    action = {"source": "t", "playbook": "fire.yml", "inventory": "inv.ini"}
+    action = {"source": "t", "playbook": "fire.yml", "inventory": "inventory.ini"}
 
     async def _go():
         async with app.db_sessionmaker() as s:
@@ -82,9 +83,13 @@ def test_firing_triggers_ansible_action(app, tmp_path):
             async with s.begin():
                 await record_metric(s, "host_agent", "srv1", "cpu_pct", 95.0)
 
-        # Let the fire-and-forget dispatch + action tasks complete.
-        pending = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
-        await asyncio.gather(*pending, return_exceptions=True)
+        # Let the dispatch, the action and the run it starts finish. The run's
+        # task is created by the action's task, so wait until none are left.
+        loop = asyncio.get_running_loop()
+        deadline = loop.time() + 120
+        while pending := [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]:
+            assert loop.time() < deadline, f"tasks still running: {pending}"
+            await asyncio.wait(pending, timeout=deadline - loop.time())
 
         async with app.db_sessionmaker() as s:
             run = (await s.execute(

tests/integration/test_alert_dispatch.py

@@ -0,0 +1,184 @@
+"""Integration (#5063): against real Postgres, notifications and actions start
+only after the evaluating transaction commits; a maintenance window mutes them
+without stopping the state from moving, and an alert still firing when the
+window ends notifies then; an alert action is held to what its source lists."""
+from __future__ import annotations
+
+import asyncio
+import os
+import uuid
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+
+import pytest
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.fixture
+def app():
+    if not os.environ.get("STEWARD_DATABASE_URL"):
+        pytest.skip("needs Postgres")
+    from steward.app import create_app
+    return create_app(testing=False)
+
+
+async def _rule(app, tag, **extra):
+    from steward.alerts import service as alert_svc
+    from steward.models.metrics import PluginMetric
+    from steward.models.users import User, UserRole
+    res = f"box-{tag}"
+    async with app.db_sessionmaker() as s:
+        async with s.begin():
+            user = User(username=f"dp-{tag}", email=f"dp-{tag}@test", password_hash="x",
+                        role=UserRole.admin)
+            s.add(user)
+            s.add(PluginMetric(source_module="host_agent", resource_name=res,
+                               metric_name="cpu_pct", value=1.0))
+            await s.flush()
+            fields = await alert_svc.rule_fields(
+                s, name=f"hot {tag}", source_module="host_agent", resource_name=res,
+                metric_name="cpu_pct", operator=">", threshold=90)
+            rule = await alert_svc.add_rule(s, fields, created_by=user.id, **extra)
+    return rule.id, res, user.id
+
+
+async def _events(app, rule_id):
+    from sqlalchemy import select
+    from steward.models.alerts import AlertEvent
+    async with app.db_sessionmaker() as s:
+        rows = (await s.execute(select(AlertEvent).where(AlertEvent.rule_id == rule_id)
+                                .order_by(AlertEvent.transitioned_at))).scalars().all()
+    return [(e.from_state, e.to_state, e.note, e.ansible_run_id) for e in rows]
+
+
+async def _settle():
+    for _ in range(5):
+        await asyncio.sleep(0)
+
+
+def test_a_rolled_back_evaluation_notifies_nobody(app):
+    from steward.core import alerts
+
+    sent = []
+
+    async def fake_dispatch(app_, notif_type, rule, value, event_id, note=None):
+        sent.append((notif_type, value))
+
+    async def _go():
+        rule_id, res, _ = await _rule(app, uuid.uuid4().hex[:8])
+
+        class Boom(Exception):
+            pass
+
+        with pytest.raises(Boom):
+            async with app.db_sessionmaker() as s:
+                async with s.begin():
+                    await alerts.record_metric(s, "host_agent", res, "cpu_pct", 95.0)
+                    raise Boom
+        await _settle()
+        assert sent == [] and await _events(app, rule_id) == []
+
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                await alerts.record_metric(s, "host_agent", res, "cpu_pct", 96.0)
+                await _settle()
+                assert sent == []  # not before the commit
+        await _settle()
+        assert sent == [("firing", 96.0)]
+
+    with patch.object(alerts, "_dispatch_notification", fake_dispatch):
+        asyncio.run(_go())
+
+
+def test_maintenance_mutes_but_the_state_moves_and_notifies_when_it_ends(app):
+    from sqlalchemy import select, update
+    from steward.core import alerts
+    from steward.models.alerts import AlertState, AlertStateEnum, MaintenanceWindow
+
+    sent = []
+
+    async def fake_dispatch(app_, notif_type, rule, value, event_id, note=None):
+        sent.append((notif_type, value))
+
+    async def _go():
+        tag = uuid.uuid4().hex[:8]
+        rule_id, res, user_id = await _rule(app, tag)
+        now = datetime.now(timezone.utc)
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                window = MaintenanceWindow(
+                    name=f"patching {tag}", start_at=now - timedelta(minutes=1),
+                    end_at=now + timedelta(hours=1), scope_source="host_agent",
+                    scope_resource=res, created_by=user_id)
+                s.add(window)
+            window_id = window.id
+
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                await alerts.record_metric(s, "host_agent", res, "cpu_pct", 95.0)
+        await _settle()
+        async with app.db_sessionmaker() as s:
+            st = (await s.execute(select(AlertState).where(
+                AlertState.rule_id == rule_id))).scalar_one()
+        assert st.state == AlertStateEnum.firing  # evaluated inside the window
+        [(frm, to, note, _)] = await _events(app, rule_id)
+        assert (frm, to, note) == ("inactive", "firing",
+                                   f"notification muted: maintenance window patching {tag}")
+        assert sent == []
+
+        await alerts.notify_after_maintenance(app)  # window still open: held
+        await _settle()
+        assert sent == []
+
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                await s.execute(update(MaintenanceWindow).where(MaintenanceWindow.id == window_id)
+                                .values(end_at=datetime.now(timezone.utc)))
+        await alerts.notify_after_maintenance(app)
+        await alerts.notify_after_maintenance(app)  # released once
+        await _settle()
+        assert sent == [("firing", 95.0)]
+        [(_, _, note, _)] = await _events(app, rule_id)
+        assert note.endswith("; notified when the window ended")
+
+    with patch.object(alerts, "_dispatch_notification", fake_dispatch):
+        asyncio.run(_go())
+
+
+def test_an_alert_action_outside_its_source_is_refused(app, tmp_path):
+    from steward.core import alerts
+    from steward.models.ansible import AnsibleRun
+    from sqlalchemy import func, select
+
+    (tmp_path / "fix.yml").write_text("- hosts: all\n  tasks: []\n")
+    (tmp_path / "inventory.ini").write_text("localhost\n")
+    app.config["ANSIBLE"] = {"sources": [{"name": "t", "type": "local", "path": str(tmp_path)}]}
+
+    async def quiet(*args, **kwargs):
+        pass
+
+    async def _go():
+        tag = uuid.uuid4().hex[:8]
+        rule_id, res, _ = await _rule(
+            app, tag, ansible_action={"source": "t", "playbook": "../../etc/evil.yml",
+                                      "inventory": "inventory.ini"})
+        async with app.db_sessionmaker() as s:
+            before = (await s.execute(select(func.count()).select_from(AnsibleRun))).scalar_one()
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                await alerts.record_metric(s, "host_agent", res, "cpu_pct", 95.0)
+        loop = asyncio.get_running_loop()
+        deadline = loop.time() + 30
+        while alerts._background:
+            assert loop.time() < deadline, "alert tasks did not finish"
+            await asyncio.wait(set(alerts._background), timeout=deadline - loop.time())
+        [(frm, to, note, run_id)] = await _events(app, rule_id)
+        assert (frm, to, run_id) == ("inactive", "firing", None)
+        assert note == "Ansible action not started: Source 't' has no playbook '../../etc/evil.yml'"
+        async with app.db_sessionmaker() as s:
+            after = (await s.execute(select(func.count()).select_from(AnsibleRun))).scalar_one()
+        assert after == before
+
+    with patch.object(alerts, "_dispatch_notification", quiet):
+        asyncio.run(_go())