Add daily registration digest email (QUI-785)

AGENTS.md

@@ -50,4 +50,24 @@ Returns per-asset expected return (μ) and current holdings for the authenticate
 }
 ```
 
-Pre-solve: falls back to the agent's configured basket with `units=0`. Post-solve: reflects actual holdings. μ is the annualised hourly expected return over the last 7 days (`MU_WINDOW_HOURS=168`).
+Pre-solve: falls back to the agent's configured basket with `units=0`. Post-solve: reflects actual holdings. μ is the annualised hourly expected return over the last 7 days (`MU_WINDOW_HOURS=168`).
+
+## Registration digest
+
+A daily email of the registered-contact list to the team, so registration stats can be handed off
+without the operator acting each time. An in-app scheduler (`orchestration/digest_scheduler.py`,
+started in the app lifespan) sends once per UTC day at/after `DIGEST_HOUR_UTC`; the projection lives
+in `reporting/registrations.py` (modeled on `persistence/leaderboard.py`).
+
+- **Enable:** set a real `SMTP_PASSWORD` (the recipient defaults to `bd@postquant.xyz`). Empty
+  recipients or no SMTP ⇒ the scheduler is inert (**fail-closed**) — the contact list never reaches
+  the logs.
+- **Config:** `DIGEST_RECIPIENTS`, `DIGEST_HOUR_UTC` (default `7`). Deployment defaults the recipient to
+  the BD alias `bd@postquant.xyz` (see `docker-compose.yml`); override with a comma-separated list.
+- **Manual send / verify:** `qtw send-digest` (real send) or `qtw send-digest --dry-run` (render the
+  summary + CSV to stdout without sending).
+- **Safety:** the CSV uses an explicit column allowlist (`reporting.registrations.CSV_COLUMNS`) — the
+  agent `id` (the secret API key) is never selected or emitted — and cells are sanitized against
+  spreadsheet formula injection. The digest adds no HTTP route and no MCP tool.
+- **Accepted risk:** exported PII then lives in recipients' inboxes; treat those addresses and any
+  saved CSVs accordingly.

backend/src/backend/api/app.py

@@ -19,6 +19,7 @@
 from .. import config
 from ..db.engine import get_engine, init_engine
 from ..db.models import Base
+from ..orchestration import digest_scheduler
 from . import routes
 from .auth import APIKeyMiddleware
 
@@ -62,7 +63,9 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
     init_engine()
     async with get_engine().begin() as conn:
         await conn.run_sync(Base.metadata.create_all)
+    digest_scheduler.start(app)
     yield
+    await digest_scheduler.stop(app)
     # Note: the engine is process-wide and intentionally NOT disposed here —
     # tests share it across many app instances and own its lifecycle.
 

backend/src/backend/cli.py

@@ -279,6 +279,57 @@ def cmd_verify_dwave(args: argparse.Namespace) -> None:
         )
 
 
+def cmd_send_digest(args: argparse.Namespace) -> None:
+    """Build and send the registration digest now (bypasses the daily schedule).
+
+    ``--dry-run`` renders the summary + CSV to stdout without sending — use it to
+    verify the payload when SMTP is not configured. A real send requires both
+    ``DIGEST_RECIPIENTS`` and ``SMTP_PASSWORD``.
+    """
+    import asyncio
+
+    asyncio.run(_send_digest_async(args.dry_run))
+
+
+async def _send_digest_async(dry_run: bool) -> None:
+    from datetime import UTC, datetime
+
+    from .db.engine import get_engine, init_engine, session_scope
+    from .db.models import Base, DigestState
+    from .email.sender import get_email_sender
+    from .orchestration import digest_scheduler
+    from .reporting import registrations
+
+    init_engine()
+    async with get_engine().begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+
+    now = datetime.now(UTC)
+    today = now.date().isoformat()
+    async with session_scope() as session:
+        rows = await registrations.fetch_rows(session)
+        state = await session.get(DigestState, 1)
+        since = state.last_sent_at if state else None
+        stats = registrations.compute_stats(rows, since)
+        subject, text, _ = registrations.render(stats, today)
+
+        if dry_run:
+            print(f"[dry-run] would send to: {config.DIGEST_RECIPIENTS or '(none set)'}")
+            print(f"[dry-run] subject: {subject}\n")
+            print(text)
+            print(f"\n--- registrations-{today}.csv ---")
+            print(registrations.rows_to_csv(rows), end="")
+            return
+
+        if not config.DIGEST_RECIPIENTS:
+            raise SystemExit("DIGEST_RECIPIENTS is empty — set it (or use --dry-run)")
+        if not config.SMTP_PASSWORD:
+            raise SystemExit("SMTP is not configured — set SMTP_PASSWORD (or use --dry-run)")
+
+        sent = await digest_scheduler.send_now(session, get_email_sender(), now)
+    print(f"sent registration digest ({sent.total} registrants) to {config.DIGEST_RECIPIENTS}")
+
+
 def _print_speedup(winner, results, winner_label) -> None:
     if winner is None or winner.solve_time_s <= 0:
         return
@@ -320,6 +371,10 @@ def main(argv: list[str] | None = None) -> None:
     _add_common_args(p_verify)
     p_verify.set_defaults(func=cmd_verify_dwave)
 
+    p_digest = sub.add_parser("send-digest", help="send the registration digest now")
+    p_digest.add_argument("--dry-run", action="store_true", help="render without sending")
+    p_digest.set_defaults(func=cmd_send_digest)
+
     args = parser.parse_args(argv)
     try:
         args.func(args)

backend/src/backend/config.py

@@ -152,3 +152,23 @@
 SMTP_PASSWORD: str = os.environ.get("SMTP_PASSWORD", "")
 SMTP_STARTTLS: bool = os.environ.get("SMTP_STARTTLS", "true").lower() != "false"
 EMAIL_FROM: str = os.environ.get("EMAIL_FROM", "Qubitrefill <noreply@quip.network>")
+
+# -----------------------------------------------------------------------------
+# Registration digest — a daily email of the registered-contact list to the team
+# -----------------------------------------------------------------------------
+
+# Comma-separated recipient addresses. EMPTY ⇒ the digest is disabled (fail-closed):
+# the scheduler never runs and no PII leaves the DB. The digest also stays inert
+# unless SMTP_PASSWORD is set, so it can never fall back to logging the list.
+# The library default is empty; the deployment default (bd@postquant.xyz) is set
+# in docker-compose.yml.
+DIGEST_RECIPIENTS: list[str] = [
+    addr.strip() for addr in os.environ.get("DIGEST_RECIPIENTS", "").split(",") if addr.strip()
+]
+# Hour of day (UTC) at/after which the daily digest may send. The scheduler wakes
+# hourly and sends the first time it is past this hour with no send recorded today.
+# Validated fail-fast: an out-of-range value would silently never fire (the guard
+# is ``now.hour < DIGEST_HOUR_UTC``), so reject it at startup instead.
+DIGEST_HOUR_UTC: int = int(os.environ.get("DIGEST_HOUR_UTC", "7"))
+if not 0 <= DIGEST_HOUR_UTC <= 23:
+    raise ValueError(f"DIGEST_HOUR_UTC must be 0–23, got {DIGEST_HOUR_UTC}")

backend/src/backend/db/models.py

@@ -37,6 +37,22 @@ class Agent(Base):
     created_at: Mapped[str] = mapped_column(String(64))
 
 
+class DigestState(Base):
+    """Single-row marker for the registration digest scheduler.
+
+    Holds the ISO-8601 UTC timestamp (``…+00:00``) of the last successful send.
+    The date prefix drives once-per-day idempotency across restarts; the full
+    timestamp is the ``new_since`` boundary, so a registrant who signs up later on
+    the send's own day is still counted next time. The row is created on first
+    send and its ``id`` is always ``1``.
+    """
+
+    __tablename__ = "digest_state"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)  # always 1
+    last_sent_at: Mapped[str | None] = mapped_column(String(32), nullable=True)
+
+
 class Job(Base):
     __tablename__ = "jobs"
 

backend/src/backend/email/sender.py

@@ -40,6 +40,16 @@ def _render(name: str, api_key: str) -> tuple[str, str, str]:
 class EmailSender(Protocol):
     async def send_api_key(self, to_email: str, name: str, api_key: str) -> None: ...
 
+    async def send_digest(
+        self,
+        recipients: list[str],
+        subject: str,
+        text: str,
+        html: str,
+        csv_bytes: bytes,
+        csv_filename: str,
+    ) -> None: ...
+
 
 class SmtpEmailSender:
     """Sends via an SMTP relay (Resend: smtp.resend.com:587, STARTTLS)."""
@@ -62,23 +72,89 @@ async def send_api_key(self, to_email: str, name: str, api_key: str) -> None:
             timeout=10.0,
         )
 
+    async def send_digest(
+        self,
+        recipients: list[str],
+        subject: str,
+        text: str,
+        html: str,
+        csv_bytes: bytes,
+        csv_filename: str,
+    ) -> None:
+        msg = EmailMessage()
+        msg["From"] = config.EMAIL_FROM
+        msg["To"] = ", ".join(recipients)
+        msg["Subject"] = subject
+        msg.set_content(text)
+        msg.add_alternative(html, subtype="html")
+        msg.add_attachment(
+            csv_bytes, maintype="text", subtype="csv", filename=csv_filename
+        )
+        await aiosmtplib.send(
+            msg,
+            hostname=config.SMTP_HOST,
+            port=config.SMTP_PORT,
+            username=config.SMTP_USERNAME,
+            password=config.SMTP_PASSWORD,
+            start_tls=config.SMTP_STARTTLS,
+            timeout=30.0,
+        )
+
 
 class ConsoleEmailSender:
     """Dev fallback — logs the key instead of sending it."""
 
     async def send_api_key(self, to_email: str, name: str, api_key: str) -> None:
         log.warning("[email:console] API key for %s <%s>: %s", name, to_email, api_key)
 
+    async def send_digest(
+        self,
+        recipients: list[str],
+        subject: str,
+        text: str,
+        html: str,
+        csv_bytes: bytes,
+        csv_filename: str,
+    ) -> None:
+        # Redacted: never log the contact list. The scheduler already refuses to
+        # run without real SMTP; this is a second line of defense.
+        log.warning(
+            "[email:console] digest suppressed (no SMTP): %d recipients, %d CSV bytes",
+            len(recipients),
+            len(csv_bytes),
+        )
+
 
 class FakeEmailSender:
     """Test double — records every send for assertion."""
 
     def __init__(self) -> None:
         self.sent: list[dict[str, str]] = []
+        self.digests: list[dict] = []
 
     async def send_api_key(self, to_email: str, name: str, api_key: str) -> None:
         self.sent.append({"email": to_email, "name": name, "api_key": api_key})
 
+    async def send_digest(
+        self,
+        recipients: list[str],
+        subject: str,
+        text: str,
+        html: str,
+        csv_bytes: bytes,
+        csv_filename: str,
+    ) -> None:
+        self.digests.append(
+            {
+                "recipients": recipients,
+                "subject": subject,
+                "text": text,
+                "html": html,
+                "csv": csv_bytes.decode("utf-8"),
+                "filename": csv_filename,
+            }
+        )
+
 
 def get_email_sender() -> EmailSender:
     if config.SMTP_PASSWORD: