feat(web): Wartungsmodus, Wartezustand und E-Mail-Einstellungen; Doku 0.4.0

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

CHANGELOG.md

@@ -8,6 +8,28 @@ Die Version steht an genau einer Stelle: `src/netmover/__init__.py`.
 
 ## [Unreleased]
 
+## [0.4.0] – 2026-10-11
+
+### Hinzugefügt
+- **Automatisch pausieren und fortsetzen:** Ist Quelle oder Ziel nicht erreichbar,
+  wartet die Regel („wartet seit …“ mit Grund) – ein Protokolleintrag statt eines
+  Fehlers alle 30 s, „Fehler in Folge“ zählt nicht hoch. Sobald die Anbindung zurück
+  ist, läuft die Regel von selbst weiter („Fortgesetzt nach … Wartezeit“).
+  Migration 0003: `rules.waiting_since`, `rules.waiting_reason`.
+- **Wartungsmodus** (Knopf auf der Statusseite): alle geplanten Läufe pausiert,
+  „Jetzt ausführen“ geht weiter, kein Alarm für Regeln in `/healthz`; gelber Hinweis
+  in der Kopfzeile.
+- **E-Mail-Benachrichtigung** per SMTP (STARTTLS, SSL/TLS oder ohne für interne
+  Relays; Zertifikate werden geprüft), Passwort verschlüsselt, mehrere Empfänger,
+  Test-Knopf. Webhook und E-Mail lassen sich gleichzeitig nutzen.
+
+### Geändert
+- Ausfälle von Anbindungen werden **einmal pro Anbindung** gemeldet, mit der Liste
+  der betroffenen Regeln, und mit Entwarnung, wenn sie zurück sind – auch wenn eine
+  Anbindung von außen ausgehängt wurde (vorher keine Meldung).
+- Einstellungen neu gegliedert: Benachrichtigungen · Webhook · E-Mail · Überwachung.
+- `/healthz`: Eine wartende Regel ist ein Problem (mit Grund); im Wartungsmodus nicht.
+
 ## [0.3.0] – 2026-10-11
 
 ### Hinzugefügt
@@ -191,7 +213,8 @@ Erste Version.
 - Löschen von Zugangsdaten, die noch von einem Mount benutzt werden, trennt den
   Mount nicht mehr stillschweigend ab, sondern wird von der Datenbank abgelehnt.
 
