diff --git a/src/scribe/services/api_keys.py b/src/scribe/services/api_keys.py
index c0ed030..d8b2021 100644
--- a/src/scribe/services/api_keys.py
+++ b/src/scribe/services/api_keys.py
@@ -13,8 +13,11 @@ def generate_key() -> str:
     return "fmcp_" + secrets.token_urlsafe(32)
 
 
-def _hash_key(key: str) -> str:
-    return hashlib.sha256(key.encode()).hexdigest()
+def hash_token(raw: str) -> str:
+    """The ONE fingerprint for every bearer secret stored by hash — API keys,
+    password-reset tokens, invitation tokens. Stored rows hold this, never
+    the raw value; a lookup hashes the presented token and compares."""
+    return hashlib.sha256(raw.encode()).hexdigest()
 
 
 def _key_prefix(key: str) -> str:
@@ -32,7 +35,7 @@ async def create_api_key(
     key = ApiKey(
         user_id=user_id,
         name=name,
-        key_hash=_hash_key(full_key),
+        key_hash=hash_token(full_key),
         key_prefix=_key_prefix(full_key),
         scope=scope,
     )
@@ -70,7 +73,7 @@ async def revoke_api_key(user_id: int, key_id: int) -> bool:
 
 async def lookup_key(raw_key: str) -> ApiKey | None:
     """Look up a non-revoked ApiKey by raw token value. Updates last_used_at."""
-    key_hash = _hash_key(raw_key)
+    key_hash = hash_token(raw_key)
     async with async_session() as session:
         result = await session.execute(
             select(ApiKey).where(
diff --git a/src/scribe/services/auth.py b/src/scribe/services/auth.py
index cee92cd..9f551df 100644
--- a/src/scribe/services/auth.py
+++ b/src/scribe/services/auth.py
@@ -1,4 +1,3 @@
-import hashlib
 import logging
 import secrets
 from datetime import datetime, timedelta, timezone
@@ -12,6 +11,8 @@ from scribe.models.invitation import InvitationToken
 from scribe.models.password_reset import PasswordResetToken
 from scribe.models.setting import Setting
 from scribe.models.user import User
+from scribe.services.api_keys import hash_token
+from scribe.services.settings import get_admin_setting
 
 logger = logging.getLogger(__name__)
 
@@ -142,16 +143,7 @@ async def is_registration_open() -> bool:
     user_count = await get_user_count()
     if user_count == 0:
         return True
-
-    async with async_session() as session:
-        # Find the admin user's registration_open setting
-        result = await session.execute(
-            select(Setting)
-            .join(User, Setting.user_id == User.id)
-            .where(User.role == "admin", Setting.key == "registration_open")
-        )
-        setting = result.scalar_one_or_none()
-        return setting.value == "true" if setting else False
+    return await get_admin_setting("registration_open", "false") == "true"
 
 
 async def list_users() -> list[User]:
@@ -211,7 +203,7 @@ async def get_user_by_email(email: str) -> User | None:
 async def create_password_reset_token(user_id: int) -> str:
     """Generate a password reset token. Returns the raw token (for the email link)."""
     raw_token = secrets.token_urlsafe(32)
-    token_hash = hashlib.sha256(raw_token.encode()).hexdigest()
+    token_hash = hash_token(raw_token)
     expires_at = datetime.now(timezone.utc) + timedelta(hours=1)
 
     async with async_session() as session:
@@ -239,7 +231,7 @@ async def create_password_reset_token(user_id: int) -> str:
 
 async def reset_password_with_token(raw_token: str, new_password: str) -> int | None:
     """Validate a reset token and update the user's password. Returns user_id on success."""
-    token_hash = hashlib.sha256(raw_token.encode()).hexdigest()
+    token_hash = hash_token(raw_token)
 
     async with async_session() as session:
         result = await session.execute(
@@ -270,7 +262,7 @@ async def reset_password_with_token(raw_token: str, new_password: str) -> int |
 async def create_invitation(email: str, invited_by: int) -> str:
     """Generate an invitation token. Returns the raw token (for the email link)."""
     raw_token = secrets.token_urlsafe(32)
-    token_hash = hashlib.sha256(raw_token.encode()).hexdigest()
+    token_hash = hash_token(raw_token)
     expires_at = datetime.now(timezone.utc) + timedelta(days=7)
 
     async with async_session() as session:
@@ -299,7 +291,7 @@ async def create_invitation(email: str, invited_by: int) -> str:
 
 async def validate_invitation_token(raw_token: str) -> InvitationToken | None:
     """Look up by hash, check not used/expired. Returns the token record with email."""
-    token_hash = hashlib.sha256(raw_token.encode()).hexdigest()
+    token_hash = hash_token(raw_token)
 
     async with async_session() as session:
         result = await session.execute(
@@ -319,7 +311,7 @@ async def validate_invitation_token(raw_token: str) -> InvitationToken | None:
 
 async def register_with_invitation(raw_token: str, username: str, password: str) -> User | None:
     """Validate token, create user with the invitation's email, mark token used."""
-    token_hash = hashlib.sha256(raw_token.encode()).hexdigest()
+    token_hash = hash_token(raw_token)
 
     async with async_session() as session:
         result = await session.execute(
@@ -398,20 +390,16 @@ async def purge_expired_auth_tokens(grace_days: int = 7) -> int:
     return removed
 
 
-async def _auth_token_retention_loop() -> None:
-    import asyncio
-    while True:
-        await asyncio.sleep(86400)  # daily
-        try:
-            removed = await purge_expired_auth_tokens()
-            if removed:
-                logger.info("Auth token retention: deleted %d expired token(s)", removed)
-        except Exception:
-            logger.exception("Error in auth token retention cleanup")
+async def _auth_token_retention_tick() -> None:
+    removed = await purge_expired_auth_tokens()
+    if removed:
+        logger.info("Auth token retention: deleted %d expired token(s)", removed)
 
 
 def start_auth_token_retention_loop() -> None:
     global _auth_retention_task
-    import asyncio
     if _auth_retention_task is None or _auth_retention_task.done():
-        _auth_retention_task = asyncio.create_task(_auth_token_retention_loop())
+        from scribe.services.background import start_periodic
+        _auth_retention_task = start_periodic(
+            86400, _auth_token_retention_tick, label="auth_token_retention",  # daily
+        )
diff --git a/src/scribe/services/background.py b/src/scribe/services/background.py
index 9f6d4ee..a81fa5d 100644
--- a/src/scribe/services/background.py
+++ b/src/scribe/services/background.py
@@ -45,6 +45,26 @@ def spawn(coro: Coroutine, *, site: str) -> None:
     task.add_done_callback(_done)
 
 
+def start_periodic(interval_s: float, work, *, label: str) -> asyncio.Task:
+    """A forever loop that sleeps ``interval_s`` then awaits ``work()``, logging
+    (never raising) when a tick fails — the one shape the hourly/daily
+    retention sweeps share (log retention, notification sweep, auth-token
+    purge). Sleeps FIRST so startup isn't a sweep; holds a strong reference
+    like spawn() so the loop cannot be garbage-collected mid-flight."""
+    async def _loop() -> None:
+        while True:
+            await asyncio.sleep(interval_s)
+            try:
+                await work()
+            except Exception:
+                logger.exception("periodic task %s failed", label)
+
+    task = asyncio.get_running_loop().create_task(_loop(), name=f"periodic-{label}")
+    _pending.add(task)
+    task.add_done_callback(_pending.discard)
+    return task
+
+
 async def drain() -> None:
     """Await everything in flight — for tests that need the writes landed."""
     while _pending:
diff --git a/src/scribe/services/backup.py b/src/scribe/services/backup.py
index 5929c23..d1ec08c 100644
--- a/src/scribe/services/backup.py
+++ b/src/scribe/services/backup.py
@@ -189,6 +189,142 @@ def _repo_binding_rows(rows) -> list[dict]:
     ]
 
 
+# Row builders for the sections both exporters carry. Pure, like the join-table
+# helpers above; the full and per-user exports used to restate every one of
+# these comprehensions side by side, and a column added to one and not the
+# other is a backup that silently drops it (#2293's shape, one layer down).
+
+def _user_rows(rows) -> list[dict]:
+    return [
+        {
+            "id": u.id, "username": u.username, "email": u.email,
+            "password_hash": u.password_hash, "oauth_sub": u.oauth_sub,
+            "role": u.role, "session_version": u.session_version,
+            "created_at": u.created_at.isoformat(),
+        }
+        for u in rows
+    ]
+
+
+def _project_rows(rows) -> list[dict]:
+    return [
+        {
+            "id": p.id, "user_id": p.user_id, "title": p.title,
+            "description": p.description, "goal": p.goal, "status": p.status,
+            "color": p.color,
+            "created_at": p.created_at.isoformat(),
+            "updated_at": p.updated_at.isoformat(),
+        }
+        for p in rows
+    ]
+
+
+def _milestone_rows(rows) -> list[dict]:
+    return [
+        {
+            "id": m.id, "user_id": m.user_id, "project_id": m.project_id,
+            "title": m.title, "description": m.description, "status": m.status,
+            "order_index": m.order_index,
+            "created_at": m.created_at.isoformat(),
+            "updated_at": m.updated_at.isoformat(),
+        }
+        for m in rows
+    ]
+
+
+def _note_rows(rows) -> list[dict]:
+    return [
+        {
+            "id": n.id, "user_id": n.user_id, "title": n.title, "body": n.body,
+            "tags": n.tags or [], "parent_id": n.parent_id,
+            "project_id": n.project_id, "milestone_id": n.milestone_id,
+            "status": n.status, "priority": n.priority,
+            "due_date": n.due_date.isoformat() if n.due_date else None,
+            "created_at": n.created_at.isoformat(),
+            "updated_at": n.updated_at.isoformat(),
+        }
+        for n in rows
+    ]
+
+
+def _task_log_rows(rows) -> list[dict]:
+    return [
+        {
+            "id": tl.id, "user_id": tl.user_id, "task_id": tl.task_id,
+            "content": tl.content, "duration_minutes": tl.duration_minutes,
+            "created_at": tl.created_at.isoformat(),
+            "updated_at": tl.updated_at.isoformat(),
+        }
+        for tl in rows
+    ]
+
+
+def _note_draft_rows(rows) -> list[dict]:
+    return [
+        {
+            "id": nd.id, "user_id": nd.user_id, "note_id": nd.note_id,
+            "proposed_body": nd.proposed_body, "original_body": nd.original_body,
+            "instruction": nd.instruction, "scope": nd.scope,
+            "created_at": nd.created_at.isoformat(),
+            "updated_at": nd.updated_at.isoformat(),
+        }
+        for nd in rows
+    ]
+
+
+def _note_version_rows(rows) -> list[dict]:
+    return [
+        {
+            "id": nv.id, "user_id": nv.user_id, "note_id": nv.note_id,
+            "title": nv.title, "body": nv.body, "tags": nv.tags or [],
+            "pin_kind": nv.pin_kind, "pin_label": nv.pin_label,
+            "created_at": nv.created_at.isoformat(),
+        }
+        for nv in rows
+    ]
+
+
+def _setting_rows(rows) -> list[dict]:
+    return [{"user_id": s.user_id, "key": s.key, "value": s.value} for s in rows]
+
+
+def _rulebook_rows(rows) -> list[dict]:
+    return [
+        {
+            "id": rb.id, "owner_user_id": rb.owner_user_id, "title": rb.title,
+            "description": rb.description, "always_on": rb.always_on,
+            "created_at": rb.created_at.isoformat(),
+            "updated_at": rb.updated_at.isoformat(),
+        }
+        for rb in rows
+    ]
+
+
+def _topic_rows(rows) -> list[dict]:
+    return [
+        {
+            "id": t.id, "rulebook_id": t.rulebook_id, "title": t.title,
+            "description": t.description, "order_index": t.order_index,
+            "created_at": t.created_at.isoformat(),
+            "updated_at": t.updated_at.isoformat(),
+        }
+        for t in rows
+    ]
+
+
+def _rule_rows(rows) -> list[dict]:
+    return [
+        {
+            "id": r.id, "topic_id": r.topic_id, "project_id": r.project_id,
+            "title": r.title, "statement": r.statement, "why": r.why,
+            "how_to_apply": r.how_to_apply, "order_index": r.order_index,
+            "created_at": r.created_at.isoformat(),
+            "updated_at": r.updated_at.isoformat(),
+        }
+        for r in rows
+    ]
+
+
 # ---------------------------------------------------------------------------
 # Export
 # ---------------------------------------------------------------------------
@@ -247,148 +383,17 @@ async def export_full_backup() -> dict:
             "Store it securely and restrict access."
         ),
         "_not_included": _NOT_INCLUDED,
-        "users": [
-            {
-                "id": u.id,
-                "username": u.username,
-                "email": u.email,
-                "password_hash": u.password_hash,
-                "oauth_sub": u.oauth_sub,
-                "role": u.role,
-                "session_version": u.session_version,
-                "created_at": u.created_at.isoformat(),
-            }
-            for u in users
-        ],
-        "projects": [
-            {
-                "id": p.id,
-                "user_id": p.user_id,
-                "title": p.title,
-                "description": p.description,
-                "goal": p.goal,
-                "status": p.status,
-                "color": p.color,
-                "created_at": p.created_at.isoformat(),
-                "updated_at": p.updated_at.isoformat(),
-            }
-            for p in projects
-        ],
-        "milestones": [
-            {
-                "id": m.id,
-                "user_id": m.user_id,
-                "project_id": m.project_id,
-                "title": m.title,
-                "description": m.description,
-                "status": m.status,
-                "order_index": m.order_index,
-                "created_at": m.created_at.isoformat(),
-                "updated_at": m.updated_at.isoformat(),
-            }
-            for m in milestones
-        ],
-        "notes": [
-            {
-                "id": n.id,
-                "user_id": n.user_id,
-                "title": n.title,
-                "body": n.body,
-                "tags": n.tags or [],
-                "parent_id": n.parent_id,
-                "project_id": n.project_id,
-                "milestone_id": n.milestone_id,
-                "status": n.status,
-                "priority": n.priority,
-                "due_date": n.due_date.isoformat() if n.due_date else None,
-                "created_at": n.created_at.isoformat(),
-                "updated_at": n.updated_at.isoformat(),
-            }
-            for n in notes
-        ],
-        "task_logs": [
-            {
-                "id": tl.id,
-                "user_id": tl.user_id,
-                "task_id": tl.task_id,
-                "content": tl.content,
-                "duration_minutes": tl.duration_minutes,
-                "created_at": tl.created_at.isoformat(),
-                "updated_at": tl.updated_at.isoformat(),
-            }
-            for tl in task_logs
-        ],
-        "note_drafts": [
-            {
-                "id": nd.id,
-                "user_id": nd.user_id,
-                "note_id": nd.note_id,
-                "proposed_body": nd.proposed_body,
-                "original_body": nd.original_body,
-                "instruction": nd.instruction,
-                "scope": nd.scope,
-                "created_at": nd.created_at.isoformat(),
-                "updated_at": nd.updated_at.isoformat(),
-            }
-            for nd in note_drafts
-        ],
-        "note_versions": [
-            {
-                "id": nv.id,
-                "user_id": nv.user_id,
-                "note_id": nv.note_id,
-                "title": nv.title,
-                "body": nv.body,
-                "tags": nv.tags or [],
-                "pin_kind": nv.pin_kind,
-                "pin_label": nv.pin_label,
-                "created_at": nv.created_at.isoformat(),
-            }
-            for nv in note_versions
-        ],
-        "settings": [
-            {"user_id": s.user_id, "key": s.key, "value": s.value}
-            for s in settings
-        ],
-        "rulebooks": [
-            {
-                "id": rb.id,
-                "owner_user_id": rb.owner_user_id,
-                "title": rb.title,
-                "description": rb.description,
-                "always_on": rb.always_on,
-                "created_at": rb.created_at.isoformat(),
-                "updated_at": rb.updated_at.isoformat(),
-            }
-            for rb in rulebooks
-        ],
-        "rulebook_topics": [
-            {
-                "id": t.id,
-                "rulebook_id": t.rulebook_id,
-                "title": t.title,
-                "description": t.description,
-                "order_index": t.order_index,
-                "created_at": t.created_at.isoformat(),
-                "updated_at": t.updated_at.isoformat(),
-            }
-            for t in topics
-        ],
-        "rules": [
-            {
-                "id": r.id,
-                "topic_id": r.topic_id,
-                "project_id": r.project_id,
-                "title": r.title,
-                "statement": r.statement,
-                "why": r.why,
-                "how_to_apply": r.how_to_apply,
-                "order_index": r.order_index,
-                "created_at": r.created_at.isoformat(),
-                "updated_at": r.updated_at.isoformat(),
-            }
-            for r in rules
-        ],
+        "users": _user_rows(users),
+        "projects": _project_rows(projects),
+        "milestones": _milestone_rows(milestones),
+        "notes": _note_rows(notes),
+        "task_logs": _task_log_rows(task_logs),
+        "note_drafts": _note_draft_rows(note_drafts),
+        "note_versions": _note_version_rows(note_versions),
+        "settings": _setting_rows(settings),
+        "rulebooks": _rulebook_rows(rulebooks),
+        "rulebook_topics": _topic_rows(topics),
+        "rules": _rule_rows(rules),
         "rulebook_subscriptions": _subscription_rows(subscriptions),
         "rule_suppressions": _rule_suppression_rows(rule_suppressions),
         "topic_suppressions": _topic_suppression_rows(topic_suppressions),
@@ -526,135 +531,16 @@ async def export_user_backup(user_id: int) -> dict:
             "role": user.role,
             "created_at": user.created_at.isoformat(),
         } if user else None,
-        "projects": [
-            {
-                "id": p.id,
-                "user_id": p.user_id,
-                "title": p.title,
-                "description": p.description,
-                "goal": p.goal,
-                "status": p.status,
-                "color": p.color,
-                "created_at": p.created_at.isoformat(),
-                "updated_at": p.updated_at.isoformat(),
-            }
-            for p in projects
-        ],
-        "milestones": [
-            {
-                "id": m.id,
-                "user_id": m.user_id,
-                "project_id": m.project_id,
-                "title": m.title,
-                "description": m.description,
-                "status": m.status,
-                "order_index": m.order_index,
-                "created_at": m.created_at.isoformat(),
-                "updated_at": m.updated_at.isoformat(),
-            }
-            for m in milestones
-        ],
-        "notes": [
-            {
-                "id": n.id,
-                "user_id": n.user_id,
-                "title": n.title,
-                "body": n.body,
-                "tags": n.tags or [],
-                "parent_id": n.parent_id,
-                "project_id": n.project_id,
-                "milestone_id": n.milestone_id,
-                "status": n.status,
-                "priority": n.priority,
-                "due_date": n.due_date.isoformat() if n.due_date else None,
-                "created_at": n.created_at.isoformat(),
-                "updated_at": n.updated_at.isoformat(),
-            }
-            for n in notes
-        ],
-        "task_logs": [
-            {
-                "id": tl.id,
-                "user_id": tl.user_id,
-                "task_id": tl.task_id,
-                "content": tl.content,
-                "duration_minutes": tl.duration_minutes,
-                "created_at": tl.created_at.isoformat(),
-                "updated_at": tl.updated_at.isoformat(),
-            }
-            for tl in task_logs
-        ],
-        "note_drafts": [
-            {
-                "id": nd.id,
-                "user_id": nd.user_id,
-                "note_id": nd.note_id,
-                "proposed_body": nd.proposed_body,
-                "original_body": nd.original_body,
-                "instruction": nd.instruction,
-                "scope": nd.scope,
-                "created_at": nd.created_at.isoformat(),
-                "updated_at": nd.updated_at.isoformat(),
-            }
-            for nd in note_drafts
-        ],
-        "note_versions": [
-            {
-                "id": nv.id,
-                "user_id": nv.user_id,
-                "note_id": nv.note_id,
-                "title": nv.title,
-                "body": nv.body,
-                "tags": nv.tags or [],
-                "pin_kind": nv.pin_kind,
-                "pin_label": nv.pin_label,
-                "created_at": nv.created_at.isoformat(),
-            }
-            for nv in note_versions
-        ],
-        "settings": [
-            {"user_id": s.user_id, "key": s.key, "value": s.value}
-            for s in settings
-        ],
-        "rulebooks": [
-            {
-                "id": rb.id,
-                "owner_user_id": rb.owner_user_id,
-                "title": rb.title,
-                "description": rb.description,
-                "always_on": rb.always_on,
-                "created_at": rb.created_at.isoformat(),
-                "updated_at": rb.updated_at.isoformat(),
-            }
-            for rb in rulebooks
-        ],
-        "rulebook_topics": [
-            {
-                "id": t.id,
-                "rulebook_id": t.rulebook_id,
-                "title": t.title,
-                "description": t.description,
-                "order_index": t.order_index,
-                "created_at": t.created_at.isoformat(),
-                "updated_at": t.updated_at.isoformat(),
-            }
-            for t in topics
-        ],
-        "rules": [
-            {
-                "id": r.id,
-                "topic_id": r.topic_id,
-                "project_id": r.project_id,
-                "title": r.title,
-                "statement": r.statement,
-                "why": r.why,
-                "how_to_apply": r.how_to_apply,
-                "order_index": r.order_index,
-                "created_at": r.created_at.isoformat(),
-                "updated_at": r.updated_at.isoformat(),
-            }
-            for r in rules
-        ],
+        "projects": _project_rows(projects),
+        "milestones": _milestone_rows(milestones),
+        "notes": _note_rows(notes),
+        "task_logs": _task_log_rows(task_logs),
+        "note_drafts": _note_draft_rows(note_drafts),
+        "note_versions": _note_version_rows(note_versions),
+        "settings": _setting_rows(settings),
+        "rulebooks": _rulebook_rows(rulebooks),
+        "rulebook_topics": _topic_rows(topics),
+        "rules": _rule_rows(rules),
         "rulebook_subscriptions": _subscription_rows(subscriptions),
         "rule_suppressions": _rule_suppression_rows(rule_suppressions),
         "topic_suppressions": _topic_suppression_rows(topic_suppressions),
diff --git a/src/scribe/services/dashboard.py b/src/scribe/services/dashboard.py
index 77c2d6f..d6942dd 100644
--- a/src/scribe/services/dashboard.py
+++ b/src/scribe/services/dashboard.py
@@ -15,6 +15,7 @@ from scribe.models import async_session
 from scribe.models.note import Note
 from scribe.models.project import Project
 from scribe.models.milestone import Milestone
+from scribe.models.base import iso
 from scribe.services import milestones as milestones_svc
 
 logger = logging.getLogger(__name__)
@@ -173,7 +174,7 @@ async def _recently_completed(user_id: int) -> list[dict]:
             .order_by(Note.completed_at.desc()).limit(RECENT_DONE_LIMIT)
         )).all()
         return [{"id": n.id, "title": n.title, "project_title": ptitle,
-                 "completed_at": n.completed_at.isoformat()} for n, ptitle in rows]
+                 "completed_at": iso(n.completed_at)} for n, ptitle in rows]
 
 
 async def _week_stats(user_id: int) -> dict:
