,
,
Markets did stuff today.
", + unsubscribe_url="https://read.markets/email/unsubscribe?token=abc", + settings_url="https://read.markets/settings", + ) + assert "Daily" in subj + assert "2026-05-25" in subj + assert "Markets did stuff today" in html + assert "abc" in html # unsubscribe link landed + assert "/settings" in html + # Plain-text fallback strips HTML. + assert "" not in text + assert "Markets did stuff today" in text + + +def test_weekly_subject_says_recap(): + subj, _, _ = render_digest_email( + kind="weekly", + date_str="2026-05-25", + content_html="
x
", + unsubscribe_url="https://x/u", + settings_url="https://x/s", + ) + assert "Weekly" in subj + assert "recap" in subj.lower() + + +def test_invalid_kind_raises(): + import pytest + with pytest.raises(ValueError): + render_digest_email( + kind="bogus", date_str="2026-05-25", + content_html="x
", + unsubscribe_url="u", settings_url="s", + ) +``` + +- [ ] **Step 2: Run the failing test** + +```bash +pytest tests/test_email_render.py -v +``` + +Expected: FAIL — `render_digest_email` doesn't exist. + +- [ ] **Step 3: Implement `render_digest_email`** + +Open `app/services/email_service.py`. Add at the bottom of the file: + +```python +# --------------------------------------------------------------------------- +# Digest email rendering +# --------------------------------------------------------------------------- + +import html as _html_lib +import re as _re + + +_DIGEST_HTML_TEMPLATE = """\ + + + + + + +|
+
+ ▰ {brand_upper} · {label_upper}
+
+
+ {content_html}
+
+ |
You're unsubscribed from email digests.
++ You can re-enable digests any time from + Settings. +
+x
"): + """Stub out the LLM call so the test never hits the network. Use a + SimpleNamespace so we don't have to know the real result class name.""" + from types import SimpleNamespace + async def _fake(_client, messages, **kwargs): + return SimpleNamespace( + content=content, model="stub", + prompt_tokens=10, completion_tokens=10, cost_usd=0.0, + ) + return _fake + + +@pytest.mark.skipif(False, reason="requires aiosqlite") +def test_daily_run_only_paid_opt_in(tmp_path): + _bootstrap(tmp_path, today_weekday=0) + from app.jobs import email_digest_job + with patch("app.jobs.email_digest_job._now", + return_value=_patch_today(0)), \ + patch("app.jobs.email_digest_job.send_email", + new=AsyncMock()) as send_mock, \ + patch("app.jobs.email_digest_job.call_llm", + new=AsyncMock(side_effect=_stub_generate())): + asyncio.run(email_digest_job.run()) + # Only user_id=1 (paid + opt-in) received an email. Two calls + # (one per tone), but only one recipient. + sent_to = {kwargs["to"] for _, _, kwargs in send_mock.mock_calls + if "to" in kwargs} + addresses_sent = {call.kwargs.get("to") for call in send_mock.await_args_list} + assert addresses_sent == {"paid_in@x"} + + +def test_weekly_run_includes_free_and_paid_opt_in(tmp_path): + _bootstrap(tmp_path, today_weekday=6) + from app.jobs import email_digest_job + with patch("app.jobs.email_digest_job._now", + return_value=_patch_today(6)), \ + patch("app.jobs.email_digest_job.send_email", + new=AsyncMock()) as send_mock, \ + patch("app.jobs.email_digest_job.call_llm", + new=AsyncMock(side_effect=_stub_generate())): + asyncio.run(email_digest_job.run()) + addresses_sent = {call.kwargs.get("to") for call in send_mock.await_args_list} + assert addresses_sent == {"paid_in@x", "free_in@x"} + + +def test_second_run_same_day_is_idempotent(tmp_path): + _bootstrap(tmp_path, today_weekday=0) + from app.jobs import email_digest_job + with patch("app.jobs.email_digest_job._now", + return_value=_patch_today(0)), \ + patch("app.jobs.email_digest_job.send_email", + new=AsyncMock()) as send_mock, \ + patch("app.jobs.email_digest_job.call_llm", + new=AsyncMock(side_effect=_stub_generate())): + asyncio.run(email_digest_job.run()) + first_count = len(send_mock.await_args_list) + asyncio.run(email_digest_job.run()) + second_count = len(send_mock.await_args_list) + assert first_count > 0 + assert second_count == first_count, "second run should not re-send" +``` + +- [ ] **Step 2: Run the failing test** + +```bash +pytest tests/test_email_digest_job.py -v +``` + +Expected: FAIL — `app.jobs.email_digest_job` doesn't exist yet. + +- [ ] **Step 3: Implement the job** + +Create `app/jobs/email_digest_job.py`: + +```python +"""Daily/weekly editorial email digest. + +Runs once a day at 06:30 UTC via the scheduler. On Sundays sends the +weekly recap to every opt-in user (free + paid). On other days sends +the daily digest to opt-in paid users only. + +Generates LLM content once per tone (NOVICE + INTERMEDIATE), then fans +out by SMTP. EmailSend audit rows guard against double-delivery if the +job is re-run within the same UTC day. +""" +from __future__ import annotations + +import asyncio +from collections import defaultdict +from datetime import datetime, timedelta, timezone + +import httpx +from sqlalchemy import desc, func, select + +from app import branding +from app.config import get_settings +from app.db import utcnow +from app.jobs._helpers import job_lifecycle, log +from app.jobs.ai_log_job import ( + REFERENCE_LINE, + _latest_quotes_by_group, + _recent_headlines_by_bucket, + _month_spend, +) +from app.models import EmailSend, User +from app.routers.email import sign_unsubscribe_token +from app.services.email_service import render_digest_email, send_email +from app.services.openrouter import ( + PROMPT_VERSION, + active_model, + build_daily_digest_prompt, + build_weekly_digest_prompt, + call_llm, + llm_configured, +) +from app.services.access import paid_status + + +# Indirection so tests can monkeypatch the "current time" without +# touching the system clock. +def _now() -> datetime: + return utcnow() + + +async def _opt_in_recipients(session, *, paid_only: bool) -> list[User]: + stmt = select(User).where(User.email_digest_opt_in.is_(True)) + rows = (await session.execute(stmt)).scalars().all() + if paid_only: + rows = [u for u in rows if paid_status(u).active] + return rows + + +async def _already_sent_today(session, user_id: int, kind: str, today: datetime) -> bool: + """True if an EmailSend row exists for this user+kind on the same UTC + day, with status in ('sent','error'). 'error' counts because we don't + want to keep retrying a bad address inside the same daily slot.""" + day_start = today.replace(hour=0, minute=0, second=0, microsecond=0) + day_end = day_start + timedelta(days=1) + stmt = select(EmailSend.id).where( + EmailSend.user_id == user_id, + EmailSend.kind == kind, + EmailSend.sent_at >= day_start, + EmailSend.sent_at < day_end, + EmailSend.status.in_(("sent", "error")), + ) + return (await session.execute(stmt)).first() is not None + + +async def _generate_variants(client, kind: str, ctx: dict) -> dict[str, str]: + """Returns {tone: html_content}. Missing tone means generation failed + for that variant — skip recipients on that tone.""" + builder = build_weekly_digest_prompt if kind == "weekly" else build_daily_digest_prompt + out: dict[str, str] = {} + for tone in ("NOVICE", "INTERMEDIATE"): + sys_, usr = builder(tone=tone, **ctx) + try: + result = await call_llm( + client, + [{"role": "system", "content": sys_}, + {"role": "user", "content": usr}], + ) + out[tone] = result.content + log.info("digest.variant_ok", kind=kind, tone=tone, + prompt_tokens=result.prompt_tokens, + completion_tokens=result.completion_tokens) + except Exception as e: + log.error("digest.variant_failed", kind=kind, tone=tone, + error=str(e)[:200]) + return out + + +def _kind_for_today(today: datetime) -> str | None: + """Sunday → weekly. Mon–Sat → daily. None means 'no run today'.""" + return "weekly" if today.weekday() == 6 else "daily" + + +async def _send_one(user: User, kind: str, content_html: str, date_str: str, + session) -> None: + settings_url = f"{branding.SITE_URL}/settings" + unsubscribe_url = ( + f"{branding.SITE_URL}/email/unsubscribe" + f"?token={sign_unsubscribe_token(user.id)}" + ) + subject, text_body, html_body = render_digest_email( + kind=kind, date_str=date_str, + content_html=content_html, + unsubscribe_url=unsubscribe_url, + settings_url=settings_url, + ) + try: + await send_email(to=user.email, subject=subject, + text_body=text_body, html_body=html_body) + status_ = "sent" + err = None + except Exception as e: + status_ = "error" + err = str(e)[:255] + log.error("digest.send_failed", user_id=user.id, error=err) + session.add(EmailSend( + user_id=user.id, kind=kind, sent_at=_now(), + status=status_, error=err, + )) + await session.commit() + + +async def run() -> None: + async with job_lifecycle("email_digest_job") as (session, jr): + if jr.status == "skipped": + return + s = get_settings() + if not llm_configured(): + log.warning("digest.skipped_no_key", provider=s.LLM_PROVIDER) + jr.status = "skipped" + return + + today = _now() + kind = _kind_for_today(today) + date_str = today.strftime("%Y-%m-%d") + + # Build the recipient list before LLM work — if it's empty, + # skip the LLM spend. + recipients = await _opt_in_recipients( + session, paid_only=(kind == "daily"), + ) + # Filter out anyone already sent today (idempotency). + fresh: list[User] = [] + for u in recipients: + if not await _already_sent_today(session, u.id, kind, today): + fresh.append(u) + if not fresh: + log.info("digest.no_fresh_recipients", kind=kind, + total=len(recipients)) + jr.status = "skipped" + return + + spent = await _month_spend(session) + if spent >= s.OPENROUTER_MONTHLY_CAP_USD: + log.warning("digest.cap_reached", spent=spent, + cap=s.OPENROUTER_MONTHLY_CAP_USD) + jr.status = "skipped" + jr.error = f"monthly cost cap reached (${spent:.2f})" + return + + quotes = await _latest_quotes_by_group(session) + news = await _recent_headlines_by_bucket( + session, hours=(168 if kind == "weekly" else 24), + ) + ctx = dict( + today=today, + quotes_by_group=quotes, + headlines_by_bucket=news, + reference_line=REFERENCE_LINE, + ) + + async with httpx.AsyncClient(follow_redirects=True) as client: + variants = await _generate_variants(client, kind, ctx) + + if not variants: + log.warning("digest.all_variants_failed", kind=kind) + jr.status = "failed" + jr.error = "all variants failed" + return + + written = 0 + for u in fresh: + tone = (u.digest_tone or "INTERMEDIATE").upper() + content = variants.get(tone) or variants.get("INTERMEDIATE") + if content is None: + # Both variants failed for this user's tone — skip. + continue + await _send_one(u, kind, content, date_str, session) + await asyncio.sleep(0.1) # gentle SMTP pacing + written += 1 + + jr.items_written = written + log.info("digest.done", kind=kind, written=written, + prompt_version=PROMPT_VERSION) + + +if __name__ == "__main__": + asyncio.run(run()) +``` + +- [ ] **Step 4: Run the tests, expect PASS** + +```bash +pytest tests/test_email_digest_job.py -v +``` + +Expected: 3 passed. + +- [ ] **Step 5: Commit** + +```bash +git add app/jobs/email_digest_job.py tests/test_email_digest_job.py +git commit -m "digest: daily/weekly job w/ EmailSend idempotency" +``` + +--- + +## Task 8: Scheduler registration + +**Files:** +- Modify: `app/scheduler_main.py` + +- [ ] **Step 1: Add the cron entry** + +Open `app/scheduler_main.py`. Locate the block where the other jobs are registered (line ~42 onward — `sched.add_job(ai_log_job.run, ...)` etc.). Add an import at the top and a new `sched.add_job` line: + +```python +from app.jobs import email_digest_job +# ... +sched.add_job( + email_digest_job.run, + CronTrigger(hour=6, minute=30), + name="email_digest_job", + id="email_digest_job", +) +``` + +- [ ] **Step 2: Verify the scheduler boots clean** + +```bash +docker compose restart scheduler +docker compose logs --tail=40 scheduler +``` + +Expected: `scheduler.started jobs=[..., 'email_digest_job']` in the log. + +- [ ] **Step 3: Commit** + +```bash +git add app/scheduler_main.py +git commit -m "scheduler: register email_digest_job at 06:30 UTC" +``` + +--- + +## Task 9: Settings page section + PATCH endpoint + +**Files:** +- Modify: `app/routers/api.py` (add PATCH /settings/digest) +- Modify: `app/templates/settings.html` +- Modify: `app/routers/pages.py` (settings_page context: pass last EmailSend) +- Create: `tests/test_settings_digest_api.py` + +- [ ] **Step 1: Write the failing test** + +Create `tests/test_settings_digest_api.py`: + +```python +"""PATCH /api/settings/digest persists opt-in + tone.""" +from __future__ import annotations + +import asyncio + +import pytest + + +def _build(tmp_path): + from fastapi import FastAPI + from fastapi.testclient import TestClient + from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + + from app import db as db_mod + from app.auth import sign_session + from app.db import Base + from app.models import User + from app.routers import api as api_router + + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/s.db") + factory = async_sessionmaker(engine, expire_on_commit=False) + db_mod._engine = engine + db_mod._session_factory = factory + + async def _seed(): + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + async with factory() as s: + s.add(User(id=1, email="u@x", tier="paid", + email_digest_opt_in=True)) + await s.commit() + + asyncio.run(_seed()) + app = FastAPI() + app.include_router(api_router.router, prefix="/api") + return TestClient(app), sign_session(1) + + +def test_patch_round_trip(tmp_path): + client, sess = _build(tmp_path) + r = client.patch( + "/api/settings/digest", + json={"opt_in": False, "tone": "NOVICE"}, + cookies={"cassandra_session": sess}, + ) + assert r.status_code == 200, r.text + assert r.json() == {"opt_in": False, "tone": "NOVICE"} + + # Round-trip GET to confirm persistence. + async def _check(): + from app import db as db_mod + from app.models import User + async with db_mod._session_factory() as s: + u = await s.get(User, 1) + assert u.email_digest_opt_in is False + assert u.digest_tone == "NOVICE" + asyncio.run(_check()) + + +def test_patch_rejects_invalid_tone(tmp_path): + client, sess = _build(tmp_path) + r = client.patch( + "/api/settings/digest", + json={"opt_in": True, "tone": "PRO"}, # not in NOVICE|INTERMEDIATE + cookies={"cassandra_session": sess}, + ) + assert r.status_code == 422 + + +def test_patch_requires_auth(tmp_path): + client, _ = _build(tmp_path) + r = client.patch("/api/settings/digest", + json={"opt_in": True, "tone": "NOVICE"}) + assert r.status_code in (401, 303) +``` + +- [ ] **Step 2: Run the failing test** + +```bash +pytest tests/test_settings_digest_api.py -v +``` + +Expected: FAIL — endpoint not defined. + +- [ ] **Step 3: Add the endpoint** + +Open `app/routers/api.py`. Near the other settings-related code (or at the bottom of the file), add: + +```python +from pydantic import BaseModel, Field +from typing import Literal + + +class DigestPrefsIn(BaseModel): + opt_in: bool + tone: Literal["NOVICE", "INTERMEDIATE"] + + +class DigestPrefsOut(BaseModel): + opt_in: bool + tone: str + + +@router.patch("/settings/digest", response_model=DigestPrefsOut) +async def patch_digest_prefs( + payload: DigestPrefsIn, + principal: CurrentUser = Depends(require_auth), + session: AsyncSession = Depends(get_session), +) -> DigestPrefsOut: + if principal.user is None: + # Admin bearer-token path: nothing to persist. + raise HTTPException(status_code=400, detail="no_user_context") + principal.user.email_digest_opt_in = payload.opt_in + principal.user.digest_tone = payload.tone + await session.commit() + return DigestPrefsOut(opt_in=payload.opt_in, tone=payload.tone) +``` + +If `require_auth`, `CurrentUser`, or `HTTPException` aren't already imported, extend the import block at the top of the file. + +- [ ] **Step 4: Run the tests, expect PASS** + +```bash +pytest tests/test_settings_digest_api.py -v +``` + +Expected: 3 passed. + +- [ ] **Step 5: Add the Settings template section** + +Open `app/templates/settings.html`. Find a sensible insertion point (after the existing tier/credit section, before the cloud-sync section). Add: + +```html +