Port eb8f039 Activity-stream logging: notifier, auth, delivery, lifecycle, 500s, slow requests from monorepo (drift reconciliation)

Claude-Session: https://claude.ai/code/session_01RdARHDJaYC1wSQRTYM6t3Z

accounts/api_accounts.py

@@ -12,6 +12,7 @@ from fastapi.concurrency import run_in_threadpool
 from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from core import audit
 from data import climate
 from data import grid
 from notifications import push
@@ -110,6 +111,8 @@ async def create_subscription(
     session.add(sub)
     await session.commit()
     await session.refresh(sub)
+    audit.log_activity("sub.create", {"user_id": str(user.id), "sub_id": sub.id,
+                                      "cell_id": cell_id, "kind": body.kind})
     return sub
 
 
@@ -121,10 +124,13 @@ async def update_subscription(
     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():
+    changed = body.model_dump(exclude_unset=True)
+    for key, value in changed.items():
         setattr(sub, key, value)
     await session.commit()
     await session.refresh(sub)
+    audit.log_activity("sub.update", {"user_id": str(user.id), "sub_id": sub.id,
+                                      "changed": list(changed.keys())})
     return sub
 
 
@@ -135,8 +141,10 @@ async def delete_subscription(
     session: AsyncSession = Depends(get_async_session),
 ):
     sub = await _owned_subscription(session, sub_id, user)
+    kind = sub.kind
     await session.delete(sub)
     await session.commit()
+    audit.log_activity("sub.delete", {"user_id": str(user.id), "sub_id": sub_id, "kind": kind})
 
 
 # --- notifications -----------------------------------------------------------
@@ -254,6 +262,8 @@ async def push_subscribe(
             )
         )
     await session.commit()
+    audit.log_activity("push.subscribe", {"user_id": str(user.id),
+                                          "refreshed": existing is not None})
     return {"ok": True}
 
 
@@ -275,6 +285,7 @@ async def push_unsubscribe(
     if row is not None:
         await session.delete(row)
         await session.commit()
+        audit.log_activity("push.unsubscribe", {"user_id": str(user.id)})
 
 
 @router.post("/push/test", status_code=202)

accounts/users.py

@@ -19,6 +19,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from accounts.db import get_async_session
 from accounts.models import AccessToken, User
+from core import audit
 from notifications import mailer
 
 BASE = "/" + os.environ.get("THERMOGRAPH_BASE", "/thermograph").strip("/")
@@ -77,11 +78,19 @@ class UserManager(UUIDIDMixin, BaseUserManager[User, uuid.UUID]):
             raise InvalidPasswordException(reason="Password is too long.")
 
     async def on_after_register(self, user: User, request: "Request | None" = None) -> None:
+        audit.log_activity("auth.register", {"user_id": str(user.id)})
         # Kick off the same token-generation path a manual "resend verification"
         # request would use, so there is exactly one place that builds the email
         # (on_after_request_verify below) rather than two copies drifting apart.
         await self.request_verify(user, request)
 
+    async def on_after_login(self, user: User, request: "Request | None" = None,
+                             response=None) -> None:
+        audit.log_activity("auth.login", {"user_id": str(user.id), "verified": user.is_verified})
+
+    async def on_after_verify(self, user: User, request: "Request | None" = None) -> None:
+        audit.log_activity("auth.verify", {"user_id": str(user.id)})
+
     async def on_after_request_verify(
         self, user: User, token: str, request: "Request | None" = None
     ) -> None:

core/audit.py

@@ -30,6 +30,7 @@ AUDIT_DIR = os.path.join(LOG_ROOT, "audit")
 ERROR_DIR = os.path.join(LOG_ROOT, "errors")
 ACCESS_DIR = os.path.join(LOG_ROOT, "access")
 HEARTBEAT_DIR = os.path.join(LOG_ROOT, "heartbeat")
+ACTIVITY_DIR = os.path.join(LOG_ROOT, "activity")
 
 _lock = threading.Lock()
 
@@ -69,6 +70,17 @@ def log_event(tag: str, record: dict) -> dict:
     return rec
 
 
+def log_activity(tag: str, record: dict) -> dict:
+    """Record a non-error application event — auth, subscriptions, notification
+    delivery, app lifecycle, slow requests — to a dedicated *activity* stream,
+    kept separate from the errors folder so these never pollute error dashboards.
+    ``tag`` names the event (e.g. ``"auth.login"``, ``"notify.pass"``). Carries no
+    PII: user ids are UUIDs, and no emails/request bodies/push endpoints. Best-effort."""
+    rec = {"ts": _now_iso(), "tag": tag, "run_id": current_run_id.get(), **record}
+    _append(os.path.join(ACTIVITY_DIR, f"activity-{_today()}.jsonl"), rec)
+    return rec
+
+
 def log_access(record: dict) -> None:
     """Append one request line (client IP, method, path, status) to the daily access
     log, so request/IP data is retained for later analysis. Best-effort; never raises

notifications/mailer.py

@@ -27,6 +27,8 @@ import ssl
 from email.message import EmailMessage
 from email.utils import formataddr, formatdate, make_msgid, parseaddr
 
+from core import audit
+
 log = logging.getLogger("thermograph.mail")
 
 # console  -> log the message, send nothing (default; dev + tests)
@@ -91,6 +93,7 @@ def send(to: str, subject: str, text: str, html: "str | None" = None) -> bool:
     if BACKEND != "smtp":
         # console (the default): prove the flow end-to-end without sending.
         log.info("mail[console] to=%s subject=%s\n%s", to, subject, text)
+        audit.log_activity("mail.send", {"ok": True, "backend": "console"})
         return True
 
     try:
@@ -100,7 +103,9 @@ def send(to: str, subject: str, text: str, html: "str | None" = None) -> bool:
             if SMTP_USER:
                 smtp.login(SMTP_USER, SMTP_PASSWORD)
             smtp.send_message(msg)
+        audit.log_activity("mail.send", {"ok": True, "backend": "smtp"})
         return True
     except Exception:  # noqa: BLE001 - deliverability is never worth a 500
         log.exception("mail: send failed to=%s host=%s:%s", to, SMTP_HOST, SMTP_PORT)
+        audit.log_activity("mail.send", {"ok": False, "backend": "smtp"})
         return False

notifications/notify.py

@@ -161,7 +161,8 @@ def _send_discord_job(job: tuple[str, str, str, str]) -> None:
     """Best-effort Discord DM of one gathered job. Never raises (send_dm itself
     doesn't). Runs off the notifier thread's send pool — no DB session here."""
     discord_id, title, body, link = job
-    discord.send_dm(discord_id, title, body, link)
+    ok = discord.send_dm(discord_id, title, body, link)
+    audit.log_activity("notif.discord", {"ok": bool(ok)})
 
 
 # How many push/Discord sends the notifier fans out at once. These are pure
@@ -302,6 +303,9 @@ def _process_cell(session, cell_id, subs, today, now, archive_budget):
             continue
         sub.last_notified_at = now
         created += 1
+        audit.log_activity("notif.emit", {"user_id": str(sub.user_id), "sub_id": sub.id,
+                                          "metric": metric, "direction": direction,
+                                          "kind": sub.kind, "percentile": graded["percentile"]})
         # Gather who to reach over Web Push and Discord DM (the in-app row above
         # is the record of truth for the bell; these are side-channels that must
         # never break it) — the actual sends happen after commit, see below.
@@ -345,9 +349,12 @@ def _cleanup_sessions(session) -> None:
 def run_pass() -> int:
     """One full evaluation sweep over all active subscriptions. Returns the number
     of notifications created."""
+    t0 = time.perf_counter()
     now = time.time()
     today = datetime.date.today()
     created = 0
+    subs: list = []
+    by_cell: dict[str, list] = {}
     archive_budget = [MAX_ARCHIVE_FETCHES_PER_PASS]   # missing archives to fetch this pass
     with sync_session_maker() as session:
         subs = session.execute(
@@ -365,6 +372,10 @@ def run_pass() -> int:
             _cleanup_sessions(session)   # opportunistic expired-token sweep
         except Exception:  # noqa: BLE001
             session.rollback()
+    audit.log_activity("notify.pass", {
+        "subs_evaluated": len(subs), "cells": len(by_cell), "created": created,
+        "archives_fetched": MAX_ARCHIVE_FETCHES_PER_PASS - archive_budget[0],
+        "duration_ms": round((time.perf_counter() - t0) * 1000.0, 1)})
     return created
 
 

tests/conftest.py

@@ -48,6 +48,8 @@ audit.ERROR_DIR = os.path.join(_TMP, "logs", "errors")
 # The request-logging middleware runs under TestClient too; keep its access log
 # out of the repo's logs/ so self-test traffic never pollutes the live monitor.
 audit.ACCESS_DIR = os.path.join(_TMP, "logs", "access")
+audit.ACTIVITY_DIR = os.path.join(_TMP, "logs", "activity")
+audit.HEARTBEAT_DIR = os.path.join(_TMP, "logs", "heartbeat")
 
 
 def _years_before(d: datetime.date, years: int) -> datetime.date:

tests/notifications/test_discord_dm.py

@@ -85,7 +85,7 @@ def _fake_session(user):
 
 
 def _sub():
-    return types.SimpleNamespace(user_id="u1", lat=47.6, lon=-122.3)
+    return types.SimpleNamespace(id="s1", user_id="u1", lat=47.6, lon=-122.3)
 
 
 def test_discord_job_gathers_linked_opted_in_user(monkeypatch):

web/app.py

@@ -15,7 +15,7 @@ from fastapi import APIRouter, FastAPI, HTTPException, Query, Request, Response
 from fastapi.concurrency import run_in_threadpool
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.middleware.gzip import GZipMiddleware
-from fastapi.responses import RedirectResponse
+from fastapi.responses import JSONResponse, RedirectResponse
 
 from accounts import api_accounts
 from api import content_routes
@@ -167,7 +167,11 @@ async def _lifespan(app):
         # Same leader gate as the notifier above — the recurring worker jobs
         # (city warming, IndexNow) must likewise run in exactly one process.
         scheduler.start()
+    # Deploy/restart marker: one line correlates a metric shift with a boot.
+    audit.log_activity("app.start", {"version": app.version, "role": ROLE, "base": BASE,
+                                     "notifier": _should_run_notifier(), "pid": os.getpid()})
     yield
+    audit.log_activity("app.stop", {"role": ROLE, "pid": os.getpid()})
     notify.stop()
     scheduler.stop()
     await _frontend_client.aclose()
@@ -202,6 +206,25 @@ if _cors_origins:
     )
 
 
+# Log a structured line for every unhandled 500. Without an exception handler the
+# route exception propagates out past the request middleware (call_next re-raises),
+# so a 500 is otherwise invisible in both the access log and metrics — this makes
+# it visible and countable (tag="http.error" in the errors stream).
+SLOW_REQUEST_MS = 2000.0   # http.slow: log any request slower than this
+
+
+@app.exception_handler(Exception)
+async def _log_unhandled_exception(request: Request, exc: Exception) -> Response:
+    try:
+        audit.log_event("http.error", {
+            "phase": "http", "path": request.url.path, "method": request.method,
+            "cat": metrics.classify_inbound(request.url.path, BASE),
+            "exc_type": type(exc).__name__, "error": repr(exc)[:500]})
+    except Exception:  # noqa: BLE001 - logging must never mask the original error
+        pass
+    return JSONResponse({"detail": "Internal server error."}, status_code=500)
+
+
 @app.get("/healthz")
 def healthz():
     """Liveness probe: the process is up and serving. Deliberately does no I/O (no
@@ -246,7 +269,9 @@ async def revalidate_static(request, call_next):
     request answered with an empty 304 instead of a full re-download — page-to-page
     navigation stops re-transferring the JS/CSS/HTML while still picking up every
     deploy immediately (no "my change isn't showing" bugs)."""
+    t0 = time.perf_counter()
     response = await call_next(request)
+    dur_ms = (time.perf_counter() - t0) * 1000.0
     path = request.url.path
     try:
         cat = metrics.classify_inbound(path, BASE)
@@ -260,6 +285,14 @@ async def revalidate_static(request, call_next):
             await run_in_threadpool(audit.log_access, {
                 "ip": _client_ip(request), "method": request.method,
                 "path": path, "status": response.status_code, "cat": cat})
+        # Threshold-gated slow-request line: RunAudit only times the 7 graded
+        # endpoints, so this is the only latency signal for auth/account/content
+        # routes. Kept rare (>2s) so it never becomes per-request hot-path noise.
+        # Also a blocking file write, so keep it off the event loop.
+        if dur_ms > SLOW_REQUEST_MS and cat != "static":
+            await run_in_threadpool(audit.log_activity, "http.slow", {
+                "path": path, "method": request.method, "cat": cat,
+                "status": response.status_code, "duration_ms": round(dur_ms, 1)})
     except Exception:  # noqa: BLE001 - never let instrumentation break a response
         pass
     pages = (BASE, f"{BASE}/", f"{BASE}/calendar", f"{BASE}/day", f"{BASE}/score",