feat: Scheduler, Laufprotokoll, Aufräumen und Gesundheitsprüfung

APScheduler-Jobs je Regel mit Startversatz und Jitter, Protokoll in
runs/run_items mit Regelzählern, Aufräumen nach Aufbewahrungsfrist,
Health-Auswertung für /healthz und Verdrahtung des Dienststarts.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

CHANGELOG.md

@@ -54,6 +54,19 @@ Die Version steht an genau einer Stelle: `src/netmover/__init__.py`.
   Inaktivitäts-Zeitlimit; ein hängender Worker blockiert nur seine eigene Regel.
 - Lokale Ziele unter dem Mount-Basisverzeichnis werden abgelehnt (dort muss die
   Anbindung als Ziel gewählt werden, damit „eingehängt?“ geprüft wird).
+- Scheduler (APScheduler): ein Job pro aktiver Regel mit zufälligem
+  Startversatz und Jitter; Läufe in eigenem Thread-Pool; Abgleich nach Änderungen.
+- Protokoll: pro Lauf eine Zeile (auch übersprungene Termine), Einzeldateien
+  in `run_items` (max. 2000 pro Lauf); Regelzähler letzter Lauf, letzter Erfolg,
+  Fehler in Folge.
+- Aufräumen der Protokolle nach konfigurierbarer Frist (Standard 30 Tage).
+- Gesundheitsprüfung: Scheduler-Herzschlag, Anbindungen im Fehlerzustand,
+  Regeln ohne Erfolg seit mehr als dem Dreifachen ihres Intervalls.
+- Webhook-Benachrichtigung mit frei wählbarer Vorlage (JSON-sicher), Modus
+  „bei jedem Fehler“ oder „nach N Fehlern in Folge“, optional Entwarnung;
+  auch für Anbindungen, deren Wiederherstellung scheitert.
+- Dienststart: Migration → Schlüssel → Anbindungen wiederherstellen → Scheduler;
+  beim Beenden werden Mounts nicht ausgehängt.
 
 ### Behoben
 - Löschen von Zugangsdaten, die noch von einem Mount benutzt werden, trennt den

DEVLOG.md

@@ -10,10 +10,37 @@ Zeiten in Ortszeit (Europe/Berlin).
 | 2026-10-10 | 17:21 | 17:27 | 0:06  | 1       | Gerüst, Datenmodell, Migrationen       |
 | 2026-10-10 | 18:35 | 18:48 | 0:13  | 2       | Mount-Schicht, Validierung, SFTP       |
 | 2026-10-10 | 18:53 | 19:07 | 0:14  | 3       | Mover-Logik                            |
-| **Summe**  |       |       | **0:33** |      |                                        |
+| 2026-10-10 | 19:10 | 19:17 | 0:07  | 4       | Scheduler, Protokoll, Benachrichtigung |
+| **Summe**  |       |       | **0:40** |      |                                        |
 
 ---
 
+## 2026-10-10 · 19:10–19:17 · Schritt 4: Scheduler und Protokollierung
+
+**Erledigt**
+- `settings_store.py` (Laufzeit-Einstellungen), `notify.py` (Webhook),
+  `runs.py` (Lauf ausführen + protokollieren, Zähler, Aufräumen),
+  `health.py`, `scheduler.py`, `app_state.py` (Verdrahtung, Start/Stop).
+- 24 neue Tests (207 gesamt). Mutationsprobe: 6/6 erkannt.
+
+**Entscheidungen**
+- Erfolg = `ok` oder `partial` (Werkzeug arbeitet); Fehlschlag = `error` oder
+  `partial` (für Benachrichtigungen). Übersprungen zählt nicht; manueller
+  Probelauf ändert keine Zähler.
+- Bezug für „zu lange kein Erfolg“: spätester von letztem Erfolg, Anlage der
+  Regel und Dienststart – sonst wäre nach einem Neustart sofort alles rot.
+- „Nach N Fehlern“ meldet genau einmal beim Erreichen der Schwelle, danach
+  Entwarnung beim ersten Erfolg (abschaltbar).
+- `max_instances=3` im Scheduler, damit überlappende Termine beim eigenen
+  Regel-Lock ankommen und protokolliert werden, statt dass APScheduler sie still verwirft.
+- Leere Läufe (alle 30 s) loggen nur auf DEBUG; in `runs` stehen sie trotzdem
+  (die Oberfläche filtert standardmäßig auf Läufe mit Aktivität).
+- Webhook-URL wird nie geloggt (enthält oft ein Token).
+
+**Stolpersteine**
+- Zwei Testfehler lagen in den Tests selbst (Fake-Symlink bei `resolve()`,
+  Jitter in der Zeitschranke vergessen).
+
 ## 2026-10-10 · 18:53–19:07 · Schritt 3: Mover-Logik
 
 **Erledigt**

README.md

@@ -3,7 +3,7 @@
 Selbst gehosteter Dienst mit Web-Oberfläche, der SMB/CIFS- und NFS-Freigaben
 verwaltet und Dateien zeitgesteuert zwischen ihnen verschiebt.
 
-> Status: in Entwicklung (Schritt 3 von 6). Vollständige Doku folgt mit Schritt 6.
+> Status: in Entwicklung (Schritt 4 von 6). Vollständige Doku folgt mit Schritt 6.
 
 ## Entwicklung
 

src/netmover/app_state.py

