fix(alerts): an alert's email waits for the real commit, not a savepoint (#5807)

SQLAlchemy fires after_commit and after_rollback for a savepoint too. The
host-agent ingest releases one after recording its metrics, so an alert
raised by a sample started its email before the outer transaction
committed; twice on 2026-10-12 the outbox row hit the event's foreign key
first and the email was lost. A savepoint rolling back (a docker persist
failing) cleared every pending notification, the outer one's included.

_after_commit now runs its jobs only on the outermost commit, and a
rolled-back savepoint drops only the jobs added inside it.

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

steward/core/alerts.py

@@ -212,28 +212,56 @@ def _spawn(coro) -> None:
     task.add_done_callback(_background.discard)
 
 
+def _within(txn, boundary) -> bool:
+    """Whether `txn` (a savepoint, or None for the outer transaction) is
+    `boundary` or opened inside it."""
+    while txn is not None:
+        if txn is boundary:
+            return True
+        txn = txn.parent
+    return False
+
+
 def _after_commit(session: AsyncSession, fn, *args) -> None:
-    """Run fn(app, *args) once the session's transaction commits, and never if
-    it rolls back: a notification or a playbook run must describe what was
-    actually stored."""
+    """Run fn(app, *args) once the session's outer transaction commits, and
+    never if what added it rolls back: a notification or a playbook run must
+    describe what was actually stored.
+
+    SQLAlchemy fires after_commit and after_rollback for a savepoint too
+    (#5807). Releasing one, as the host-agent ingest does after its metrics,
+    is not the commit: a job started then raced the real commit, and an email
+    queued for an event not yet visible broke its foreign key. A savepoint
+    rolling back drops only the jobs added inside it."""
     if _app is None:
         return
     info = session.info
+    sync = session.sync_session
     if _AFTER_COMMIT not in info:
         info[_AFTER_COMMIT] = []
         app = _app
 
-        def _committed(_sync_session) -> None:
+        def _committed(sync_session) -> None:
+            if sync_session.get_nested_transaction() is not None:
+                return  # a savepoint released: its jobs wait for the outer commit
             jobs, info[_AFTER_COMMIT] = info[_AFTER_COMMIT], []
-            for job_fn, job_args in jobs:
+            for _owner, job_fn, job_args in jobs:
                 _spawn(job_fn(app, *job_args))
 
-        def _rolled_back(_sync_session) -> None:
-            info[_AFTER_COMMIT] = []
+        def _rolled_back(_sync_session, previous_transaction) -> None:
+            # What was actually rolled back: the nearest savepoint, or the
+            # outer transaction (a failed flush rolls back to one of them).
+            boundary = previous_transaction
+            while boundary.parent is not None and not boundary.nested:
+                boundary = boundary.parent
+            if not boundary.nested:
+                info[_AFTER_COMMIT] = []
+            else:
+                info[_AFTER_COMMIT] = [job for job in info[_AFTER_COMMIT]
+                                       if not _within(job[0], boundary)]
 
-        sa_event.listen(session.sync_session, "after_commit", _committed)
-        sa_event.listen(session.sync_session, "after_rollback", _rolled_back)
-    info[_AFTER_COMMIT].append((fn, args))
+        sa_event.listen(sync, "after_commit", _committed)
+        sa_event.listen(sync, "after_soft_rollback", _rolled_back)
+    info[_AFTER_COMMIT].append((sync.get_nested_transaction(), fn, args))
 
 
 LATEST_CHUNK = 1000

tests/integration/test_alert_after_commit_savepoints.py

@@ -0,0 +1,126 @@
+"""Integration (#5807): against real Postgres, a savepoint is not the commit.
+
+SQLAlchemy fires after_commit and after_rollback for a savepoint too. The
+host-agent ingest releases one after recording its metrics, so an alert's
+email started then, before the outer commit, and twice on 2026-10-12 its
+outbox row hit the event's foreign key before the event was visible. A
+savepoint rolling back (a docker persist failing) cleared every pending
+notification, the outer transaction's included."""
+from __future__ import annotations
+
+import asyncio
+import os
+import uuid
+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):
+    from steward.alerts import service as alert_svc
+    from steward.core.alerts import record_metrics
+    from steward.models.metrics import PluginMetric
+    from steward.models.users import User, UserRole
+    res = f"sp-{tag}"
+    async with app.db_sessionmaker() as s:
+        async with s.begin():
+            user = User(username=f"sp-{tag}", email=f"sp-{tag}@test", password_hash="x",
+                        role=UserRole.admin)
+            s.add(user)
+            await record_metrics(s, [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)
+            await alert_svc.add_rule(s, fields, created_by=user.id)
+    return res
+
+
+async def _settle():
+    for _ in range(5):
+        await asyncio.sleep(0)
+
+
+class Boom(Exception):
+    pass
+
+
+def _recorder():
+    sent = []
+
+    async def fake_dispatch(app_, notif_type, rule, value, event_id, note=None):
+        sent.append((rule.resource_name, notif_type, value))
+
+    return sent, fake_dispatch
+
+
+def test_a_released_savepoint_holds_the_email_until_the_outer_commit(app):
+    from steward.core import alerts
+    sent, fake_dispatch = _recorder()
+
+    async def _go():
+        res = await _rule(app, uuid.uuid4().hex[:8])
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                await alerts.record_metric(s, "host_agent", res, "cpu_pct", 95.0)
+                async with s.begin_nested():  # as the ingest's maintenance step
+                    pass
+                await _settle()
+                assert sent == []  # the event isn't committed yet
+        await _settle()
+        assert sent == [(res, "firing", 95.0)]
+
+    with patch.object(alerts, "_dispatch_notification", fake_dispatch):
+        asyncio.run(_go())
+
+
+def test_a_savepoint_rolled_back_after_the_alert_keeps_its_email(app):
+    from steward.core import alerts
+    sent, fake_dispatch = _recorder()
+
+    async def _go():
+        res = await _rule(app, uuid.uuid4().hex[:8])
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                await alerts.record_metric(s, "host_agent", res, "cpu_pct", 95.0)
+                with pytest.raises(Boom):
+                    async with s.begin_nested():  # as a docker persist failing
+                        raise Boom
+        await _settle()
+        assert sent == [(res, "firing", 95.0)]
+
+    with patch.object(alerts, "_dispatch_notification", fake_dispatch):
+        asyncio.run(_go())
+
+
+def test_an_alert_inside_a_rolled_back_savepoint_is_dropped_alone(app):
+    from steward.core import alerts
+    sent, fake_dispatch = _recorder()
+
+    async def _go():
+        kept = await _rule(app, uuid.uuid4().hex[:8])
+        undone = await _rule(app, uuid.uuid4().hex[:8])
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                await alerts.record_metric(s, "host_agent", kept, "cpu_pct", 95.0)
+                with pytest.raises(Boom):
+                    async with s.begin_nested():
+                        async with s.begin_nested():  # released into the one undone
+                            await alerts.record_metric(s, "host_agent", undone, "cpu_pct", 97.0)
+                        raise Boom
+        await _settle()
+        assert sent == [(kept, "firing", 95.0)]
+
+    with patch.object(alerts, "_dispatch_notification", fake_dispatch):
+        asyncio.run(_go())