-[Unreleased]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.3.0...HEAD
+[Unreleased]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.4.0...HEAD
+[0.4.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.3.0...v0.4.0
 [0.3.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.2.1...v0.3.0
 [0.2.1]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.2.0...v0.2.1
 [0.2.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.1.1...v0.2.0

DEVLOG.md

@@ -17,10 +17,38 @@ Zeiten in Ortszeit (Europe/Berlin).
 | 2026-10-10 | 22:45 | 23:00 | 0:15  | –       | Echtes NAS/Paperless anbinden          |
 | 2026-10-10 | 23:31 | 23:58 | 0:27  | –       | Schlüsselverwaltung, Update-Skript, 0.2.x |
 | 2026-10-11 | 00:13 | 00:32 | 0:19  | –       | Menü, Anleitung, Roadmap, Uptime Kuma (0.3.0) |
-| **Summe**  |       |       | **2:36** |      |                                        |
+| 2026-10-11 | 00:31 | 00:50 | 0:19  | –       | Pausieren, Wartungsmodus, E-Mail (0.4.0) |
+| **Summe**  |       |       | **2:55** |      |                                        |
 
 ---
 
+## 2026-10-11 · 00:31–00:50 · Pausieren, Wartungsmodus, E-Mail (0.4.0)
+
+**Erledigt**
+- Wartezustand (`rules.waiting_since/_reason`, Migration 0003), Wartungsmodus,
+  E-Mail (`mail.py`, SMTP), eine Meldung pro Anbindung mit betroffenen Regeln.
+- 0.4.0/0.5.0 aufgeteilt: kleinere Updates, schneller Nutzen.
+- 21 neue Tests (299 gesamt), u. a. SMTP gegen echten Server (aiosmtpd) mit
+  STARTTLS, SSL, falschem Passwort und nicht vertrauenswürdigem Zertifikat.
+
+**Entscheidungen**
+- Warten statt Deaktivieren; kein Fehlerzähler und nur ein Protokolleintrag beim
+  Eintritt; Ende des Wartens automatisch beim nächsten durchkommenden Lauf.
+- Benachrichtigung pro Anbindung (nicht pro Regel) – sonst bei drei Regeln am NAS
+  drei gleiche Meldungen. Lücke geschlossen: Aushängen von außen meldete nichts.
+- Wartungsmodus unterdrückt nur Regel-Alarme, Anbindungen werden weiter geprüft.
+- SMTP prüft Zertifikate immer; für Tests ist nur `mail.tls_context()` ersetzbar.
+
+**Stolpersteine (alle im Testaufbau)**
+- Globales Monkeypatchen von `ssl.create_default_context` traf auch den
+  Testserver → eigene Funktion `tls_context()`.
+- aiosmtpd: `AuthResult(success=False)` ohne `handled=False` schickt keine Antwort →
+  Client hängt bis zum Zeitlimit.
+
+**Migrationen** – 0003 baut `rules` neu; getestet von 0001 und von 0002 mit allen
+verknüpften Tabellen sowie an der befüllten Demo-Datenbank; vor dem Release an einer
+Kopie der Live-Datenbank auf der VM.
+
 ## 2026-10-11 · 00:13–00:32 · Menü, Anleitung, Roadmap, Uptime Kuma (0.3.0)
 
 **Erledigt**

ROADMAP.md

@@ -6,18 +6,12 @@ stehen mit Version im [CHANGELOG](CHANGELOG.md); Details zu Entscheidungen im
 
 Legende: ✅ erledigt · 🔨 in Arbeit · 📋 geplant · 💡 Idee (noch nicht entschieden)
 
-## In Arbeit
-
-Gerade nichts – als Nächstes kommt 0.4.0.
-
-## Als Nächstes (0.4.0)
+## In Arbeit (0.5.0)
 
 | Status | Thema | Warum |
 |---|---|---|
-| 📋 | **SFTP als Quelle**: Scan, Stabilitätsprüfung und Löschen der Quelle über SFTP, mit denselben Garantien wie bei SMB/NFS | Quellen, die nur per SSH erreichbar sind (Wunsch aus dem Praxisbetrieb) |
-| 📋 | **Gespeicherte Zugangsdaten** für SMB und SFTP-Passwörter, zusammen mit den SSH-Schlüsseln auf einer Seite „Zugangsdaten“ | Mehrere Freigaben mit demselben NAS-Benutzer an einer Stelle pflegen |
-| 📋 | **Automatisch pausieren und fortsetzen**: Ist Quelle oder Ziel nicht erreichbar, wechselt die Regel in „wartet auf Anbindung X“ – ein Protokolleintrag und eine Benachrichtigung statt eines Fehlers alle 30 s; fortsetzen von selbst, sobald die Anbindung zurück ist. Dazu manuelles Pausieren (Wartungsmodus) | Weniger Rauschen im Protokoll und in Benachrichtigungen, kein Handgriff nach NAS-Neustarts |
-| 📋 | **E-Mail-Benachrichtigung** (SMTP mit STARTTLS/SSL, verschlüsseltes Passwort, mehrere Empfänger) zusätzlich zum Webhook | Benachrichtigung ohne eigenen Webhook-Dienst |
+| 🔨 | **SFTP als Quelle**: Scan, Stabilitätsprüfung und Löschen der Quelle über SFTP, mit denselben Garantien wie bei SMB/NFS | Quellen, die nur per SSH erreichbar sind |
+| 🔨 | **Gespeicherte Zugangsdaten** für SMB und SFTP-Passwörter, zusammen mit den SSH-Schlüsseln auf einer Seite „Zugangsdaten“ | Mehrere Freigaben mit demselben NAS-Benutzer an einer Stelle pflegen |
 
 ## Geplant
 
@@ -41,6 +35,7 @@ Gerade nichts – als Nächstes kommt 0.4.0.
 
 | Version | Thema |
 |---|---|
+| 0.4.0 | Automatisch pausieren/fortsetzen bei nicht erreichbaren Anbindungen; Wartungsmodus; E-Mail-Benachrichtigung; eine Meldung pro ausgefallener Anbindung |
 | 0.3.0 | Neues Menü (gruppierte Seitenleiste, Statusampel, Handy-Menü); Anleitung im Repo und als Hilfe-Seite; Roadmap; Uptime Kuma (Einzel-Endpunkte, Push-Monitor) |
 | 0.2.1 | Migrationen in einer Transaktion und ohne Fremdschlüsselfehler auf befüllten Datenbanken |
 | 0.2.0 | Schlüsselverwaltung für SFTP; `netmover-update` mit Sicherung; Installation als Git-Klon |

compose.yml

@@ -4,7 +4,7 @@
 services:
   netmover:
     build: .
-    image: netmover:0.3.0
+    image: netmover:0.4.0
     container_name: netmover
     restart: unless-stopped
     cap_add:

docs/ANLEITUNG.md

@@ -9,7 +9,7 @@ stehen im [README](../README.md).
 
 **Inhalt:** [Begriffe](#begriffe) · [Erste Anmeldung](#erste-anmeldung) ·
 [Anbindungen](#anbindungen) · [Schlüssel](#schluessel) · [Regeln](#regeln) ·
-[Probelauf](#probelauf) · [Status](#status) · [Protokoll](#protokoll) ·
+[Probelauf](#probelauf) · [Status](#status) · [Wartungsmodus](#wartungsmodus) · [Protokoll](#protokoll) ·
 [Einstellungen](#einstellungen) · [Uptime Kuma](#uptime-kuma) · [Beispiele](#beispiele) · [Häufige Fragen](#fragen)
 
 ## Die vier Begriffe {#begriffe}
@@ -99,6 +99,8 @@ Server neu aufgesetzt: bei der Anbindung **Hostschlüssel zurücksetzen** und ne
 
 Regeln, deren Anbindung nicht verfügbar ist, **brechen ab, ohne etwas anzufassen**.
 Es wird nie in ein leeres Verzeichnis geschrieben, nur weil die Freigabe fehlt.
+Solche Regeln **warten** automatisch, bis die Anbindung zurück ist (siehe
+[Status](#status)), und laufen dann von selbst weiter.
 
 ## Schlüssel {#schluessel}
 
@@ -202,6 +204,28 @@ Eine Regel wird **gelb**, wenn ihr letzter erfolgreicher Lauf länger als das
 Dreifache ihres Intervalls zurückliegt. So fällt auf, wenn NetMover still ausfällt
 – statt dass es aussieht, als gäbe es nichts zu tun.
 
+### Regel wartet auf eine Anbindung {#wartet}
+
+Ist die Quelle oder das Ziel einer Regel nicht erreichbar (NAS neu gestartet, Server
+aus), **pausiert die Regel automatisch**: Sie zeigt „wartet seit …“ mit dem Grund.
+
+- Im Protokoll steht **ein** Eintrag „Pausiert, bis die Anbindung zurück ist“ – nicht
+  alle 30 Sekunden ein neuer Fehler.
+- Es kommt **eine** Benachrichtigung pro Anbindung, mit der Liste der betroffenen
+  Regeln.
+- Sobald die Anbindung zurück ist, **läuft die Regel von selbst weiter**; der nächste
+  Lauf vermerkt „Fortgesetzt nach … Wartezeit“, und es kommt eine Entwarnung.
+
+Man muss also nach einem NAS-Neustart nichts tun.
+
+### Wartungsmodus {#wartungsmodus}
+
+Der Knopf **Wartungsmodus** auf der Statusseite pausiert **alle geplanten Läufe** –
+etwa während eines NAS-Updates. Oben erscheint dann gelb „Wartungsmodus“. „Jetzt
+ausführen“ geht weiterhin. Für Regeln meldet `/healthz` in dieser Zeit keinen Alarm
+(Anbindungen werden weiter geprüft). Mit **Wartungsmodus beenden** läuft alles
+wieder nach Plan.
+
 ## Protokoll {#protokoll}
 
 *Automatisierung → Protokoll* zeigt jeden Lauf: Zeitpunkt, Dauer, gesehen /
@@ -218,7 +242,8 @@ Häufige Einträge:
 | *Quelle hat sich während der Übertragung geändert* | Abgebrochen, nichts gelöscht; nächster Lauf versucht es erneut |
 | *n bereits übertragen* | Kopiermodus: diese Dateiversionen sind schon kopiert |
 | *Termin übersprungen – voriger Lauf läuft noch* | Ein Lauf dauert länger als das Intervall |
-| *nicht eingehängt – Lauf abgebrochen, nichts angefasst* | Anbindung fehlt; siehe [Zustände](#anbindung-zustaende) |
+| *Pausiert, bis die Anbindung zurück ist* | Quelle oder Ziel nicht erreichbar; die Regel wartet, siehe [Regel wartet](#wartet) |
+| *Fortgesetzt nach … Wartezeit* | Die Anbindung ist zurück, die Regel läuft wieder |
 
 Protokolle werden nach der eingestellten Frist gelöscht (Standard 30 Tage).
 
@@ -228,15 +253,31 @@ Protokolle werden nach der eingestellten Frist gelöscht (Standard 30 Tage).
 
 ### Benachrichtigungen {#benachrichtigungen}
 
-NetMover kann bei Fehlern eine Nachricht an einen **Webhook** schicken – z. B. an
-ntfy, Gotify oder Home Assistant.
+NetMover meldet Probleme per **E-Mail** und/oder an einen **Webhook** (z. B. ntfy,
+Gotify, Home Assistant). Beides lässt sich gleichzeitig nutzen.
+
+| Feld | Bedeutung |
+|---|---|
+| **Benachrichtigen** | nie / bei jedem fehlgeschlagenen Lauf / erst nach N Fehlschlägen in Folge. Fällt eine **Anbindung** aus, kommt die Meldung immer sofort (einmal, mit den betroffenen Regeln). |
+| **Entwarnung** | Nachricht, wenn eine Regel oder Anbindung wieder funktioniert |
+
+**E-Mail:**
+
+| Feld | Bedeutung |
+|---|---|
+| **SMTP-Server, Port, Verschlüsselung** | z. B. `smtp.example.org`, `587` mit STARTTLS oder `465` mit SSL/TLS. „keine“ nur für interne Relays. Zertifikate werden immer geprüft. |
+| **Benutzer, Passwort** | Anmeldung am SMTP-Server; leer = ohne Anmeldung. Das Passwort wird verschlüsselt gespeichert und nie angezeigt. |
+| **Absender, Empfänger** | Empfänger mit Komma getrennt |
+
+*Test-E-Mail senden* prüft die gespeicherten Einstellungen und zeigt bei Fehlern die
+Antwort des Servers.
+
+**Webhook:**
 
 | Feld | Bedeutung |
 |---|---|
-| **Benachrichtigen** | nie / bei jedem fehlgeschlagenen Lauf / erst nach N Fehlschlägen in Folge |
 | **Webhook-URL** | Adresse des Empfängers. Wird nur gekürzt angezeigt, weil sie oft ein Token enthält. |
 | **Content-Type, Vorlage** | Format der Nachricht; Platzhalter `$title`, `$message`, `$rule`, `$event`, `$time`, `$host` |
-| **Entwarnung** | Nachricht, wenn eine Regel oder Anbindung wieder funktioniert |
 
 Beispiele: ntfy → URL `https://ntfy.sh/<thema>`, Content-Type `text/plain`, Vorlage
 `$title: $message`. Gotify → URL `https://gotify.lan/message?token=…`, Content-Type
@@ -315,7 +356,11 @@ bei regulären Ausdrücken nicht), Datei wird noch geschrieben oder ist größer
 
 **Die Anbindung zeigt „Fehler“.** – Die Meldung darunter nennt die Ursache und meist
 die Abhilfe. Ist das NAS nur neu gestartet, hängt NetMover die Freigabe innerhalb
-einer Minute selbst wieder ein.
+einer Minute selbst wieder ein; die Regeln warten so lange und laufen dann weiter.
+
+**Ich mache gleich ein NAS-Update.** – Vorher auf der Statusseite *Wartungsmodus*
+einschalten, danach wieder beenden. Nötig ist es nicht (die Regeln warten ohnehin
+von selbst), aber so gibt es keine Alarme.
 
 **„mount error(95): Operation not supported“** – NAS und NetMover finden keine
 gemeinsame SMB-Version. Am NAS das maximale SMB-Protokoll auf SMB3 stellen

pyproject.toml

@@ -36,6 +36,7 @@ dev = [
     "types-paramiko",
     "types-Markdown",
     "shellcheck-py",
+    "aiosmtpd",
 ]
 
 [project.scripts]

src/netmover/__init__.py

@@ -1,3 +1,3 @@
 """NetMover – NFS/SMB-Mounts verwalten und Dateien zeitgesteuert verschieben."""
 
-__version__ = "0.3.0"
+__version__ = "0.4.0"

src/netmover/app_state.py

@@ -50,7 +50,7 @@ class NetMover:
         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.notifier = notifier or Notifier(self.session_factory, secrets=self.secrets)
         self.mounts = MountManager(
             self.session_factory,
             self.backend,
@@ -93,4 +93,5 @@ class NetMover:
             scheduler_running=self.scheduler.running,
             last_heartbeat=self.scheduler.last_heartbeat,
             service_started=self.started_at,
+            maintenance=load_settings(self.session_factory).maintenance,
         )

src/netmover/health.py

@@ -49,6 +49,7 @@ class Health:
     problems: list[str] = field(default_factory=list)
     stale_rules: set[int] = field(default_factory=set)
     checks: list[Check] = field(default_factory=list)
+    maintenance: bool = False
 
     def add(self, check: Check) -> None:
         self.checks.append(check)
@@ -86,12 +87,22 @@ def _mount_check(m: Mount) -> Check:
     return mk(True, f"Anbindung „{m.name}“ in Ordnung")
 
 
-def _rule_check(r: Rule, service_started: datetime, now: datetime) -> Check:
+def _rule_check(
+    r: Rule, service_started: datetime, now: datetime, maintenance: bool = False
+) -> Check:
     def mk(ok: bool, message: str) -> Check:
         return Check("rule", r.name, ok, message, id=r.id)
 
     if not r.enabled:
         return mk(True, f"Regel „{r.name}“ deaktiviert")
+    if maintenance:
+        return mk(True, f"Regel „{r.name}“ pausiert (Wartungsmodus)")
+    if r.waiting_since is not None:
+        return mk(
+            False,
+            f"Regel „{r.name}“ wartet seit {r.waiting_since:%Y-%m-%d %H:%M} UTC auf eine "
+            f"Anbindung: {r.waiting_reason or ''}".strip(),
+        )
     if rule_is_stale(r, service_started, now):
         ago = (
             "noch nie"
@@ -113,9 +124,11 @@ def check_health(
     last_heartbeat: datetime | None,
     service_started: datetime,
     now: datetime | None = None,
+    maintenance: bool = False,
 ) -> Health:
     now = now or utcnow()
     health = Health()
+    health.maintenance = maintenance
     if not scheduler_running:
         health.add(Check("scheduler", "Scheduler", False, "Scheduler läuft nicht"))
     elif last_heartbeat is None or now - last_heartbeat > HEARTBEAT_MAX_AGE:
@@ -128,8 +141,8 @@ def check_health(
         for m in s.scalars(select(Mount).order_by(Mount.name)):
             health.add(_mount_check(m))
         for r in s.scalars(select(Rule).order_by(Rule.name)):
-            check = _rule_check(r, service_started, now)
-            if r.enabled and rule_is_stale(r, service_started, now):
+            check = _rule_check(r, service_started, now, maintenance)
+            if r.enabled and not maintenance and rule_is_stale(r, service_started, now):
                 health.stale_rules.add(r.id)
             health.add(check)
     return health

src/netmover/mail.py

@@ -0,0 +1,91 @@
+"""E-Mail-Benachrichtigung per SMTP.
+
+* Verschlüsselung: ``starttls`` (Port 587, Standard), ``ssl`` (Port 465) oder ``none``
+  (nur für interne Relays ohne TLS). Zertifikate werden immer geprüft.
+* Das Passwort liegt Fernet-verschlüsselt in den Einstellungen.
+"""
+
+from __future__ import annotations
+
+import re
+import smtplib
+import socket
+import ssl
+from datetime import datetime
+from email.message import EmailMessage
+from email.utils import formatdate, make_msgid
+
+from netmover.settings_store import AppSettings
+
+SECURITY_MODES = ("starttls", "ssl", "none")
+_ADDRESS = re.compile(r"^[^@\s,;<>]+@[^@\s,;<>]+\.[^@\s,;<>]+$")
+
+
+class MailConfigError(ValueError):
+    pass
+
+
+def parse_recipients(raw: str) -> list[str]:
+    recipients = [r.strip() for r in re.split(r"[,;\n]", raw) if r.strip()]
+    bad = [r for r in recipients if not _ADDRESS.fullmatch(r)]
+    if bad:
+        raise MailConfigError(f"Ungültige Adresse: {', '.join(bad)}")
+    return recipients
+
+
+def valid_address(value: str) -> bool:
+    return bool(_ADDRESS.fullmatch(value.strip()))
+
+
+def mail_configured(cfg: AppSettings) -> bool:
+    return bool(cfg.email_enabled and cfg.smtp_host and cfg.smtp_from and cfg.smtp_to)
+
+
+def build_message(cfg: AppSettings, subject: str, body: str) -> EmailMessage:
+    msg = EmailMessage()
+    msg["Subject"] = subject
+    msg["From"] = cfg.smtp_from
+    msg["To"] = ", ".join(parse_recipients(cfg.smtp_to))
+    msg["Date"] = formatdate(localtime=True)
+    msg["Message-ID"] = make_msgid(domain=cfg.smtp_from.rsplit("@", 1)[-1])
+    msg["X-Mailer"] = "NetMover"
+    footer = (
+        f"\n\n-- \nNetMover auf {socket.gethostname()}, "
+        f"{datetime.now().astimezone():%d.%m.%Y %H:%M:%S %Z}"
+    )
+    msg.set_content(body + footer)
+    return msg
+
+
+def tls_context() -> ssl.SSLContext:
+    """Zertifikatsprüfung gegen die System-CAs (eigene Funktion, damit Tests ihr
+    Test-Zertifikat einsetzen können, ohne die Prüfung abzuschalten)."""
+    return ssl.create_default_context()
+
+
+def smtp_send(
+    cfg: AppSettings, password: str | None, subject: str, body: str, timeout: float = 20
+) -> None:
+    """Verschickt eine Nachricht; Fehler werden als Ausnahme weitergereicht."""
+    if cfg.smtp_security not in SECURITY_MODES:
+        raise MailConfigError(f"Unbekannte Verschlüsselung: {cfg.smtp_security}")
+    msg = build_message(cfg, subject, body)
+    context = tls_context()
+    smtp: smtplib.SMTP
+    if cfg.smtp_security == "ssl":
+        smtp = smtplib.SMTP_SSL(cfg.smtp_host, cfg.smtp_port, timeout=timeout, context=context)
+    else:
+        smtp = smtplib.SMTP(cfg.smtp_host, cfg.smtp_port, timeout=timeout)
+    try:
+        smtp.ehlo()
+        if cfg.smtp_security == "starttls":
+            smtp.starttls(context=context)
+            smtp.ehlo()
+        if cfg.smtp_user:
+            smtp.login(cfg.smtp_user, password or "")
+        smtp.send_message(msg)
+    finally:
+        try:
+            smtp.quit()
+        except smtplib.SMTPException:
+            smtp.close()

src/netmover/migrations/versions/0003_rule_waiting_state.py

@@ -0,0 +1,32 @@
+"""rule waiting state
+
+Revision ID: 0003
+Revises: 0002
+Create Date: 2026-10-11
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+import netmover.models
+
+revision: str = "0003"
+down_revision: str | None = "0002"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("rules", schema=None) as batch_op:
+        batch_op.add_column(
+            sa.Column("waiting_since", netmover.models.UTCDateTime(), nullable=True)
+        )
+        batch_op.add_column(sa.Column("waiting_reason", sa.Text(), nullable=True))
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("rules", schema=None) as batch_op:
+        batch_op.drop_column("waiting_reason")
+        batch_op.drop_column("waiting_since")

src/netmover/models.py

@@ -10,6 +10,8 @@ Abweichungen bzw. Ergänzungen gegenüber dem Mindestumfang aus der Spezifikatio
 * ``credentials.managed`` + ``mounts.username`` (Migration 0002) –
   Schlüsselverwaltung: gespeicherte SSH-Schlüssel, die mehrere Anbindungen mit
   jeweils eigenem Benutzernamen nutzen.
+* ``rules.waiting_since`` / ``rules.waiting_reason`` (Migration 0003) – Regel wartet
+  auf eine nicht erreichbare Anbindung (automatisch pausiert).
 * ``mounts.state_since`` – für „hängt seit …“ / „gestört seit …“ auf der Statusseite.
 * ``mounts.write_ok`` / ``mounts.root_squash`` / ``mounts.last_test_*`` – Ergebnis
   des letzten Mount-Tests; ``root_squash`` entscheidet, ob ``chown`` auf NFS-Zielen
@@ -260,6 +262,9 @@ class Rule(Base):
     last_run_at: Mapped[datetime | None] = mapped_column(UTCDateTime)
     last_success_at: Mapped[datetime | None] = mapped_column(UTCDateTime)
     consecutive_failures: Mapped[int] = mapped_column(Integer, default=0)
+    # Automatisch pausiert, solange Quelle oder Ziel nicht erreichbar ist (Migration 0003)
+    waiting_since: Mapped[datetime | None] = mapped_column(UTCDateTime)
+    waiting_reason: Mapped[str | None] = mapped_column(Text)
     created_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow)
 
     src_mount: Mapped[Mount] = relationship(foreign_keys=[src_mount_id])

src/netmover/mounting/manager.py

@@ -32,7 +32,7 @@ from sqlalchemy.orm import Session, sessionmaker
 from netmover.crypto import SecretBox, redact
 from netmover.db import session_scope
 from netmover.logging_setup import event, get_logger
-from netmover.models import CredentialKind, Mount, MountState, MountType, utcnow
+from netmover.models import CredentialKind, Mount, MountState, MountType, Rule, utcnow
 from netmover.mounting.backend import CifsCredentials, MountBackend, MountError, MountSpec
 from netmover.mounting.report import CheckReport, assess_write
 from netmover.mounting.sftp import SftpConfig, SftpError, check_connection, connect
@@ -203,13 +203,32 @@ class MountManager:
         with self._locks_guard:
             return self._locks.setdefault(mount_id, threading.RLock())
 
-    def _notify(self, kind: str, name: str, message: str) -> None:
-        if self._notifier is not None:
+    def _notify(self, kind: str, name: str, message: str, mount_id: int | None = None) -> None:
+        if self._notifier is None:
+            return
+        if mount_id is not None:
+            rules = self._affected_rules(mount_id)
+            if rules and kind != "mount.recovered":
+                waiting = "\nBetroffene Regeln (warten, bis die Anbindung zurück ist): "
+                message += waiting + ", ".join(rules)
+            elif rules:
+                message += f"\nRegeln laufen wieder: {', '.join(rules)}"
         try:
             self._notifier(kind, name, message)
         except Exception:
             log.exception("notifier.failed")
 
+    def _affected_rules(self, mount_id: int) -> list[str]:
+        with session_scope(self.session_factory) as s:
+            return list(
+                s.scalars(
+                    select(Rule.name)
+                    .where(Rule.enabled.is_(True))
+                    .where((Rule.src_mount_id == mount_id) | (Rule.dst_mount_id == mount_id))
+                    .order_by(Rule.name)
+                )
+            )
+
     def mountpoint_for(self, slug: str) -> Path:
         if slugify(slug) != slug:
             raise ValidationError("slug", f"Ungültiger Slug: {slug!r}")
@@ -347,7 +366,7 @@ class MountManager:
                 # z. B. Stromausfall: die VM bootet schneller als das NAS
                 self._retry_at[mount_id] = self._clock() + self.retry_interval
                 results.append(RestoreResult(name, False, str(exc)))
-                self._notify("mount.restore_failed", name, str(exc))
+                self._notify("mount.restore_failed", name, str(exc), mount_id)
         for r in results:
             event(
                 log,
@@ -531,6 +550,8 @@ class MountManager:
                     self._retry_mount(mount_id, m, cred, mp)
                     return mp
                 msg = f"„{m.name}“ ist nicht eingehängt – Lauf abgebrochen, nichts angefasst."
+                if m.state is not MountState.ERROR:
+                    self._notify("mount.error", m.name, msg, mount_id)
                 self._set_state(mount_id, MountState.ERROR, msg)
                 raise MountUnavailable(msg)
             try:
@@ -563,8 +584,10 @@ class MountManager:
             msg = f"„{m.name}“ nicht erreichbar: {exc}"
             self._set_state(mount_id, MountState.ERROR, msg)
             if m.state is not MountState.ERROR:
-                self._notify("mount.error", m.name, msg)
+                self._notify("mount.error", m.name, msg, mount_id)
             raise MountUnavailable(msg) from None
+        if m.state is MountState.ERROR:
+            self._notify("mount.recovered", m.name, f"„{m.name}“ ist wieder erreichbar.", mount_id)
         if m.state is not MountState.MOUNTED:
             self._set_state(mount_id, MountState.MOUNTED)
 
@@ -601,7 +624,7 @@ class MountManager:
             self._retry_at[mount_id] = self._clock() + self.retry_interval
             msg += f"\n→ Neuer Versuch automatisch alle {self.retry_interval:g} s."
         self._set_state(mount_id, MountState.ERROR, msg)
-        self._notify("mount.error", m.name, msg)
+        self._notify("mount.error", m.name, msg, mount_id)
         raise MountUnavailable(msg)
 
     def retry_pending(self) -> list[str]:
@@ -663,7 +686,7 @@ class MountManager:
         del self._retry_at[mount_id]
         self._set_state(mount_id, MountState.MOUNTED, mounted=True)
         event(log, "mount.recovered", mount=m.name, attempt="retry")
-        self._notify("mount.recovered", m.name, f"„{m.name}“ ist wieder eingehängt.")
+        self._notify("mount.recovered", m.name, f"„{m.name}“ ist wieder eingehängt.", mount_id)
 
     def usage(self, mount_id: int, timeout: float | None = None) -> tuple[int, int] | None:
         """(gesamt, frei) in Bytes – nur bei eingehängten Kernel-Mounts."""

src/netmover/mover/engine.py

@@ -92,6 +92,8 @@ class RunResult:
     failed: int = 0
     message: str | None = None
     items: list[ItemResult] = field(default_factory=list)
+    # Quelle oder Ziel nicht erreichbar → Regel wartet (automatisch pausiert)
+    unavailable: bool = False
 
     def add(self, item: ItemResult) -> None:
         self.items.append(item)

src/netmover/mover/runner.py

@@ -178,7 +178,7 @@ class RuleRunner:
                 else None
             )
         except MountUnavailable as exc:
-            return RunResult(RunStatus.ERROR, dry, message=str(exc))
+            return RunResult(RunStatus.ERROR, dry, message=str(exc), unavailable=True)
         if src_mp is None:  # nur bei SFTP, oben bereits ausgeschlossen
             return RunResult(RunStatus.ERROR, dry, message="Quelle ist kein Kernel-Mount")
         src_root = src_mp / plan.src_subpath if plan.src_subpath else src_mp

src/netmover/notify.py

@@ -22,7 +22,9 @@ from string import Template
 import httpx
 from sqlalchemy.orm import Session, sessionmaker
 
+from netmover.crypto import SecretBox
 from netmover.logging_setup import event, get_logger
+from netmover.mail import mail_configured, smtp_send
 from netmover.settings_store import AppSettings, NotifyMode, load_settings
 
 log = get_logger(__name__)
@@ -60,34 +62,71 @@ def http_sender(url: str, content_type: str, body: str) -> None:
     resp.raise_for_status()
 
 
+Mailer = Callable[[AppSettings, "str | None", str, str], None]  # (cfg, passwort, betreff, text)
+
+
 class Notifier:
+    """Verteilt Benachrichtigungen auf die eingerichteten Kanäle (Webhook, E-Mail)."""
+
     def __init__(
         self,
         session_factory: sessionmaker[Session],
         *,
         sender: Sender = http_sender,
+        mailer: Mailer | None = None,
+        secrets: SecretBox | None = None,
         background: bool = True,
     ) -> None:
         self.session_factory = session_factory
         self._sender = sender
+        self._mailer: Mailer = mailer or smtp_send
+        self._secrets = secrets
         self._background = background
 
     def settings(self) -> AppSettings:
         return load_settings(self.session_factory)
 
+    def smtp_password(self, cfg: AppSettings) -> str | None:
+        if not cfg.smtp_password_enc or self._secrets is None:
+            return None
+        return self._secrets.decrypt(cfg.smtp_password_enc)
+
+    def send_mail_now(self, subject: str, body: str, settings: AppSettings | None = None) -> None:
+        """Synchron senden (Test-Knopf); Fehler werden weitergereicht."""
+        cfg = settings or self.settings()
+        self._mailer(cfg, self.smtp_password(cfg), subject, body)
+
     def send(self, n: Notification, settings: AppSettings | None = None) -> None:
         cfg = settings or self.settings()
-        if cfg.notify_mode is NotifyMode.OFF or not cfg.webhook_url:
+        if cfg.notify_mode is NotifyMode.OFF:
             return
+        jobs: list[tuple[str, Callable[[], None]]] = []
+        if cfg.webhook_url:
             body = render(cfg.webhook_template, cfg.webhook_content_type, n)
+            jobs.append(
+                ("webhook", lambda: self._sender(cfg.webhook_url, cfg.webhook_content_type, body))
+            )
+        if mail_configured(cfg):
+            password = self.smtp_password(cfg)
+            jobs.append(("email", lambda: self._mailer(cfg, password, n.title, n.message)))
+        if not jobs:
+            return
 
         def deliver() -> None:
+            for channel, job in jobs:
                 try:
-                self._sender(cfg.webhook_url, cfg.webhook_content_type, body)
-                event(log, "notify.sent", kind=n.event, rule=n.rule)
+                    job()
+                    event(log, "notify.sent", kind=n.event, rule=n.rule, channel=channel)
                 except Exception as exc:  # Benachrichtigung darf nie den Dienst stören
-                # URL nicht protokollieren – sie enthält oft ein Token.
-                event(log, "notify.failed", level=30, kind=n.event, error=type(exc).__name__)
+                    # Weder URL noch Zugangsdaten protokollieren.
+                    event(
+                        log,
+                        "notify.failed",
+                        level=30,
+                        kind=n.event,
+                        channel=channel,
+                        error=type(exc).__name__,
+                    )
 
         if self._background:
             threading.Thread(target=deliver, name="nm-notify", daemon=True).start()

src/netmover/runs.py

@@ -8,6 +8,14 @@ die Einzeldateien in ``run_items``. Danach die Regelzähler:
 * ``consecutive_failures`` – ``error`` und ``partial`` zählen hoch, ``ok`` setzt zurück.
 
 Ein manuell gestarteter Probelauf verändert keine Zähler.
+
+Automatisch pausieren (0.4.0): Ist Quelle oder Ziel nicht erreichbar, wechselt die
+Regel in „wartet auf Anbindung“. Protokolliert wird nur der Beginn, nicht jeder
+weitere Termin, und „Fehler in Folge“ zählt nicht hoch – die Benachrichtigung
+kommt einmal pro Anbindung vom MountManager. Sobald ein Lauf wieder durchkommt,
+endet das Warten von selbst, und der Lauf vermerkt die Wartezeit.
+
+Wartungsmodus: geplante Läufe werden gar nicht erst ausgeführt.
 """
 
 from __future__ import annotations
@@ -51,6 +59,8 @@ class RunService:
     ) -> int | None:
         """Führt die Regel aus, protokolliert den Lauf und liefert die Run-ID
         (None, wenn die Regel inzwischen gelöscht wurde)."""
+        if trigger is RunTrigger.SCHEDULE and load_settings(self.session_factory).maintenance:
+            return None
         started = utcnow()
         result = self.runner.run(rule_id, dry_run=dry_run)
         finished = utcnow()
@@ -73,6 +83,29 @@ class RunService:
             if rule is None:
                 return None
             message = result.message
+            if update_counters and result.unavailable:
+                if rule.waiting_since is not None:
+                    # wartet bereits – nur den Grund aktualisieren, kein neuer Eintrag
+                    rule.waiting_reason = result.message
+                    event(log, "rule.still_waiting", level=10, rule=rule.name)
+                    return None
+                rule.waiting_since = finished
+                rule.waiting_reason = result.message
+                rule.last_run_at = finished
+                message = f"Pausiert, bis die Anbindung zurück ist: {result.message}"
+                update_counters = False  # „Fehler in Folge“ nicht hochzählen
+                event(log, "rule.waiting", level=30, rule=rule.name, reason=result.message or "")
+            elif (
+                update_counters
+                and rule.waiting_since is not None
+                and result.status is not RunStatus.SKIPPED
+            ):
+                waited = _human(finished - rule.waiting_since)
+                rule.waiting_since = None
+                rule.waiting_reason = None
+                note = f"Fortgesetzt nach {waited} Wartezeit"
+                message = f"{note}; {message}" if message else note
+                event(log, "rule.resumed", rule=rule.name, waited=waited)
             items = result.items
             if len(items) > MAX_ITEMS_PER_RUN:
                 extra = len(items) - MAX_ITEMS_PER_RUN
@@ -153,6 +186,15 @@ class RunService:
         return "\n".join(lines)
 
 
+def _human(delta: timedelta) -> str:
+    seconds = int(delta.total_seconds())
+    if seconds < 90:
+        return f"{seconds} s"
+    if seconds < 5400:
+        return f"{round(seconds / 60)} min"
+    return f"{seconds / 3600:.1f} h".replace(".", ",")
+
+
 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."""

src/netmover/settings_store.py

@@ -35,6 +35,17 @@ class AppSettings:
     webhook_template: str = DEFAULT_TEMPLATE
     webhook_content_type: str = "application/json"
     kuma_push_url: str = ""
+    # Wartungsmodus: geplante Läufe pausiert, kein Alarm für Regeln
+    maintenance: bool = False
+    # E-Mail (SMTP); Passwort Fernet-verschlüsselt
+    email_enabled: bool = False
+    smtp_host: str = ""
+    smtp_port: int = 587
+    smtp_security: str = "starttls"
+    smtp_user: str = ""
+    smtp_password_enc: str = ""
+    smtp_from: str = ""
+    smtp_to: str = ""
 
 
 _PREFIX = "app."

src/netmover/web/app.py

@@ -27,6 +27,7 @@ from fastapi import APIRouter, Depends, FastAPI, Request
 from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, RedirectResponse, Response
 from fastapi.staticfiles import StaticFiles
 from fastapi.templating import Jinja2Templates
+from markupsafe import escape
 from sqlalchemy import func, or_, select
 from starlette.concurrency import run_in_threadpool
 from starlette.datastructures import UploadFile
@@ -52,6 +53,7 @@ from netmover.db import session_scope
 from netmover.keys import delete_key, generate_key, import_key, list_keys, rename_key
 from netmover.kuma import push as kuma_push
 from netmover.logging_setup import event, get_logger
+from netmover.mail import SECURITY_MODES, MailConfigError, parse_recipients, valid_address
 from netmover.models import (
     CollisionPolicy,
     Credential,
@@ -185,6 +187,7 @@ class NavState:
     keys: int
     rules: int
     rules_active: int
+    maintenance: bool = False
 
 
 def _nav_state(request: Request) -> NavState | None:
@@ -208,6 +211,7 @@ def _nav_state(request: Request) -> NavState | None:
         keys=keys or 0,
         rules=len(rules),
         rules_active=sum(1 for (en,) in rules if en),
+        maintenance=health.maintenance,
     )
 
 
@@ -459,6 +463,23 @@ async def status_partial(request: Request) -> Response:
     return render(request, "partials/status_body.html", **await _status_context(request))
 
 
+@protected.post("/maintenance")
+async def toggle_maintenance(request: Request) -> Response:
+    state = nm(request)
+    cfg = load_settings(state.session_factory)
+    cfg.maintenance = not cfg.maintenance
+    save_settings(state.session_factory, cfg)
+    event(log, "maintenance", enabled=cfg.maintenance)
+    flash(
+        request,
+        "Wartungsmodus eingeschaltet – geplante Läufe sind pausiert."
+        if cfg.maintenance
+        else "Wartungsmodus beendet – Regeln laufen wieder nach Plan.",
+        "info",
+    )
+    return redirect("/")
+
+
 @protected.post("/rules/{rule_id}/run")
 async def rule_run_now(request: Request, rule_id: int) -> Response:
     state = nm(request)
@@ -1000,6 +1021,32 @@ async def settings_save(request: Request) -> Response:
         if url and not url.startswith(("http://", "https://")):
             errors["webhook_url"] = "URL muss mit http:// oder https:// beginnen."
         cfg.webhook_url = url
+    cfg.email_enabled = form.get("email_enabled") == "on"
+    cfg.smtp_host = (form.get("smtp_host") or "").strip()
+    cfg.smtp_user = (form.get("smtp_user") or "").strip()
+    cfg.smtp_from = (form.get("smtp_from") or "").strip()
+    cfg.smtp_to = (form.get("smtp_to") or "").strip()
+    cfg.smtp_security = form.get("smtp_security") or "starttls"
+    try:
+        cfg.smtp_port = int(form.get("smtp_port") or 587)
+        if not 1 <= cfg.smtp_port <= 65535:
+            raise ValueError
+    except ValueError:
+        errors["smtp_port"] = "Port zwischen 1 und 65535."
+    if cfg.smtp_security not in SECURITY_MODES:
+        errors["smtp_security"] = "Ungültig."
+    if cfg.smtp_host and not re.fullmatch(r"[A-Za-z0-9.-]+", cfg.smtp_host):
+        errors["smtp_host"] = "Hostname oder IP-Adresse."
+    if cfg.smtp_from and not valid_address(cfg.smtp_from):
+        errors["smtp_from"] = "Keine gültige E-Mail-Adresse."
+    try:
+        parse_recipients(cfg.smtp_to)
+    except MailConfigError as exc:
+        errors["smtp_to"] = str(exc)
+    if cfg.email_enabled and not (cfg.smtp_host and cfg.smtp_from and cfg.smtp_to):
+        errors["smtp_host"] = "Für E-Mail werden Server, Absender und Empfänger benötigt."
+    if form.get("smtp_password"):
+        cfg.smtp_password_enc = state.secrets.encrypt(form["smtp_password"])
     kuma = (form.get("kuma_push_url") or "").strip()
     if kuma != auth_mask(cfg.kuma_push_url):
         if kuma and not kuma.startswith(("http://", "https://")):
@@ -1050,6 +1097,25 @@ async def settings_webhook_test(request: Request) -> Response:
     return HTMLResponse('<p class="msg ok">Testnachricht gesendet.</p>')
 
 
+@protected.post("/settings/mail-test", response_class=HTMLResponse)
+async def settings_mail_test(request: Request) -> Response:
+    state = nm(request)
+    cfg = load_settings(state.session_factory)
+    if not (cfg.smtp_host and cfg.smtp_from and cfg.smtp_to):
+        return HTMLResponse('<p class="msg error">Server, Absender und Empfänger speichern.</p>')
+    try:
+        await run_in_threadpool(
+            state.notifier.send_mail_now,
+            "NetMover: Testnachricht",
+            "Die E-Mail-Benachrichtigung funktioniert.",
+            cfg,
+        )
+    except Exception as exc:
+        detail = escape(str(exc))[:300] or type(exc).__name__
+        return HTMLResponse(f'<p class="msg error">Fehlgeschlagen: {detail}</p>')
+    return HTMLResponse(f'<p class="msg ok">Gesendet an {escape(cfg.smtp_to)}.</p>')
+
+
 @protected.post("/settings/kuma-test", response_class=HTMLResponse)
 async def settings_kuma_test(request: Request) -> Response:
     state = nm(request)

src/netmover/web/static/app.css

@@ -30,6 +30,7 @@ pre { white-space: pre-wrap; word-break: break-word; margin: 0; }
 .health .dot { width: 9px; height: 9px; border-radius: 50%; background: currentColor; }
 .health.ok { color: var(--ok); background: var(--ok-bg); }
 .health.bad { color: var(--err); background: var(--err-bg); }
+.health.warn { color: var(--warn); background: var(--warn-bg); }
 .help-link { display: inline-flex; align-items: center; gap: 6px; font-size: .9em; text-decoration: none; }
 .menu-toggle { display: none; align-items: center; gap: 6px; }
 .ico { width: 18px; height: 18px; flex: none; fill: none; stroke: currentColor; stroke-width: 2;

src/netmover/web/templates/base.html

@@ -23,6 +23,7 @@
   {% if nav %}
   <a class="health {{ 'ok' if nav.healthy else 'bad' }}" href="/" title="Gesundheitszustand">
     <span class="dot"></span>{{ 'Alles in Ordnung' if nav.healthy else (nav.problems ~ (' Problem' if nav.problems == 1 else ' Probleme')) }}</a>
+  {% if nav.maintenance %}<a class="health warn" href="/" title="Geplante Läufe sind pausiert"><span class="dot"></span>Wartungsmodus</a>{% endif %}
   {% endif %}
   <span class="spacer"></span>
   {% if help_anchor %}<a class="help-link" href="/help#{{ help_anchor }}">{{ icon('help') }}<span>Hilfe zu dieser Seite</span></a>{% endif %}

src/netmover/web/templates/macros.html

@@ -21,8 +21,8 @@
 </div>
 {%- endmacro %}
 
-{% macro check(name, label, checked, hint='') -%}
-<div class="field check">
+{% macro check(name, label, checked, hint='', wide=false) -%}
+<div class="field check{{ ' wide' if wide }}">
   <input id="f-{{ name }}" type="checkbox" name="{{ name }}" {{ 'checked' if checked }}>
   <label for="f-{{ name }}">{{ label }}</label>
   {% if hint %}<span class="hint">{{ hint }}</span>{% endif %}

src/netmover/web/templates/partials/status_body.html

@@ -33,17 +33,19 @@
   <thead><tr><th>Regel</th><th>Letztes Ergebnis</th><th>Letzter Erfolg</th><th>Nächster Lauf</th><th></th></tr></thead>
   <tbody>
   {% for v in rules %}{% set r = v.r %}
-  <tr class="{{ 'warn' if v.stale }}">
+  <tr class="{{ 'warn' if v.stale or r.waiting_since }}">
     <td data-label="Regel"><strong>{{ r.name }}</strong>{% if not r.enabled %} <span class="badge">aus</span>{% endif %}{% if r.dry_run %} <span class="badge warn">Probelauf</span>{% endif %}
       <div class="muted small mono">{{ v.target }}</div></td>
     <td data-label="Letztes Ergebnis">
-      {% if v.running_since %}<span class="badge warn">läuft seit {{ v.running_since|ago|replace('vor ', '') }}</span>
+      {% if r.waiting_since %}<span class="badge warn">wartet seit {{ r.waiting_since|ago|replace('vor ', '') }}</span>
+        <div class="small text-warn">{{ r.waiting_reason }}</div>
+      {% elif v.running_since %}<span class="badge warn">läuft seit {{ v.running_since|ago|replace('vor ', '') }}</span>
       {% elif v.last_run %}{{ run_badge(v.last_run.status) }}
         <span class="small muted">{{ v.last_run.finished_at|ago }}{% if v.last_run.seen %} · {{ v.last_run.moved }}/{{ v.last_run.seen }}{% endif %}</span>
         {% if v.last_run.status in (RunStatus.ERROR, RunStatus.PARTIAL) and v.last_run.message %}<div class="small text-err">{{ v.last_run.message }}</div>{% endif %}
       {% else %}<span class="muted">–</span>{% endif %}</td>
     <td data-label="Letzter Erfolg">{{ r.last_success_at|ago }}{% if v.stale %}<div class="small text-warn">länger als 3 × Intervall!</div>{% endif %}
-      {% if r.consecutive_failures %}<div class="small text-err">{{ r.consecutive_failures }} Fehler in Folge</div>{% endif %}</td>
+      {% if r.consecutive_failures and not r.waiting_since %}<div class="small text-err">{{ r.consecutive_failures }} Fehler in Folge</div>{% endif %}</td>
     <td data-label="Nächster Lauf">{{ v.next_run|ago if v.next_run else '–' }}</td>
     <td><form class="inline" method="post" action="/rules/{{ r.id }}/run">
       <input type="hidden" name="csrf_token" value="{{ csrf }}">

src/netmover/web/templates/rules.html

@@ -22,6 +22,8 @@
     </div>
   </div>
   <div class="muted small mono">{{ v.target }} · Filter {{ r.pattern }}{{ ' (rekursiv)' if r.recursive }}</div>
+  {% if r.waiting_since %}<p class="msg warn small">Wartet seit {{ r.waiting_since|localdt }} ({{ r.waiting_since|ago }}) – {{ r.waiting_reason }}
+    Die Regel läuft von selbst weiter, sobald die Anbindung zurück ist.</p>{% endif %}
   <div class="small">Letzter Lauf: {% if v.running_since %}<span class="badge warn">läuft seit {{ v.running_since|ago|replace('vor ', '') }}</span>{% elif v.last_run %}{{ run_badge(v.last_run.status) }} {{ v.last_run.finished_at|ago }}{% else %}–{% endif %}
     · letzter Erfolg: <span class="{{ 'text-warn' if v.stale }}">{{ r.last_success_at|ago }}</span>
     · nächster: {{ v.next_run|ago if v.next_run else '–' }}</div>

src/netmover/web/templates/settings.html

@@ -8,18 +8,37 @@
   <fieldset><legend>Protokoll</legend><div class="fields">
     {{ field('retention_days', 'Aufbewahrung (Tage)', cfg.retention_days, type='number', errors=errors, attrs='min="1" max="3650"') }}
   </div></fieldset>
-  <fieldset><legend>Benachrichtigung (Webhook) <a class="legend-help" href="/help#benachrichtigungen" title="Hilfe">?</a></legend><div class="fields">
-    {{ select('notify_mode', 'Benachrichtigen', [('off', 'nie'), ('every', 'bei jedem fehlgeschlagenen Lauf'), ('after_n', 'nach N Fehlschlägen in Folge')], cfg.notify_mode.value, errors=errors) }}
+  <fieldset><legend>Benachrichtigungen <a class="legend-help" href="/help#benachrichtigungen" title="Hilfe">?</a></legend><div class="fields">
+    {{ select('notify_mode', 'Benachrichtigen', [('off', 'nie'), ('every', 'bei jedem fehlgeschlagenen Lauf'), ('after_n', 'nach N Fehlschlägen in Folge')], cfg.notify_mode.value, errors=errors, hint='Gilt für Webhook und E-Mail. Ausfälle von Anbindungen werden immer sofort gemeldet (außer bei „nie“).') }}
     <div data-show-for="notify_mode:after_n">{{ field('notify_after', 'N', cfg.notify_after, type='number', errors=errors, attrs='min="1"') }}</div>
-    {{ field('webhook_url', 'Webhook-URL', mask_url(cfg.webhook_url), errors=errors, wide=true, placeholder='https://ntfy.sh/mein-topic', hint='Wird nur gekürzt angezeigt; unverändert lassen = behalten, leeren = entfernen.') }}
+    {{ check('notify_recovery', 'Entwarnung senden, wenn es wieder läuft', cfg.notify_recovery, wide=true) }}
+  </div></fieldset>
+
+  <fieldset><legend>Webhook</legend><div class="fields">
+    {{ field('webhook_url', 'Webhook-URL', mask_url(cfg.webhook_url), errors=errors, wide=true, placeholder='https://ntfy.sh/mein-topic', hint='Leer = kein Webhook. Wird nur gekürzt angezeigt; unverändert lassen = behalten.') }}
     {{ field('webhook_content_type', 'Content-Type', cfg.webhook_content_type, errors=errors) }}
-    {{ check('notify_recovery', 'Entwarnung senden, wenn es wieder läuft', cfg.notify_recovery) }}
     <div class="field wide"><label for="f-tpl">Vorlage</label>
       <textarea id="f-tpl" name="webhook_template" class="mono">{{ cfg.webhook_template }}</textarea>
       <span class="hint">Platzhalter: $event $title $message $rule $time $host – bei JSON automatisch maskiert.</span></div>
   </div>
   <div class="row"><button type="button" hx-post="/settings/webhook-test" hx-target="#hook-result">Test senden (gespeicherte Einstellungen)</button><span id="hook-result"></span></div>
   </fieldset>
+
+  <fieldset><legend>E-Mail</legend><div class="fields">
+    {{ check('email_enabled', 'E-Mail-Benachrichtigung einschalten', cfg.email_enabled, wide=true) }}
+    {{ field('smtp_host', 'SMTP-Server', cfg.smtp_host, errors=errors, placeholder='smtp.example.org') }}
+    {{ field('smtp_port', 'Port', cfg.smtp_port, type='number', errors=errors, attrs='min="1" max="65535"', hint='587 (STARTTLS) oder 465 (SSL)') }}
+    {{ select('smtp_security', 'Verschlüsselung', [('starttls', 'STARTTLS'), ('ssl', 'SSL/TLS'), ('none', 'keine (nur interne Relays)')], cfg.smtp_security, errors=errors) }}
+    {{ field('smtp_user', 'Benutzer', cfg.smtp_user, errors=errors, attrs='autocomplete="off"', hint='leer = ohne Anmeldung') }}
+    <div class="field{{ ' has-error' if errors.get('smtp_password') }}"><label for="f-smtp_password">Passwort</label>
+      <input id="f-smtp_password" type="password" name="smtp_password" autocomplete="new-password" placeholder="{{ '••••••' if cfg.smtp_password_enc }}">
+      <span class="hint">{{ 'Leer lassen = unverändert.' if cfg.smtp_password_enc else '' }} Wird verschlüsselt gespeichert.</span></div>
+    {{ field('smtp_from', 'Absender', cfg.smtp_from, errors=errors, placeholder='netmover@example.org') }}
+    {{ field('smtp_to', 'Empfänger', cfg.smtp_to, errors=errors, wide=true, placeholder='ich@example.org, admin@example.org', hint='mehrere mit Komma getrennt') }}
+  </div>
+  <div class="row"><button type="button" hx-post="/settings/mail-test" hx-target="#mail-result">Test-E-Mail senden (gespeicherte Einstellungen)</button><span id="mail-result"></span></div>
+  </fieldset>
+
   <fieldset><legend>Überwachung (Uptime Kuma) <a class="legend-help" href="/help#uptime-kuma" title="Hilfe">?</a></legend>
     <div class="fields">
       {{ field('kuma_push_url', 'Push-URL (Kuma-Monitor vom Typ „Push“)', mask_url(cfg.kuma_push_url), errors=errors, wide=true, placeholder='https://kuma.lan/api/push/AbCdEf?status=up&msg=OK&ping=', hint='NetMover meldet jede Minute den Zustand. Leer = aus. Wird nur gekürzt angezeigt.') }}

src/netmover/web/templates/status.html

@@ -2,7 +2,14 @@
 {% block title %}Status{% endblock %}
 {% block content %}
 <div class="row spread"><h1>Status</h1>
-  <span class="muted small">aktualisiert sich alle 15 s</span></div>
+  <div class="row"><span class="muted small">aktualisiert sich alle 15 s</span>
+    <form class="inline" method="post" action="/maintenance"
+          {% if not (nav and nav.maintenance) %}data-confirm="Alle geplanten Läufe pausieren?"{% endif %}>
+      <input type="hidden" name="csrf_token" value="{{ csrf }}">
+      <button class="small">{{ 'Wartungsmodus beenden' if nav and nav.maintenance else 'Wartungsmodus' }}</button>
+    </form></div></div>
+{% if nav and nav.maintenance %}<p class="msg warn"><strong>Wartungsmodus:</strong> Geplante Läufe sind pausiert.
+  „Jetzt ausführen“ geht weiterhin. Für Regeln schlägt <code>/healthz</code> keinen Alarm.</p>{% endif %}
 <div id="status-body" hx-get="/partials/status" hx-trigger="every 15s" hx-swap="innerHTML">
   {% include "partials/status_body.html" %}
 </div>

tests/test_mail.py

@@ -0,0 +1,304 @@
+"""E-Mail-Benachrichtigung gegen einen echten SMTP-Server im Testprozess (aiosmtpd)."""
+
+from __future__ import annotations
+
+import datetime
+import socket
+import ssl
+from collections.abc import Iterator
+from email import message_from_bytes
+from email.policy import default
+from pathlib import Path
+from typing import Any
+
+import pytest
+from aiosmtpd.controller import Controller
+from aiosmtpd.smtp import AuthResult, LoginPassword
+from cryptography import x509
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric import ec
+from cryptography.x509.oid import NameOID
+from fastapi.testclient import TestClient
+from sqlalchemy.orm import Session, sessionmaker
+
+from netmover.crypto import SecretBox
+from netmover.mail import MailConfigError, parse_recipients, smtp_send
+from netmover.notify import Notification, Notifier
+from netmover.settings_store import AppSettings, NotifyMode, load_settings, save_settings
+from tests.webutil import post
+
+
+class Inbox:
+    def __init__(self) -> None:
+        self.messages: list[tuple[list[str], Any]] = []
+
+    async def handle_DATA(self, server: Any, session: Any, envelope: Any) -> str:
+        self.messages.append(
+            (envelope.rcpt_tos, message_from_bytes(envelope.content, policy=default))
+        )
+        return "250 OK"
+
+
+def _free_port() -> int:
+    s = socket.socket()
+    s.bind(("127.0.0.1", 0))
+    port: int = s.getsockname()[1]
+    s.close()
+    return port
+
+
+def _auth(server: Any, session: Any, envelope: Any, mechanism: str, data: Any) -> AuthResult:
+    ok = (
+        isinstance(data, LoginPassword)
+        and data.login == b"netmover"
+        and data.password == b"Pa$$,wort"
+    )
+    return AuthResult(success=ok, handled=False)
+
+
+@pytest.fixture(scope="module")
+def tls_files(tmp_path_factory: pytest.TempPathFactory) -> tuple[Path, Path]:
+    d = tmp_path_factory.mktemp("tls")
+    key = ec.generate_private_key(ec.SECP256R1())
+    name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "localhost")])
+    now = datetime.datetime.now(datetime.UTC)
+    cert = (
+        x509.CertificateBuilder()
+        .subject_name(name)
+        .issuer_name(name)
+        .public_key(key.public_key())
+        .serial_number(x509.random_serial_number())
+        .not_valid_before(now - datetime.timedelta(days=1))
+        .not_valid_after(now + datetime.timedelta(days=1))
+        .add_extension(x509.SubjectAlternativeName([x509.DNSName("localhost")]), critical=False)
+        .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True)
+        .sign(key, hashes.SHA256())
+    )
+    (d / "cert.pem").write_bytes(cert.public_bytes(serialization.Encoding.PEM))
+    (d / "key.pem").write_bytes(
+        key.private_bytes(
+            serialization.Encoding.PEM,
+            serialization.PrivateFormat.PKCS8,
+            serialization.NoEncryption(),
+        )
+    )
+    return d / "cert.pem", d / "key.pem"
+
+
+@pytest.fixture
+def trust(tls_files: tuple[Path, Path], monkeypatch: pytest.MonkeyPatch) -> None:
+    """Client vertraut unserem Test-Zertifikat (Prüfung bleibt eingeschaltet)."""
+    monkeypatch.setattr(
+        "netmover.mail.tls_context", lambda: ssl.create_default_context(cafile=str(tls_files[0]))
+    )
+
+
+def _server(mode: str, tls_files: tuple[Path, Path]) -> tuple[Controller, Inbox, int]:
+    inbox, port = Inbox(), _free_port()
+    ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
+    ctx.load_cert_chain(*map(str, tls_files))
+    kw: dict[str, Any] = {
+        "hostname": "localhost",
+        "port": port,
+        "authenticator": _auth,
+        "auth_exclude_mechanism": ["LOGIN"],
+    }
+    if mode == "starttls":
+        kw.update(tls_context=ctx, require_starttls=True)
+    elif mode == "ssl":
+        # implizites TLS erkennt aiosmtpd nicht als „verschlüsselt“ – AUTH trotzdem anbieten
+        kw.update(ssl_context=ctx, auth_require_tls=False)
+    else:
+        kw.update(auth_require_tls=False)
+    controller = Controller(inbox, **kw)
+    controller.start()
+    return controller, inbox, port
+
+
+@pytest.fixture(params=["starttls", "ssl", "none"])
+def smtp_server(
+    request: pytest.FixtureRequest, tls_files: tuple[Path, Path], trust: None
+) -> Iterator[tuple[str, Inbox, int]]:
+    controller, inbox, port = _server(request.param, tls_files)
+    yield request.param, inbox, port
+    controller.stop()
+
+
+def _cfg(mode: str, port: int, **kw: Any) -> AppSettings:
+    data: dict[str, Any] = {
+        "email_enabled": True,
+        "smtp_host": "localhost",
+        "smtp_port": port,
+        "smtp_security": mode,
+        "smtp_user": "netmover",
+        "smtp_from": "netmover@example.org",
+        "smtp_to": "ich@example.org; admin@example.org",
+        "notify_mode": NotifyMode.EVERY,
+    }
+    data.update(kw)
+    return AppSettings(**data)
+
+
+def test_smtp_send_all_security_modes(smtp_server: tuple[str, Inbox, int]) -> None:
+    mode, inbox, port = smtp_server
+    smtp_send(_cfg(mode, port), "Pa$$,wort", "NetMover: Anbindung „NAS“ gestört", "Host is down")
+    ((rcpts, msg),) = inbox.messages
+    assert rcpts == ["ich@example.org", "admin@example.org"]
+    assert msg["Subject"] == "NetMover: Anbindung „NAS“ gestört"
+    assert msg["From"] == "netmover@example.org"
+    body = msg.get_content()
+    assert "Host is down" in body and "NetMover auf" in body
+
+
+def test_wrong_password_raises(smtp_server: tuple[str, Inbox, int]) -> None:
+    import smtplib
+
+    mode, inbox, port = smtp_server
+    with pytest.raises(smtplib.SMTPAuthenticationError):
+        smtp_send(_cfg(mode, port), "falsch", "x", "y")
+    assert inbox.messages == []
+
+
+def test_certificate_is_verified(tls_files: tuple[Path, Path]) -> None:
+    """Ohne Vertrauen in das Zertifikat wird abgelehnt – keine stille Herabstufung."""
+    controller, inbox, port = _server("starttls", tls_files)
+    try:
+        with pytest.raises(ssl.SSLCertVerificationError):
+            smtp_send(_cfg("starttls", port), "Pa$$,wort", "x", "y")
+    finally:
+        controller.stop()
+    assert inbox.messages == []
+
+
+def test_recipients() -> None:
+    assert parse_recipients("a@b.de, c@d.org;e@f.net\n") == ["a@b.de", "c@d.org", "e@f.net"]
+    with pytest.raises(MailConfigError, match="kein-mail"):
+        parse_recipients("a@b.de, kein-mail")
+
+
+def test_notifier_sends_webhook_and_mail(
+    session_factory: sessionmaker[Session], secretbox: SecretBox
+) -> None:
+    hooks: list[Any] = []
+    mails: list[Any] = []
+    cfg = _cfg(
+        "starttls",
+        587,
+        smtp_password_enc=secretbox.encrypt("Pa$$,wort"),
+        webhook_url="http://hook.invalid/x",
+    )
+    save_settings(session_factory, cfg)
+    n = Notifier(
+        session_factory,
+        sender=lambda *a: hooks.append(a),
+        mailer=lambda *a: mails.append(a),
+        secrets=secretbox,
+        background=False,
+    )
+    n.send(Notification("mount.error", "NetMover: Anbindung „NAS“ gestört", "weg"))
+    assert len(hooks) == 1 and len(mails) == 1
+    _cfg_used, password, subject, body = mails[0]
+    assert password == "Pa$$,wort" and subject.endswith("gestört") and body == "weg"
+
+
+def test_mail_only_and_mode_off(
+    session_factory: sessionmaker[Session], secretbox: SecretBox
+) -> None:
+    mails: list[Any] = []
+    n = Notifier(
+        session_factory, mailer=lambda *a: mails.append(a), secrets=secretbox, background=False
+    )
+    save_settings(session_factory, _cfg("starttls", 587, notify_mode=NotifyMode.OFF))
+    n.send(Notification("x", "t", "m"))
+    save_settings(session_factory, _cfg("starttls", 587, email_enabled=False))
+    n.send(Notification("x", "t", "m"))
+    assert mails == []
+    save_settings(session_factory, _cfg("starttls", 587))  # nur E-Mail, kein Webhook
+    n.send(Notification("x", "t", "m"))
+    assert len(mails) == 1
+
+
+def test_mail_failure_is_swallowed(
+    session_factory: sessionmaker[Session], secretbox: SecretBox
+) -> None:
+    def broken(*a: Any) -> None:
+        raise ConnectionRefusedError("weg")
+
+    save_settings(session_factory, _cfg("starttls", 587))
+    Notifier(session_factory, mailer=broken, secrets=secretbox, background=False).send(
+        Notification("x", "t", "m")
+    )
+
+
+# -- Einstellungsseite ------------------------------------------------------------------------
+
+
+def _form(**kw: str) -> dict[str, str]:
+    data = {
+        "retention_days": "30",
+        "notify_mode": "every",
+        "notify_after": "3",
+        "webhook_url": "",
+        "webhook_content_type": "application/json",
+        "webhook_template": "$message",
+        "email_enabled": "on",
+        "smtp_host": "smtp.example.org",
+        "smtp_port": "587",
+        "smtp_security": "starttls",
+        "smtp_user": "netmover",
+        "smtp_password": "Geheim,123!",
+        "smtp_from": "netmover@example.org",
+        "smtp_to": "ich@example.org",
+    }
+    data.update(kw)
+    return data
+
+
+def test_settings_store_encrypted_password(client: TestClient, state: Any) -> None:
+    assert post(client, "/settings", _form()).status_code == 303
+    cfg = load_settings(state.session_factory)
+    assert cfg.email_enabled and "Geheim" not in cfg.smtp_password_enc
+    assert state.secrets.decrypt(cfg.smtp_password_enc) == "Geheim,123!"
+    page = client.get("/settings").text
+    assert "Geheim,123!" not in page and 'placeholder="••••••"' in page
+    # leer lassen = unverändert
+    post(client, "/settings", _form(smtp_password=""))
+    assert (
+        state.secrets.decrypt(load_settings(state.session_factory).smtp_password_enc)
+        == "Geheim,123!"
+    )
+
+
+def test_settings_validation(client: TestClient) -> None:
+    r = post(
+        client,
+        "/settings",
+        _form(smtp_to="ich@example.org, quatsch", smtp_port="99999", smtp_from="kein-absender"),
+    )
+    assert r.status_code == 400
+    for msg in ("Ungültige Adresse: quatsch", "Port zwischen", "Keine gültige E-Mail-Adresse"):
+        assert msg in r.text
+
+
+def test_mail_test_button(
+    client: TestClient, state: Any, tls_files: tuple[Path, Path], trust: None
+) -> None:
+    controller, inbox, port = _server("starttls", tls_files)
+    try:
+        post(
+            client,
+            "/settings",
+            _form(smtp_host="localhost", smtp_port=str(port), smtp_password="Pa$$,wort"),
+        )
+        r = post(client, "/settings/mail-test")
+        assert "Gesendet an ich@example.org" in r.text
+        post(
+            client,
+            "/settings",
+            _form(smtp_host="localhost", smtp_port=str(port), smtp_password="falsch"),
+        )
+        r = post(client, "/settings/mail-test")
+        assert "Fehlgeschlagen" in r.text and "falsch" not in r.text
+    finally:
+        controller.stop()
+    assert len(inbox.messages) == 1

tests/test_migrations.py

@@ -18,6 +18,14 @@ from alembic import command
 from netmover.db import alembic_config, upgrade
 
 
+def _head() -> str:
+    from alembic.script import ScriptDirectory
+
+    head = ScriptDirectory.from_config(alembic_config("sqlite://")).get_current_head()
+    assert head is not None
+    return head
+
+
 def _seed(db: Path) -> None:
     """Datenbank auf Stand 0001 mit verknüpften Daten, wie auf einem echten System."""
     c = sqlite3.connect(db)
@@ -80,7 +88,7 @@ def test_upgrade_with_linked_data(db_0001: tuple[str, Path]) -> None:
     url, db = db_0001
     upgrade(url)
     st = _state(db)
-    assert st["version"] == "0002"
+    assert st["version"] == _head()
     assert st["tmp_tables"] == [] and st["fk_violations"] == []
     assert "managed" in st["cred_columns"]  # type: ignore[operator]
     assert st["mounts"] == [("NAS", 1), ("paperless", 2)]
@@ -122,7 +130,7 @@ def test_failed_migration_leaves_database_untouched(
 
     monkeypatch.setattr(DefaultImpl, "drop_table", real_drop)
     upgrade(url)  # danach klappt es ohne Handarbeit
-    assert _state(db)["version"] == "0002"
+    assert _state(db)["version"] == _head()
 
 
 def test_app_engine_still_enforces_foreign_keys(db_url: str) -> None:
@@ -134,3 +142,36 @@ def test_app_engine_still_enforces_foreign_keys(db_url: str) -> None:
     with engine.connect() as conn:
         assert conn.execute(text("PRAGMA foreign_keys")).scalar() == 1
     engine.dispose()
+
+
+def test_upgrade_from_0002_keeps_everything_linked_to_rules(tmp_path: Path) -> None:
+    """0003 baut ``rules`` neu auf – Läufe, Buchführung und Fehlerzähler verweisen darauf."""
+    db = tmp_path / "v02.db"
+    url = f"sqlite:///{db}"
+    upgrade(url, "0001")
+    _seed(db)
+    upgrade(url, "0002")
+    c = sqlite3.connect(db)
+    c.execute("PRAGMA foreign_keys=ON")
+    c.executescript(
+        """
+        UPDATE credentials SET managed = 1 WHERE id = 2;
+        UPDATE mounts SET username = 'paperless' WHERE id = 2;
+        INSERT INTO copied_files (rule_id, src_relpath, size, mtime_ns, dst_path, copied_at)
+            VALUES (1, 'a.pdf', 1, 1, '/x/a.pdf', '2026-10-10 20:00:00');
+        INSERT INTO file_failures (rule_id, src_relpath, size, mtime_ns, attempts, last_attempt_at)
+            VALUES (1, 'b.pdf', 1, 1, 2, '2026-10-10 20:00:00');
+        """
+    )
+    c.commit()
+    c.close()
+    upgrade(url)
+    st = _state(db)
+    assert st["version"] == _head() and st["fk_violations"] == [] and st["tmp_tables"] == []
+    assert st["rules"] == [("Scans → Paperless", 1, 2)] and st["run_items"] == 1
+    c = sqlite3.connect(db)
+    assert c.execute("SELECT COUNT(*) FROM copied_files").fetchone()[0] == 1
+    assert c.execute("SELECT attempts FROM file_failures").fetchone()[0] == 2
+    assert c.execute("SELECT waiting_since FROM rules").fetchone()[0] is None
+    assert c.execute("SELECT username FROM mounts WHERE id = 2").fetchone()[0] == "paperless"
+    c.close()

tests/test_waiting.py

@@ -0,0 +1,184 @@
+"""0.4.0: automatisch pausieren/fortsetzen, Wartungsmodus, eine Meldung pro Anbindung."""
+
+from __future__ import annotations
+
+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 Rule, Run, RunStatus, RunTrigger, utcnow
+from netmover.mounting.fake import FakeMountBackend
+from netmover.mounting.manager import MountManager
+from netmover.mover.engine import RunResult
+from netmover.mover.runner import RuleRunner
+from netmover.runs import RunService
+from netmover.settings_store import AppSettings, save_settings
+from tests.conftest import MountFactory
+
+
+class FakeRunner:
+    def __init__(self) -> None:
+        self.results: list[RunResult] = []
+        self.calls = 0
+
+    def run(self, rule_id: int, *, dry_run: bool | None = None) -> RunResult:
+        self.calls += 1
+        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
+
+
+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 _runs(sf: sessionmaker[Session]) -> list[Run]:
+    with sf() as s:
+        return list(s.scalars(select(Run).order_by(Run.id)))
+
+
+def unavailable() -> RunResult:
+    return RunResult(
+        RunStatus.ERROR,
+        message="„NAS“ ist nicht eingehängt – Lauf abgebrochen, nichts angefasst.",
+        unavailable=True,
+    )
+
+
+def test_waiting_logs_once_and_resumes_by_itself(
+    session_factory: sessionmaker[Session], rule_id: int
+) -> None:
+    runner = FakeRunner()
+    service = RunService(session_factory, runner)  # type: ignore[arg-type]
+    runner.results += [unavailable(), unavailable(), unavailable(), RunResult(RunStatus.OK)]
+
+    first = service.execute(rule_id)
+    assert first is not None
+    r = _rule(session_factory, rule_id)
+    assert r.waiting_since is not None and "nicht eingehängt" in (r.waiting_reason or "")
+    assert r.consecutive_failures == 0  # Warten ist kein Fehlschlag
+
+    assert service.execute(rule_id) is None  # kein neuer Eintrag alle 30 s
+    assert service.execute(rule_id) is None
+    assert len(_runs(session_factory)) == 1
+    assert _runs(session_factory)[0].message.startswith("Pausiert, bis die Anbindung zurück ist")
+
+    service.execute(rule_id)  # Anbindung wieder da
+    r = _rule(session_factory, rule_id)
+    assert r.waiting_since is None and r.waiting_reason is None and r.last_success_at is not None
+    runs = _runs(session_factory)
+    assert len(runs) == 2 and runs[1].message and runs[1].message.startswith("Fortgesetzt nach")
+    assert r.consecutive_failures == 0
+
+
+def test_real_errors_still_count(session_factory: sessionmaker[Session], rule_id: int) -> None:
+    runner = FakeRunner()
+    service = RunService(session_factory, runner)  # type: ignore[arg-type]
+    runner.results += [RunResult(RunStatus.ERROR, message="kaputt")] * 2
+    service.execute(rule_id)
+    service.execute(rule_id)
+    r = _rule(session_factory, rule_id)
+    assert r.consecutive_failures == 2 and r.waiting_since is None
+    assert len(_runs(session_factory)) == 2
+
+
+def test_manual_dry_run_does_not_start_waiting(
+    session_factory: sessionmaker[Session], rule_id: int
+) -> None:
+    runner = FakeRunner()
+    service = RunService(session_factory, runner)  # type: ignore[arg-type]
+    runner.results.append(unavailable())
+    service.execute(rule_id, RunTrigger.MANUAL, dry_run=True)
+    assert _rule(session_factory, rule_id).waiting_since is None
+    assert len(_runs(session_factory)) == 1  # Probelauf sieht man trotzdem
+
+
+def test_maintenance_skips_scheduled_runs_but_not_manual(
+    session_factory: sessionmaker[Session], rule_id: int
+) -> None:
+    runner = FakeRunner()
+    service = RunService(session_factory, runner)  # type: ignore[arg-type]
+    save_settings(session_factory, AppSettings(maintenance=True))
+    assert service.execute(rule_id) is None
+    assert runner.calls == 0 and _runs(session_factory) == []
+    runner.results.append(RunResult(RunStatus.OK))
+    assert service.execute(rule_id, RunTrigger.MANUAL) is not None
+    assert runner.calls == 1
+
+
+def _health(sf: sessionmaker[Session], **kw: object):  # type: ignore[no-untyped-def]
+    now = utcnow()
+    return check_health(
+        sf,
+        scheduler_running=True,
+        last_heartbeat=now,
+        service_started=now - timedelta(hours=1),
+        now=now,
+        **kw,
+    )  # type: ignore[arg-type]
+
+
+def test_health_waiting_and_maintenance(
+    session_factory: sessionmaker[Session], rule_id: int
+) -> None:
+    with session_scope(session_factory) as s:
+        r = s.get(Rule, rule_id)
+        assert r is not None
+        r.created_at = utcnow() - timedelta(days=1)
+        r.last_success_at = utcnow() - timedelta(hours=2)
+        r.waiting_since = utcnow() - timedelta(minutes=12)
+        r.waiting_reason = "„NAS“ nicht verfügbar"
+    h = _health(session_factory)
+    assert not h.ok and "wartet seit" in h.problems[0] and "NAS" in h.problems[0]
+    h = _health(session_factory, maintenance=True)
+    assert h.ok and h.maintenance and h.stale_rules == set()
+    check = h.find("rule", id=rule_id)
+    assert check is not None and "Wartungsmodus" in check.message
+
+
+# -- Integration mit MountManager und Runner ------------------------------------------------
+
+
+def test_vanished_mount_notifies_once_with_affected_rules(
+    manager: MountManager,
+    fake_backend: FakeMountBackend,
+    make_mount: MountFactory,
+    session_factory: sessionmaker[Session],
+    notifications: list[tuple[str, str, str]],
+    tmp_path: Path,
+) -> None:
+    fake_backend.export("nas", "scans")
+    mid = make_mount("nas", share="scans")
+    manager.mount(mid)
+    with session_scope(session_factory) as s:
+        s.add(Rule(name="Scans → Paperless", src_mount_id=mid, dst_path=str(tmp_path)))
+        s.add(Rule(name="Archiv", src_mount_id=mid, dst_path=str(tmp_path), enabled=False))
+    fake_backend.vanish(manager.mountpoint_for("nas"))
+
+    runner = RuleRunner(session_factory, manager, sleep=lambda _: None)
+    with session_factory() as s:
+        rid = s.scalars(select(Rule.id).where(Rule.name == "Scans → Paperless")).one()
+    for _ in range(3):
+        result = runner.run(rid)
+        assert result.unavailable and result.status is RunStatus.ERROR
+    errors = [n for n in notifications if n[0] == "mount.error"]
+    assert len(errors) == 1
+    assert "Betroffene Regeln" in errors[0][2] and "Scans → Paperless" in errors[0][2]
+    assert "Archiv" not in errors[0][2]  # deaktivierte Regel nicht erwähnt