@@ -0,0 +1,83 @@
+"""Verdrahtung aller Bausteine – einmal pro Prozess.
+
+Start: Datenbank migrieren → Schlüssel laden → Anbindungen mit ``auto_mount``
+wiederherstellen (Ergebnis protokollieren) → erst dann Scheduler starten.
+Ende: Scheduler stoppen, Mounts **nicht** aushängen (andere könnten sie nutzen).
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from netmover.config import Settings
+from netmover.crypto import SecretBox, load_or_create_key
+from netmover.db import make_engine, make_session_factory, upgrade
+from netmover.health import Health, check_health
+from netmover.logging_setup import event, get_logger
+from netmover.models import utcnow
+from netmover.mounting.backend import MountBackend
+from netmover.mounting.fake import FakeMountBackend
+from netmover.mounting.linux import LinuxMountBackend
+from netmover.mounting.manager import MountManager, RestoreResult
+from netmover.mover.runner import RuleRunner
+from netmover.notify import Notifier
+from netmover.runs import RunService
+from netmover.scheduler import RuleScheduler
+
+log = get_logger(__name__)
+
+
+def make_backend(settings: Settings) -> MountBackend:
+    if settings.mount_backend == "fake":
+        return FakeMountBackend(settings.data_dir / "fake")
+    return LinuxMountBackend(settings.run_dir)
+
+
+class NetMover:
+    def __init__(
+        self,
+        settings: Settings,
+        backend: MountBackend | None = None,
+        notifier: Notifier | None = None,
+    ) -> None:
+        self.settings = settings
+        settings.data_dir.mkdir(parents=True, exist_ok=True)
+        url = settings.effective_database_url
+        upgrade(url)
+        self.db_engine = make_engine(url)
+        self.session_factory = make_session_factory(self.db_engine)
+        self.secrets = SecretBox(load_or_create_key(settings.secret_key_path))
+        self.backend = backend or make_backend(settings)
+        self.notifier = notifier or Notifier(self.session_factory)
+        self.mounts = MountManager(
+            self.session_factory,
+            self.backend,
+            self.secrets,
+            settings.mount_base,
+            notifier=self.notifier.mount_event,
+        )
+        self.runner = RuleRunner(self.session_factory, self.mounts)
+        self.runs = RunService(self.session_factory, self.runner, self.notifier)
+        self.scheduler = RuleScheduler(self.session_factory, self.runs)
+        self.started_at: datetime = utcnow()
+        self.restore_results: list[RestoreResult] = []
+
+    def start(self) -> None:
+        """Muss im laufenden Event-Loop aufgerufen werden (AsyncIOScheduler)."""
+        self.started_at = utcnow()
+        self.restore_results = self.mounts.restore_all()
+        ok = sum(r.ok for r in self.restore_results)
+        event(log, "startup.mounts_restored", ok=ok, failed=len(self.restore_results) - ok)
+        self.scheduler.start()
+
+    def stop(self) -> None:
+        self.scheduler.shutdown()
+        self.db_engine.dispose()
+
+    def health(self) -> Health:
+        return check_health(
+            self.session_factory,
+            scheduler_running=self.scheduler.running,
+            last_heartbeat=self.scheduler.last_heartbeat,
+            service_started=self.started_at,
+        )

src/netmover/health.py

@@ -0,0 +1,75 @@
+"""Gesundheitszustand für ``/healthz`` und die Statusseite (Fallstrick 9).
+
+Ein Werkzeug, das nichts mehr tut, darf nicht wie eines aussehen, das nichts zu
+tun hat. Deshalb:
+
+* der Scheduler muss leben (Herzschlag nicht älter als 3 Minuten),
+* keine aktive Anbindung im Fehlerzustand,
+* keine aktive Regel, deren letzter erfolgreicher Lauf länger als das Dreifache
+  ihres Intervalls zurückliegt. Bezugspunkt ist der späteste von letztem Erfolg,
+  Anlage der Regel und Dienststart – sonst wäre direkt nach einem Neustart alles rot.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session, sessionmaker
+
+from netmover.db import session_scope
+from netmover.models import Mount, MountState, Rule, utcnow
+
+STALE_FACTOR = 3
+HEARTBEAT_MAX_AGE = timedelta(minutes=3)
+
+
+@dataclass
+class Health:
+    ok: bool = True
+    problems: list[str] = field(default_factory=list)
+    stale_rules: set[int] = field(default_factory=set)
+
+    def fail(self, problem: str) -> None:
+        self.ok = False
+        self.problems.append(problem)
+
+
+def rule_is_stale(rule: Rule, service_started: datetime, now: datetime) -> bool:
+    if not rule.enabled:
+        return False
+    reference = max(
+        d for d in (rule.last_success_at, rule.created_at, service_started) if d is not None
+    )
+    return now - reference > timedelta(seconds=STALE_FACTOR * rule.interval_seconds)
+
+
+def check_health(
+    session_factory: sessionmaker[Session],
+    *,
+    scheduler_running: bool,
+    last_heartbeat: datetime | None,
+    service_started: datetime,
+    now: datetime | None = None,
+) -> Health:
+    now = now or utcnow()
+    health = Health()
+    if not scheduler_running:
+        health.fail("Scheduler läuft nicht")
+    elif last_heartbeat is None or now - last_heartbeat > HEARTBEAT_MAX_AGE:
+        health.fail("Scheduler reagiert nicht (kein Herzschlag)")
+    with session_scope(session_factory) as s:
+        for m in s.scalars(select(Mount).where(Mount.enabled.is_(True))):
+            if m.state is MountState.ERROR:
+                health.fail(f"Anbindung „{m.name}“ im Fehlerzustand: {m.last_error or ''}".strip())
+        for r in s.scalars(select(Rule).where(Rule.enabled.is_(True))):
+            if rule_is_stale(r, service_started, now):
+                health.stale_rules.add(r.id)
+                ago = (
+                    "noch nie"
+                    if r.last_success_at is None
+                    else f"seit {r.last_success_at:%Y-%m-%d %H:%M} UTC"
+                )
+                health.fail(f"Regel „{r.name}“: kein erfolgreicher Lauf {ago}")
+    return health