diff --git a/src/scribe/services/db_maintenance.py b/src/scribe/services/db_maintenance.py
index b01277c..349ea7e 100644
--- a/src/scribe/services/db_maintenance.py
+++ b/src/scribe/services/db_maintenance.py
@@ -21,6 +21,7 @@ from datetime import datetime, timezone
 from sqlalchemy import text
 
 from scribe.models import async_session, engine
+from scribe.models.base import iso
 from scribe.services.settings import get_admin_setting, set_admin_setting
 
 logger = logging.getLogger(__name__)
@@ -128,10 +129,6 @@ _HEALTH_SQL = text("""
 """)
 
 
-def _iso(value) -> str | None:
-    return value.isoformat() if value is not None else None
-
-
 async def get_table_health() -> dict:
     """Per-table health from Postgres statistics + the total database size.
 
@@ -156,8 +153,8 @@ async def get_table_health() -> dict:
             "dead_pct": float(r["dead_pct"] or 0),
             "total_bytes": int(r["total_bytes"] or 0),
             "mod_since_analyze": int(r["mod_since_analyze"] or 0),
-            "last_vacuum": _iso(r["last_vacuum"]),
-            "last_analyze": _iso(r["last_analyze"]),
+            "last_vacuum": iso(r["last_vacuum"]),
+            "last_analyze": iso(r["last_analyze"]),
         }
         for r in rows
     ]
diff --git a/src/scribe/services/db_maintenance_scheduler.py b/src/scribe/services/db_maintenance_scheduler.py
index 83cc6ff..35ff69a 100644
--- a/src/scribe/services/db_maintenance_scheduler.py
+++ b/src/scribe/services/db_maintenance_scheduler.py
@@ -1,9 +1,8 @@
 """Daily APScheduler cron for basic DB maintenance (targeted VACUUM ANALYZE).
 
