feat: add job cleanup loop, Scheduler task definitions, and registrar

- job_cleanup_loop (src/jobs/job_manager.py): hourly in-process pass over
  JobManager.cleanup_expired_jobs, started at app startup and cancelled
  at shutdown; Redis job payloads auto-expire but set memberships do not.
- docs/scheduler-tasks.md: the four production Scheduler task payloads
  for the deploy checklist - nightly integrity check 04:30, weekly
  quality report Sunday 03:00 (day_of_week=6, 0=Monday), daily Paperless
  orphan-cleanup 05:00 hitting the existing
  /maintenance/cleanup/paperless?user=jpmschweitzer&dry_run=false
  endpoint, and disabling test_example_task - with exact HTTP bodies
  (explicit user=jpmschweitzer, Authorization: Bearer ${LIBRARY_API_KEY}
  placeholder).
- scripts/register_scheduler_tasks.py: reads SCHEDULER_URL from env,
  DRY-RUN BY DEFAULT (prints the exact payloads, provably contacts
  nothing), --execute gated and requiring LIBRARY_API_KEY to fill the
  placeholder. NOT executed - definitions delivered for the deploy
  checklist only.

9 new offline tests (loop passes/error-resilience/cancellation, payload
schedules, explicit production user, placeholder, dry-run default).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QbFZyDvYksazX6nYQYZ67L

tests/test_job_plumbing.py

@@ -0,0 +1,121 @@
+"""
+Offline unit tests for job + Scheduler task plumbing (Phase C item 5).
+
+Covers:
+- job_cleanup_loop: invokes cleanup_expired_jobs per pass, survives
+  transient errors, honors cancellation
+- register_scheduler_tasks.py: dry-run default, payload contents
+  (explicit production user, auth placeholder, schedules)
+"""
+
+import asyncio
+from unittest.mock import AsyncMock
+
+import pytest
+
+from src.jobs.job_manager import job_cleanup_loop
+
+
+class TestJobCleanupLoop:
+    @pytest.mark.asyncio
+    async def test_invokes_cleanup_each_pass(self):
+        manager = AsyncMock()
+
+        passes = await job_cleanup_loop(manager, interval_seconds=0, max_iterations=3)
+
+        assert passes == 3
+        assert manager.cleanup_expired_jobs.await_count == 3
+
+    @pytest.mark.asyncio
+    async def test_transient_error_does_not_kill_loop(self):
+        manager = AsyncMock()
+        manager.cleanup_expired_jobs = AsyncMock(
+            side_effect=[RuntimeError("redis hiccup"), None]
+        )
+
+        passes = await job_cleanup_loop(manager, interval_seconds=0, max_iterations=2)
+
+        assert passes == 2
+        assert manager.cleanup_expired_jobs.await_count == 2
+
+    @pytest.mark.asyncio
+    async def test_cancellation_stops_loop(self):
+        manager = AsyncMock()
+
+        task = asyncio.create_task(job_cleanup_loop(manager, interval_seconds=60))
+        await asyncio.sleep(0)  # let it start sleeping
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+
+class TestSchedulerTaskDefinitions:
+    def _load_module(self):
+        import importlib
+        return importlib.import_module("scripts.register_scheduler_tasks")
+
+    def test_four_production_payloads_defined(self):
+        mod = self._load_module()
+        names = {t["task_name"] for t in mod.TASKS}
+        assert names == {
+            "library_integrity_check",
+            "library_quality_report",
+            "library_paperless_orphan_cleanup",
+        }
+        updates = {u["task_name"]: u["updates"] for u in mod.TASK_UPDATES}
+        assert updates == {"test_example_task": {"enabled": False}}
+
+    def test_schedules(self):
+        mod = self._load_module()
+        by_name = {t["task_name"]: t for t in mod.TASKS}
+
+        integrity = by_name["library_integrity_check"]
+        assert (integrity["hour"], integrity["minute"], integrity["day_of_week"]) == (4, 30, -1)
+
+        quality = by_name["library_quality_report"]
+        # Sunday 03:00 (Scheduler: 0 = Monday .. 6 = Sunday)
+        assert (quality["hour"], quality["minute"], quality["day_of_week"]) == (3, 0, 6)
+
+        paperless = by_name["library_paperless_orphan_cleanup"]
+        assert (paperless["hour"], paperless["minute"], paperless["day_of_week"]) == (5, 0, -1)
+
+    def test_payloads_use_explicit_production_user_and_placeholder(self):
+        mod = self._load_module()
+        for task in mod.TASKS:
+            config = task["config"]
+            auth = config["headers"]["Authorization"]
+            assert auth == f"Bearer {mod.API_KEY_PLACEHOLDER}"
+            # Explicit production tenant in body or query string (Phase B)
+            body_user = config.get("body", {}).get("user")
+            assert body_user == "jpmschweitzer" or "user=jpmschweitzer" in config["url"]
+
+    def test_paperless_task_hits_existing_endpoint(self):
+        mod = self._load_module()
+        task = next(t for t in mod.TASKS
+                    if t["task_name"] == "library_paperless_orphan_cleanup")
+        assert "/maintenance/cleanup/paperless" in task["config"]["url"]
+        assert "dry_run=false" in task["config"]["url"]
+
+    def test_substitute_api_key_replaces_placeholder_without_mutating(self):
+        mod = self._load_module()
+        original = mod.TASKS[0]
+        resolved = mod.substitute_api_key(original, "sekret")
+        assert resolved["config"]["headers"]["Authorization"] == "Bearer sekret"
+        # The module-level definition keeps the placeholder
+        assert mod.API_KEY_PLACEHOLDER in original["config"]["headers"]["Authorization"]
+
+    def test_dry_run_is_default_and_sends_nothing(self, capsys, monkeypatch):
+        mod = self._load_module()
+        monkeypatch.setattr("sys.argv", ["register_scheduler_tasks.py"])
+        monkeypatch.setenv("SCHEDULER_URL", "http://scheduler.test:8090")
+
+        def _boom(*args, **kwargs):  # any HTTP client construction = failure
+            raise AssertionError("dry-run must not contact the Scheduler")
+
+        monkeypatch.setattr(mod.httpx, "Client", _boom)
+
+        assert mod.main() == 0
+        out = capsys.readouterr().out
+        assert "DRY RUN" in out
+        assert "library_integrity_check" in out
+        assert mod.API_KEY_PLACEHOLDER in out  # placeholder, never a real key