src/netmover/runs.py

@@ -0,0 +1,171 @@
+"""Läufe ausführen und protokollieren.
+
+Pro Lauf eine Zeile in ``runs`` (auch für übersprungene Termine, Fallstrick 8),
+die Einzeldateien in ``run_items``. Danach die Regelzähler:
+
+* ``last_run_at`` – jeder Lauf außer übersprungenen,
+* ``last_success_at`` – Status ``ok`` oder ``partial`` (das Werkzeug hat gearbeitet),
+* ``consecutive_failures`` – ``error`` und ``partial`` zählen hoch, ``ok`` setzt zurück.
+
+Ein manuell gestarteter Probelauf verändert keine Zähler.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+
+from sqlalchemy import delete, select
+from sqlalchemy.orm import Session, sessionmaker
+
+from netmover.db import session_scope
+from netmover.logging_setup import event, get_logger
+from netmover.models import FileFailure, Rule, Run, RunItem, RunStatus, RunTrigger, utcnow
+from netmover.mover.engine import RunResult
+from netmover.mover.runner import RuleRunner
+from netmover.notify import Notifier
+from netmover.settings_store import load_settings
+
+log = get_logger(__name__)
+
+MAX_ITEMS_PER_RUN = 2000
+SUCCESS = frozenset({RunStatus.OK, RunStatus.PARTIAL})
+FAILURE = frozenset({RunStatus.ERROR, RunStatus.PARTIAL})
+
+
+class RunService:
+    def __init__(
+        self,
+        session_factory: sessionmaker[Session],
+        runner: RuleRunner,
+        notifier: Notifier | None = None,
+    ) -> None:
+        self.session_factory = session_factory
+        self.runner = runner
+        self.notifier = notifier
+
+    def execute(
+        self,
+        rule_id: int,
+        trigger: RunTrigger = RunTrigger.SCHEDULE,
+        dry_run: bool | None = None,
+    ) -> int | None:
+        """Führt die Regel aus, protokolliert den Lauf und liefert die Run-ID
+        (None, wenn die Regel inzwischen gelöscht wurde)."""
+        started = utcnow()
+        result = self.runner.run(rule_id, dry_run=dry_run)
+        finished = utcnow()
+        count_it = not (trigger is RunTrigger.MANUAL and dry_run)
+        return self.record(rule_id, trigger, result, started, finished, update_counters=count_it)
+
+    def record(
+        self,
+        rule_id: int,
+        trigger: RunTrigger,
+        result: RunResult,
+        started: datetime,
+        finished: datetime,
+        *,
+        update_counters: bool = True,
+    ) -> int | None:
+        notify: tuple[str, bool, int, str] | None = None
+        with session_scope(self.session_factory) as s:
+            rule = s.get(Rule, rule_id)
+            if rule is None:
+                return None
+            message = result.message
+            items = result.items
+            if len(items) > MAX_ITEMS_PER_RUN:
+                extra = len(items) - MAX_ITEMS_PER_RUN
+                items = items[:MAX_ITEMS_PER_RUN]
+                note = f"{extra} weitere Dateien nicht einzeln protokolliert"
+                message = f"{message}; {note}" if message else note
+            run = Run(
+                rule_id=rule_id,
+                started_at=started,
+                finished_at=finished,
+                status=result.status,
+                trigger=trigger,
+                dry_run=result.dry_run,
+                seen=result.seen,
+                moved=result.moved,
+                skipped=result.skipped,
+                failed=result.failed,
+                message=message,
+            )
+            s.add(run)
+            s.flush()
+            s.add_all(
+                RunItem(
+                    run_id=run.id,
+                    src=i.src,
+                    dst=i.dst,
+                    status=i.status,
+                    size=i.size,
+                    message=i.message,
+                )
+                for i in items
+            )
+            if update_counters and result.status is not RunStatus.SKIPPED:
+                previous = rule.consecutive_failures
+                rule.last_run_at = finished
+                if result.status in SUCCESS:
+                    rule.last_success_at = finished
+                if result.status in FAILURE:
+                    rule.consecutive_failures += 1
+                    notify = (rule.name, True, rule.consecutive_failures, self._summary(result))
+                else:
+                    rule.consecutive_failures = 0
+                    if previous:
+                        notify = (rule.name, False, previous, "")
+            run_id = run.id
+            rule_name = rule.name
+
+        level = 20 if result.status in (RunStatus.OK, RunStatus.SKIPPED) else 30
+        if result.status is RunStatus.OK and not result.items:
+            level = 10  # leere Läufe alle 30 s nur auf DEBUG
+        event(
+            log,
+            "run.finished",
+            level=level,
+            rule=rule_name,
+            run=run_id,
+            status=result.status.value,
+            trigger=trigger.value,
+            dry_run=result.dry_run,
+            seen=result.seen,
+            moved=result.moved,
+            skipped=result.skipped,
+            failed=result.failed,
+            took_ms=int((finished - started).total_seconds() * 1000),
+            message=result.message or "",
+        )
+        if notify is not None and self.notifier is not None:
+            self.notifier.rule_result(*notify)
+        return run_id
+
+    @staticmethod
+    def _summary(result: RunResult) -> str:
+        if result.status is RunStatus.ERROR:
+            return result.message or "Lauf fehlgeschlagen"
+        failed = [i for i in result.items if i.status.value == "failed"][:5]
+        lines = [f"{result.failed} Datei(en) fehlgeschlagen:"]
+        lines += [f"• {i.src}: {i.message}" for i in failed]
+        return "\n".join(lines)
+
+
+def cleanup(session_factory: sessionmaker[Session], now: datetime | None = None) -> dict[str, int]:
+    """Protokolle älter als die Aufbewahrungsfrist löschen (Items per CASCADE),
+    dazu Fehlerzähler, die seitdem nicht mehr angefasst wurden."""
+    days = load_settings(session_factory).retention_days
+    cutoff = (now or utcnow()) - timedelta(days=days)
+    with session_scope(session_factory) as s:
+        old_ids = list(s.scalars(select(Run.id).where(Run.started_at < cutoff)))
+        for i in range(0, len(old_ids), 500):
+            chunk = old_ids[i : i + 500]
+            s.execute(delete(RunItem).where(RunItem.run_id.in_(chunk)))
+            s.execute(delete(Run).where(Run.id.in_(chunk)))
+        failures = s.execute(delete(FileFailure).where(FileFailure.last_attempt_at < cutoff))
+        removed_failures = getattr(failures, "rowcount", 0) or 0
+    stats = {"runs": len(old_ids), "file_failures": removed_failures}
+    event(log, "cleanup.done", retention_days=days, **stats)
+    return stats

src/netmover/scheduler.py

@@ -0,0 +1,146 @@
+"""Zeitsteuerung mit APScheduler (``AsyncIOScheduler``).
+
+* Ein Job pro aktiver Regel im jeweiligen Intervall. Erster Start mit kleinem
+  zufälligem Versatz, dazu Jitter pro Termin – damit nicht alle Regeln
+  gleichzeitig starten (Fallstrick 8).
+* Die Läufe selbst sind blockierend und laufen in einem eigenen Thread-Pool,
+  nie im Event-Loop. Überlappung regelt der Lock im :class:`RuleRunner`, der den
+  übersprungenen Termin auch protokolliert – deshalb ``max_instances`` > 1.
+* Herzschlag-Job für ``/healthz`` und stündliches Aufräumen der Protokolle.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import random
+from concurrent.futures import ThreadPoolExecutor
+from datetime import UTC, datetime, timedelta
+
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+from apscheduler.triggers.interval import IntervalTrigger
+from sqlalchemy import select
+from sqlalchemy.orm import Session, sessionmaker
+
+from netmover.db import session_scope
+from netmover.logging_setup import event, get_logger
+from netmover.models import Rule, RunTrigger, utcnow
+from netmover.runs import RunService, cleanup
+
+log = get_logger(__name__)
+
+MAX_START_OFFSET = 10.0
+HEARTBEAT_SECONDS = 30
+CLEANUP_SECONDS = 3600
+
+
+def _job_id(rule_id: int) -> str:
+    return f"rule:{rule_id}"
+
+
+class RuleScheduler:
+    def __init__(
+        self,
+        session_factory: sessionmaker[Session],
+        service: RunService,
+        *,
+        rng: random.Random | None = None,
+        workers: int = 8,
+    ) -> None:
+        self.session_factory = session_factory
+        self.service = service
+        self._rng = rng or random.Random()  # noqa: S311 – nur Startversatz
+        self._executor = ThreadPoolExecutor(max_workers=workers, thread_name_prefix="nm-run")
+        self.scheduler = AsyncIOScheduler(
+            timezone=UTC,
+            job_defaults={"coalesce": True, "max_instances": 3},
+        )
+        self._intervals: dict[int, int] = {}
+        self.last_heartbeat: datetime | None = None
+
+    # -- Lebenszyklus ---------------------------------------------------------------
+
+    def start(self) -> None:
+        self.scheduler.start()
+        self.heartbeat()
+        self.scheduler.add_job(
+            self.heartbeat, IntervalTrigger(seconds=HEARTBEAT_SECONDS), id="heartbeat"
+        )
+        self.scheduler.add_job(
+            self._cleanup,
+            IntervalTrigger(seconds=CLEANUP_SECONDS, start_date=utcnow() + timedelta(minutes=1)),
+            id="cleanup",
+        )
+        self.sync()
+        event(log, "scheduler.started", rules=len(self._intervals))
+
+    def shutdown(self) -> None:
+        if self.scheduler.running:
+            self.scheduler.shutdown(wait=False)
+        # Laufende Worker nicht abwarten: abgebrochene Übertragungen hinterlassen
+        # höchstens .part-Dateien, die Quelle bleibt unangetastet.
+        self._executor.shutdown(wait=False, cancel_futures=True)
+        event(log, "scheduler.stopped")
+
+    @property
+    def running(self) -> bool:
+        return bool(self.scheduler.running)
+
+    def heartbeat(self) -> None:
+        self.last_heartbeat = utcnow()
+
+    # -- Regeln -------------------------------------------------------------------------
+
+    def sync(self) -> None:
+        """Jobs an die Regeln in der Datenbank angleichen (nach jeder Änderung)."""
+        with session_scope(self.session_factory) as s:
+            wanted = {
+                rid: interval
+                for rid, interval in s.execute(
+                    select(Rule.id, Rule.interval_seconds).where(Rule.enabled.is_(True))
+                ).all()
+            }
+        for rid in list(self._intervals):
+            if rid not in wanted:
+                if self.scheduler.get_job(_job_id(rid)):
+                    self.scheduler.remove_job(_job_id(rid))
+                del self._intervals[rid]
+        for rid, interval in wanted.items():
+            if self._intervals.get(rid) == interval and self.scheduler.get_job(_job_id(rid)):
+                continue
+            offset = self._rng.uniform(0, min(float(interval), MAX_START_OFFSET))
+            trigger = IntervalTrigger(
+                seconds=interval,
+                start_date=utcnow() + timedelta(seconds=offset),
+                jitter=max(1, min(5, interval // 10)),
+            )
+            self.scheduler.add_job(
+                self._run_rule,
+                trigger,
+                args=[rid],
+                id=_job_id(rid),
+                name=f"Regel {rid}",
+                replace_existing=True,
+                misfire_grace_time=interval,
+            )
+            self._intervals[rid] = interval
+
+    def next_run(self, rule_id: int) -> datetime | None:
+        job = self.scheduler.get_job(_job_id(rule_id))
+        return job.next_run_time if job is not None else None
+
+    async def _run_rule(self, rule_id: int) -> None:
+        loop = asyncio.get_running_loop()
+        await loop.run_in_executor(
+            self._executor, self.service.execute, rule_id, RunTrigger.SCHEDULE
+        )
+
+    async def run_now(self, rule_id: int, dry_run: bool | None = None) -> int | None:
+        """„Jetzt ausführen“ / „Probelauf“ aus der Oberfläche."""
+        loop = asyncio.get_running_loop()
+        return await loop.run_in_executor(
+            self._executor, self.service.execute, rule_id, RunTrigger.MANUAL, dry_run
+        )
+
+    async def _cleanup(self) -> None:
+        loop = asyncio.get_running_loop()
+        await loop.run_in_executor(self._executor, cleanup, self.session_factory)

tests/test_runs.py

@@ -0,0 +1,395 @@
+"""Protokollierung, Regelzähler, Benachrichtigung, Aufräumen, Health (Fallstrick 9)."""
+
+from __future__ import annotations
+
+import json
+from datetime import timedelta
+from pathlib import Path
+
+import pytest
+from sqlalchemy import select
+from sqlalchemy.orm import Session, sessionmaker
+
+from netmover.db import session_scope
+from netmover.health import check_health
+from netmover.models import (
+    FileFailure,
+    ItemStatus,
+    Mount,
+    MountState,
+    Rule,
+    Run,
+    RunItem,
+    RunStatus,
+    RunTrigger,
+    utcnow,
+)
+from netmover.mover.engine import ItemResult, RunResult
+from netmover.mover.runner import RuleRunner
+from netmover.notify import Notification, Notifier, render
+from netmover.runs import RunService, cleanup
+from netmover.settings_store import AppSettings, NotifyMode, load_settings, save_settings
+from tests.conftest import MountFactory
+
+
+class FakeRunner:
+    def __init__(self) -> None:
+        self.results: list[RunResult] = []
+
+    def run(self, rule_id: int, *, dry_run: bool | None = None) -> RunResult:
+        return self.results.pop(0)
+
+
+@pytest.fixture
+def rule_id(make_mount: MountFactory, session_factory: sessionmaker[Session]) -> int:
+    mid = make_mount()
+    with session_scope(session_factory) as s:
+        r = Rule(name="scans", src_mount_id=mid, dst_path="/x", interval_seconds=30)
+        s.add(r)
+        s.flush()
+        return r.id
+
+
+@pytest.fixture
+def sent() -> list[tuple[str, str, str]]:
+    return []
+
+
+@pytest.fixture
+def notifier(session_factory: sessionmaker[Session], sent: list[tuple[str, str, str]]) -> Notifier:
+    return Notifier(session_factory, sender=lambda *a: sent.append(a), background=False)
+
+
+@pytest.fixture
+def fake_runner() -> FakeRunner:
+    return FakeRunner()
+
+
+@pytest.fixture
+def service(
+    session_factory: sessionmaker[Session], fake_runner: FakeRunner, notifier: Notifier
+) -> RunService:
+    return RunService(session_factory, fake_runner, notifier)  # type: ignore[arg-type]
+
+
+def _rule(sf: sessionmaker[Session], rid: int) -> Rule:
+    with sf() as s:
+        r = s.get(Rule, rid)
+        assert r is not None
+        s.expunge(r)
+        return r
+
+
+def _enable_webhook(sf: sessionmaker[Session], mode: NotifyMode, after: int = 3) -> None:
+    save_settings(
+        sf, AppSettings(webhook_url="http://hook.invalid/x", notify_mode=mode, notify_after=after)
+    )
+
+
+def ok() -> RunResult:
+    return RunResult(
+        RunStatus.OK, seen=1, moved=1, items=[ItemResult("a.pdf", ItemStatus.MOVED, "/x/a.pdf", 3)]
+    )
+
+
+def err() -> RunResult:
+    return RunResult(RunStatus.ERROR, message="„scans“ ist nicht eingehängt")
+
+
+def partial() -> RunResult:
+    return RunResult(
+        RunStatus.PARTIAL,
+        seen=2,
+        moved=1,
+        failed=1,
+        items=[ItemResult("b.pdf", ItemStatus.FAILED, None, 1, "kaputt (Versuch 1)")],
+    )
+
+
+def skipped() -> RunResult:
+    return RunResult(RunStatus.SKIPPED, message="läuft noch")
+
+
+def test_run_and_items_are_recorded(
+    service: RunService,
+    fake_runner: FakeRunner,
+    rule_id: int,
+    session_factory: sessionmaker[Session],
+) -> None:
+    fake_runner.results.append(ok())
+    run_id = service.execute(rule_id)
+    with session_factory() as s:
+        run = s.get(Run, run_id)
+        assert run is not None
+        assert (run.status, run.seen, run.moved, run.trigger) == (
+            RunStatus.OK,
+            1,
+            1,
+            RunTrigger.SCHEDULE,
+        )
+        assert run.finished_at is not None and run.finished_at >= run.started_at
+        items = s.scalars(select(RunItem)).all()
+        assert [(i.src, i.dst, i.status) for i in items] == [
+            ("a.pdf", "/x/a.pdf", ItemStatus.MOVED)
+        ]
+    r = _rule(session_factory, rule_id)
+    assert r.last_success_at is not None and r.consecutive_failures == 0
+
+
+def test_counters(
+    service: RunService,
+    fake_runner: FakeRunner,
+    rule_id: int,
+    session_factory: sessionmaker[Session],
+) -> None:
+    fake_runner.results += [err(), partial(), skipped()]
+    service.execute(rule_id)
+    r = _rule(session_factory, rule_id)
+    assert r.consecutive_failures == 1 and r.last_success_at is None and r.last_run_at is not None
+    service.execute(rule_id)
+    r = _rule(session_factory, rule_id)
+    assert r.consecutive_failures == 2 and r.last_success_at is not None  # partial: Werkzeug lebt
+    last_run = r.last_run_at
+    service.execute(rule_id)  # übersprungen: protokolliert, Zähler unverändert
+    r = _rule(session_factory, rule_id)
+    assert r.consecutive_failures == 2 and r.last_run_at == last_run
+    with session_factory() as s:
+        assert [x.status for x in s.scalars(select(Run).order_by(Run.id))] == [
+            RunStatus.ERROR,
+            RunStatus.PARTIAL,
+            RunStatus.SKIPPED,
+        ]
+
+
+def test_manual_dry_run_does_not_touch_counters(
+    service: RunService,
+    fake_runner: FakeRunner,
+    rule_id: int,
+    session_factory: sessionmaker[Session],
+) -> None:
+    fake_runner.results.append(RunResult(RunStatus.ERROR, dry_run=True, message="x"))
+    service.execute(rule_id, RunTrigger.MANUAL, dry_run=True)
+    r = _rule(session_factory, rule_id)
+    assert r.consecutive_failures == 0 and r.last_run_at is None
+
+
+def test_deleted_rule_is_not_recorded(service: RunService, fake_runner: FakeRunner) -> None:
+    fake_runner.results.append(err())
+    assert service.execute(4711) is None
+
+
+def test_items_are_capped(
+    service: RunService,
+    fake_runner: FakeRunner,
+    rule_id: int,
+    session_factory: sessionmaker[Session],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr("netmover.runs.MAX_ITEMS_PER_RUN", 3)
+    fake_runner.results.append(
+        RunResult(RunStatus.OK, items=[ItemResult(f"{i}.pdf", ItemStatus.MOVED) for i in range(5)])
+    )
+    run_id = service.execute(rule_id)
+    with session_factory() as s:
+        assert len(s.scalars(select(RunItem)).all()) == 3
+        run = s.get(Run, run_id)
+        assert run is not None and "2 weitere" in (run.message or "")
+
+
+# -- Benachrichtigung -------------------------------------------------------------------------
+
+
+def test_notify_every_failure(
+    service: RunService,
+    fake_runner: FakeRunner,
+    rule_id: int,
+    session_factory: sessionmaker[Session],
+    sent: list[tuple[str, str, str]],
+) -> None:
+    _enable_webhook(session_factory, NotifyMode.EVERY)
+    fake_runner.results += [err(), err(), ok()]
+    for _ in range(3):
+        service.execute(rule_id)
+    bodies = [json.loads(b) for _, _, b in sent]
+    assert [b["event"] for b in bodies] == ["rule.failed", "rule.failed", "rule.recovered"]
+    assert bodies[0]["message"] == "„scans“ ist nicht eingehängt"
+    assert "2× in Folge" in bodies[1]["title"]
+
+
+def test_notify_after_n_only_once_at_threshold(
+    service: RunService,
+    fake_runner: FakeRunner,
+    rule_id: int,
+    session_factory: sessionmaker[Session],
+    sent: list[tuple[str, str, str]],
+) -> None:
+    _enable_webhook(session_factory, NotifyMode.AFTER_N, after=3)
+    fake_runner.results += [err(), err(), err(), err(), partial(), ok()]
+    events = []
+    for _ in range(6):
+        service.execute(rule_id)
+        events.append(len(sent))
+    assert events == [0, 0, 1, 1, 1, 2]
+    assert json.loads(sent[-1][2])["event"] == "rule.recovered"
+
+
+def test_no_recovery_message_below_threshold(
+    service: RunService,
+    fake_runner: FakeRunner,
+    rule_id: int,
+    session_factory: sessionmaker[Session],
+    sent: list[tuple[str, str, str]],
+) -> None:
+    _enable_webhook(session_factory, NotifyMode.AFTER_N, after=3)
+    fake_runner.results += [err(), ok()]
+    service.execute(rule_id)
+    service.execute(rule_id)
+    assert sent == []
+
+
+def test_notify_off_by_default(
+    service: RunService, fake_runner: FakeRunner, rule_id: int, sent: list[tuple[str, str, str]]
+) -> None:
+    fake_runner.results.append(err())
+    service.execute(rule_id)
+    assert sent == []
+
+
+def test_render_json_escapes_values() -> None:
+    n = Notification("rule.failed", 'Titel mit "Anführung"', "Zeile 1\nZeile 2 \\ $x", "r")
+    body = render('{"t": "$title", "m": "$message", "x": "$unbekannt"}', "application/json", n)
+    data = json.loads(body)
+    assert data == {"t": 'Titel mit "Anführung"', "m": "Zeile 1\nZeile 2 \\ $x", "x": "$unbekannt"}
+    assert render("$message", "text/plain", n) == "Zeile 1\nZeile 2 \\ $x"
+
+
+def test_sender_failure_is_swallowed(session_factory: sessionmaker[Session]) -> None:
+    def broken(*a: object) -> None:
+        raise ConnectionError("weg")
+
+    _enable_webhook(session_factory, NotifyMode.EVERY)
+    Notifier(session_factory, sender=broken, background=False).send(Notification("x", "t", "m"))
+
+
+def test_settings_roundtrip(session_factory: sessionmaker[Session]) -> None:
+    assert load_settings(session_factory) == AppSettings()
+    save_settings(session_factory, AppSettings(retention_days=7, notify_mode=NotifyMode.EVERY))
+    loaded = load_settings(session_factory)
+    assert loaded.retention_days == 7 and loaded.notify_mode is NotifyMode.EVERY
+
+
+# -- Aufräumen ---------------------------------------------------------------------------------
+
+
+def test_cleanup_removes_old_runs_with_items(
+    service: RunService,
+    fake_runner: FakeRunner,
+    rule_id: int,
+    session_factory: sessionmaker[Session],
+) -> None:
+    fake_runner.results += [ok(), ok()]
+    old = service.execute(rule_id)
+    new = service.execute(rule_id)
+    with session_scope(session_factory) as s:
+        run = s.get(Run, old)
+        assert run is not None
+        run.started_at = utcnow() - timedelta(days=31)
+        s.add(
+            FileFailure(
+                rule_id=rule_id,
+                src_relpath="x",
+                size=1,
+                mtime_ns=1,
+                attempts=1,
+                last_attempt_at=utcnow() - timedelta(days=40),
+            )
+        )
+    stats = cleanup(session_factory)
+    assert stats == {"runs": 1, "file_failures": 1}
+    with session_factory() as s:
+        assert [r.id for r in s.scalars(select(Run))] == [new]
+        assert len(s.scalars(select(RunItem)).all()) == 1
+
+
+# -- Health (Fallstrick 9) -------------------------------------------------------------------
+
+
+def _health(sf: sessionmaker[Session], started_ago: float = 3600, **kw: object):  # type: ignore[no-untyped-def]
+    now = utcnow()
+    args: dict[str, object] = {
+        "scheduler_running": True,
+        "last_heartbeat": now,
+        "service_started": now - timedelta(seconds=started_ago),
+        "now": now,
+    }
+    args.update(kw)
+    return check_health(sf, **args)  # type: ignore[arg-type]
+
+
+def _age_rule(sf: sessionmaker[Session], rid: int, seconds: float | None) -> None:
+    with session_scope(sf) as s:
+        r = s.get(Rule, rid)
+        assert r is not None
+        r.created_at = utcnow() - timedelta(days=1)
+        r.last_success_at = None if seconds is None else utcnow() - timedelta(seconds=seconds)
+
+
+def test_health_ok(session_factory: sessionmaker[Session], rule_id: int) -> None:
+    _age_rule(session_factory, rule_id, 60)  # 2× Intervall
+    h = _health(session_factory)
+    assert h.ok, h.problems
+
+
+def test_health_stale_rule_after_three_intervals(
+    session_factory: sessionmaker[Session], rule_id: int
+) -> None:
+    _age_rule(session_factory, rule_id, 91)
+    h = _health(session_factory)
+    assert not h.ok and h.stale_rules == {rule_id}
+    assert "kein erfolgreicher Lauf" in h.problems[0]
+
+
+def test_health_never_succeeded_but_service_just_started(
+    session_factory: sessionmaker[Session], rule_id: int
+) -> None:
+    _age_rule(session_factory, rule_id, None)
+    assert _health(session_factory, started_ago=30).ok
+    assert not _health(session_factory, started_ago=120).ok
+
+
+def test_health_disabled_rule_is_ignored(
+    session_factory: sessionmaker[Session], rule_id: int
+) -> None:
+    _age_rule(session_factory, rule_id, None)
+    with session_scope(session_factory) as s:
+        r = s.get(Rule, rule_id)
+        assert r is not None
+        r.enabled = False
+    assert _health(session_factory).ok
+
+
+def test_health_mount_error_only_for_enabled_mounts(
+    session_factory: sessionmaker[Session], make_mount: MountFactory
+) -> None:
+    mid = make_mount("b", share="b", enabled=False)
+    with session_scope(session_factory) as s:
+        m = s.get(Mount, mid)
+        assert m is not None
+        m.state = MountState.ERROR
+    assert _health(session_factory).ok
+    with session_scope(session_factory) as s:
+        m = s.get(Mount, mid)
+        assert m is not None
+        m.enabled = True
+    h = _health(session_factory)
+    assert not h.ok and "Fehlerzustand" in h.problems[0]
+
+
+def test_health_scheduler(session_factory: sessionmaker[Session]) -> None:
+    assert not _health(session_factory, scheduler_running=False).ok
+    assert not _health(session_factory, last_heartbeat=utcnow() - timedelta(minutes=5)).ok
+
+
+def test_runner_type_is_compatible(session_factory: sessionmaker[Session], tmp_path: Path) -> None:
+    # FakeRunner oben muss dieselbe Schnittstelle bieten wie RuleRunner.run
+    assert RuleRunner.run.__code__.co_kwonlyargcount == 1

tests/test_scheduler.py

@@ -0,0 +1,177 @@
+"""Scheduler: Jobs je Regel, Versatz, Neuabgleich, manuelle Läufe, Dienststart."""
+
+from __future__ import annotations
+
+import asyncio
+import random
+from collections.abc import Callable, Coroutine
+from datetime import timedelta
+from pathlib import Path
+from typing import Any
+
+import pytest
+from sqlalchemy import select
+from sqlalchemy.orm import Session, sessionmaker
+
+from netmover.app_state import NetMover
+from netmover.config import Settings
+from netmover.db import session_scope
+from netmover.models import Mount, MountState, MountType, Rule, Run, RunStatus, RunTrigger, utcnow
+from netmover.mounting.fake import FakeMountBackend
+from netmover.mounting.manager import MountManager
+from netmover.mover.runner import RuleRunner
+from netmover.runs import RunService
+from netmover.scheduler import MAX_START_OFFSET, RuleScheduler
+from tests.conftest import MountFactory
+
+
+def arun(fn: Callable[[], Coroutine[Any, Any, None]]) -> None:
+    asyncio.run(fn())
+
+
+@pytest.fixture
+def rules(
+    make_mount: MountFactory,
+    session_factory: sessionmaker[Session],
+    fake_backend: FakeMountBackend,
+    manager: MountManager,
+) -> list[int]:
+    fake_backend.export("nas", "scans")
+    mid = make_mount()
+    manager.mount(mid)
+    ids = []
+    with session_scope(session_factory) as s:
+        for name, interval, enabled in [("a", 30, True), ("b", 60, True), ("c", 30, False)]:
+            r = Rule(
+                name=name,
+                src_mount_id=mid,
+                dst_path=str(fake_backend.remote_root),
+                interval_seconds=interval,
+                enabled=enabled,
+                stability_seconds=0,
+            )
+            s.add(r)
+            s.flush()
+            ids.append(r.id)
+    return ids
+
+
+@pytest.fixture
+def scheduler(session_factory: sessionmaker[Session], manager: MountManager) -> RuleScheduler:
+    runner = RuleRunner(session_factory, manager, sleep=lambda _: None)
+    return RuleScheduler(session_factory, RunService(session_factory, runner), rng=random.Random(1))  # noqa: S311
+
+
+def test_jobs_follow_rules(
+    scheduler: RuleScheduler, rules: list[int], session_factory: sessionmaker[Session]
+) -> None:
+    a, b, c = rules
+
+    async def main() -> None:
+        scheduler.start()
+        try:
+            assert scheduler.next_run(a) is not None
+            assert scheduler.next_run(b) is not None
+            assert scheduler.next_run(c) is None
+            # erster Termin mit kleinem Versatz
+            for rid in (a, b):
+                nxt = scheduler.next_run(rid)
+                assert nxt is not None
+                # Versatz ≤ 10 s plus Jitter ≤ 5 s
+                assert nxt - utcnow() <= timedelta(seconds=MAX_START_OFFSET + 5 + 1)
+
+            with session_scope(session_factory) as s:
+                ra, rc = s.get(Rule, a), s.get(Rule, c)
+                assert ra is not None and rc is not None
+                ra.enabled = False
+                rc.enabled = True
+                rb = s.get(Rule, b)
+                assert rb is not None
+                rb.interval_seconds = 120
+            scheduler.sync()
+            assert scheduler.next_run(a) is None
+            assert scheduler.next_run(c) is not None
+            job = scheduler.scheduler.get_job(f"rule:{b}")
+            assert job is not None and job.trigger.interval == timedelta(seconds=120)
+        finally:
+            scheduler.shutdown()
+
+    arun(main)
+
+
+def test_run_now_records_manual_run(
+    scheduler: RuleScheduler, rules: list[int], session_factory: sessionmaker[Session]
+) -> None:
+    a = rules[0]
+
+    async def main() -> None:
+        scheduler.start()
+        try:
+            run_id = await scheduler.run_now(a, dry_run=True)
+        finally:
+            scheduler.shutdown()
+        with session_factory() as s:
+            run = s.get(Run, run_id)
+            assert run is not None
+            assert run.trigger is RunTrigger.MANUAL and run.dry_run and run.status is RunStatus.OK
+
+    arun(main)
+
+
+def test_scheduled_job_runs_in_thread_pool(
+    scheduler: RuleScheduler, rules: list[int], session_factory: sessionmaker[Session]
+) -> None:
+    a = rules[0]
+
+    async def main() -> None:
+        scheduler.start()
+        try:
+            await scheduler._run_rule(a)
+            assert scheduler.running and scheduler.last_heartbeat is not None
+        finally:
+            scheduler.shutdown()
+        with session_factory() as s:
+            runs = s.scalars(select(Run)).all()
+            assert [(r.rule_id, r.trigger) for r in runs] == [(a, RunTrigger.SCHEDULE)]
+
+    arun(main)
+
+
+def test_app_startup_restores_mounts_before_scheduler(tmp_path: Path) -> None:
+    settings = Settings(
+        data_dir=tmp_path / "lib",
+        run_dir=tmp_path / "run",
+        mount_base=tmp_path / "mnt",
+        mount_backend="fake",
+    )
+    backend = FakeMountBackend(tmp_path / "fake")
+    backend.export("nas", "scans")
+    app = NetMover(settings, backend=backend)
+    with session_scope(app.session_factory) as s:
+        s.add(
+            Mount(
+                name="scans",
+                slug="scans",
+                type=MountType.CIFS,
+                server="nas",
+                share="scans",
+                mountpoint=str((tmp_path / "mnt" / "scans").resolve()),
+            )
+        )
+    assert (tmp_path / "lib" / "secret.key").stat().st_mode & 0o777 == 0o600
+
+    async def main() -> None:
+        app.start()
+        try:
+            assert [r.ok for r in app.restore_results] == [True]
+            assert app.scheduler.running
+            assert app.health().ok
+        finally:
+            app.stop()
+
+    arun(main)
+    # Beim Beenden NICHT aushängen
+    assert backend.is_mounted((tmp_path / "mnt").resolve() / "scans")
+    with app.session_factory() as s:
+        m = s.scalars(select(Mount)).one()
+        assert m.state is MountState.MOUNTED