feat(alerts): alerts that arrive together go out as one email (#5465)

#5465 "One email per burst of alerts". When a switch reboots, every host's
agent-offline rule fires at once and each account got one email per rule.
Alert email now waits 30 s from the first alert: each recipient then gets one
message listing every alert in the window, oldest first, in their own zone
("[Steward] 3 alerts: 2 firing, 1 resolved"). One alert alone keeps its
usual email. Webhooks still post per alert, at once. Every alert's event
records the email's outcome. A burst held when Steward stops is not sent;
the alerts stay on the record and on the Alerts page.

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

tests/core/test_alert_burst.py

@@ -0,0 +1,79 @@
+"""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"]]