feat(email): delivery you can see: each alert's email in history, outgoing email on Settings -> System (#5509)

#5509 "Delivery you can see: each alert's email state in history, and
outgoing email on Settings -> System" (milestone 501).

- Alerts -> History: an alert's email reads from the outbox (#5506):
  "gathering" with when it sends, "sending", "retrying" with the attempt,
  next try and the server's last answer, "sent", or "failed" naming each
  reader it didn't reach and why (outbox.delivery / email_states). Events
  from before the outbox keep what they recorded. The Notified column
  shows once an alert has a queued email.
- Settings -> System gets "Outgoing email": every email of the last 7
  days not yet sent, failures first, then the longest retrying, with
  recipient, kind, state and last error, and a count of what was sent.
  Failures and anything still retrying after 5 minutes lead the page's
  verdict (outbox.outgoing).
- get_alert_history (MCP) adds `email`: the same delivery state.

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

steward/core/outbox.py

@@ -356,3 +356,81 @@ def said(rows: list[EmailOutbox]) -> tuple[bool, str]:
         ([f"Not sent yet to {'; '.join(later)}; Steward will try again for up to an hour."]
          if later else [])
     return bool(rows) and not failed and not later, " ".join(parts)
+
+
+# ── What the pages show (#5509) ───────────────────────────────────────────────
+
+KIND_LABELS = {"alert": "Alert", "report": "Weekly report", "invite": "Invite",
+               "reset": "Password reset", "test": "Test"}
+# Retrying longer than this is worth a line in Settings → System's verdict.
+SLOW_S = 300
+SHOWN_DAYS = 7
+
+
+def delivery(rows: list[EmailOutbox]) -> dict:
+    """One alert event's email, over every reader's copy, as history says
+    it: {"state", "at", "attempts", "error"}. Gathering while any copy is
+    held (at: when it sends); sending/retrying while any is still due (at:
+    the next try); failed when any reader wasn't reached (error names each);
+    else sent (at: the last acceptance)."""
+    if held := [r for r in rows if r.state == "held"]:
+        return {"state": "held", "at": min(r.send_after for r in held), "attempts": 0, "error": None}
+    if due := [r for r in rows if r.state in ("pending", "sending")]:
+        worst = max(due, key=lambda r: r.attempts)
+        return {"state": "retrying" if worst.attempts else "sending",
+                "at": min((r.next_attempt_at for r in due if r.next_attempt_at), default=None),
+                "attempts": worst.attempts, "error": worst.last_error}
+    if failed := [r for r in rows if r.state == "failed"]:
+        return {"state": "failed", "at": max(r.first_attempt_at or r.created_at for r in failed),
+                "attempts": max(r.attempts for r in failed),
+                "error": "; ".join(f"{r.recipient}: {r.last_error}" for r in failed)}
+    return {"state": "sent", "at": max((r.sent_at for r in rows if r.sent_at), default=None),
+            "attempts": max(r.attempts for r in rows), "error": None}
+
+
+async def email_states(db, event_ids: list[str]) -> dict[str, dict]:
+    """{event id: delivery(...)} for the alert events whose email went
+    through the outbox; an event with none (no reader, or from before it)
+    is absent."""
+    if not event_ids:
+        return {}
+    link = email_outbox_events.c
+    by: dict[str, list] = {}
+    for event_id, row in (await db.execute(
+            select(link.event_id, EmailOutbox).join(EmailOutbox, EmailOutbox.id == link.outbox_id)
+            .where(link.event_id.in_(event_ids)))).all():
+        by.setdefault(event_id, []).append(row)
+    return {eid: delivery(rows) for eid, rows in by.items()}
+
+
+async def outgoing(db, now: datetime | None = None, limit: int = 20) -> dict:
+    """Settings → System's "Outgoing email": every email of the last
+    SHOWN_DAYS not yet sent (failed first, then retrying longest, then
+    sending and gathering), how many were sent, and the problems for the
+    page's verdict. {"rows": [EmailOutbox], "more": n, "sent": n,
+    "problems": [str]}."""
+    from sqlalchemy import case, func
+    now = now or datetime.now(timezone.utc)
+    since = now - timedelta(days=SHOWN_DAYS)
+    open_states = ("failed", "pending", "sending", "held")
+    order = case({"failed": 0, "pending": 1, "sending": 2, "held": 3}, value=EmailOutbox.state)
+    base = select(EmailOutbox).where(EmailOutbox.created_at >= since,
+                                     EmailOutbox.state.in_(open_states))
+    rows = list((await db.execute(
+        base.order_by(order, EmailOutbox.first_attempt_at.asc().nulls_last(),
+                      EmailOutbox.created_at.desc()).limit(limit))).scalars())
+    counts = dict((await db.execute(
+        select(EmailOutbox.state, func.count()).where(EmailOutbox.created_at >= since)
+        .group_by(EmailOutbox.state))).all())
+    slow = (await db.execute(select(func.count()).select_from(EmailOutbox).where(
+        EmailOutbox.state.in_(("pending", "sending")),
+        EmailOutbox.first_attempt_at < now - timedelta(seconds=SLOW_S)))).scalar_one()
+    problems = []
+    if failed := counts.get("failed", 0):
+        problems.append(f"{failed} email{'s' if failed != 1 else ''} failed in the last {SHOWN_DAYS} days")
+    if slow:
+        problems.append(f"{slow} email{'s' if slow != 1 else ''} still not accepted by the mail "
+                        f"server after {SLOW_S // 60} minutes of trying")
+    shown = sum(counts.get(s, 0) for s in open_states)
+    return {"rows": rows, "more": max(shown - len(rows), 0), "sent": counts.get("sent", 0),
+            "problems": problems}