-Mirrors trash_scheduler.py: a single global BackgroundScheduler job bridges
-into the asyncio loop to run the async maintenance. Scheduled for 04:00 UTC by
-default — after the 03:30 trash purge — so it collects the dead tuples that
-night's delete sweeps leave behind.
+One ScheduledJob (services/scheduler.py). Scheduled for 04:00 UTC by default
+— after the 03:30 trash purge — so it collects the dead tuples that night's
+delete sweeps leave behind.
 
 Two things are operator-tunable from the admin Settings card:
   - db_maintenance_enabled ("true"/"false") — checked at fire time, so toggling
@@ -16,9 +15,9 @@ from __future__ import annotations
 import asyncio
 import logging
 
-from apscheduler.schedulers.background import BackgroundScheduler
 from apscheduler.triggers.cron import CronTrigger
 
+from scribe.services.scheduler import ScheduledJob
 from scribe.services.settings import get_admin_setting
 
 logger = logging.getLogger(__name__)
@@ -26,9 +25,6 @@ logger = logging.getLogger(__name__)
 _JOB_ID = "db_maintenance_vacuum"
 _DEFAULT_HOUR = 4
 
-_scheduler: BackgroundScheduler | None = None
-_loop: asyncio.AbstractEventLoop | None = None
-
 
 async def get_maintenance_hour() -> int:
     """The configured run-hour (UTC, 0–23), clamped; default 04:00."""
