diff --git a/api_accounts.py b/api_accounts.py
new file mode 100644
index 0000000..86a3011
--- /dev/null
+++ b/api_accounts.py
@@ -0,0 +1,196 @@
+"""Authenticated API for subscriptions and notifications.
+
+All routes require a logged-in user (the fastapi-users cookie session) and are
+scoped to that user — a row that isn't theirs reads as 404, never 403, so the API
+doesn't leak which ids exist. Mounted under {BASE}/api/v2 alongside the rest of v2.
+"""
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+import climate
+import grid
+from db import get_async_session
+from models import Notification, Subscription
+from schemas import (
+    NotificationList,
+    NotificationOut,
+    SubscriptionIn,
+    SubscriptionOut,
+    SubscriptionPatch,
+)
+from users import current_active_user
+
+router = APIRouter(tags=["accounts"])
+
+
+async def _owned_subscription(session: AsyncSession, sub_id: int, user) -> Subscription:
+    sub = (
+        await session.execute(
+            select(Subscription).where(
+                Subscription.id == sub_id, Subscription.user_id == user.id
+            )
+        )
+    ).scalar_one_or_none()
+    if sub is None:
+        raise HTTPException(status_code=404, detail="Subscription not found.")
+    return sub
+
+
+# --- subscriptions -----------------------------------------------------------
+@router.get("/subscriptions", response_model=list[SubscriptionOut])
+async def list_subscriptions(
+    user=Depends(current_active_user),
+    session: AsyncSession = Depends(get_async_session),
+):
+    rows = (
+        await session.execute(
+            select(Subscription)
+            .where(Subscription.user_id == user.id)
+            .order_by(Subscription.created_at)
+        )
+    ).scalars().all()
+    return rows
+
+
+@router.post("/subscriptions", response_model=SubscriptionOut, status_code=201)
+async def create_subscription(
+    body: SubscriptionIn,
+    user=Depends(current_active_user),
+    session: AsyncSession = Depends(get_async_session),
+):
+    cell = grid.snap(body.lat, body.lon)
+    cell_id = cell["id"]
+
+    dup = (
+        await session.execute(
+            select(Subscription).where(
+                Subscription.user_id == user.id,
+                Subscription.cell_id == cell_id,
+                Subscription.kind == body.kind,
+            )
+        )
+    ).scalar_one_or_none()
+    if dup is not None:
+        raise HTTPException(
+            status_code=409,
+            detail=f"You already have a {body.kind} alert for this location.",
+        )
+
+    label = body.label
+    if not label:
+        # Cache-only lookup — never block the event loop on Nominatim. The frontend
+        # normally supplies the label from /api/v2/place; this is just a fallback.
+        _, label = climate.reverse_geocode_cached(cell["center_lat"], cell["center_lon"])
+
+    sub = Subscription(
+        user_id=user.id,
+        cell_id=cell_id,
+        label=label,
+        lat=body.lat,
+        lon=body.lon,
+        threshold=body.threshold,
+        metrics=body.metrics,
+        kind=body.kind,
+        two_sided=body.two_sided,
+    )
+    session.add(sub)
+    await session.commit()
+    await session.refresh(sub)
+    return sub
+
+
+@router.patch("/subscriptions/{sub_id}", response_model=SubscriptionOut)
+async def update_subscription(
+    sub_id: int,
+    body: SubscriptionPatch,
+    user=Depends(current_active_user),
+    session: AsyncSession = Depends(get_async_session),
+):
+    sub = await _owned_subscription(session, sub_id, user)
+    for key, value in body.model_dump(exclude_unset=True).items():
+        setattr(sub, key, value)
+    await session.commit()
+    await session.refresh(sub)
+    return sub
+
+
+@router.delete("/subscriptions/{sub_id}", status_code=204)
+async def delete_subscription(
+    sub_id: int,
+    user=Depends(current_active_user),
+    session: AsyncSession = Depends(get_async_session),
+):
+    sub = await _owned_subscription(session, sub_id, user)
+    await session.delete(sub)
+    await session.commit()
+
+
+# --- notifications -----------------------------------------------------------
+@router.get("/notifications", response_model=NotificationList)
+async def list_notifications(
+    unread: bool = Query(False, description="only return unread notifications"),
+    limit: int = Query(50, ge=1, le=200),
+    user=Depends(current_active_user),
+    session: AsyncSession = Depends(get_async_session),
+):
+    q = select(Notification).where(Notification.user_id == user.id)
+    if unread:
+        q = q.where(Notification.read_at.is_(None))
+    q = q.order_by(Notification.created_at.desc()).limit(limit)
+    rows = (await session.execute(q)).scalars().all()
+
+    unread_count = (
+        await session.execute(
+            select(func.count())
+            .select_from(Notification)
+            .where(Notification.user_id == user.id, Notification.read_at.is_(None))
+        )
+    ).scalar_one()
+    return NotificationList(
+        notifications=[NotificationOut.model_validate(r) for r in rows],
+        unread_count=unread_count,
+    )
+
+
+@router.post("/notifications/{notif_id}/read", status_code=204)
+async def mark_notification_read(
+    notif_id: int,
+    user=Depends(current_active_user),
+    session: AsyncSession = Depends(get_async_session),
+):
+    import time
+
+    notif = (
+        await session.execute(
+            select(Notification).where(
+                Notification.id == notif_id, Notification.user_id == user.id
+            )
+        )
+    ).scalar_one_or_none()
+    if notif is None:
+        raise HTTPException(status_code=404, detail="Notification not found.")
+    if notif.read_at is None:
+        notif.read_at = time.time()
+        await session.commit()
+
+
+@router.post("/notifications/read-all", status_code=204)
+async def mark_all_read(
+    user=Depends(current_active_user),
+    session: AsyncSession = Depends(get_async_session),
+):
+    import time
+
+    rows = (
+        await session.execute(
+            select(Notification).where(
+                Notification.user_id == user.id, Notification.read_at.is_(None)
+            )
+        )
+    ).scalars().all()
+    now = time.time()
+    for n in rows:
+        n.read_at = now
+    if rows:
+        await session.commit()
diff --git a/app.py b/app.py
index 8f7e61a..6351d97 100644
--- a/app.py
+++ b/app.py
@@ -15,12 +15,17 @@ from fastapi.middleware.gzip import GZipMiddleware
 from fastapi.responses import RedirectResponse
 from fastapi.staticfiles import StaticFiles
 
