feat(security): R4 S3 — abuse & leak hardening

- C-3.7: fold a per-link password_version into the signed watch grant (migration
  0059 adds the column) and bump it on every password change, so rotating a share
  link's password invalidates outstanding grants at once instead of letting them
  live out the 6h TTL. make_grant/check_grant take the version; update_link bumps.
- C-3.8: the share recipient picker and share-by-email now reuse the messageable
  filter (not-demo AND active AND not-suspended) instead of only excluding demo,
  so suspended/deactivated addresses aren't leaked or reachable as targets.
- C-3.9: GET /keys/{user_id} adopts get_thread's MB2 guard — a non-messageable
  target with no shared history returns the same "User not found" as a missing id,
  so it can't be probed to enumerate users or fetch suspended users' keys.
- C-3.12: _register_account catches IntegrityError on the insert (the select-then-
  insert TOCTOU) and treats it as the already-registered no-op.
- C-3.13: trigger_job claims the running flag atomically under _activity_lock
  (compare-and-set) instead of a bare check, closing the double-start window.
- Move _messageable/is_messageable_user into a shared app/userscope.py so downloads
  doesn't import the messages route module.

backend/app/scheduler.py

@@ -380,6 +380,18 @@ def _is_running(job_id: str) -> bool:
         return bool(_activity.get(job_id, {}).get("running"))
 
 
+def _claim_running(job_id: str) -> bool:
+    """Atomically claim a job as running: True if it was idle (and marks it running now), False if
+    it was already running. Holding the lock across BOTH the check and the set closes the TOCTOU
+    window where two rapid manual triggers each saw 'idle' (the running flag is otherwise only set
+    later inside the spawned thread) and both started a run."""
+    with _activity_lock:
+        if _activity.get(job_id, {}).get("running"):
+            return False
+        _activity.setdefault(job_id, {})["running"] = True
+        return True
+
+
 def running_job_ids() -> set[str]:
     """Ids of jobs executing right now (any trigger). Lets non-admin surfaces — e.g. the
     header's sync indicator — reflect real activity instead of just pending-work counts."""
@@ -399,7 +411,9 @@ def trigger_job(job_id: str, actor_id: int | None = None) -> str:
     fn = JOB_FUNCS.get(job_id)
     if fn is None:
         raise KeyError(job_id)
-    if _is_running(job_id):
+    # Atomic claim (not a bare _is_running check) so two rapid clicks can't both start — the loser
+    # gets "already_running". _job re-affirms running=True and clears it when the run ends.
+    if not _claim_running(job_id):
         return "already_running"
 
     def run() -> None:
@@ -407,7 +421,12 @@ def trigger_job(job_id: str, actor_id: int | None = None) -> str:
             _manual_actor.set(actor_id)
         fn()
 
-    threading.Thread(target=run, name=f"trigger-{job_id}", daemon=True).start()
+    try:
+        threading.Thread(target=run, name=f"trigger-{job_id}", daemon=True).start()
+    except BaseException:
+        # Never got the worker going — release the claim so the job isn't wedged "running" forever.
+        _record(job_id, running=False)
+        raise
     return "started"