feat(alerts): alert email waits 2 minutes in the outbox and each alert records its outcome (#5507)

#5507 "Alert emails through the outbox, gathered for 2 minutes"
(milestone 501). Replaces the in-memory core/alert_burst.py.

- outbox.hold_alert: an alert joins each recipient's held alert email, or
  opens one due 2 minutes later (System 156 ruling). A Postgres advisory
  lock keeps two alerts at once from opening two windows for one reader.
- email_outbox_events.alert_json (migration 0043) keeps the alert as it
  fired, so the email written when the window closes says what it said.
- write_held writes it then, in the reader's zone: the usual alert email
  for one alert, the list for several (outbox.alert_words).
- after_settle records on each AlertEvent, once every email carrying it
  has settled: email_sent, or email_error naming each reader it didn't
  reach and the server's reason.
- A held email survives a restart and goes out when its window closes.
- A refusal of a message's one recipient no longer repeats the address.

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

tests/core/test_alert_burst.py

@@ -1,79 +0,0 @@
-"""Alert emails gathered into one message per burst (#5465): alerts that
-arrive within the window go out as one email per recipient listing each of
-them; one alert alone keeps its usual email; a failed send says so for every
-alert in the burst; an alert arriving after a window closes opens the next."""
-import asyncio
-from email.message import EmailMessage
-from zoneinfo import ZoneInfo
-
-from steward.core import alert_burst, mail
-from steward.core import notifications as n
-
-NY = ZoneInfo("America/New_York")
-ANN = mail.Recipient("ann@example.com", NY, False, "ann")
-BOB = mail.Recipient("bob@example.com", NY, False, "bob")
-CFG = {"host": "mx", "port": 587, "from": "alerts@example.com", "base_url": "https://steward.example/"}
-
-
-def _alert(host: str, state: str = "FIRING", value: float = 400.0) -> dict:
-    return {"rule_name": f"Agent offline: {host}", "state": state, "metric": "seconds_since_seen",
-            "value": value, "operator": ">", "threshold": 180.0, "resource": host,
-            "source_module": "host_agent", "note": None, "timestamp": "2026-10-09T12:00:00+00:00"}
-
-
-def _capture(monkeypatch) -> list[EmailMessage]:
-    out: list[EmailMessage] = []
-    monkeypatch.setattr(n, "_smtp_send_sync", lambda c, msgs: out.extend(msgs))
-    return out
-
-
-async def test_one_alert_alone_keeps_its_usual_email(monkeypatch):
-    sent = _capture(monkeypatch)
-    assert await alert_burst.email_burst(CFG, [_alert("web01")], [ANN]) == {"sent": True, "error": None}
-    assert len(sent) == 1
-    assert sent[0]["Subject"] == mail.alert_email(CFG, _alert("web01"), ANN)[0]
-
-
-async def test_a_burst_is_one_email_per_recipient_listing_each_alert(monkeypatch):
-    sent = _capture(monkeypatch)
-    burst = [_alert("web01"), _alert("web02"), _alert("db01", "RESOLVED", 12.0)]
-    assert (await alert_burst.email_burst(CFG, burst, [ANN, BOB]))["sent"]
-    assert [m["To"] for m in sent] == ["ann@example.com", "bob@example.com"]
-    assert sent[0]["Subject"] == "[Steward] 3 alerts: 2 firing, 1 resolved"
-    text = sent[0].get_body(("plain",)).get_content()
-    html = sent[0].get_body(("html",)).get_content()
-    for host in ("web01", "web02", "db01"):
-        assert f"Agent offline: {host}" in text and f"Agent offline: {host}" in html
-    assert text.index("web01") < text.index("web02") < text.index("db01")   # oldest first
-    assert "Fri 9 Oct 2026 08:00 EDT" in text                                # the reader's zone
-
-
-async def test_a_failed_send_is_said_for_the_burst(monkeypatch):
-    def boom(cfg, msgs):
-        raise OSError("connection refused")
-    monkeypatch.setattr(n, "_smtp_send_sync", boom)
-    got = await alert_burst.email_burst(CFG, [_alert("a"), _alert("b")], [ANN])
-    assert got == {"sent": False, "error": "connection refused"}
-
-
-async def test_nobody_to_send_to_sends_nothing(monkeypatch):
-    sent = _capture(monkeypatch)
-    assert await alert_burst.email_burst(CFG, [_alert("a")], []) is None and sent == []
-
-
-async def test_alerts_within_the_window_are_flushed_together(monkeypatch):
-    flushed: list[list[str]] = []
-
-    async def fake_flush(app):
-        flushed.append([a["resource"] for a, _ in alert_burst._held])
-        alert_burst._held.clear()
-
-    monkeypatch.setattr(alert_burst, "BURST_WINDOW_S", 0.05)
-    monkeypatch.setattr(alert_burst, "flush", fake_flush)
-    alert_burst._held.clear()
-    for host in ("a", "b", "c"):
-        alert_burst.hold(None, _alert(host), f"ev-{host}")
-    await asyncio.sleep(0.15)
-    alert_burst.hold(None, _alert("d"), "ev-d")      # after the window: the next burst
-    await asyncio.sleep(0.15)
-    assert flushed == [["a", "b", "c"], ["d"]]

tests/core/test_outbox.py

@@ -46,7 +46,9 @@ def test_reason_is_what_the_server_said():
     assert outbox.reason(smtplib.SMTPDataError(554, b"5.7.1  rejected\r\n as spam")) \
         == "554 5.7.1 rejected as spam"
     assert outbox.reason(smtplib.SMTPRecipientsRefused({"a@x": (550, b"no such user")})) \
-        == "a@x: 550 no such user"
+        == "550 no such user"
+    assert outbox.reason(smtplib.SMTPRecipientsRefused({"a@x": (550, b"no"), "b@x": (450, b"busy")})) \
+        == "a@x: 550 no; b@x: 450 busy"
     assert outbox.reason(TimeoutError()) == "The mail server didn't answer in time"
 
 
@@ -61,7 +63,7 @@ def test_a_refusal_fails_at_once_with_the_servers_words():
     row = _row()
     outbox.settle(row, smtplib.SMTPRecipientsRefused({"ann@example.com": (550, b"no such user")}), NOW)
     assert row.state == "failed" and row.next_attempt_at is None
-    assert row.last_error == "ann@example.com: 550 no such user"
+    assert row.last_error == "550 no such user"
 
 
 def test_a_temporary_failure_waits_and_tries_again():
@@ -160,3 +162,27 @@ def test_no_connection_is_every_messages_outcome(monkeypatch):
 
     monkeypatch.setattr(notifications, "smtp_open", _open)
     assert outbox.deliver_sync({}, _msgs(2)) == [refused, refused]
+
+
+CFG = {"host": "mx", "port": 587, "from": "alerts@example.com", "base_url": "https://steward.example/"}
+ANN = mail.Recipient("ann@example.com", ZoneInfo("America/New_York"), False, "ann")
+
+
+def _alert(host: str, state: str = "FIRING", value: float = 400.0) -> dict:
+    return {"rule_name": f"Agent offline: {host}", "state": state, "metric": "seconds_since_seen",
+            "value": value, "operator": ">", "threshold": 180.0, "resource": host,
+            "source_module": "host_agent", "note": None, "timestamp": "2026-10-09T12:00:00+00:00"}
+
+
+def test_one_alert_alone_keeps_its_usual_email():
+    assert outbox.alert_words(CFG, [_alert("web01")], ANN) == mail.alert_email(CFG, _alert("web01"), ANN)
+
+
+def test_several_alerts_are_one_email_listing_each_oldest_first():
+    burst = [_alert("web01"), _alert("web02"), _alert("db01", "RESOLVED", 12.0)]
+    subject, text, html = outbox.alert_words(CFG, burst, ANN)
+    assert subject == "[Steward] 3 alerts: 2 firing, 1 resolved"
+    for host in ("web01", "web02", "db01"):
+        assert f"Agent offline: {host}" in text and f"Agent offline: {host}" in html
+    assert text.index("web01") < text.index("web02") < text.index("db01")
+    assert "Fri 9 Oct 2026 08:00 EDT" in text                                # the reader's zone

tests/integration/test_alert_email_outbox.py

@@ -0,0 +1,169 @@
+"""Integration: alert email through the outbox (#5507), against Postgres.
+
+Alerts that fire within 2 minutes of the first join one held email per
+reader; when the window closes it is written as one message listing them,
+oldest first; each alert's event records the outcome once every email
+carrying it has settled, naming a reader it didn't reach. Only the SMTP
+conversation is replaced.
+"""
+from __future__ import annotations
+
+import asyncio
+import os
+import smtplib
+import uuid
+from datetime import datetime, timedelta, timezone
+
+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)
+
+
+def _alert(host: str, at: datetime) -> dict:
+    return {"rule_name": f"Agent offline: {host}", "state": "FIRING", "metric": "seconds_since_seen",
+            "value": 400.0, "operator": ">", "threshold": 180.0, "resource": host,
+            "source_module": "host_agent", "note": None, "timestamp": at.isoformat()}
+
+
+async def _events(app, tag: str, times: list[datetime]) -> list[str]:
+    from steward.alerts import service as alert_svc
+    from steward.core.alerts import record_metrics
+    from steward.models.alerts import AlertEvent
+    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"ob-{tag}", email=f"ob-{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)
+            rule = await alert_svc.add_rule(s, fields, created_by=user.id)
+            events = [AlertEvent(rule_id=rule.id, from_state="inactive", to_state="firing",
+                                 metric_value=95.0, transitioned_at=at) for at in times]
+            s.add_all(events)
+            await s.flush()
+            return [e.id for e in events]
+
+
+async def _emails_for(app, event_ids: list[str]):
+    from sqlalchemy import select
+    from steward.models.outbox import EmailOutbox, email_outbox_events as link
+    async with app.db_sessionmaker() as s:
+        rows = (await s.execute(
+            select(EmailOutbox).where(EmailOutbox.id.in_(
+                select(link.c.outbox_id).where(link.c.event_id.in_(event_ids)))))).scalars().all()
+        links = (await s.execute(select(link.c.outbox_id, link.c.event_id)
+                                 .where(link.c.event_id.in_(event_ids)))).all()
+    return rows, links
+
+
+async def _event(app, event_id):
+    from steward.models.alerts import AlertEvent
+    async with app.db_sessionmaker() as s:
+        return await s.get(AlertEvent, event_id)
+
+
+def test_alerts_in_one_window_are_one_email_and_each_event_records_it(app, monkeypatch):
+    from steward.core import outbox
+
+    tag = uuid.uuid4().hex[:8]
+    extra = f"oncall-{tag}@test"
+    app.config["SMTP"] = {"host": "mx.test", "port": 587, "from": "steward@test",
+                          "recipients": [extra], "tz": "America/New_York"}
+    t0 = datetime(2026, 10, 9, 12, 0, tzinfo=timezone.utc)
+    sent = []
+
+    def fake_deliver(cfg, messages):
+        sent.extend(messages)
+        return [smtplib.SMTPRecipientsRefused({m["To"]: (550, b"no such user")})
+                if m["To"] == extra and "web03" in m["Subject"] else None for m in messages]
+
+    monkeypatch.setattr(outbox, "deliver_sync", fake_deliver)
+
+    async def _go():
+        ev1, ev2, ev3 = await _events(app, tag, [t0, t0 + timedelta(seconds=30),
+                                                 t0 + timedelta(minutes=5)])
+        assert await outbox.hold_alert(app, _alert("web01", t0), ev1, now=t0) >= 1
+        await outbox.hold_alert(app, _alert("web02", t0), ev2, now=t0 + timedelta(seconds=30))
+
+        rows, links = await _emails_for(app, [ev1, ev2])
+        mine = [r for r in rows if r.recipient == extra]
+        assert len(mine) == 1, "both alerts joined the one held email"
+        (held,) = mine
+        assert held.state == "held" and held.subject is None
+        assert held.next_attempt_at == t0 + timedelta(minutes=2)
+        assert {e for o, e in links if o == held.id} == {ev1, ev2}
+        # Each reader has exactly one email carrying the pair.
+        assert len({r.recipient for r in rows}) == len(rows)
+
+        # The window closes: written, sent, and recorded on both events.
+        await outbox.attempt(app, [r.id for r in rows], now=t0 + timedelta(minutes=2))
+        (msg,) = [m for m in sent if m["To"] == extra]
+        assert msg["Subject"] == "[Steward] 2 alerts: 2 firing"
+        body = msg.get_body(("plain",)).get_content()
+        assert body.index("web01") < body.index("web02")
+        assert "Fri 9 Oct 2026 08:00 EDT" in body          # the Also-send-to zone
+        for ev in (ev1, ev2):
+            event = await _event(app, ev)
+            assert (event.email_sent, event.email_error) == (True, None)
+
+        # A later alert is its own email; refused for one reader, it is
+        # recorded as not sent, naming who and why.
+        await outbox.hold_alert(app, _alert("web03", t0), ev3, now=t0 + timedelta(minutes=5))
+        rows, _ = await _emails_for(app, [ev3])
+        assert all(r.state == "held" for r in rows)
+        await outbox.attempt(app, [r.id for r in rows], now=t0 + timedelta(minutes=7))
+        (refused,) = [m for m in sent if m["To"] == extra and "web03" in m["Subject"]]
+        assert refused["Subject"].startswith("[Steward]")   # one alert: its usual email
+        event = await _event(app, ev3)
+        assert event.email_sent is False
+        assert event.email_error == f"{extra}: 550 no such user"
+
+    asyncio.run(_go())
+
+
+def test_an_email_whose_alerts_were_deleted_fails_and_says_so(app, monkeypatch):
+    from sqlalchemy import delete
+    from steward.core import outbox
+    from steward.models.alerts import AlertEvent
+
+    tag = uuid.uuid4().hex[:8]
+    extra = f"gone-{tag}@test"
+    app.config["SMTP"] = {"host": "mx.test", "port": 587, "from": "steward@test", "recipients": [extra]}
+    monkeypatch.setattr(outbox, "deliver_sync", lambda cfg, msgs: [None] * len(msgs))
+    t0 = datetime.now(timezone.utc)
+
+    async def _go():
+        (ev,) = await _events(app, tag, [t0])
+        await outbox.hold_alert(app, _alert("web09", t0), ev, now=t0)
+        rows, _ = await _emails_for(app, [ev])
+        (row,) = [r for r in rows if r.recipient == extra]
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                await s.execute(delete(AlertEvent).where(AlertEvent.id == ev))
+        states = await outbox.attempt(app, [row.id])
+        return states[row.id], row.id
+
+    state, rid = asyncio.run(_go())
+    assert state == "failed"
+
+    async def _reason():
+        from steward.models.outbox import EmailOutbox
+        async with app.db_sessionmaker() as s:
+            return (await s.get(EmailOutbox, rid)).last_error
+
+    assert asyncio.run(_reason()) == "The alerts this email carried were deleted"

tests/integration/test_email_outbox.py

@@ -72,7 +72,7 @@ def test_the_sender_settles_each_row_on_what_the_server_said(app, monkeypatch):
     by = {k: rows[to[k]] for k in to}
     assert by["ok"].state == "sent" and by["ok"].sent_at is not None and by["ok"].attempts == 1
     assert by["refused"].state == "failed"
-    assert by["refused"].last_error == f"{to['refused']}: 550 no such user"
+    assert by["refused"].last_error == "550 no such user"
     later = by["later"]
     assert later.state == "pending" and later.attempts == 1 and later.last_error == "451 greylisted"
     assert later.next_attempt_at > later.first_attempt_at