+import api_accounts
 import audit
 import climate
+import db
 import grid
+import notify
 import places
 import store
+import users
 import views
+from schemas import UserCreate, UserRead, UserUpdate
 
 FRONTEND_DIR = os.path.join(os.path.dirname(__file__), "..", "frontend")
 
@@ -105,9 +110,15 @@ async def _lifespan(app):
     # hook (not import time) so offline importers (tests, the migrate script's
     # dependencies) don't kick off a GeoNames download. The neighbor warmer is
     # also a server concern: enqueues before startup just wait in the queue.
+    # Create the account DB tables (users/sessions/subscriptions/notifications)
+    # before serving — a fast no-op once they exist.
+    await db.create_db_and_tables()
     places.start_loading()
     threading.Thread(target=_neighbor_warmer, name="neighbor-warmer", daemon=True).start()
+    # Background subscription evaluator (in-process; single-worker deploy assumed).
+    notify.start()
     yield
+    notify.stop()
 
 
 app = FastAPI(title="Thermograph", version="0.2.0", lifespan=_lifespan)
@@ -127,7 +138,7 @@ async def revalidate_static(request, call_next):
     deploy immediately (no "my change isn't showing" bugs)."""
     response = await call_next(request)
     path = request.url.path
-    pages = (BASE, f"{BASE}/", f"{BASE}/calendar", f"{BASE}/day", f"{BASE}/compare", f"{BASE}/legend")
+    pages = (BASE, f"{BASE}/", f"{BASE}/calendar", f"{BASE}/day", f"{BASE}/compare", f"{BASE}/legend", f"{BASE}/alerts")
     if path.endswith((".js", ".css", ".html")) or path in pages:
         response.headers["Cache-Control"] = "no-cache"
     return response
@@ -542,6 +553,25 @@ app.include_router(v1, prefix=f"{BASE}/api")        # legacy unversioned == v1 (
 app.include_router(v1, prefix=f"{BASE}/api/v1")
 app.include_router(v2, prefix=f"{BASE}/api/v2")
 
+# --- accounts (fastapi-users) ----------------------------------------------
+# Library-provided routers: /auth/login + /auth/logout (cookie session),
+# /auth/register (signup), and /users/me (session check / profile). All under the
+# same v2 prefix as the rest of the API, so the frontend calls them base-relative.
+app.include_router(
+    users.fastapi_users.get_auth_router(users.auth_backend),
+    prefix=f"{BASE}/api/v2/auth", tags=["auth"],
+)
+app.include_router(
+    users.fastapi_users.get_register_router(UserRead, UserCreate),
+    prefix=f"{BASE}/api/v2/auth", tags=["auth"],
+)
+app.include_router(
+    users.fastapi_users.get_users_router(UserRead, UserUpdate),
+    prefix=f"{BASE}/api/v2/users", tags=["users"],
+)
+# Subscriptions + notifications (authed, user-scoped).
+app.include_router(api_accounts.router, prefix=f"{BASE}/api/v2")
+
 
 # --- static frontend (served under BASE) -----------------------------------
 def _page(name):
@@ -576,6 +606,7 @@ app.add_api_route(f"{BASE}/calendar", _page("calendar.html"), methods=["GET", "H
 app.add_api_route(f"{BASE}/day", _page("day.html"), methods=["GET", "HEAD"], include_in_schema=False)
 app.add_api_route(f"{BASE}/compare", _page("compare.html"), methods=["GET", "HEAD"], include_in_schema=False)
 app.add_api_route(f"{BASE}/legend", _page("legend.html"), methods=["GET", "HEAD"], include_in_schema=False)
+app.add_api_route(f"{BASE}/alerts", _page("subscriptions.html"), methods=["GET", "HEAD"], include_in_schema=False)
 
 # Everything else under BASE (app.js, style.css, nav.js, …) is a static asset.
 # Registered last so the explicit page routes above win.
diff --git a/db.py b/db.py
new file mode 100644
index 0000000..660fcc7
--- /dev/null
+++ b/db.py
@@ -0,0 +1,75 @@
+"""Authoritative account database — SQLAlchemy over data/accounts.sqlite (WAL).
+
+This is deliberately a *separate* database from data/thermograph.sqlite (store.py).
+That file is a disposable accelerator — "deleting it is a safe reset" — because
+everything in it recomputes from the parquet source of truth. Accounts,
+subscriptions and notifications have no source to recompute from, so they live in
+their own file with their own (stricter) rules: foreign keys on, errors surface
+rather than get swallowed, and it should be backed up (it is not regenerable).
+
+Two engines share the one file:
+
+* an **async** engine (``sqlite+aiosqlite``) drives the request/auth path, because
+  fastapi-users is async; and
+* a **sync** engine drives the background notifier thread (notify.py), which is a
+  plain daemon thread with no event loop.
+
+WAL mode lets the async readers/writers and the sync notifier coexist on the same
+file without blocking each other.
+"""
+import os
+
+from sqlalchemy import create_engine, event
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.orm import DeclarativeBase, sessionmaker
+
+DB_PATH = os.path.abspath(
+    os.path.join(os.path.dirname(__file__), "..", "data", "accounts.sqlite")
+)
+os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
+
+
+class Base(DeclarativeBase):
+    """Declarative base shared by every account-domain table (models.py)."""
+
+
+def _apply_pragmas(dbapi_conn, _rec):
+    # Per-connection SQLite setup: WAL for concurrent async/sync access, NORMAL
+    # sync for a good durability/speed tradeoff, and foreign_keys ON so the
+    # ON DELETE CASCADE relationships (user -> subscriptions -> notifications)
+    # are actually enforced (SQLite defaults them off).
+    cur = dbapi_conn.cursor()
+    cur.execute("PRAGMA journal_mode=WAL")
+    cur.execute("PRAGMA synchronous=NORMAL")
+    cur.execute("PRAGMA foreign_keys=ON")
+    cur.close()
+
+
+# --- async engine (web / auth path) -----------------------------------------
+async_engine = create_async_engine(f"sqlite+aiosqlite:///{DB_PATH}", future=True)
+event.listen(async_engine.sync_engine, "connect", _apply_pragmas)
+async_session_maker = async_sessionmaker(async_engine, expire_on_commit=False)
+
+
+async def get_async_session() -> AsyncSession:
+    """FastAPI dependency: one AsyncSession per request."""
+    async with async_session_maker() as session:
+        yield session
+
+
+# --- sync engine (background notifier thread) -------------------------------
+sync_engine = create_engine(f"sqlite:///{DB_PATH}", future=True)
+event.listen(sync_engine, "connect", _apply_pragmas)
+sync_session_maker = sessionmaker(sync_engine, expire_on_commit=False)
+
+
+async def create_db_and_tables() -> None:
+    """Create any missing tables. Called once at startup (app lifespan).
+
+    Imported for its side effect of registering the mapped classes on Base.metadata
+    before create_all runs.
+    """
+    import models  # noqa: F401  (registers tables on Base.metadata)
+
+    async with async_engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
diff --git a/models.py b/models.py
new file mode 100644
index 0000000..86da3ca
--- /dev/null
+++ b/models.py
@@ -0,0 +1,111 @@
+"""ORM tables for the account domain (data/accounts.sqlite).
+
+``User`` / ``AccessToken`` are fastapi-users' base tables (UUID primary keys);
+the access-token table backs a database session strategy, so logins survive a
+process restart and are individually revocable. ``Subscription`` and
+``Notification`` are our own, keyed to a user and cascading on delete.
+"""
+import time
+import uuid
+
+from fastapi_users_db_sqlalchemy import SQLAlchemyBaseUserTableUUID
+from fastapi_users_db_sqlalchemy.access_token import SQLAlchemyBaseAccessTokenTableUUID
+from fastapi_users_db_sqlalchemy.generics import GUID
+from sqlalchemy import (
+    JSON,
+    Boolean,
+    CheckConstraint,
+    Float,
+    ForeignKey,
+    Index,
+    Integer,
+    String,
+    Text,
+    UniqueConstraint,
+)
+from sqlalchemy.orm import Mapped, mapped_column
+
+from db import Base
+
+
+class User(SQLAlchemyBaseUserTableUUID, Base):
+    # Inherits id (UUID), email (unique), hashed_password, is_active,
+    # is_superuser, is_verified. One optional extra:
+    display_name: Mapped[str | None] = mapped_column(String(120), nullable=True)
+
+
+class AccessToken(SQLAlchemyBaseAccessTokenTableUUID, Base):
+    # Inherits token (PK), user_id (FK -> user.id), created_at. Rows here ARE the
+    # sessions: DatabaseStrategy looks a cookie's token up in this table.
+    pass
+
+
+class Subscription(Base):
+    __tablename__ = "subscription"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    user_id: Mapped[uuid.UUID] = mapped_column(
+        GUID, ForeignKey("user.id", ondelete="CASCADE"), nullable=False
+    )
+    # grid.snap(lat, lon)["id"] — the stable per-location key used across the app.
+    cell_id: Mapped[str] = mapped_column(String(40), nullable=False)
+    label: Mapped[str | None] = mapped_column(String(200), nullable=True)
+    lat: Mapped[float] = mapped_column(Float, nullable=False)
+    lon: Mapped[float] = mapped_column(Float, nullable=False)
+    # Unusualness cutoff the user picked; the low tail mirrors it at 100-threshold.
+    threshold: Mapped[int] = mapped_column(Integer, nullable=False)
+    # Grading metric keys this subscription watches, e.g. ["tmax", "feels", "precip"].
+    metrics: Mapped[list] = mapped_column(JSON, nullable=False, default=list)
+    # 'observed' (a recorded day crossed) or 'forecast' (an upcoming day is projected to).
+    kind: Mapped[str] = mapped_column(String(16), nullable=False, default="observed")
+    # Also alert the cold/low tail for temperature-like metrics (precip stays one-sided).
+    two_sided: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
+    active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
+    # Epoch seconds of the last notification emitted — enforces the weekly cap.
+    last_notified_at: Mapped[float | None] = mapped_column(Float, nullable=True)
+    created_at: Mapped[float] = mapped_column(Float, nullable=False, default=time.time)
+
+    __table_args__ = (
+        CheckConstraint("threshold BETWEEN 95 AND 99", name="ck_sub_threshold"),
+        CheckConstraint("kind IN ('observed','forecast')", name="ck_sub_kind"),
+        # One observed + one forecast subscription per location per user.
+        UniqueConstraint("user_id", "cell_id", "kind", name="uq_sub_user_cell_kind"),
+        Index("idx_sub_user", "user_id"),
+        Index("idx_sub_active", "active"),
+    )
+
+
+class Notification(Base):
+    __tablename__ = "notification"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    user_id: Mapped[uuid.UUID] = mapped_column(
+        GUID, ForeignKey("user.id", ondelete="CASCADE"), nullable=False
+    )
+    subscription_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("subscription.id", ondelete="CASCADE"), nullable=False
+    )
+    event_date: Mapped[str] = mapped_column(String(10), nullable=False)  # YYYY-MM-DD
+    metric: Mapped[str] = mapped_column(String(16), nullable=False)
+    direction: Mapped[str] = mapped_column(String(4), nullable=False)  # 'high' | 'low'
+    kind: Mapped[str] = mapped_column(String(16), nullable=False, default="observed")
+    percentile: Mapped[float] = mapped_column(Float, nullable=False)
+    value: Mapped[float | None] = mapped_column(Float, nullable=True)
+    grade: Mapped[str | None] = mapped_column(String(40), nullable=True)
+    title: Mapped[str] = mapped_column(String(200), nullable=False)
+    body: Mapped[str | None] = mapped_column(Text, nullable=True)
+    # 'inapp' today; the seam for future 'email' / 'push' delivery.
+    channel: Mapped[str] = mapped_column(String(16), nullable=False, default="inapp")
+    created_at: Mapped[float] = mapped_column(Float, nullable=False, default=time.time)
+    read_at: Mapped[float | None] = mapped_column(Float, nullable=True)  # NULL == unread
+
+    __table_args__ = (
+        # The dedup key: a given event (day+metric+direction+kind) notifies a
+        # subscription at most once, so re-running the evaluator never repeats it.
+        UniqueConstraint(
+            "subscription_id", "event_date", "metric", "direction", "kind",
+            name="uq_notif_event",
+        ),
+        Index("idx_notif_user_created", "user_id", "created_at"),
+        Index("idx_notif_user_read", "user_id", "read_at"),
+    )
diff --git a/notify.py b/notify.py
new file mode 100644
index 0000000..36d499e
--- /dev/null
+++ b/notify.py
@@ -0,0 +1,251 @@
+"""Subscription evaluation engine — the background worker that turns unusual
+weather into notifications.
+
+A daemon thread (modeled on app.py's neighbor warmer) wakes on an interval and, for
+every active subscription, checks whether the subscribed cell's recent/forecast
+weather crossed the user's percentile threshold on any watched metric. Crossings
+become rows in the notifications table.
+
+Two guards keep it quiet and cheap:
+
+* **Per-event dedup** — the notifications table has a UNIQUE(subscription_id,
+  event_date, metric, direction, kind); an INSERT OR IGNORE means re-running the
+  pass never re-notifies the same event.
+* **Weekly cap** — a subscription that has notified within the last 7 days is
+  skipped entirely (Subscription.last_notified_at), so one alert = at most one
+  notification per week, as specified.
+
+Quota safety: the loop reads history from the parquet cache only
+(climate.load_cached_history) and treats a forecast-fetch failure as "skip this
+cell", so it never spends archive quota or dies on an upstream rate limit.
+"""
+import datetime
+import os
+import threading
+import time
+
+import pandas as pd
+from sqlalchemy import delete, select
+
+import climate
+import grading
+import grid
+from db import sync_session_maker
+from models import AccessToken, Notification, Subscription
+from views import OBS_COLS
+
+WEEK_SECONDS = 7 * 86400
+INTERVAL = int(os.environ.get("THERMOGRAPH_NOTIFY_INTERVAL", "900"))  # 15 min default
+SESSION_TTL_SECONDS = int(os.environ.get("THERMOGRAPH_SESSION_TTL_DAYS", "30")) * 86400
+LOOKBACK_DAYS = 3        # re-check the last few observed days (dedup makes it safe)
+FORECAST_HORIZON_DAYS = 7
+
+# Metrics whose low tail also counts as unusual when a subscription is two-sided
+# (cold snaps, unusually calm/dry). Precipitation is one-directional (high only).
+TWO_SIDED_METRICS = set(grading.TEMP_METRICS)
+
+METRIC_NOUN = {
+    "tmax": "daytime high",
+    "tmin": "overnight low",
+    "feels": "feels-like temperature",
+    "humid": "humidity",
+    "wind": "wind speed",
+    "gust": "wind gusts",
+    "precip": "rainfall",
+}
+METRIC_UNIT = {
+    "tmax": "°F", "tmin": "°F", "feels": "°F",
+    "humid": " g/m³", "wind": " mph", "gust": " mph", "precip": " in",
+}
+# A readable descriptor per (metric, direction) so titles don't read like
+# "unusually high high temperature". Precipitation is high-only.
+METRIC_PHRASE = {
+    ("tmax", "high"): "unusually hot day", ("tmax", "low"): "unusually cool day",
+    ("tmin", "high"): "unusually warm night", ("tmin", "low"): "unusually cold night",
+    ("feels", "high"): "unusually hot conditions", ("feels", "low"): "unusually cold conditions",
+    ("humid", "high"): "unusually humid air", ("humid", "low"): "unusually dry air",
+    ("wind", "high"): "unusually strong wind", ("wind", "low"): "unusually calm wind",
+    ("gust", "high"): "unusually strong gusts", ("gust", "low"): "unusually light gusts",
+    ("precip", "high"): "unusually heavy rain",
+}
+
+_STOP = threading.Event()
+_thread = None
+
+
+# --- message building --------------------------------------------------------
+def _place(sub: Subscription) -> str:
+    return sub.label or f"{sub.lat:.2f}, {sub.lon:.2f}"
+
+
+def _compose(sub, event_date, metric, direction, graded):
+    noun = METRIC_NOUN.get(metric, metric)
+    pct = graded["percentile"]
+    grade = graded.get("grade")
+    value = graded.get("value")
+    unit = METRIC_UNIT.get(metric, "")
+    phrase = METRIC_PHRASE.get((metric, direction), f"unusual {noun}")
+    title = f"{_place(sub)}: {phrase}"
+    val_txt = f" ({value}{unit})" if value is not None else ""
+    if sub.kind == "forecast":
+        body = (f"Forecast for {event_date}: the {noun} is projected at the "
+                f"{pct:g}th percentile{val_txt} — {grade}.")
+    else:
+        body = (f"On {event_date}, the {noun} hit the {pct:g}th "
+                f"percentile{val_txt} — {grade}.")
+    return title, body
+
+
+# --- per-cell evaluation -----------------------------------------------------
+def _obs_from_row(row) -> dict:
+    return {k: row[k] for k in OBS_COLS if k in row}
+
+
+def _candidate_rows(recent: pd.DataFrame, kind: str, today: pd.Timestamp):
+    """Rows to grade for a subscription of the given kind, most-relevant first."""
+    if recent is None or recent.empty or "date" not in recent.columns:
+        return []
+    dates = pd.to_datetime(recent["date"]).dt.normalize()
+    if kind == "forecast":
+        mask = (dates > today) & (dates <= today + pd.Timedelta(days=FORECAST_HORIZON_DAYS))
+        ordered = recent[mask].assign(_d=dates[mask]).sort_values("_d")  # soonest first
+    else:
+        mask = (dates <= today) & (dates >= today - pd.Timedelta(days=LOOKBACK_DAYS))
+        ordered = recent[mask].assign(_d=dates[mask]).sort_values("_d", ascending=False)  # newest first
+    return [row for _, row in ordered.iterrows()]
+
+
+def _first_trigger(sub: Subscription, rows, history, grade_cache):
+    """The first (event_date, metric, direction, graded) that crosses, or None."""
+    for row in rows:
+        date_key = pd.Timestamp(row["date"]).date().isoformat()
+        graded = grade_cache.get(date_key)
+        if graded is None:
+            graded = grading.grade_day(history, row["date"], _obs_from_row(row))
+            grade_cache[date_key] = graded
+        for metric in sub.metrics:
+            g = graded.get(metric)
+            if not g or g.get("percentile") is None:
+                continue
+            pct = g["percentile"]
+            if pct >= sub.threshold:
+                return (date_key, metric, "high", g)
+            if sub.two_sided and metric in TWO_SIDED_METRICS and pct <= 100 - sub.threshold:
+                return (date_key, metric, "low", g)
+    return None
+
+
+def _process_cell(session, cell_id, subs, today, now):
+    """Evaluate every subscription on one cell, inserting any new notifications."""
+    try:
+        cell = grid.from_id(cell_id)
+    except Exception:  # noqa: BLE001 - a malformed cell_id shouldn't kill the pass
+        return 0
+    history = climate.load_cached_history(cell)
+    if history is None or history.empty:
+        return 0   # nothing cached yet — never spend archive quota from here
+    try:
+        recent = climate.get_recent_forecast(cell)
+    except climate.WeatherUnavailable:
+        return 0   # upstream rate-limited — skip this cell this pass
+    except Exception:  # noqa: BLE001
+        return 0
+
+    grade_cache = {}
+    created = 0
+    for sub in subs:
+        # Weekly cap: one notification per subscription per 7 days.
+        if sub.last_notified_at and now - sub.last_notified_at < WEEK_SECONDS:
+            continue
+        rows = _candidate_rows(recent, sub.kind, today)
+        hit = _first_trigger(sub, rows, history, grade_cache)
+        if hit is None:
+            continue
+        event_date, metric, direction, graded = hit
+        title, body = _compose(sub, event_date, metric, direction, graded)
+        notif = Notification(
+            user_id=sub.user_id,
+            subscription_id=sub.id,
+            event_date=event_date,
+            metric=metric,
+            direction=direction,
+            kind=sub.kind,
+            percentile=graded["percentile"],
+            value=graded.get("value"),
+            grade=graded.get("grade"),
+            title=title,
+            body=body,
+            channel="inapp",
+            created_at=now,
+        )
+        session.add(notif)
+        try:
+            session.flush()   # trips the UNIQUE dedup constraint if already seen
+        except Exception:  # noqa: BLE001 - already notified for this exact event
+            session.rollback()
+            continue
+        sub.last_notified_at = now
+        created += 1
+    session.commit()
+    return created
+
+
+# --- housekeeping ------------------------------------------------------------
+def _cleanup_sessions(session) -> None:
+    """Drop expired login sessions (access tokens past their lifetime)."""
+    cutoff = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(
+        seconds=SESSION_TTL_SECONDS
+    )
+    session.execute(delete(AccessToken).where(AccessToken.created_at < cutoff))
+    session.commit()
+
+
+# --- pass + loop -------------------------------------------------------------
+def run_pass() -> int:
+    """One full evaluation sweep over all active subscriptions. Returns the number
+    of notifications created."""
+    now = time.time()
+    today = pd.Timestamp(datetime.date.today())
+    created = 0
+    with sync_session_maker() as session:
+        subs = session.execute(
+            select(Subscription).where(Subscription.active.is_(True))
+        ).scalars().all()
+        by_cell: dict[str, list] = {}
+        for sub in subs:
+            by_cell.setdefault(sub.cell_id, []).append(sub)
+        for cell_id, cell_subs in by_cell.items():
+            try:
+                created += _process_cell(session, cell_id, cell_subs, today, now)
+            except Exception:  # noqa: BLE001 - one bad cell must not abort the pass
+                session.rollback()
+        try:
+            _cleanup_sessions(session)   # opportunistic expired-token sweep
+        except Exception:  # noqa: BLE001
+            session.rollback()
+    return created
+
+
+def run_loop():
+    # Wait first, then run — gives the app a moment to finish booting.
+    while not _STOP.wait(INTERVAL):
+        try:
+            run_pass()
+        except Exception:  # noqa: BLE001 - a bad pass must never kill the loop
+            pass
+
+
+def start():
+    """Start the notifier daemon thread (unless disabled via env)."""
+    global _thread
+    if os.environ.get("THERMOGRAPH_ENABLE_NOTIFIER", "1") == "0":
+        return
+    if _thread is not None:
+        return
+    _STOP.clear()
+    _thread = threading.Thread(target=run_loop, name="subscription-notifier", daemon=True)
+    _thread.start()
+
+
+def stop():
+    _STOP.set()
diff --git a/requirements.txt b/requirements.txt
index 022d732..77e1a84 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -4,3 +4,6 @@ httpx==0.28.1
 pandas==2.2.3
 pyarrow==18.1.0
 numpy==2.2.1
+# Accounts + notification subscriptions (see backend/db.py, users.py, notify.py).
+fastapi-users[sqlalchemy]==15.0.5
+aiosqlite==0.22.1
diff --git a/schemas.py b/schemas.py
new file mode 100644
index 0000000..54eb7f4
--- /dev/null
+++ b/schemas.py
@@ -0,0 +1,114 @@
+"""Pydantic request/response models for accounts, subscriptions, notifications.
+
+The User* schemas are fastapi-users' base schemas extended with display_name; the
+Subscription*/Notification* models are the app's first request bodies and shape the
+JSON the frontend sends and receives.
+"""
+import uuid
+from typing import Literal
+
+from fastapi_users import schemas
+from pydantic import BaseModel, Field, field_validator
+
+import grading
+
+# Canonical metric keys a subscription may watch (kept in lockstep with grading).
+ALLOWED_METRICS = tuple(grading.CLIMO_METRICS)
+DEFAULT_METRICS = ["tmax", "feels", "precip"]
+
+
+# --- users (fastapi-users) ---------------------------------------------------
+class UserRead(schemas.BaseUser[uuid.UUID]):
+    display_name: str | None = None
+
+
+class UserCreate(schemas.BaseUserCreate):
+    display_name: str | None = None
+
+
+class UserUpdate(schemas.BaseUserUpdate):
+    display_name: str | None = None
+
+
+# --- subscriptions -----------------------------------------------------------
+def _check_metrics(v: list[str]) -> list[str]:
+    if not v:
+        raise ValueError("pick at least one metric")
+    bad = [m for m in v if m not in ALLOWED_METRICS]
+    if bad:
+        raise ValueError(f"unknown metric(s): {', '.join(bad)}")
+    # de-dupe, preserve order
+    seen, out = set(), []
+    for m in v:
+        if m not in seen:
+            seen.add(m)
+            out.append(m)
+    return out
+
+
+class SubscriptionIn(BaseModel):
+    lat: float = Field(ge=-90, le=90)
+    lon: float = Field(ge=-180, le=180)
+    label: str | None = None
+    threshold: int = Field(ge=95, le=99)
+    metrics: list[str] = Field(default_factory=lambda: list(DEFAULT_METRICS))
+    kind: Literal["observed", "forecast"] = "observed"
+    two_sided: bool = True
+
+    @field_validator("metrics")
+    @classmethod
+    def _metrics(cls, v):
+        return _check_metrics(v)
+
+
+class SubscriptionPatch(BaseModel):
+    threshold: int | None = Field(default=None, ge=95, le=99)
+    metrics: list[str] | None = None
+    two_sided: bool | None = None
+    active: bool | None = None
+
+    @field_validator("metrics")
+    @classmethod
+    def _metrics(cls, v):
+        return _check_metrics(v) if v is not None else v
+
+
+class SubscriptionOut(BaseModel):
+    id: int
+    cell_id: str
+    label: str | None
+    lat: float
+    lon: float
+    threshold: int
+    metrics: list[str]
+    kind: str
+    two_sided: bool
+    active: bool
+    last_notified_at: float | None
+    created_at: float
+
+    model_config = {"from_attributes": True}
+
+
+# --- notifications -----------------------------------------------------------
+class NotificationOut(BaseModel):
+    id: int
+    subscription_id: int
+    event_date: str
+    metric: str
+    direction: str
+    kind: str
+    percentile: float
+    value: float | None
+    grade: str | None
+    title: str
+    body: str | None
+    created_at: float
+    read_at: float | None
+
+    model_config = {"from_attributes": True}
+
+
+class NotificationList(BaseModel):
+    notifications: list[NotificationOut]
+    unread_count: int
diff --git a/tests/test_notify.py b/tests/test_notify.py
new file mode 100644
index 0000000..123758c
--- /dev/null
+++ b/tests/test_notify.py
@@ -0,0 +1,80 @@
+"""Unit tests for the subscription evaluation engine's pure logic (no DB, no
+network): trigger detection against a synthetic climatology and message wording."""
+import types
+
+import numpy as np
+import pandas as pd
+
+import notify
+
+
+def _history() -> pd.DataFrame:
+    dates = pd.date_range("1990-01-01", "2020-12-31", freq="D")
+    rng = np.random.default_rng(1)
+    n = len(dates)
+    df = pd.DataFrame({"date": dates})
+    df["doy"] = df["date"].dt.dayofyear
+    df["tmax"] = 70 + rng.normal(0, 8, n)
+    df["tmin"] = 48 + rng.normal(0, 7, n)
+    df["precip"] = rng.exponential(0.05, n)
+    return df
+
+
+def _sub(**kw):
+    d = dict(metrics=["tmax"], threshold=95, two_sided=False, kind="observed",
+             label="Testville", lat=1.0, lon=2.0)
+    d.update(kw)
+    return types.SimpleNamespace(**d)
+
+
+def _row(date, **vals):
+    r = {"date": pd.Timestamp(date)}
+    r.update(vals)
+    return r
+
+
+def test_high_trigger():
+    row = _row("2021-07-15", tmax=115.0, tmin=50.0, precip=0.0)
+    hit = notify._first_trigger(_sub(metrics=["tmax"], threshold=95), [row], _history(), {})
+    assert hit is not None
+    _, metric, direction, g = hit
+    assert metric == "tmax" and direction == "high" and g["percentile"] >= 95
+
+
+def test_no_trigger_when_normal():
+    row = _row("2021-07-15", tmax=70.0, tmin=48.0, precip=0.0)
+    assert notify._first_trigger(_sub(metrics=["tmax"], threshold=95), [row], _history(), {}) is None
+
+
+def test_low_trigger_only_when_two_sided():
+    row = _row("2021-07-15", tmax=70.0, tmin=10.0, precip=0.0)   # extreme low overnight
+    assert notify._first_trigger(
+        _sub(metrics=["tmin"], threshold=95, two_sided=False), [row], _history(), {}) is None
+    hit = notify._first_trigger(
+        _sub(metrics=["tmin"], threshold=95, two_sided=True), [row], _history(), {})
+    assert hit is not None and hit[1] == "tmin" and hit[2] == "low"
+
+
+def test_precip_never_low_side():
+    # A dry day has no rain-percentile, so a precip subscription never fires low
+    # even when two-sided (rain is one-directional).
+    row = _row("2021-07-15", tmax=70.0, precip=0.0)
+    assert notify._first_trigger(
+        _sub(metrics=["precip"], threshold=95, two_sided=True), [row], _history(), {}) is None
+
+
+def test_compose_observed_wording():
+    g = {"percentile": 99.0, "value": 115.0, "grade": "Near Record"}
+    title, body = notify._compose(_sub(kind="observed", label="Phoenix"),
+                                  "2026-07-14", "tmax", "high", g)
+    assert title == "Phoenix: unusually hot day"
+    assert body.startswith("On 2026-07-14")
+    assert "99th percentile" in body and "Near Record" in body
+
+
+def test_compose_forecast_wording():
+    g = {"percentile": 2.0, "value": 5.0, "grade": "Near Record"}
+    title, body = notify._compose(_sub(kind="forecast", label="Nome"),
+                                  "2026-01-02", "tmin", "low", g)
+    assert title == "Nome: unusually cold night"
+    assert body.startswith("Forecast for 2026-01-02")
diff --git a/users.py b/users.py
new file mode 100644
index 0000000..ff0aed4
--- /dev/null
+++ b/users.py
@@ -0,0 +1,86 @@
+"""fastapi-users wiring: password hashing, cookie sessions, the auth dependency.
+
+We use the library end-to-end (no hand-rolled crypto): pwdlib hashing under the
+hood, an HttpOnly cookie transport, and a *database* session strategy (tokens live
+in the access_token table) so logins survive a restart and can be revoked.
+"""
+import os
+import secrets
+import uuid
+
+from fastapi import Depends
+from fastapi_users import BaseUserManager, FastAPIUsers, InvalidPasswordException, UUIDIDMixin
+from fastapi_users.authentication import AuthenticationBackend, CookieTransport
+from fastapi_users.authentication.strategy.db import DatabaseStrategy
+from fastapi_users_db_sqlalchemy import SQLAlchemyUserDatabase
+from fastapi_users_db_sqlalchemy.access_token import SQLAlchemyAccessTokenDatabase
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from db import get_async_session
+from models import AccessToken, User
+
+BASE = "/" + os.environ.get("THERMOGRAPH_BASE", "/thermograph").strip("/")
+# Only used to sign reset/verification tokens — features that stay dormant until
+# email is wired up. A per-boot random secret is fine while they're unused.
+SECRET = os.environ.get("THERMOGRAPH_AUTH_SECRET") or secrets.token_urlsafe(32)
+SESSION_TTL_SECONDS = int(os.environ.get("THERMOGRAPH_SESSION_TTL_DAYS", "30")) * 86400
+
+
+def _cookie_secure() -> bool:
+    # The app serves plain HTTP on the LAN by default; a Secure cookie would never
+    # be sent over HTTP, silently breaking auth. So default OFF and let the TLS/VPS
+    # deploy opt in with THERMOGRAPH_COOKIE_SECURE=1. Accepted tradeoff: LAN traffic
+    # is unencrypted on a trusted network.
+    v = os.environ.get("THERMOGRAPH_COOKIE_SECURE", "").strip().lower()
+    return v in ("1", "true", "yes", "always", "on")
+
+
+async def get_user_db(session: AsyncSession = Depends(get_async_session)):
+    yield SQLAlchemyUserDatabase(session, User)
+
+
+async def get_access_token_db(session: AsyncSession = Depends(get_async_session)):
+    yield SQLAlchemyAccessTokenDatabase(session, AccessToken)
+
+
+class UserManager(UUIDIDMixin, BaseUserManager[User, uuid.UUID]):
+    reset_password_token_secret = SECRET
+    verification_token_secret = SECRET
+
+    async def validate_password(self, password: str, user) -> None:
+        if len(password) < 8:
+            raise InvalidPasswordException(reason="Password must be at least 8 characters.")
+        if len(password) > 1024:
+            # Guard against a scrypt/argon DoS from a megabyte "password".
+            raise InvalidPasswordException(reason="Password is too long.")
+
+
+async def get_user_manager(user_db=Depends(get_user_db)):
+    yield UserManager(user_db)
+
+
+cookie_transport = CookieTransport(
+    cookie_name="tg_session",
+    cookie_max_age=SESSION_TTL_SECONDS,
+    cookie_path=BASE,          # scope the cookie to /thermograph, not the whole domain
+    cookie_secure=_cookie_secure(),
+    cookie_httponly=True,      # unreadable by JS -> XSS can't steal the session
+    cookie_samesite="lax",     # blocks cross-site form posts (CSRF), allows top-level nav
+)
+
+
+def get_database_strategy(access_token_db=Depends(get_access_token_db)) -> DatabaseStrategy:
+    return DatabaseStrategy(access_token_db, lifetime_seconds=SESSION_TTL_SECONDS)
+
+
+auth_backend = AuthenticationBackend(
+    name="cookie",
+    transport=cookie_transport,
+    get_strategy=get_database_strategy,
+)
+
+fastapi_users = FastAPIUsers[User, uuid.UUID](get_user_manager, [auth_backend])
+
+# Dependencies handlers use to require / peek at the logged-in user.
+current_active_user = fastapi_users.current_user(active=True)
+current_user_optional = fastapi_users.current_user(active=True, optional=True)