@@ -45,23 +41,20 @@ async def is_maintenance_enabled() -> bool:
     return (await get_admin_setting("db_maintenance_enabled", "true")) != "false"
 
 
-def _run_maintenance_threadsafe() -> None:
-    """APScheduler invokes this from a worker thread; bridge into the loop."""
-    if _loop is None:
-        logger.warning("db maintenance scheduler: no loop registered")
+async def _run_maintenance() -> None:
+    if not await is_maintenance_enabled():
+        logger.debug("db maintenance: disabled, skipping scheduled run")
         return
+    from scribe.services.db_maintenance import run_maintenance
+    await run_maintenance()
 
-    async def _runner():
-        try:
-            if not await is_maintenance_enabled():
-                logger.debug("db maintenance: disabled, skipping scheduled run")
-                return
-            from scribe.services.db_maintenance import run_maintenance
-            await run_maintenance()
-        except Exception:
-            logger.exception("db maintenance run failed")
 
-    asyncio.run_coroutine_threadsafe(_runner(), _loop)
+_JOB = ScheduledJob(_JOB_ID, _run_maintenance, label="DB maintenance")
+
+
+def _trigger(hour: int) -> CronTrigger:
+    hour = hour if 0 <= hour <= 23 else _DEFAULT_HOUR
+    return CronTrigger(hour=hour, minute=0, timezone="UTC")
 
 
 def start_db_maintenance_scheduler(
@@ -72,36 +65,15 @@ def start_db_maintenance_scheduler(
     in rather than read here so we never block the event loop at startup. The
     job's enabled-gate is re-checked at every fire, so only the hour is needed
     up front."""
-    global _scheduler, _loop
-    if _scheduler is not None:
-        return
-    _loop = loop
     hour = hour if 0 <= hour <= 23 else _DEFAULT_HOUR
-    _scheduler = BackgroundScheduler()
-    _scheduler.add_job(
-        _run_maintenance_threadsafe,
-        trigger=CronTrigger(hour=hour, minute=0, timezone="UTC"),
-        id=_JOB_ID,
-        replace_existing=True,
-    )
-    _scheduler.start()
-    logger.info("DB maintenance scheduler started (daily %02d:00 UTC)", hour)
+    _JOB.start(loop, _trigger(hour), describe=f"daily {hour:02d}:00 UTC")
 
 
 def reschedule_db_maintenance(hour: int) -> None:
     """Move the live job to a new UTC hour (called when the admin changes it)."""
-    if _scheduler is None:
-        return
     hour = hour if 0 <= hour <= 23 else _DEFAULT_HOUR
-    _scheduler.reschedule_job(
-        _JOB_ID, trigger=CronTrigger(hour=hour, minute=0, timezone="UTC")
-    )
-    logger.info("DB maintenance scheduler rescheduled to %02d:00 UTC", hour)
+    _JOB.reschedule(_trigger(hour), describe=f"{hour:02d}:00 UTC")
 
 
 def stop_db_maintenance_scheduler() -> None:
-    global _scheduler
-    if _scheduler is not None:
-        _scheduler.shutdown(wait=False)
-        _scheduler = None
-        logger.info("DB maintenance scheduler stopped")
+    _JOB.stop()
diff --git a/src/scribe/services/dedup.py b/src/scribe/services/dedup.py
index 103ad96..72c2c41 100644
--- a/src/scribe/services/dedup.py
+++ b/src/scribe/services/dedup.py
@@ -32,6 +32,7 @@ from scribe.models import async_session
 from scribe.models.embedding import NoteEmbedding
 from scribe.models.note import Note
 from scribe.models.rulebook import Rule
+from scribe.models.base import iso
 from scribe.services import embeddings as embeddings_svc
 # Imported rather than redeclared: no service imports this module (the create
 # gate is called from the routes/tools layer), so there is no cycle to dodge,
@@ -592,8 +593,8 @@ async def find_duplicate_records(
                 titles[int(i)] = t
                 records[int(i)] = d or {}
                 meta[int(i)] = {
-                    "created_at": created.isoformat() if created else None,
-                    "updated_at": updated.isoformat() if updated else None,
+                    "created_at": iso(created),
+                    "updated_at": iso(updated),
                     "task_kind": task_kind,
                 }
     except Exception:
diff --git a/src/scribe/services/knowledge.py b/src/scribe/services/knowledge.py
index 10b644c..e0a74f2 100644
--- a/src/scribe/services/knowledge.py
+++ b/src/scribe/services/knowledge.py
@@ -22,6 +22,7 @@ from sqlalchemy import and_, func, or_, select
 
 from scribe.models import async_session
 from scribe.models.note import Note
+from scribe.models.base import iso
 from scribe.services.access import browsable_notes_clause, readable_notes_clause
 
 logger = logging.getLogger(__name__)
@@ -211,8 +212,8 @@ def _note_to_item(note: Note) -> dict:
         # These lists now include records shared with the caller, so the client
         # needs the owner to tell "mine" from "someone else's" in a mixed list.
         "user_id": note.user_id,
-        "created_at": note.created_at.isoformat(),
-        "updated_at": note.updated_at.isoformat(),
+        "created_at": iso(note.created_at),
+        "updated_at": iso(note.updated_at),
     }
     # Drift verdict (#2086), when one has been recorded. Included here rather
     # than decorated on by the snippet layer because `current` is derivable from
@@ -249,7 +250,7 @@ def _note_to_item(note: Note) -> dict:
         item["task_kind"] = note.task_kind
         item["status"] = note.status
         item["priority"] = note.priority
-        item["due_date"] = note.due_date.isoformat() if note.due_date else None
+        item["due_date"] = iso(note.due_date)
 
     return item
 
diff --git a/src/scribe/services/logging.py b/src/scribe/services/logging.py
index 3123920..1f560e9 100644
--- a/src/scribe/services/logging.py
+++ b/src/scribe/services/logging.py
@@ -194,18 +194,14 @@ async def delete_old_logs(retention_days: int) -> int:
         return result.rowcount
 
 
-async def _retention_loop() -> None:
-    while True:
-        await asyncio.sleep(3600)  # hourly
-        try:
-            deleted = await delete_old_logs(Config.LOG_RETENTION_DAYS)
-            if deleted:
-                logger.info("Log retention: deleted %d old log entries", deleted)
-        except Exception:
-            logger.exception("Error in log retention cleanup")
+async def _retention_tick() -> None:
+    deleted = await delete_old_logs(Config.LOG_RETENTION_DAYS)
+    if deleted:
+        logger.info("Log retention: deleted %d old log entries", deleted)
 
 
 def start_log_retention_loop() -> None:
     global _retention_task
     if _retention_task is None or _retention_task.done():
-        _retention_task = asyncio.create_task(_retention_loop())
+        from scribe.services.background import start_periodic
+        _retention_task = start_periodic(3600, _retention_tick, label="log_retention")  # hourly
diff --git a/src/scribe/services/milestones.py b/src/scribe/services/milestones.py
index 679aae5..5de7b59 100644
--- a/src/scribe/services/milestones.py
+++ b/src/scribe/services/milestones.py
@@ -235,12 +235,6 @@ async def get_project_milestone_summaries(
 
 
 async def get_project_milestone_summary(user_id: int, project_id: int) -> list[dict]:
-    """Return ordered list of milestones with their progress stats."""
-    milestones = await list_milestones(user_id, project_id)
-    result = []
-    for m in milestones:
-        progress = await get_milestone_progress(m.id)
-        entry = m.to_dict()
-        entry.update(progress)
-        result.append(entry)
-    return result
+    """Ordered milestones with progress — the one-project view of
+    get_project_milestone_summaries (two queries, not N+1)."""
+    return (await get_project_milestone_summaries(user_id, [project_id])).get(project_id, [])
diff --git a/src/scribe/services/note_usage.py b/src/scribe/services/note_usage.py
index 6355c7d..cf67d5c 100644
--- a/src/scribe/services/note_usage.py
+++ b/src/scribe/services/note_usage.py
@@ -36,6 +36,7 @@ from sqlalchemy import case, func, select
 
 from scribe.models import async_session
 from scribe.models.note_usage import PULLED, SURFACED, NoteUsageEvent
+from scribe.models.base import iso
 
 logger = logging.getLogger(__name__)
 
@@ -232,13 +233,13 @@ async def usage_for_notes(note_ids: list[int]) -> dict[int, dict]:
             slot["ambient_count"] = int(n)
         elif event == SURFACED:
             slot["surfaced_count"] = int(n)
-            slot["last_surfaced_at"] = last_at.isoformat() if last_at else None
+            slot["last_surfaced_at"] = iso(last_at)
         elif event == PULLED:
             # Pulls are pulls regardless of what surfaced the record — the
             # question a pull answers ("did anyone ever open this?") doesn't
             # depend on how it was found.
             slot["pull_count"] = slot["pull_count"] + int(n)
-            latest = last_at.isoformat() if last_at else None
+            latest = iso(last_at)
             if latest and (slot["last_pulled_at"] or "") < latest:
                 slot["last_pulled_at"] = latest
     return out
diff --git a/src/scribe/services/notifications.py b/src/scribe/services/notifications.py
index cd171d6..00fb0cd 100644
--- a/src/scribe/services/notifications.py
+++ b/src/scribe/services/notifications.py
@@ -3,17 +3,20 @@
 import asyncio
 import json
 import logging
-from datetime import date, datetime, time, timezone
+from datetime import date, datetime, time, timedelta, timezone
 
-from sqlalchemy import func, select, text
+from sqlalchemy import delete as sa_delete, func, select, text
+from sqlalchemy import update as sa_update
 
 from scribe.models import async_session
 from scribe.models.app_log import AppLog
 from scribe.models.note import Note
-from scribe.models.setting import Setting
+from scribe.models.notification import Notification
 from scribe.models.user import User
+from scribe.models.base import iso
 from scribe.services.email import _email_html, is_smtp_configured, send_email
 from scribe.services.logging import log_audit
+from scribe.services.settings import get_setting
 
 logger = logging.getLogger(__name__)
 
@@ -29,13 +32,7 @@ SECURITY_EVENT_LABELS = {
 
 async def _get_user_notification_pref(user_id: int, key: str) -> bool:
     """Check if a user has a notification preference enabled (default True)."""
-    async with async_session() as session:
-        result = await session.execute(
-            select(Setting).where(Setting.user_id == user_id, Setting.key == key)
-        )
-        setting = result.scalar_one_or_none()
-        # Default to enabled
-        return setting.value != "false" if setting else True
+    return await get_setting(user_id, key, "true") != "false"
 
 
 async def _get_user_email(user_id: int) -> str | None:
@@ -222,7 +219,7 @@ async def check_due_tasks() -> None:
                 for task in user_tasks:
                     overdue = task.due_date < today if task.due_date else False
                     date_color = "#ef4444" if overdue else "#374151"
-                    date_label = f'<span style="color: {date_color};">{task.due_date.isoformat()}</span>' if task.due_date else ""
+                    date_label = f'<span style="color: {date_color};">{iso(task.due_date)}</span>' if task.due_date else ""
                     overdue_badge = ' <span style="color:#ef4444;font-weight:600;font-size:11px;">(overdue)</span>' if overdue else ""
                     task_rows += (
                         f'<tr>'
@@ -261,13 +258,10 @@ _NOTIFICATION_RETENTION_DAYS = 30
 
 async def purge_old_read_notifications(retention_days: int = _NOTIFICATION_RETENTION_DAYS) -> int:
     """Delete already-read in-app notifications older than retention_days."""
-    from datetime import timedelta
-    from sqlalchemy import delete
-    from scribe.models.notification import Notification
     cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days)
     async with async_session() as session:
         result = await session.execute(
-            delete(Notification).where(
+            sa_delete(Notification).where(
                 Notification.read_at.isnot(None),
                 Notification.read_at < cutoff,
             )
@@ -276,25 +270,21 @@ async def purge_old_read_notifications(retention_days: int = _NOTIFICATION_RETEN
         return result.rowcount or 0
 
 
-async def _notification_loop() -> None:
-    while True:
-        await asyncio.sleep(3600)  # hourly
-        try:
-            await check_due_tasks()
-        except Exception:
-            logger.exception("Error in notification loop")
-        try:
-            removed = await purge_old_read_notifications()
-            if removed:
-                logger.info("Notification retention: deleted %d read notification(s)", removed)
-        except Exception:
-            logger.exception("Error in notification retention cleanup")
+async def _notification_tick() -> None:
+    try:
+        await check_due_tasks()
+    except Exception:
+        logger.exception("Error in notification loop")
+    removed = await purge_old_read_notifications()
+    if removed:
+        logger.info("Notification retention: deleted %d read notification(s)", removed)
 
 
 def start_notification_loop() -> None:
     global _notification_task
     if _notification_task is None or _notification_task.done():
-        _notification_task = asyncio.create_task(_notification_loop())
+        from scribe.services.background import start_periodic
+        _notification_task = start_periodic(3600, _notification_tick, label="notifications")  # hourly
 
 
 # ---------------------------------------------------------------------------
@@ -303,7 +293,6 @@ def start_notification_loop() -> None:
 
 async def create_in_app_notification(user_id: int, notif_type: str, payload: dict):
     """Create an in-app Notification record."""
-    from scribe.models.notification import Notification
     async with async_session() as session:
         n = Notification(user_id=user_id, type=notif_type, payload=payload)
         session.add(n)
@@ -316,11 +305,10 @@ async def _fire_share_email(user_id: int, subject: str, body_text: str) -> None:
     try:
         if not await is_smtp_configured():
             return
-        async with async_session() as session:
-            user = await session.get(User, user_id)
-        if user and user.email:
+        email = await _get_user_email(user_id)
+        if email:
             html = _email_html(subject, f"<p>{body_text.replace(chr(10), '<br>')}</p>")
-            await send_email(user.email, subject, html)
+            await send_email(email, subject, html)
     except Exception:
         logger.exception("Share email notification failed for user %d", user_id)
 
@@ -427,7 +415,6 @@ async def notify_group_added(
 
 
 async def list_in_app_notifications(user_id: int, unread_only: bool = True) -> list[dict]:
-    from scribe.models.notification import Notification
     async with async_session() as session:
         q = select(Notification).where(Notification.user_id == user_id)
         if unread_only:
@@ -438,7 +425,6 @@ async def list_in_app_notifications(user_id: int, unread_only: bool = True) -> l
 
 
 async def unread_notification_count(user_id: int) -> int:
-    from scribe.models.notification import Notification
     async with async_session() as session:
         result = await session.execute(
             select(func.count()).where(
@@ -450,8 +436,6 @@ async def unread_notification_count(user_id: int) -> int:
 
 
 async def mark_notification_read(user_id: int, notification_id: int) -> bool:
-    from scribe.models.notification import Notification
-    from datetime import timezone as tz
     async with async_session() as session:
         n = (await session.execute(
             select(Notification).where(
@@ -461,21 +445,17 @@ async def mark_notification_read(user_id: int, notification_id: int) -> bool:
         )).scalar_one_or_none()
         if not n:
             return False
-        from datetime import datetime
-        n.read_at = datetime.now(tz.utc)
+        n.read_at = datetime.now(timezone.utc)
         await session.commit()
         return True
 
 
 async def mark_all_notifications_read(user_id: int) -> int:
-    from scribe.models.notification import Notification
-    from datetime import datetime, timezone as tz
-    from sqlalchemy import update as sa_update
     async with async_session() as session:
         result = await session.execute(
             sa_update(Notification)
             .where(Notification.user_id == user_id, Notification.read_at.is_(None))
-            .values(read_at=datetime.now(tz.utc))
+            .values(read_at=datetime.now(timezone.utc))
             .returning(Notification.id)
         )
         await session.commit()
diff --git a/src/scribe/services/projects.py b/src/scribe/services/projects.py
index cfae26e..afdfafd 100644
--- a/src/scribe/services/projects.py
+++ b/src/scribe/services/projects.py
@@ -208,54 +208,9 @@ async def get_project_summaries(
 
 
 async def get_project_summary(user_id: int, project_id: int) -> dict:
-    """Return task counts by status, note count, and last activity."""
-    async with async_session() as session:
-        # Task counts by status
-        task_rows = await session.execute(
-            select(Note.status, func.count(Note.id))
-            .where(
-                Note.user_id == user_id,
-                Note.project_id == project_id,
-                Note.status.isnot(None),
-                Note.deleted_at.is_(None),
-            )
-            .group_by(Note.status)
-        )
-        # Initialise all three lifecycle keys to 0 so consumers can sum them
-        # safely without `?? 0` guards. Frontend interface declares all three
-        # as required; rendering `undefined + N` yields NaN.
-        task_counts: dict[str, int] = {"todo": 0, "in_progress": 0, "done": 0}
-        for status, count in task_rows.fetchall():
-            task_counts[status] = count
-
-        # Note count (non-tasks)
-        note_count_result = await session.scalar(
-            select(func.count(Note.id)).where(
-                Note.user_id == user_id,
-                Note.project_id == project_id,
-                Note.status.is_(None),
-                Note.deleted_at.is_(None),
-            )
-        )
-        note_count = note_count_result or 0
-
-        # Last activity
-        last_activity_result = await session.scalar(
-            select(func.max(Note.updated_at)).where(
-                Note.user_id == user_id,
-                Note.project_id == project_id,
-            )
-        )
-
-        from scribe.services.milestones import get_project_milestone_summary
-        milestone_summary = await get_project_milestone_summary(user_id, project_id)
-
-        return {
-            "task_counts": task_counts,
-            "note_count": note_count,
-            "last_activity": last_activity_result.isoformat() if last_activity_result else None,
-            "milestone_summary": milestone_summary,
-        }
+    """Return task counts by status, note count, and last activity — the
+    one-project view of get_project_summaries (one rule, not two copies)."""
+    return (await get_project_summaries(user_id, [project_id]))[project_id]
 
 
 # ---------------------------------------------------------------------------
@@ -279,8 +234,6 @@ async def list_projects_for_user(user_id: int, status: str | None = None) -> lis
     """Owned projects + shared projects, each dict has 'permission' field."""
     from scribe.models.group import GroupMembership
     from scribe.models.share import ProjectShare
-    from scribe.services.access import PERMISSION_RANK
-
     owned = await list_projects(user_id, status)
     owned_ids = {p.id for p in owned}
 
@@ -307,13 +260,14 @@ async def list_projects_for_user(user_id: int, status: str | None = None) -> lis
                 )
             )).scalars().all()
 
-    seen: dict[int, str] = {}
-    for share in list(shared_direct) + list(shared_group):
-        if share.project_id in owned_ids:
-            continue
-        prev = seen.get(share.project_id)
-        if prev is None or PERMISSION_RANK[share.permission] > PERMISSION_RANK[prev]:
-            seen[share.project_id] = share.permission
+    from scribe.services.sharing import best_permission_by
+    seen = {
+        pid: perm
+        for pid, perm in best_permission_by(
+            list(shared_direct) + list(shared_group), "project_id"
+        ).items()
+        if pid not in owned_ids
+    }
 
     for pid, perm in seen.items():
         if status:
diff --git a/src/scribe/services/recurrence_scheduler.py b/src/scribe/services/recurrence_scheduler.py
index 082751a..127569e 100644
--- a/src/scribe/services/recurrence_scheduler.py
+++ b/src/scribe/services/recurrence_scheduler.py
@@ -4,7 +4,7 @@ Every 15 minutes, creates the next occurrence of any recurring task whose spawn
 time has arrived — draining `recurrence_next_spawn_at`, which is armed on task
 completion. Without this job, recurring tasks would never recur.
 
-Uses the BackgroundScheduler pattern shared with the other *_scheduler modules.
+One ScheduledJob (services/scheduler.py), like the other *_scheduler modules.
 (Formerly event_scheduler.py, which also ran event reminders + CalDAV sync;
 those were removed when the calendar surface was retired.)
 """
@@ -13,52 +13,27 @@ from __future__ import annotations
 import asyncio
 import logging
 
-from apscheduler.schedulers.background import BackgroundScheduler
 from apscheduler.triggers.interval import IntervalTrigger
 
-logger = logging.getLogger(__name__)
+from scribe.services.scheduler import ScheduledJob
 
-_scheduler: BackgroundScheduler | None = None
-_loop: asyncio.AbstractEventLoop | None = None
+logger = logging.getLogger(__name__)
 
 
 async def _run_recurrence_spawn() -> None:
     from scribe.services.recurrence import spawn_recurring_tasks  # noqa: PLC0415
-    try:
-        await spawn_recurring_tasks()
-    except Exception:
-        logger.warning("Recurring-task spawn job failed", exc_info=True)
+    await spawn_recurring_tasks()
 
 
-def _run_recurrence_spawn_threadsafe(loop: asyncio.AbstractEventLoop) -> None:
-    asyncio.run_coroutine_threadsafe(_run_recurrence_spawn(), loop)
+_JOB = ScheduledJob("recurrence_spawn", _run_recurrence_spawn, label="Recurring-task spawn")
 
 
 def start_recurrence_scheduler(loop: asyncio.AbstractEventLoop) -> None:
-    global _scheduler, _loop
-    if _scheduler is not None:
-        return
-    _loop = loop
-    _scheduler = BackgroundScheduler()
-
     # Spawn the next occurrence of due recurring tasks every 15 minutes.
     # Without this job, recurrence_next_spawn_at is armed on completion but
     # never drained, so recurring tasks never recur.
-    _scheduler.add_job(
-        _run_recurrence_spawn_threadsafe,
-        trigger=IntervalTrigger(minutes=15),
-        args=[loop],
-        id="recurrence_spawn",
-        replace_existing=True,
-    )
-
-    _scheduler.start()
-    logger.info("Recurrence scheduler started (recurring-task spawn every 15m)")
+    _JOB.start(loop, IntervalTrigger(minutes=15), describe="recurring-task spawn every 15m")
 
 
 def stop_recurrence_scheduler() -> None:
-    global _scheduler
-    if _scheduler is not None:
-        _scheduler.shutdown(wait=False)
-        _scheduler = None
-        logger.info("Recurrence scheduler stopped")
+    _JOB.stop()
diff --git a/src/scribe/services/scheduler.py b/src/scribe/services/scheduler.py
new file mode 100644
index 0000000..1489930
--- /dev/null
+++ b/src/scribe/services/scheduler.py
@@ -0,0 +1,78 @@
+"""One APScheduler job bridged into the asyncio loop — the shape the four
+*_scheduler modules (recurrence spawn, auto-pin scan, trash purge, DB
+maintenance) each used to carry a private copy of.
+
+APScheduler's BackgroundScheduler fires from a worker thread; the work is
+async and must run on the app's loop, so the fire is bridged with
+``run_coroutine_threadsafe``. Each job is a module-level singleton: start is
+idempotent, stop shuts the scheduler down, and a job whose trigger the
+operator can change (the maintenance hour) reschedules the live job instead
+of restarting.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+from collections.abc import Awaitable, Callable
+
+from apscheduler.schedulers.background import BackgroundScheduler
+
+logger = logging.getLogger(__name__)
+
+
+class ScheduledJob:
+    """A named APScheduler job that awaits ``work()`` on the asyncio loop.
+
+    ``work`` is an async callable; exceptions it raises are logged under
+    ``label`` and never propagate into APScheduler's thread.
+    """
+
+    def __init__(self, job_id: str, work: Callable[[], Awaitable[None]], *, label: str) -> None:
+        self.job_id = job_id
+        self._work = work
+        self.label = label
+        self._scheduler: BackgroundScheduler | None = None
+        self._loop: asyncio.AbstractEventLoop | None = None
+
+    @property
+    def running(self) -> bool:
+        return self._scheduler is not None
+
+    def _fire(self) -> None:
+        """APScheduler invokes this from its worker thread; bridge into the loop."""
+        if self._loop is None:
+            logger.warning("%s scheduler: no loop registered", self.label)
+            return
+
+        async def _runner() -> None:
+            try:
+                await self._work()
+            except Exception:
+                logger.exception("%s run failed", self.label)
+
+        asyncio.run_coroutine_threadsafe(_runner(), self._loop)
+
+    def start(self, loop: asyncio.AbstractEventLoop, trigger, *, describe: str = "") -> None:
+        """Start the job on ``trigger``. Idempotent — a second start is a no-op."""
+        if self._scheduler is not None:
+            return
+        self._loop = loop
+        self._scheduler = BackgroundScheduler()
+        self._scheduler.add_job(
+            self._fire, trigger=trigger, id=self.job_id, replace_existing=True,
+        )
+        self._scheduler.start()
+        logger.info("%s scheduler started%s", self.label, f" ({describe})" if describe else "")
+
+    def reschedule(self, trigger, *, describe: str = "") -> None:
+        """Move the live job to a new trigger; a no-op when not running."""
+        if self._scheduler is None:
+            return
+        self._scheduler.reschedule_job(self.job_id, trigger=trigger)
+        logger.info("%s scheduler rescheduled%s", self.label, f" to {describe}" if describe else "")
+
+    def stop(self) -> None:
+        if self._scheduler is not None:
+            self._scheduler.shutdown(wait=False)
+            self._scheduler = None
+            logger.info("%s scheduler stopped", self.label)
diff --git a/src/scribe/services/shape_ledger.py b/src/scribe/services/shape_ledger.py
index bbe89c7..deadb38 100644
--- a/src/scribe/services/shape_ledger.py
+++ b/src/scribe/services/shape_ledger.py
@@ -31,6 +31,7 @@ from sqlalchemy import select
 
 from scribe.models import async_session
 from scribe.models.code_shape import CodeShape, CodeShapeEvent
+from scribe.models.base import iso
 
 logger = logging.getLogger(__name__)
 
@@ -1318,9 +1319,9 @@ async def shape_history(
                 "classified_by": r.classified_by, "reason": r.reason,
                 "first_seen_commit": r.first_seen_commit,
                 "last_seen_commit": r.last_seen_commit,
-                "first_seen_at": r.created_at.isoformat() if r.created_at else None,
-                "vanished_at": r.vanished_at.isoformat() if r.vanished_at else None,
-                "recheck_at": r.recheck_at.isoformat() if r.recheck_at else None,
+                "first_seen_at": iso(r.created_at),
+                "vanished_at": iso(r.vanished_at),
+                "recheck_at": iso(r.recheck_at),
                 "diverges_from": r.diverges_from,
             }
             for r in rows
diff --git a/src/scribe/services/sharing.py b/src/scribe/services/sharing.py
index 036c43c..3081e4e 100644
--- a/src/scribe/services/sharing.py
+++ b/src/scribe/services/sharing.py
@@ -10,6 +10,7 @@ from scribe.models.note import Note
 from scribe.models.project import Project
 from scribe.models.share import NoteShare, ProjectShare
 from scribe.models.user import User
+from scribe.models.base import iso
 
 logger = logging.getLogger(__name__)
 
@@ -31,7 +32,7 @@ async def _enrich_shares(session, shares) -> list[dict]:
     return result
 
 
-def _deduplicate_by_permission(shares, id_attr: str) -> dict[int, str]:
+def best_permission_by(shares, id_attr: str) -> dict[int, str]:
     """Return {resource_id: best_permission} keeping the highest-ranked permission per resource."""
     from scribe.services.access import PERMISSION_RANK
     seen: dict[int, str] = {}
@@ -210,7 +211,7 @@ async def list_shared_with_me(user_id: int) -> dict:
                 )
             )).scalars().all()
 
-        seen_projects = _deduplicate_by_permission(list(proj_direct) + list(proj_group), "project_id")
+        seen_projects = best_permission_by(list(proj_direct) + list(proj_group), "project_id")
 
         projects = []
         for pid, perm in seen_projects.items():
@@ -223,7 +224,7 @@ async def list_shared_with_me(user_id: int) -> dict:
                     "description": proj.description,
                     "status": proj.status,
                     "color": proj.color,
-                    "updated_at": proj.updated_at.isoformat(),
+                    "updated_at": iso(proj.updated_at),
                     "owner_username": owner.username if owner else None,
                     "permission": perm,
                 })
@@ -241,7 +242,7 @@ async def list_shared_with_me(user_id: int) -> dict:
                 )
             )).scalars().all()
 
-        seen_notes = _deduplicate_by_permission(list(note_direct) + list(note_group), "note_id")
+        seen_notes = best_permission_by(list(note_direct) + list(note_group), "note_id")
 
         notes = []
         for nid, perm in seen_notes.items():
@@ -253,7 +254,7 @@ async def list_shared_with_me(user_id: int) -> dict:
                     "title": note.title,
                     "is_task": note.is_task,
                     "project_id": note.project_id,
-                    "updated_at": note.updated_at.isoformat(),
+                    "updated_at": iso(note.updated_at),
                     "owner_username": owner.username if owner else None,
                     "permission": perm,
                 })
diff --git a/src/scribe/services/trash.py b/src/scribe/services/trash.py
index 8c171a8..03d4654 100644
--- a/src/scribe/services/trash.py
+++ b/src/scribe/services/trash.py
@@ -8,15 +8,16 @@ trashed rows via `alive()`.
 from __future__ import annotations
 
 import uuid
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 
-from sqlalchemy import or_, select, update
+from sqlalchemy import delete as sql_delete, or_, select, update
 
 from scribe.models import async_session
 from scribe.models.note import Note
 from scribe.models.project import Project
 from scribe.models.milestone import Milestone
 from scribe.models.rulebook import Rulebook, RulebookTopic, Rule
+from scribe.models.base import iso
 
 # entity_type -> Model. Used to resolve which table a trash op targets.
 _MODEL_FOR = {
@@ -87,16 +88,15 @@ async def _cascade(session, user_id: int, etype: str, eid: int, batch: str, now)
         # FK CASCADE would handle a full DELETE on the project row, but the
         # soft-delete path keeps the project row alive; this guarantees the
         # rows are gone whether or not the project ever gets purged.
-        from sqlalchemy import delete as _sql_delete
         from scribe.models.rulebook import (
             project_rule_suppressions, project_topic_suppressions,
         )
         await session.execute(
-            _sql_delete(project_rule_suppressions)
+            sql_delete(project_rule_suppressions)
             .where(project_rule_suppressions.c.project_id == eid)
         )
         await session.execute(
-            _sql_delete(project_topic_suppressions)
+            sql_delete(project_topic_suppressions)
             .where(project_topic_suppressions.c.project_id == eid)
         )
         await _set(session, Project, [Project.user_id == user_id, Project.id == eid], batch, now)
@@ -213,7 +213,6 @@ async def restore_entity(user_id: int, entity_type: str, entity_id: int) -> int
 
 async def purge(user_id: int, batch_id: str) -> int:
     """Hard-delete every row in the batch. Irreversible."""
-    from sqlalchemy import delete as sql_delete
     n = 0
     async with async_session() as session:
         for model in _ALL:
@@ -241,7 +240,7 @@ async def list_trash(user_id: int) -> list[dict]:
                 grp = batches.setdefault(
                     r.deleted_batch_id,
                     {"batch_id": r.deleted_batch_id,
-                     "deleted_at": r.deleted_at.isoformat() if r.deleted_at else None,
+                     "deleted_at": iso(r.deleted_at),
                      "items": []},
                 )
                 grp["items"].append({
@@ -265,8 +264,6 @@ async def purge_expired(user_id: int, retention_days: int) -> int:
     user's short window prematurely destroy another's data.
     retention_days <= 0 disables auto-purge (returns 0 without touching anything).
     """
-    from datetime import timedelta
-    from sqlalchemy import delete as sql_delete
     if retention_days <= 0:
         return 0
     cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days)
diff --git a/src/scribe/services/trash_scheduler.py b/src/scribe/services/trash_scheduler.py
index 3d5a5fa..63cbf51 100644
--- a/src/scribe/services/trash_scheduler.py
+++ b/src/scribe/services/trash_scheduler.py
@@ -1,80 +1,50 @@
 """Daily APScheduler cron that purges expired trash.
 
-Mirrors version_pinning_scheduler.py: a single global BackgroundScheduler job
-at 03:30 UTC bridges into the asyncio loop to run the async purge. Iterates
-every user and applies that user's own `trash_retention_days` setting; 0
-disables auto-purge for that user.
+A single job at 03:30 UTC (services/scheduler.py). Iterates every user and
+applies that user's own `trash_retention_days` setting; 0 disables auto-purge
+for that user.
 """
 from __future__ import annotations
 
 import asyncio
 import logging
 
-from apscheduler.schedulers.background import BackgroundScheduler
 from apscheduler.triggers.cron import CronTrigger
+from sqlalchemy import select
 
+from scribe.models import async_session
+from scribe.models.user import User
 from scribe.services import trash as trash_svc
+from scribe.services.scheduler import ScheduledJob
 from scribe.services.settings import get_setting
 
 logger = logging.getLogger(__name__)
 
-_scheduler: BackgroundScheduler | None = None
-_loop: asyncio.AbstractEventLoop | None = None
 
+async def _run_purge() -> None:
+    async with async_session() as session:
+        user_ids = (await session.execute(select(User.id))).scalars().all()
 
-def _run_purge_threadsafe() -> None:
-    """APScheduler invokes this from a worker thread; bridge into the loop."""
-    if _loop is None:
-        logger.warning("trash scheduler: no loop registered")
-        return
-
-    async def _runner():
+    purged = 0
+    for uid in user_ids:
+        raw = await get_setting(uid, "trash_retention_days", "90")
         try:
-            from sqlalchemy import select
+            days = int(raw)
+        except (TypeError, ValueError):
+            days = 90
+        purged += await trash_svc.purge_expired(uid, days)
+    if purged:
+        logger.info("trash purge: removed %d expired row(s)", purged)
+    else:
+        logger.debug("trash purge: nothing expired")
 
-            from scribe.models import async_session
-            from scribe.models.user import User
 
-            async with async_session() as session:
-                user_ids = (await session.execute(select(User.id))).scalars().all()
-
-            purged = 0
-            for uid in user_ids:
-                raw = await get_setting(uid, "trash_retention_days", "90")
-                try:
-                    days = int(raw)
-                except (TypeError, ValueError):
-                    days = 90
-                purged += await trash_svc.purge_expired(uid, days)
-            if purged:
-                logger.info("trash purge: removed %d expired row(s)", purged)
-            else:
-                logger.debug("trash purge: nothing expired")
-        except Exception:
-            logger.exception("trash purge run failed")
-
-    asyncio.run_coroutine_threadsafe(_runner(), _loop)
+_JOB = ScheduledJob("trash_retention_purge", _run_purge, label="Trash retention")
 
 
 def start_trash_scheduler(loop: asyncio.AbstractEventLoop) -> None:
-    global _scheduler, _loop
-    if _scheduler is not None:
-        return
-    _loop = loop
-    _scheduler = BackgroundScheduler()
-    _scheduler.add_job(
-        _run_purge_threadsafe,
-        trigger=CronTrigger(hour=3, minute=30, timezone="UTC"),
-        id="trash_retention_purge",
-        replace_existing=True,
-    )
-    _scheduler.start()
-    logger.info("Trash retention scheduler started (daily 03:30 UTC)")
+    _JOB.start(loop, CronTrigger(hour=3, minute=30, timezone="UTC"), describe="daily 03:30 UTC")
 
 
 def stop_trash_scheduler() -> None:
-    global _scheduler
-    if _scheduler is not None:
-        _scheduler.shutdown(wait=False)
-        _scheduler = None
-        logger.info("Trash retention scheduler stopped")
+    _JOB.stop()
diff --git a/src/scribe/services/version_pinning_scheduler.py b/src/scribe/services/version_pinning_scheduler.py
index 7189a86..a12f607 100644
--- a/src/scribe/services/version_pinning_scheduler.py
+++ b/src/scribe/services/version_pinning_scheduler.py
@@ -5,68 +5,39 @@ system promotes stable note versions before they get aged out of the
 rolling cap. Off-hours by design — the scan is cheap but not time-
 critical and doesn't need to interrupt regular activity.
 
-Mirrors the BackgroundScheduler + threadsafe-async-call pattern used by
-journal_scheduler.py.
+One ScheduledJob (services/scheduler.py), like the other *_scheduler modules.
 """
 from __future__ import annotations
 
 import asyncio
 import logging
 
-from apscheduler.schedulers.background import BackgroundScheduler
 from apscheduler.triggers.cron import CronTrigger
 
+from scribe.services.scheduler import ScheduledJob
 from scribe.services.version_pinning import scan_all_users_for_auto_pins
 
 logger = logging.getLogger(__name__)
 
-_scheduler: BackgroundScheduler | None = None
-_loop: asyncio.AbstractEventLoop | None = None
+
+async def _run_scan() -> None:
+    results = await scan_all_users_for_auto_pins()
+    total = sum(results.values())
+    if total > 0:
+        logger.info(
+            "auto-pin scan: pinned %d version(s) across %d user(s)",
+            total, len(results),
+        )
+    else:
+        logger.debug("auto-pin scan: no new pins")
 
 
-def _run_scan_threadsafe() -> None:
-    """APScheduler invokes this from a worker thread; bridge into the
-    asyncio loop so the scan can await its DB operations."""
-    if _loop is None:
-        logger.warning("version_pinning scheduler: no loop registered")
-        return
-
-    async def _runner():
-        try:
-            results = await scan_all_users_for_auto_pins()
-            total = sum(results.values())
-            if total > 0:
-                logger.info(
-                    "auto-pin scan: pinned %d version(s) across %d user(s)",
-                    total, len(results),
-                )
-            else:
-                logger.debug("auto-pin scan: no new pins")
-        except Exception:
-            logger.exception("auto-pin scan run failed")
-
-    asyncio.run_coroutine_threadsafe(_runner(), _loop)
+_JOB = ScheduledJob("version_pinning_auto_scan", _run_scan, label="Version pinning")
 
 
 def start_version_pinning_scheduler(loop: asyncio.AbstractEventLoop) -> None:
-    global _scheduler, _loop
-    if _scheduler is not None:
-        return
-    _loop = loop
-    _scheduler = BackgroundScheduler()
-    _scheduler.add_job(
-        _run_scan_threadsafe,
-        trigger=CronTrigger(hour=3, minute=0, timezone="UTC"),
-        id="version_pinning_auto_scan",
-        replace_existing=True,
-    )
-    _scheduler.start()
-    logger.info("Version pinning scheduler started (daily 03:00 UTC)")
+    _JOB.start(loop, CronTrigger(hour=3, minute=0, timezone="UTC"), describe="daily 03:00 UTC")
 
 
 def stop_version_pinning_scheduler() -> None:
-    global _scheduler
-    if _scheduler is not None:
-        _scheduler.shutdown(wait=False)
-        _scheduler = None
-        logger.info("Version pinning scheduler stopped")
+    _JOB.stop()
diff --git a/tests/test_api_keys.py b/tests/test_api_keys.py
index 7a0e312..5708bdf 100644
--- a/tests/test_api_keys.py
+++ b/tests/test_api_keys.py
@@ -5,7 +5,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 
 from scribe.services.api_keys import (
-    _hash_key,
+    hash_token,
     _key_prefix,
     generate_key,
     create_api_key,
@@ -29,7 +29,7 @@ def test_generate_key_uniqueness():
 
 def test_hash_key_is_sha256():
     key = "fmcp_testkey"
-    h = _hash_key(key)
+    h = hash_token(key)
     expected = hashlib.sha256(key.encode()).hexdigest()
     assert h == expected
 
@@ -79,7 +79,7 @@ async def test_lookup_key_returns_none_for_unknown():
 
 def test_hash_key_deterministic():
     key = "fmcp_some_test_key_value"
-    assert _hash_key(key) == _hash_key(key)
+    assert hash_token(key) == hash_token(key)
 
 
 @pytest.mark.asyncio